test and tweak for config & status enums

testplan/common/config/base.py

@@ -167,9 +167,7 @@ def __getattr__(self, name):
             )
 
     def __repr__(self):
-        return "{}{}".format(
-            self.__class__.__name__, self._cfg_input or self._options
-        )
+        return "{}{}".format(self.__class__.__name__, self._options)
 
     def get_local(self, name, default=None):
         """Returns a local config setting (not from container)"""
@@ -206,12 +204,6 @@ def denormalize(self):
         new_options = {}
         for key in self._options:
             value = getattr(self, key)
-            if inspect.isroutine(value):
-                # Skipping non-serializable classes and routines.
-                logger.TESTPLAN_LOGGER.debug(
-                    "Skip denormalizing option: %s", key
-                )
-                continue
             try:
                 new_options[copy.deepcopy(key)] = copy.deepcopy(value)
             except Exception as exc:
@@ -221,7 +213,6 @@ def denormalize(self):
 
         # XXX: we have transformed options, which should not be validated
         # XXX: against schema again
-        # TODO: refactor
         new = object.__new__(self.__class__)
         setattr(new, "_parent", None)
         setattr(new, "_cfg_input", new_options)

testplan/report/testing/base.py

@@ -49,7 +49,7 @@
 from collections import Counter
 from enum import Enum
 from functools import reduce
-from typing import Callable, Optional, Dict, List
+from typing import Callable, Dict, List, Optional
 
 from typing_extensions import Self
 
@@ -87,13 +87,29 @@ def precedent(cls, stats):
         """
         return min(stats, key=lambda stat: RUNTIMESTATUS_PRECEDENCE[stat])
 
+    def __lt__(self, other: Self) -> bool:
+        lhs, rhs = (
+            RUNTIMESTATUS_PRECEDENCE[self],
+            RUNTIMESTATUS_PRECEDENCE[other],
+        )
+        if lhs == rhs and self != other:
+            return NotImplemented
+        return lhs < rhs
+
+    def __le__(self, other: Self) -> bool:
+        lhs, rhs = (
+            RUNTIMESTATUS_PRECEDENCE[self],
+            RUNTIMESTATUS_PRECEDENCE[other],
+        )
+        if lhs == rhs and self != other:
+            return NotImplemented
+        return lhs <= rhs
+
+    precede = __lt__
+
     def __bool__(self):
         return self != self.NONE
 
-    @staticmethod
-    def all_statuses() -> List["RuntimeStatus"]:
-        return list(RUNTIMESTATUS_PRECEDENCE.keys())
-
 
 RUNTIMESTATUS_PRECEDENCE = {
     RuntimeStatus.RUNNING: 0,
@@ -138,29 +154,30 @@ def _cmp(x: Self, y: Self) -> Self:
             try:
                 r = x < y
             except TypeError:
-                # NotImplemented was returned
                 return x.normalised()
             else:
                 return x if r else y
 
         return reduce(_cmp, stats, cls.NONE)
 
-    def __le__(self, other: Self) -> bool:
+    def __lt__(self, other: Self) -> bool:
         lhs, rhs = STATUS_PRECEDENCE[self], STATUS_PRECEDENCE[other]
         if lhs == rhs and self != other:
             return NotImplemented
-        return lhs <= rhs
+        return lhs < rhs
 
-    def __lt__(self, other: Self) -> bool:
+    def __le__(self, other: Self) -> bool:
         lhs, rhs = STATUS_PRECEDENCE[self], STATUS_PRECEDENCE[other]
         if lhs == rhs and self != other:
             return NotImplemented
-        return lhs < rhs
+        return lhs <= rhs
 
     def precede(self, other: Self) -> bool:
-        # a grep-friendly and more-intuitive version
-        r = self < other
-        return False if r is NotImplemented else r
+        # a (slightly) more intuitive & exception-free version
+        try:
+            return self < other
+        except TypeError:
+            return False
 
     def normalised(self) -> Self:
         return STATUS_NORMED[STATUS_PRECEDENCE[self] // 10]
@@ -171,7 +188,7 @@ def __bool__(self) -> bool:
 
 # Status Precedence encoded by numeric value and
 # Status categorization done through list indices.
-STATUS_PRECEDENCE: Dict[Status, int] = {
+STATUS_PRECEDENCE = {
     Status.ERROR: 9,
     Status.INCOMPLETE: 18,
     Status.XPASS_STRICT: 18,
@@ -1019,8 +1036,7 @@ def merge(self, report, strict=True):
         if self.suite_related and self.status.precede(report.status):
             return
 
-        # FIXME
-        self.status_override = report.status_override
+        self.status_override = Status.precedent([self.status, report.status])
         self.runtime_status = report.runtime_status
         self.logs = report.logs
         self.entries = report.entries

testplan/runnable/base.py

@@ -231,7 +231,7 @@ def get_options(cls):
                 "test_breaker_thres", default=common.TestBreakerThres.null()
             ): Or(
                 And(str, Use(common.TestBreakerThres.parse)),
-                Use(lambda _: common.TestBreakerThres.null()),
+                And(None, Use(lambda _: common.TestBreakerThres.null())),
             ),
         }
 

tests/unit/testplan/common/config/test_generic_config.py

@@ -3,7 +3,7 @@
 import os
 import re
 
-from schema import Optional, Or, SchemaError
+from schema import And, Optional, Or, SchemaError, Use
 
 from testplan.common.config import Config, ConfigOption
 from testplan.common.entity import Entity
@@ -51,18 +51,33 @@ def get_options(cls):
         }
 
 
+class Complex(Second):
+    @classmethod
+    def get_options(cls):
+        return {
+            ConfigOption("e", default=complex("-j")): And(str, Use(complex)),
+            "f": And(callable, Use(lambda x: lambda y: complex(x(y)))),
+        }
+
+
 def test_basic_config():
     """Basic config operations."""
     item = First()
     assert (1, 2, 3) == (item.a, item.b, item.c)
 
-    item = First(a=5, b=4)
-    assert (5, 4, 3) == (item.a, item.b, item.c)
+    item = Complex(a=0.5, b=4, e="10-10j", f=lambda x: x.replace(" ", ""))
+    assert (0.5, 4, 9.0, complex("10-10j")) == (item.a, item.b, item.c, item.e)
 
     clone = item.denormalize()
     assert id(clone) != id(item)
     assert clone.parent is None
-    assert (clone.a, clone.b, clone.c) == (item.a, item.b, item.c)
+    assert (clone.a, clone.b, clone.c, clone.e) == (
+        item.a,
+        item.b,
+        item.c,
+        item.e,
+    )
+    assert clone.f("- 10j ") == complex(real=0, imag=-10)
 
 
 def test_basic_schema_fail():

tests/unit/testplan/report/test_testing.py

@@ -24,6 +24,26 @@
 DummyReportGroup = functools.partial(BaseReportGroup, name="dummy")
 
 
+def test_report_status_basic_op():
+    assert Status.ERROR <= Status.ERROR
+    assert Status.FAILED > Status.ERROR
+    assert Status.INCOMPLETE < Status.FAILED
+    with pytest.raises(TypeError):
+        Status.INCOMPLETE < Status.XPASS_STRICT
+    with pytest.raises(TypeError):
+        Status.XFAIL >= Status.SKIPPED
+    assert Status.XFAIL != Status.XPASS
+    assert Status.XFAIL is not Status.XPASS
+    assert Status.UNKNOWN < Status.NONE
+    assert not Status.NONE
+
+    assert Status.XPASS_STRICT.normalised() is Status.FAILED
+    assert Status.PASSED.normalised() is Status.PASSED
+
+    assert not Status.INCOMPLETE.precede(Status.XPASS_STRICT)
+    assert Status.INCOMPLETE.precede(Status.FAILED)
+
+
 def test_report_status_precedent():
     """
     `precedent` should return the value with the
