diff --git a/python/lib/sift_client/_tests/util/test_test_results_utils.py b/python/lib/sift_client/_tests/util/test_test_results_utils.py index 61d08a8d2..82bea7c0c 100644 --- a/python/lib/sift_client/_tests/util/test_test_results_utils.py +++ b/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. diff --git a/python/lib/sift_client/util/test_results/context_manager.py b/python/lib/sift_client/util/test_results/context_manager.py index 38feeb65e..da7de8c65 100644 --- a/python/lib/sift_client/util/test_results/context_manager.py +++ b/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)