Capture assertion errors in pytest fixture.

python/lib/sift_client/_tests/conftest.py

@@ -32,21 +32,6 @@ def sift_client() -> SiftClient:
     return client
 
 
-@pytest.fixture(scope="session")
-def client_has_connection(sift_client):
-    """Check if the SiftClient has a connection to the Sift server.
-
-    Can be used to skip tests that require a connection to the Sift server.
-    """
-    has_connection = False
-    try:
-        sift_client.ping.ping()
-        has_connection = True
-    except Exception:
-        has_connection = False
-    return has_connection
-
-
 @pytest.fixture
 def mock_client():
     """Create a mock SiftClient for unit testing."""
@@ -89,4 +74,16 @@ def ci_pytest_tag(sift_client):
     return tag
 
 
-from sift_client.util.test_results import module_substep, report_context, step  # noqa: F401
+from sift_client.util.test_results import (
+    client_has_connection,  # noqa: F401
+    pytest_runtest_makereport,  # noqa: F401
+)
+from sift_client.util.test_results import (
+    module_substep_check_connection as module_substep,  # noqa: F401
+)
+from sift_client.util.test_results import (
+    report_context_check_connection as report_context,  # noqa: F401
+)
+from sift_client.util.test_results import (
+    step_check_connection as step,  # noqa: F401
+)

python/lib/sift_client/_tests/util/test_test_results_utils.py

@@ -123,6 +123,24 @@ def test_measurement_update(self, report_context):
         assert test_step.measurements[2].boolean_value == True
         assert test_step.measurements[2].measurement_type == TestMeasurementType.BOOLEAN
 
+    def test_bad_assert(self, report_context, step):
+        test_step = None
+        parent_step_path = step.current_step.step_path
+        initial_open_step_result = report_context.open_step_results.get(parent_step_path, True)
+        initial_any_failures = report_context.any_failures
+
+        with step.substep("Test Bad Assert", "Test Bad Assert Description") as new_step:
+            test_step = new_step.current_step
+            assert False == True
+
+        assert test_step.status == TestStatus.ERROR
+        assert test_step.error_info is not None
+        assert "AssertionError" in test_step.error_info.error_message
+        if initial_open_step_result:
+            report_context.open_step_results[parent_step_path] = True
+        if not initial_any_failures:
+            report_context.any_failures = False
+
     def test_error_info(self, report_context, step):
         test_step = None
         parent_step_path = step.current_step.step_path
@@ -268,20 +286,39 @@ def test_evaluate_measurement_bounds_boolean_matching(self):
         """Test boolean value matching expected string."""
         measurement = TestMeasurementUpdate()
         result = evaluate_measurement_bounds(measurement, True, "true")
+        result2 = evaluate_measurement_bounds(measurement, True, "True")
+        result3 = evaluate_measurement_bounds(measurement, True, "TRUE")
+        result4 = evaluate_measurement_bounds(measurement, True, True)
+        result5 = evaluate_measurement_bounds(measurement, True, 1)
+        result6 = evaluate_measurement_bounds(measurement, 1, True)
         assert result == True
+        assert result2 == True
+        assert result3 == True
+        assert result4 == True
+        assert result5 == True
+        assert result6 == True
         assert measurement.passed == True
         assert measurement.boolean_value == True
-        assert measurement.string_expected_value == "true"
-        assert measurement.measurement_type == TestMeasurementType.BOOLEAN
+        assert measurement.string_expected_value.lower() == "true"
 
     def test_evaluate_measurement_bounds_boolean_not_matching(self):
         """Test boolean value not matching expected string."""
         measurement = TestMeasurementUpdate()
         result = evaluate_measurement_bounds(measurement, False, "true")
+        result2 = evaluate_measurement_bounds(measurement, False, "tRuE")
+        result3 = evaluate_measurement_bounds(measurement, False, "TRUE")
+        result4 = evaluate_measurement_bounds(measurement, False, True)
+        result5 = evaluate_measurement_bounds(measurement, 0, True)
+        result6 = evaluate_measurement_bounds(measurement, "False", True)
         assert result == False
+        assert result2 == False
+        assert result3 == False
+        assert result4 == False
+        assert result5 == False
+        assert result6 == False
         assert measurement.passed == False
         assert measurement.boolean_value == False
-        assert measurement.string_expected_value == "true"
+        assert measurement.string_expected_value.lower() == "true"
 
     def test_evaluate_measurement_bounds_boolean_case_insensitive(self):
         """Test boolean comparison is case insensitive."""

python/lib/sift_client/util/test_results/__init__.py

@@ -9,7 +9,7 @@
 ### Example
 
 ```python
-client = SiftClient(api_key=api_key,grpc_url=grpc_url,rest_url=rest_url)
+client = SiftClient(api_key=api_key, grpc_url=grpc_url, rest_url=rest_url)
 with ReportContext(client, name="Example Report") as rc:
     with rc.new_step(name="Setup") as step:
         controller_setup(step)
@@ -28,7 +28,7 @@
 For a larger class or script, consider creating the context in a setup method and passing it to the test functions.
 ```python
 def main(self):
