pyhon(feat): Add more funcs to test results context managers (#440)

Author: ian-sift

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

@@ -1,8 +1,11 @@
 from datetime import datetime, timezone
 
+import numpy as np
+import pandas as pd
 import pytest
 
 from sift_client.sift_types.test_report import (
+    NumericBounds,
     TestMeasurementCreate,
     TestMeasurementType,
     TestMeasurementUpdate,
@@ -141,16 +144,220 @@ def test_measurement_update(self, report_context):
             new_step.measure(name="Test Measurement 2", value="string value", bounds="string value")
             new_step.measure(name="Test Measurement 3", value=True, bounds="true")
 
-        assert len(test_step.measurements) == 3
-        assert test_step.measurements[0].name == "Test Measurement"
-        assert test_step.measurements[0].numeric_value == 10
-        assert test_step.measurements[0].measurement_type == TestMeasurementType.DOUBLE
-        assert test_step.measurements[1].name == "Test Measurement 2"
-        assert test_step.measurements[1].string_value == "string value"
-        assert test_step.measurements[1].measurement_type == TestMeasurementType.STRING
-        assert test_step.measurements[2].name == "Test Measurement 3"
-        assert test_step.measurements[2].boolean_value == True
-        assert test_step.measurements[2].measurement_type == TestMeasurementType.BOOLEAN
+        measurements = test_step.measurements
+        assert len(measurements) == 3
+        assert measurements[0].name == "Test Measurement"
+        assert measurements[0].numeric_value == 10
+        assert measurements[0].measurement_type == TestMeasurementType.DOUBLE
+        assert measurements[1].name == "Test Measurement 2"
+        assert measurements[1].string_value == "string value"
+        assert measurements[1].measurement_type == TestMeasurementType.STRING
+        assert measurements[2].name == "Test Measurement 3"
+        assert measurements[2].boolean_value == True
+        assert measurements[2].measurement_type == TestMeasurementType.BOOLEAN
+
+    def test_measure_avg_list_within_bounds(self, step):
+        """Test measure_avg with a list of values where average is within bounds."""
+        result = step.measure_avg(
+            name="Avg Temperature",
+            values=[10.0, 20.0, 30.0],  # avg = 20.0
+            bounds={"min": 15.0, "max": 25.0},
+        )
+        assert result == True
+        assert step.current_step.measurements[0].name == "Avg Temperature"
+        assert step.current_step.measurements[0].numeric_value == 20.0
+        assert step.current_step.measurements[0].passed == True
+
+    def test_measure_avg_list_outside_bounds(self, report_context, step):
+        """Test measure_avg with a list where average is outside bounds."""
+        # Capture initial state to restore after test
+        current_step_path = step.current_step.step_path
+        initial_open_step_result = report_context.open_step_results.get(current_step_path, True)
+        initial_any_failures = report_context.any_failures
+
+        result = step.measure_avg(
+            name="Avg Temperature Fail",
+            values=[50.0, 60.0, 70.0],  # avg = 60.0
+            bounds={"min": 15.0, "max": 25.0},
+        )
+        assert result == False
+        assert step.current_step.measurements[0].numeric_value == 60.0
+        assert step.current_step.measurements[0].passed == False
+
+        # Restore state
+        if initial_open_step_result:
+            report_context.open_step_results[current_step_path] = True
+        if not initial_any_failures:
+            report_context.any_failures = False
+
+    def test_measure_avg_numpy_array(self, step):
+        """Test measure_avg with a numpy array."""
+        result = step.measure_avg(
+            name="Avg Pressure",
+            values=np.array([100.0, 200.0, 300.0]),  # avg = 200.0
+            bounds={"min": 150.0, "max": 250.0},
+        )
+        assert result == True
+        assert step.current_step.measurements[0].numeric_value == 200.0
+        assert step.current_step.measurements[0].passed == True
+
+    def test_measure_avg_pandas_series(self, step):
+        """Test measure_avg with a pandas Series."""
+        series = pd.Series([5.0, 10.0, 15.0])  # avg = 10.0
+        result = step.measure_avg(
+            name="Avg Voltage",
+            values=series,
+            bounds={"min": 5.0, "max": 15.0},
+        )
+        assert result == True
+        assert step.current_step.measurements[0].numeric_value == 10.0
+        assert step.current_step.measurements[0].passed == True
+
+    def test_measure_avg_with_numeric_bounds_object(self, step):
+        """Test measure_avg with NumericBounds object instead of dict."""
+        result = step.measure_avg(
+            name="Avg Current",
+            values=[1.0, 2.0, 3.0],  # avg = 2.0
+            bounds=NumericBounds(min=1.0, max=3.0),
+        )
+        assert result == True
+        assert step.current_step.measurements[0].numeric_value == 2.0
+        assert step.current_step.measurements[0].passed == True
+
+    def test_measure_avg_invalid_type(self, step):
+        """Test measure_avg raises ValueError for invalid value type."""
+        with pytest.raises(ValueError, match="Invalid value type"):
+            step.measure_avg(
+                name="Invalid",
+                values="not a list",  # type: ignore
+                bounds={"min": 0.0, "max": 10.0},
+            )
+
+    def test_measure_avg_with_integers(self, step):
+        """Test measure_avg with integer values in list."""
+        result = step.measure_avg(
+            name="Avg Count",
+            values=[1, 2, 3, 4, 5],  # avg = 3.0
+            bounds={"min": 2.0, "max": 4.0},
+        )
+        assert result == True
+        assert step.current_step.measurements[0].numeric_value == 3.0
+        assert step.current_step.measurements[0].passed == True
+
+    def test_measure_all_list_within_bounds(self, step):
+        """Test measure_all with a list of values all within bounds."""
+        result = step.measure_all(
+            name="All Temperatures",
+            values=[10.0, 15.0, 20.0],
+            bounds={"min": 5.0, "max": 25.0},
+        )
+        assert result == True
+
+    def test_measure_all_list_some_outside_bounds(self, report_context, step):
+        """Test measure_all with a list where some values are outside bounds."""
+        # Capture initial state to restore after test
+        current_step_path = step.current_step.step_path
+        initial_open_step_result = report_context.open_step_results.get(current_step_path, True)
+        initial_any_failures = report_context.any_failures
+
+        result = step.measure_all(
+            name="temp",
+            values=[10.0, 50.0, 20.0, -1.0],  # 50.0 and -1.0 are outside
+            bounds={"min": 5.0, "max": 25.0},
+            unit="C",
+        )
+        assert result == False
+        test_step = step.current_step
+        measurements = test_step.measurements
+        measurements.sort(key=lambda x: x.numeric_value)
+        assert len(measurements) == 2
+        assert measurements[0].numeric_value == -1.0
+        assert measurements[0].passed == False
+        assert measurements[1].numeric_value == 50.0
+        assert measurements[1].passed == False
+
+        # Restore state
+        if initial_open_step_result:
+            report_context.open_step_results[current_step_path] = True
+        if not initial_any_failures:
+            report_context.any_failures = False
+
+    def test_measure_all_numpy_array(self, step):
+        """Test measure_all with a numpy array."""
+        result = step.measure_all(
+            name="All Pressures",
+            values=np.array([100.0, 150.0, 200.0]),
+            bounds={"min": 50.0, "max": 250.0},
+        )
+        assert result == True
+
+    def test_measure_all_pandas_series(self, step):
+        """Test measure_all with a pandas Series."""
+        series = pd.Series([5.0, 10.0, 15.0])
+        result = step.measure_all(
+            name="All Voltages",
+            values=series,
+            bounds={"min": 0.0, "max": 20.0},
+        )
+        assert result == True
+
+    def test_measure_all_with_numeric_bounds_object(self, step):
+        """Test measure_all with NumericBounds object instead of dict."""
+        result = step.measure_all(
+            name="All Currents",
+            values=[1.0, 2.0, 3.0],
+            bounds=NumericBounds(min=0.0, max=5.0),
+        )
+        assert result == True
+
+    def test_measure_all_invalid_type(self, step):
+        """Test measure_all raises ValueError for invalid value type."""
+        with pytest.raises(ValueError, match="Invalid value type"):
+            step.measure_all(
+                name="Invalid",
+                values="not a list",  # type: ignore
+                bounds={"min": 0.0, "max": 10.0},
+            )
+
+    def test_measure_all_no_bounds(self, step):
+        """Test measure_all raises ValueError when no bounds provided."""
+        with pytest.raises(ValueError, match="No bounds provided"):
+            step.measure_all(
+                name="No Bounds",
+                values=[1.0, 2.0, 3.0],
+                bounds={},  # Empty bounds dict
+            )
+
+    def test_measure_all_min_only(self, step):
+        """Test measure_all with only minimum bound."""
+        result = step.measure_all(
+            name="Min Only",
+            values=[10.0, 20.0, 30.0],
+            bounds={"min": 5.0},
+        )
+        assert result == True
+
+    def test_measure_all_max_only(self, step):
+        """Test measure_all with only maximum bound."""
+        result = step.measure_all(
+            name="Max Only",
+            values=[10.0, 20.0, 30.0],
+            bounds={"max": 50.0},
+        )
+        assert result == True
+
+    def test_report_outcome(self, report_context, step):
+        # Capture current state of report context's failures so we can keep things passed at a high level if the test's induced failures happen as expected.
+        current_step_path = step.current_step.step_path
+        initial_open_step_result = report_context.open_step_results.get(current_step_path, True)
+        initial_any_failures = report_context.any_failures
+        assert step.report_outcome("Test Pass Outcome", True, "Test Pass Description") == True
+        assert step.report_outcome("Test Fail Outcome", False, "Test Failure Description") == False
+        # If this test was successful, mark that at a high level.
+        if initial_open_step_result:
+            report_context.open_step_results[current_step_path] = True
+        if not initial_any_failures:
+            report_context.any_failures = False
 
     def test_bad_assert(self, report_context, step):
         # Capture current state of report context's failures so we can keep things passed at a high level if the test's induced failures happen as expected.

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

@@ -8,10 +8,12 @@
 from datetime import datetime, timezone
 from typing import TYPE_CHECKING
 
+import numpy as np
+import pandas as pd
+
 from sift_client.sift_types.test_report import (
     ErrorInfo,
     NumericBounds,
-    TestMeasurement,
     TestMeasurementCreate,
     TestReport,
     TestReportCreate,
@@ -25,6 +27,8 @@
 )
 
 if TYPE_CHECKING:
+    from numpy.typing import NDArray
+
     from sift_client.client import SiftClient
 
 
@@ -138,10 +142,10 @@ def create_step(self, name: str, description: str | None = None) -> TestStep:
 
         return step
 
-    def report_measurement(self, measurement: TestMeasurement, step: TestStep):
+    def record_step_outcome(self, outcome: bool, step: TestStep):
         """Report a failure to the report context."""
         # Failures will be propogated when the step exits.
-        if not measurement.passed:
+        if not outcome:
             self.open_step_results[step.step_path] = False
             self.any_failures = True
 
@@ -303,10 +307,118 @@ def measure(
         )
         evaluate_measurement_bounds(create, value, bounds)
         measurement = self.client.test_results.create_measurement(create)
-        self.report_context.report_measurement(measurement, self.current_step)
+        self.report_context.record_step_outcome(measurement.passed, self.current_step)
 
         return measurement.passed
 
+    def measure_avg(
+        self,
+        *,
+        name: str,
+        values: list[float | int] | NDArray[np.float64] | pd.Series,
+        bounds: dict[str, float] | NumericBounds,
+        timestamp: datetime | None = None,
+        unit: str | None = None,
+    ) -> bool:
+        """Calculate the average of a list of values, measure the average against given bounds, and return the result.
+
+        Args:
+            name: The name of the measurement.
+            values: The list of values to measure the average of.
+            bounds: The bounds to compare the value to.
+            timestamp: [Optional] The timestamp of the measurement. Defaults to the current time.
+            unit: [Optional] The unit of the measurement.
+
+        returns: The true if the average of the values is within the bounds, false otherwise.
+        """
+        timestamp = timestamp if timestamp else datetime.now(timezone.utc)
+        np_array = None
+        if isinstance(values, list):
+            np_array = np.array(values)
+        elif isinstance(values, np.ndarray):
+            np_array = values
+        elif isinstance(values, pd.Series):
+            np_array = values.to_numpy()
+        else:
+            raise ValueError(f"Invalid value type: {type(values)}")
+        avg = float(np.mean(np_array))
+        result = self.measure(name=name, value=avg, bounds=bounds, timestamp=timestamp, unit=unit)
+        assert self.current_step is not None
+        self.report_context.record_step_outcome(result, self.current_step)
+
+        return result
+
+    def measure_all(
+        self,
+        *,
+        name: str,
+        values: list[float | int] | NDArray[np.float64] | pd.Series,
+        bounds: dict[str, float] | NumericBounds,
+        timestamp: datetime | None = None,
+        unit: str | None = None,
+    ) -> bool:
+        """Ensure that all values in a list are within bounds and return the result. Records measurements for all values outside the bounds.
+
+        Note: Measurements will only be recorded for values outside the bounds. To record measurements for all values, just call measure for each value.
+
+        Args:
+            name: The name of the measurement.
+            values: The list of values to measure the average of.
+            bounds: The bounds to compare the value to.
+            timestamp: [Optional] The timestamp of the measurement. Defaults to the current time.
+            unit: [Optional] The unit of the measurement.
+
+        returns: The true if all values are within the bounds, false otherwise.
+        """
+        timestamp = timestamp if timestamp else datetime.now(timezone.utc)
+        np_array = None
+        if isinstance(values, list):
+            np_array = np.array(values)
+        elif isinstance(values, np.ndarray):
+            np_array = values
+        elif isinstance(values, pd.Series):
+            np_array = values.to_numpy()
+        else:
+            raise ValueError(f"Invalid value type: {type(values)}")
+
+        numeric_bounds = bounds
+        if isinstance(numeric_bounds, dict):
+            numeric_bounds = NumericBounds(min=bounds.get("min"), max=bounds.get("max"))  # type: ignore
+
+        # Construct a mask of the values that are outside the bounds.
+        mask = None
+        if numeric_bounds.min is not None:
+            mask = np_array < numeric_bounds.min
+        if numeric_bounds.max is not None:
+            val_above_max = np_array > numeric_bounds.max
+            mask = mask | val_above_max if mask is not None else val_above_max
+        if mask is None:
+            raise ValueError("No bounds provided")
+
+        rows_outside_bounds = np_array[mask]
+        for row in rows_outside_bounds:
+            self.measure(name=name, value=row, bounds=bounds, timestamp=timestamp, unit=unit)
+
+        result = rows_outside_bounds.size == 0
+        assert self.current_step is not None
+        self.report_context.record_step_outcome(result, self.current_step)
+
+        return result
+
+    def report_outcome(self, name: str, result: bool, reason: str | None = None) -> bool:
+        """Report an outcome from some action or measurement. Creates a substep that is pass/fail with the optional reason as the description.
+
+        Args:
+            name: The name of the substep.
+            result: True if the action or measurement passed, False otherwise.
+            reason: [Optional] The context to include in the description of the substep.
+
+        returns: The given result so the function can be used in line.
+        """
+        with self.substep(name=name, description=reason) as substep:
+            self.report_context.record_step_outcome(result, substep.current_step)
+        return result
+
     def substep(self, name: str, description: str | None = None) -> NewStep:
         """Alias to return a new step context manager from the current step. The ReportContext will manage nesting of steps."""
         return self.report_context.new_step(name=name, description=description)