@@ -569,8 +589,16 @@ def iter_report_entries(report):
             yield from iter_report_entries(entry)
 
 
+def test_runtime_status_basic_op():
+    assert RuntimeStatus.WAITING < RuntimeStatus.READY
+    assert RuntimeStatus.RESETTING >= RuntimeStatus.RUNNING
+    assert RuntimeStatus.RUNNING.precede(RuntimeStatus.FINISHED)
+    assert RuntimeStatus.NOT_RUN < RuntimeStatus.NONE
+    assert not RuntimeStatus.NONE
+
+
 def test_runtime_status_setting(dummy_test_plan_report):
-    for status in RuntimeStatus.all_statuses()[:-1]:
+    for status in list(RuntimeStatus)[:-1]:
         dummy_test_plan_report.runtime_status = status
         assert dummy_test_plan_report.runtime_status == status
         for entry in iter_report_entries(dummy_test_plan_report):
@@ -592,14 +620,14 @@ def test_runtime_status_setting_filtered(dummy_test_plan_report):
         "test_case_1",
         "test_case_2",
     ]
-    for status in RuntimeStatus.all_statuses()[:-1]:
+    for status in list(RuntimeStatus)[:-1]:
         dummy_test_plan_report.set_runtime_status_filtered(
             status, filtered_entries
         )
         # Due to precedence logic, as soon as we hit NOT_RUN and FINISHED
         # the not run entry's READY status will be returned in the getter.
         # This is expected behavior.
-        if status in RuntimeStatus.all_statuses()[:-3]:
+        if status in list(RuntimeStatus)[:-3]:
             assert (
                 dummy_test_plan_report["Test Group 2"].runtime_status == status
             )