-    self.sift_client = SiftClient(api_key=api_key,grpc_url=grpc_url,rest_url=rest_url)
+    self.sift_client = SiftClient(api_key=api_key, grpc_url=grpc_url, rest_url=rest_url)
     with ReportContext(self.sift_client, name="Test Class", description="Test Class") as rc:
         setup(rc)
         test_one(rc)
@@ -57,11 +57,11 @@ def sift_client() -> SiftClient:
     rest_url = os.getenv("SIFT_REST_URI", "localhost:8080")
     api_key = os.getenv("SIFT_API_KEY", "")
 
-    client = SiftClient(api_key=api_key,grpc_url=grpc_url,rest_url=rest_url)
+    client = SiftClient(api_key=api_key, grpc_url=grpc_url, rest_url=rest_url)
 
     return client
 
-from sift_client.util.test_results import report_context, step, module_substep
+from sift_client.util.test_results import pytest_runtest_makereport, report_context, step, module_substep
 ```
 
 ###### Then in your test file:
@@ -83,6 +83,26 @@ def test_example(report_context, step):
 """
 
 from .context_manager import NewStep, ReportContext
-from .pytest_util import module_substep, report_context, step
-
-__all__ = ["NewStep", "ReportContext", "module_substep", "report_context", "step"]
+from .pytest_util import (
+    client_has_connection,
+    module_substep,
+    module_substep_check_connection,
+    pytest_runtest_makereport,
+    report_context,
+    report_context_check_connection,
+    step,
+    step_check_connection,
+)
+
+__all__ = [
+    "NewStep",
+    "ReportContext",
+    "client_has_connection",
+    "module_substep",
+    "module_substep_check_connection",
+    "pytest_runtest_makereport",
+    "report_context",
+    "report_context_check_connection",
+    "step",
+    "step_check_connection",
+]

python/lib/sift_client/util/test_results/bounds.py

@@ -35,14 +35,14 @@ def assign_value_to_measurement(
 def evaluate_measurement_bounds(
     measurement: TestMeasurement | TestMeasurementCreate | TestMeasurementUpdate,
     value: float | str | bool,
-    bounds: dict[str, float] | NumericBounds | str | None,
+    bounds: dict[str, float] | NumericBounds | str | bool | None,
 ) -> bool:
     """Update a measurement with the resolved bounds type and result of evaluating the given value against those bounds.
 
     Args:
         measurement: The measurement to update.
         value: The value to evaluate the bounds of.
-        bounds: The bounds to evaluate the value against.
+        bounds: The bounds to evaluate the value against. Either a dictionary with "min" and "max" keys, a NumericBounds object, a string, a boolean, or None.
 
     Returns:
         True if the value is within the bounds, False otherwise.
@@ -53,7 +53,13 @@ def evaluate_measurement_bounds(
 
     if isinstance(bounds, dict):
         bounds = NumericBounds(min=bounds.get("min"), max=bounds.get("max"))
-    if isinstance(bounds, str):
+    if isinstance(bounds, bool):
+        if isinstance(value, str):
+            measurement.passed = str(value).lower() == str(bounds).lower()
+        else:
+            measurement.passed = bool(value) == bounds
+        return bool(measurement.passed)
+    elif isinstance(bounds, str):
         if not (isinstance(value, str) or isinstance(value, bool)):
             raise ValueError("Value must be a string if bounds provided is a string")
         measurement.string_expected_value = bounds

python/lib/sift_client/util/test_results/context_manager.py

@@ -211,22 +211,39 @@ def __enter__(self):
         """
         return self
 
-    def __exit__(self, exc, exc_value, tb):
+    def update_step_from_result(
+        self,
+        exc: type[Exception] | None,
+        exc_value: Exception | None,
+        tb: traceback.TracebackException | None,
+    ):
+        """Update the step based on its substeps and if there was an exception while executing the step.
+
+        Args:
+            exc: The class of Exception that was raised.
+            exc_value: The exception value.
+            tb: The traceback object.
+        """
         error_info = None
         if exc:
-            trace = "".join(traceback.format_exception(exc, exc_value, tb, limit=10))
+            stack = traceback.format_exception(exc, exc_value, tb)  # type: ignore
+            stack = [stack[0], *stack[-10:]] if len(stack) > 10 else stack
+            trace = "".join(stack)
             error_info = ErrorInfo(
                 error_code=1,
                 error_message=trace,
             )
+        assert self.current_step is not None
 
         # Resolve the status of this step (i.e. fail if children failed) and propagate the result to the parent step.
         result = self.report_context.resolve_and_propagate_step_result(
             self.current_step, self.parent_step, error_info
         )
 
         # Mark the step as completed
-        status = TestStatus.PASSED if result else TestStatus.FAILED
+        status = self.current_step.status
+        if status == TestStatus.IN_PROGRESS:
+            status = TestStatus.PASSED if result else TestStatus.FAILED
         if error_info:
             status = TestStatus.ERROR
         self.current_step.update(
@@ -237,6 +254,9 @@ def __exit__(self, exc, exc_value, tb):
             }
         )
 
+    def __exit__(self, exc, exc_value, tb):
+        self.update_step_from_result(exc, exc_value, tb)
+
         # Now that the step is updated. Let the report context handle removing it from the stack and updating the report context.
         self.report_context.exit_step(self.current_step)