From da23f2d26ef34e2f7d99545b001d9f89b384c8b5 Mon Sep 17 00:00:00 2001 From: Ian Later Date: Wed, 17 Dec 2025 19:56:08 -0800 Subject: [PATCH 1/5] python(feat): Add report_outcome(result, reason) functionality to test step context managers. --- .../_tests/util/test_test_results_utils.py | 13 ++++++++++++ .../util/test_results/context_manager.py | 20 ++++++++++++++++--- 2 files changed, 30 insertions(+), 3 deletions(-) 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..7a0bc2d4d 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 @@ -152,6 +152,19 @@ 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_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. current_step_path = step.current_step.step_path 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..09219b223 100644 --- a/python/lib/sift_client/util/test_results/context_manager.py +++ b/python/lib/sift_client/util/test_results/context_manager.py @@ -138,10 +138,10 @@ def create_step(self, name: str, description: str | None = None) -> TestStep: return step - def report_measurement(self, measurement: TestMeasurement, step: TestStep): + def report_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 +303,24 @@ 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.report_outcome(measurement.passed, self.current_step) return measurement.passed + 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.report_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) From 49d57ff0d67ec93c0615775f74d89f324423c354 Mon Sep 17 00:00:00 2001 From: Ian Later Date: Thu, 18 Dec 2025 16:30:16 -0800 Subject: [PATCH 2/5] python(feat): Add more measurement functions for test result utils. --- .../_tests/util/test_test_results_utils.py | 240 ++++++++++++++++++ .../util/test_results/context_manager.py | 104 +++++++- 2 files changed, 340 insertions(+), 4 deletions(-) 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 7a0bc2d4d..4f86b9953 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, @@ -152,6 +155,243 @@ 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_measure_avg_list_within_bounds(self, report_context): + """Test measure_avg with a list of values where average is within bounds.""" + with report_context.new_step( + "Test Measure Avg List", "Test measure_avg with list" + ) as new_step: + result = new_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 + test_step = new_step.current_step + assert len(test_step.measurements) == 1 + assert test_step.measurements[0].name == "Avg Temperature" + assert test_step.measurements[0].numeric_value == 20.0 + assert test_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 + + with report_context.new_step( + "Test Measure Avg Outside", "Test measure_avg outside bounds" + ) as new_step: + result = new_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 + test_step = new_step.current_step + assert len(test_step.measurements) == 1 + assert test_step.measurements[0].numeric_value == 60.0 + assert test_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, report_context): + """Test measure_avg with a numpy array.""" + with report_context.new_step( + "Test Measure Avg Numpy", "Test measure_avg with numpy array" + ) as new_step: + result = new_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 + test_step = new_step.current_step + assert test_step.measurements[0].numeric_value == 200.0 + assert test_step.measurements[0].passed == True + + def test_measure_avg_pandas_dataframe(self, report_context): + """Test measure_avg with a pandas DataFrame.""" + with report_context.new_step( + "Test Measure Avg DataFrame", "Test measure_avg with DataFrame" + ) as new_step: + df = pd.DataFrame({"values": [5.0, 10.0, 15.0]}) # avg = 10.0 + result = new_step.measure_avg( + name="Avg Voltage", + values=df, + bounds={"min": 5.0, "max": 15.0}, + ) + assert result == True + test_step = new_step.current_step + assert test_step.measurements[0].numeric_value == 10.0 + assert test_step.measurements[0].passed == True + + def test_measure_avg_with_numeric_bounds_object(self, report_context): + """Test measure_avg with NumericBounds object instead of dict.""" + with report_context.new_step( + "Test Measure Avg NumericBounds", "Test measure_avg with NumericBounds" + ) as new_step: + result = new_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 + test_step = new_step.current_step + assert test_step.measurements[0].numeric_value == 2.0 + assert test_step.measurements[0].passed == True + + def test_measure_avg_invalid_type(self, report_context): + """Test measure_avg raises ValueError for invalid value type.""" + with report_context.new_step( + "Test Measure Avg Invalid", "Test measure_avg with invalid type" + ) as new_step: + with pytest.raises(ValueError, match="Invalid value type"): + new_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, report_context): + """Test measure_avg with integer values in list.""" + with report_context.new_step( + "Test Measure Avg Integers", "Test measure_avg with integers" + ) as new_step: + result = new_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 + test_step = new_step.current_step + assert test_step.measurements[0].numeric_value == 3.0 + assert test_step.measurements[0].passed == True + + def test_measure_all_list_within_bounds(self, report_context): + """Test measure_all with a list of values all within bounds.""" + with report_context.new_step( + "Test Measure All List", "Test measure_all with list" + ) as new_step: + result = new_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 + + with report_context.new_step( + "Test Measure All Outside", "Test measure_all with values outside bounds" + ) as new_step: + result = new_step.measure_all( + name="All Temperatures Fail", + values=[10.0, 50.0, 20.0, 60.0], # 50.0 and 60.0 are outside + bounds={"min": 5.0, "max": 25.0}, + ) + assert result == 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, report_context): + """Test measure_all with a numpy array.""" + with report_context.new_step( + "Test Measure All Numpy", "Test measure_all with numpy array" + ) as new_step: + result = new_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_dataframe(self, report_context): + """Test measure_all with a pandas DataFrame.""" + with report_context.new_step( + "Test Measure All DataFrame", "Test measure_all with DataFrame" + ) as new_step: + df = pd.DataFrame({"values": [5.0, 10.0, 15.0]}) + result = new_step.measure_all( + name="All Voltages", + values=df, + bounds={"min": 0.0, "max": 20.0}, + ) + assert result == True + + def test_measure_all_with_numeric_bounds_object(self, report_context): + """Test measure_all with NumericBounds object instead of dict.""" + with report_context.new_step( + "Test Measure All NumericBounds", "Test measure_all with NumericBounds" + ) as new_step: + result = new_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, report_context): + """Test measure_all raises ValueError for invalid value type.""" + with report_context.new_step( + "Test Measure All Invalid", "Test measure_all with invalid type" + ) as new_step: + with pytest.raises(ValueError, match="Invalid value type"): + new_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, report_context): + """Test measure_all raises ValueError when no bounds provided.""" + with report_context.new_step( + "Test Measure All No Bounds", "Test measure_all with no bounds" + ) as new_step: + with pytest.raises(ValueError, match="No bounds provided"): + new_step.measure_all( + name="No Bounds", + values=[1.0, 2.0, 3.0], + bounds={}, # Empty bounds dict + ) + + def test_measure_all_min_only(self, report_context): + """Test measure_all with only minimum bound.""" + with report_context.new_step( + "Test Measure All Min Only", "Test measure_all with min only" + ) as new_step: + result = new_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, report_context): + """Test measure_all with only maximum bound.""" + with report_context.new_step( + "Test Measure All Max Only", "Test measure_all with max only" + ) as new_step: + result = new_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 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 09219b223..6abe0ef9b 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, @@ -26,6 +28,7 @@ if TYPE_CHECKING: from sift_client.client import SiftClient + from numpy.typing import NDArray class ReportContext(AbstractContextManager): @@ -138,7 +141,7 @@ def create_step(self, name: str, description: str | None = None) -> TestStep: return step - def report_outcome(self, outcome: bool, 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 outcome: @@ -237,6 +240,7 @@ def update_step_from_result( error_code=1, error_message=trace, ) + # print("".join(stack)) 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. @@ -303,10 +307,102 @@ def measure( ) evaluate_measurement_bounds(create, value, bounds) measurement = self.client.test_results.create_measurement(create) - self.report_context.report_outcome(measurement.passed, 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.DataFrame, + bounds: dict[str, float] | NumericBounds, + timestamp: datetime | None = None, + unit: str | None = None, + ) -> bool: + """Measure the average of a list of values 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.DataFrame): + np_array = values.values + 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.DataFrame, + 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. + + 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.DataFrame): + np_array = values.values + 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. @@ -318,7 +414,7 @@ def report_outcome(self, name: str, result: bool, reason: str | None = None) -> returns: The given result so the function can be used in line. """ with self.substep(name=name, description=reason) as substep: - self.report_context.report_outcome(result, substep.current_step) + self.report_context.record_step_outcome(result, substep.current_step) return result def substep(self, name: str, description: str | None = None) -> NewStep: From d7678883f59952a729778a52a59217aa7ec66e6b Mon Sep 17 00:00:00 2001 From: Ian Later Date: Mon, 22 Dec 2025 17:23:38 -0600 Subject: [PATCH 3/5] rm unused print --- python/lib/sift_client/util/test_results/context_manager.py | 1 - 1 file changed, 1 deletion(-) 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 6abe0ef9b..14fc06924 100644 --- a/python/lib/sift_client/util/test_results/context_manager.py +++ b/python/lib/sift_client/util/test_results/context_manager.py @@ -240,7 +240,6 @@ def update_step_from_result( error_code=1, error_message=trace, ) - # print("".join(stack)) 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. From 6bcfda68a703edefcd3a13b2cfb2f9fb40cc3f99 Mon Sep 17 00:00:00 2001 From: Ian Later Date: Mon, 22 Dec 2025 17:44:21 -0600 Subject: [PATCH 4/5] cleanup --- .../_tests/util/test_test_results_utils.py | 344 ++++++++---------- .../util/test_results/context_manager.py | 9 +- 2 files changed, 155 insertions(+), 198 deletions(-) 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 4f86b9953..127cab13e 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 @@ -144,33 +144,29 @@ 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 - - def test_measure_avg_list_within_bounds(self, report_context): + 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.""" - with report_context.new_step( - "Test Measure Avg List", "Test measure_avg with list" - ) as new_step: - result = new_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 - test_step = new_step.current_step - assert len(test_step.measurements) == 1 - assert test_step.measurements[0].name == "Avg Temperature" - assert test_step.measurements[0].numeric_value == 20.0 - assert test_step.measurements[0].passed == True + 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.""" @@ -179,19 +175,14 @@ def test_measure_avg_list_outside_bounds(self, report_context, step): initial_open_step_result = report_context.open_step_results.get(current_step_path, True) initial_any_failures = report_context.any_failures - with report_context.new_step( - "Test Measure Avg Outside", "Test measure_avg outside bounds" - ) as new_step: - result = new_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 - test_step = new_step.current_step - assert len(test_step.measurements) == 1 - assert test_step.measurements[0].numeric_value == 60.0 - assert test_step.measurements[0].passed == False + 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: @@ -199,90 +190,68 @@ def test_measure_avg_list_outside_bounds(self, report_context, step): if not initial_any_failures: report_context.any_failures = False - def test_measure_avg_numpy_array(self, report_context): + def test_measure_avg_numpy_array(self, step): """Test measure_avg with a numpy array.""" - with report_context.new_step( - "Test Measure Avg Numpy", "Test measure_avg with numpy array" - ) as new_step: - result = new_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 - test_step = new_step.current_step - assert test_step.measurements[0].numeric_value == 200.0 - assert test_step.measurements[0].passed == True + 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_dataframe(self, report_context): + def test_measure_avg_pandas_dataframe(self, step): """Test measure_avg with a pandas DataFrame.""" - with report_context.new_step( - "Test Measure Avg DataFrame", "Test measure_avg with DataFrame" - ) as new_step: - df = pd.DataFrame({"values": [5.0, 10.0, 15.0]}) # avg = 10.0 - result = new_step.measure_avg( - name="Avg Voltage", - values=df, - bounds={"min": 5.0, "max": 15.0}, - ) - assert result == True - test_step = new_step.current_step - assert test_step.measurements[0].numeric_value == 10.0 - assert test_step.measurements[0].passed == True + df = pd.DataFrame({"values": [5.0, 10.0, 15.0]}) # avg = 10.0 + result = step.measure_avg( + name="Avg Voltage", + values=df, + 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, report_context): + def test_measure_avg_with_numeric_bounds_object(self, step): """Test measure_avg with NumericBounds object instead of dict.""" - with report_context.new_step( - "Test Measure Avg NumericBounds", "Test measure_avg with NumericBounds" - ) as new_step: - result = new_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 - test_step = new_step.current_step - assert test_step.measurements[0].numeric_value == 2.0 - assert test_step.measurements[0].passed == True + 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, report_context): + def test_measure_avg_invalid_type(self, step): """Test measure_avg raises ValueError for invalid value type.""" - with report_context.new_step( - "Test Measure Avg Invalid", "Test measure_avg with invalid type" - ) as new_step: - with pytest.raises(ValueError, match="Invalid value type"): - new_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, report_context): - """Test measure_avg with integer values in list.""" - with report_context.new_step( - "Test Measure Avg Integers", "Test measure_avg with integers" - ) as new_step: - result = new_step.measure_avg( - name="Avg Count", - values=[1, 2, 3, 4, 5], # avg = 3.0 - bounds={"min": 2.0, "max": 4.0}, + 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}, ) - assert result == True - test_step = new_step.current_step - assert test_step.measurements[0].numeric_value == 3.0 - assert test_step.measurements[0].passed == True - def test_measure_all_list_within_bounds(self, report_context): + 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.""" - with report_context.new_step( - "Test Measure All List", "Test measure_all with list" - ) as new_step: - result = new_step.measure_all( - name="All Temperatures", - values=[10.0, 15.0, 20.0], - bounds={"min": 5.0, "max": 25.0}, - ) - assert result == True + 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.""" @@ -291,15 +260,21 @@ def test_measure_all_list_some_outside_bounds(self, report_context, step): initial_open_step_result = report_context.open_step_results.get(current_step_path, True) initial_any_failures = report_context.any_failures - with report_context.new_step( - "Test Measure All Outside", "Test measure_all with values outside bounds" - ) as new_step: - result = new_step.measure_all( - name="All Temperatures Fail", - values=[10.0, 50.0, 20.0, 60.0], # 50.0 and 60.0 are outside - bounds={"min": 5.0, "max": 25.0}, - ) - assert result == False + 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: @@ -307,90 +282,69 @@ def test_measure_all_list_some_outside_bounds(self, report_context, step): if not initial_any_failures: report_context.any_failures = False - def test_measure_all_numpy_array(self, report_context): + def test_measure_all_numpy_array(self, step): """Test measure_all with a numpy array.""" - with report_context.new_step( - "Test Measure All Numpy", "Test measure_all with numpy array" - ) as new_step: - result = new_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 + 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_dataframe(self, report_context): + def test_measure_all_pandas_dataframe(self, step): """Test measure_all with a pandas DataFrame.""" - with report_context.new_step( - "Test Measure All DataFrame", "Test measure_all with DataFrame" - ) as new_step: - df = pd.DataFrame({"values": [5.0, 10.0, 15.0]}) - result = new_step.measure_all( - name="All Voltages", - values=df, - bounds={"min": 0.0, "max": 20.0}, - ) - assert result == True + df = pd.DataFrame({"values": [5.0, 10.0, 15.0]}) + result = step.measure_all( + name="All Voltages", + values=df, + bounds={"min": 0.0, "max": 20.0}, + ) + assert result == True - def test_measure_all_with_numeric_bounds_object(self, report_context): + def test_measure_all_with_numeric_bounds_object(self, step): """Test measure_all with NumericBounds object instead of dict.""" - with report_context.new_step( - "Test Measure All NumericBounds", "Test measure_all with NumericBounds" - ) as new_step: - result = new_step.measure_all( - name="All Currents", - values=[1.0, 2.0, 3.0], - bounds=NumericBounds(min=0.0, max=5.0), - ) - assert result == True + 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, report_context): + def test_measure_all_invalid_type(self, step): """Test measure_all raises ValueError for invalid value type.""" - with report_context.new_step( - "Test Measure All Invalid", "Test measure_all with invalid type" - ) as new_step: - with pytest.raises(ValueError, match="Invalid value type"): - new_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, report_context): + 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 report_context.new_step( - "Test Measure All No Bounds", "Test measure_all with no bounds" - ) as new_step: - with pytest.raises(ValueError, match="No bounds provided"): - new_step.measure_all( - name="No Bounds", - values=[1.0, 2.0, 3.0], - bounds={}, # Empty bounds dict - ) - - def test_measure_all_min_only(self, report_context): - """Test measure_all with only minimum bound.""" - with report_context.new_step( - "Test Measure All Min Only", "Test measure_all with min only" - ) as new_step: - result = new_step.measure_all( - name="Min Only", - values=[10.0, 20.0, 30.0], - bounds={"min": 5.0}, + 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 ) - assert result == True - def test_measure_all_max_only(self, report_context): + 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.""" - with report_context.new_step( - "Test Measure All Max Only", "Test measure_all with max only" - ) as new_step: - result = new_step.measure_all( - name="Max Only", - values=[10.0, 20.0, 30.0], - bounds={"max": 50.0}, - ) - assert result == True + 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. 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 14fc06924..51d5c85c2 100644 --- a/python/lib/sift_client/util/test_results/context_manager.py +++ b/python/lib/sift_client/util/test_results/context_manager.py @@ -27,9 +27,10 @@ ) if TYPE_CHECKING: - from sift_client.client import SiftClient from numpy.typing import NDArray + from sift_client.client import SiftClient + class ReportContext(AbstractContextManager): """Context manager for a new TestReport. See usage example in __init__.py.""" @@ -319,7 +320,7 @@ def measure_avg( timestamp: datetime | None = None, unit: str | None = None, ) -> bool: - """Measure the average of a list of values and return the result. + """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. @@ -358,6 +359,8 @@ def measure_all( ) -> bool: """Ensure that all values in a list are within bounds and return the result. + 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. @@ -380,7 +383,7 @@ def measure_all( numeric_bounds = bounds if isinstance(numeric_bounds, dict): - numeric_bounds = NumericBounds(min=bounds.get("min"), max=bounds.get("max")) # type: ignore + 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 From 417d92d868c6ed97d4bd623b8486ea047b920730 Mon Sep 17 00:00:00 2001 From: Ian Later Date: Tue, 23 Dec 2025 11:55:47 -0600 Subject: [PATCH 5/5] use series instead of dataframe --- .../_tests/util/test_test_results_utils.py | 16 ++++++++-------- .../util/test_results/context_manager.py | 14 +++++++------- 2 files changed, 15 insertions(+), 15 deletions(-) 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 127cab13e..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 @@ -201,12 +201,12 @@ def test_measure_avg_numpy_array(self, step): assert step.current_step.measurements[0].numeric_value == 200.0 assert step.current_step.measurements[0].passed == True - def test_measure_avg_pandas_dataframe(self, step): - """Test measure_avg with a pandas DataFrame.""" - df = pd.DataFrame({"values": [5.0, 10.0, 15.0]}) # avg = 10.0 + 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=df, + values=series, bounds={"min": 5.0, "max": 15.0}, ) assert result == True @@ -291,12 +291,12 @@ def test_measure_all_numpy_array(self, step): ) assert result == True - def test_measure_all_pandas_dataframe(self, step): - """Test measure_all with a pandas DataFrame.""" - df = pd.DataFrame({"values": [5.0, 10.0, 15.0]}) + 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=df, + values=series, bounds={"min": 0.0, "max": 20.0}, ) assert result == True 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 51d5c85c2..da7de8c65 100644 --- a/python/lib/sift_client/util/test_results/context_manager.py +++ b/python/lib/sift_client/util/test_results/context_manager.py @@ -315,7 +315,7 @@ def measure_avg( self, *, name: str, - values: list[float | int] | NDArray[np.float64] | pd.DataFrame, + values: list[float | int] | NDArray[np.float64] | pd.Series, bounds: dict[str, float] | NumericBounds, timestamp: datetime | None = None, unit: str | None = None, @@ -337,8 +337,8 @@ def measure_avg( np_array = np.array(values) elif isinstance(values, np.ndarray): np_array = values - elif isinstance(values, pd.DataFrame): - np_array = values.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)) @@ -352,12 +352,12 @@ def measure_all( self, *, name: str, - values: list[float | int] | NDArray[np.float64] | pd.DataFrame, + 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. + """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. @@ -376,8 +376,8 @@ def measure_all( np_array = np.array(values) elif isinstance(values, np.ndarray): np_array = values - elif isinstance(values, pd.DataFrame): - np_array = values.values + elif isinstance(values, pd.Series): + np_array = values.to_numpy() else: raise ValueError(f"Invalid value type: {type(values)}")