Python(feat): add metadata, description, and associated channel support to measurements (#561)

Author: alexluck-sift

python/lib/sift_client/_internal/low_level_wrappers/calculated_channels.py

@@ -220,16 +220,16 @@ async def list_calculated_channel_versions(
         Raises:
             ValueError: If neither calculated_channel_id nor client_key is provided.
         """
-        request_kwargs = {}
+        request_kwargs: dict[str, Any] = {}
         if calculated_channel_id:
-            request_kwargs = {"calculated_channel_id": calculated_channel_id}
+            request_kwargs["calculated_channel_id"] = calculated_channel_id
         elif client_key:
-            request_kwargs = {"client_key": client_key}
+            request_kwargs["client_key"] = client_key
         else:
             raise ValueError("Either calculated_channel_id or client_key must be provided")
 
         if page_size is not None:
-            request_kwargs["page_size"] = str(page_size)
+            request_kwargs["page_size"] = page_size
         if page_token is not None:
             request_kwargs["page_token"] = page_token
         if query_filter is not None:
@@ -239,7 +239,7 @@ async def list_calculated_channel_versions(
         if organization_id is not None:
             request_kwargs["organization_id"] = organization_id
 
-        request = ListCalculatedChannelVersionsRequest(**request_kwargs)  # type: ignore # mypy thinks we should pass an int
+        request = ListCalculatedChannelVersionsRequest(**request_kwargs)
         response = await self._grpc_client.get_stub(
             CalculatedChannelServiceStub
         ).ListCalculatedChannelVersions(request)

python/lib/sift_client/_internal/low_level_wrappers/test_results.py

@@ -320,6 +320,14 @@ def simulate_update_test_measurement_response(
                 updates["string_expected_value"] = proto.string_bounds.expected_value
             else:
                 updates["string_expected_value"] = None
+        if "description" in update_mask_paths:
+            updates["description"] = proto.description if proto.description else None
+        if "metadata" in update_mask_paths:
+            from sift_client.util.metadata import metadata_proto_to_dict
+
+            updates["metadata"] = metadata_proto_to_dict(proto.metadata) if proto.metadata else None  # type: ignore[arg-type]
+        if "channel_names" in update_mask_paths:
+            updates["channel_names"] = list(proto.channel_names) if proto.channel_names else None
 
         return existing.model_copy(update=updates)
 
@@ -1259,6 +1267,9 @@ def _measurement_create_from_simulated(
             unit=simulated.unit,
             numeric_bounds=simulated.numeric_bounds,
             string_expected_value=simulated.string_expected_value,
+            description=simulated.description,
+            metadata=simulated.metadata,
+            channel_names=simulated.channel_names,
         )
 
 

python/lib/sift_client/_tests/resources/test_test_results.py

@@ -64,6 +64,9 @@ def compare_test_measurement_fields(simulated: TestMeasurement, actual: TestMeas
     assert simulated.boolean_value == actual.boolean_value
     assert simulated.passed == actual.passed
     assert simulated.timestamp == actual.timestamp
+    assert simulated.description == actual.description
+    assert simulated.metadata == actual.metadata
+    assert simulated.channel_names == actual.channel_names
 
 
 def test_client_binding(sift_client):
@@ -281,6 +284,9 @@ def test_create_test_measurements(self, sift_client, tmp_path):
             unit="Celsius",
             passed=True,
             timestamp=step1.start_time,
+            description="Expected nominal: 25.0C",
+            metadata={"sensor": "thermocouple_a", "channel_index": 1},
+            channel_names=["temperature_celsius"],
         )
 
         # Create simulated measurement first
@@ -359,6 +365,9 @@ def test_update_test_measurements(self, sift_client, tmp_path):
             "passed": False,
             "string_expected_value": "1.10.4",
             "unit": "C",
+            "description": "Updated note after recalibration",
+            "metadata": {"part_number": "PN-002"},
+            "channel_names": ["firmware_version_channel"],
         }
 
         # Test update with log_file first
@@ -384,6 +393,9 @@ def test_update_test_measurements(self, sift_client, tmp_path):
         assert measurement2.passed == False
         assert measurement2.string_expected_value == "1.10.4"
         assert measurement2.unit == "C"
+        assert measurement2.description == "Updated note after recalibration"
+        assert measurement2.metadata == {"part_number": "PN-002"}
+        assert measurement2.channel_names == ["firmware_version_channel"]
         # Update the measurement using class function.
         measurement4 = measurement4.update(
             {

python/lib/sift_client/_tests/sift_types/test_results.py

@@ -3,15 +3,21 @@
 from __future__ import annotations
 
 import tempfile
+import warnings
 from datetime import datetime, timedelta, timezone
 from unittest.mock import MagicMock, call
 
 import pytest
+from sift.test_reports.v1.test_reports_pb2 import (
+    TestMeasurement as TestMeasurementProto,
+)
 
+from sift_client.sift_types.channel import Channel, ChannelDataType
 from sift_client.sift_types.test_report import (
     ErrorInfo,
     NumericBounds,
     TestMeasurement,
+    TestMeasurementCreate,
     TestMeasurementType,
     TestReport,
     TestStatus,
@@ -84,6 +90,9 @@ def mock_test_measurement(mock_client):
         unit="Celsius",
         passed=True,
         timestamp=simulated_time,
+        description="Expected nominal: 25.0C",
+        metadata={"part_number": "PN-001", "serial_number": "SN-42"},
+        channel_names=["temperature_celsius"],
     )
     test_measurement._apply_client_to_instance(mock_client)
     return test_measurement
@@ -298,3 +307,135 @@ def test_upload_attachment(self, mock_test_report, mock_test_step, mock_client):
                 ),
             ]
         )
+
+    def test_measurement_description_truncates_with_warning(self):
+        """Description over the server limit is truncated and a UserWarning is raised."""
+        over_limit = "x" * 2001
+        with pytest.warns(UserWarning, match="exceeds 2000 characters"):
+            create = TestMeasurementCreate(
+                test_step_id="step_123",
+                name="m",
+                passed=True,
+                timestamp=datetime.now(timezone.utc),
+                numeric_value=1.0,
+                description=over_limit,
+            )
+        assert create.description is not None
+        assert len(create.description) == 2000
+
+    def test_measurement_description_at_limit_is_not_truncated(self):
+        """A description exactly at the limit should not warn or truncate."""
+        at_limit = "x" * 2000
+        with warnings.catch_warnings():
+            warnings.simplefilter("error")  # any warning fails the test
+            create = TestMeasurementCreate(
+                test_step_id="step_123",
+                name="m",
+                passed=True,
+                timestamp=datetime.now(timezone.utc),
+                numeric_value=1.0,
+                description=at_limit,
+            )
+        assert create.description == at_limit
+
+    def test_measurement_channel_names_accepts_strings(self):
+        """channel_names accepts a homogeneous list of channel name strings."""
+        create = TestMeasurementCreate(
+            test_step_id="step_123",
+            name="m",
+            passed=True,
+            timestamp=datetime.now(timezone.utc),
+            numeric_value=1.0,
+            channel_names=["temperature_celsius", "pressure_psi"],
+        )
+        assert create.channel_names == ["temperature_celsius", "pressure_psi"]
+        proto = create.to_proto()
+        assert list(proto.channel_names) == ["temperature_celsius", "pressure_psi"]
+
+    def test_measurement_channel_names_accepts_channels(self):
+        """channel_names accepts a homogeneous list of Channel instances; names are extracted at serialization."""
+        now = datetime.now(timezone.utc)
+
+        def _channel(name: str) -> Channel:
+            return Channel(
+                proto=MagicMock(),
+                id_=f"channel_{name}",
+                name=name,
+                data_type=ChannelDataType.DOUBLE,
+                description="",
+                unit="",
+                bit_field_elements=[],
+                enum_types={},
+                asset_id="asset_1",
+                created_date=now,
+                modified_date=now,
+                created_by_user_id="user1",
+                modified_by_user_id="user1",
+            )
+
+        create = TestMeasurementCreate(
+            test_step_id="step_123",
+            name="m",
+            passed=True,
+            timestamp=now,
+            numeric_value=1.0,
+            channel_names=[_channel("temperature_celsius"), _channel("pressure_psi")],
+        )
+        proto = create.to_proto()
+        assert list(proto.channel_names) == ["temperature_celsius", "pressure_psi"]
+
+    def test_measurement_create_to_proto_writes_new_fields(self):
+        """to_proto carries description, metadata, and channel_names onto the proto."""
+        create = TestMeasurementCreate(
+            test_step_id="step_123",
+            name="m",
+            passed=True,
+            timestamp=datetime.now(timezone.utc),
+            numeric_value=1.0,
+            description="note",
+            metadata={"pn": "PN-001", "count": 3, "flag": True},
+            channel_names=["chan_a", "chan_b"],
+        )
+        proto = create.to_proto()
+        assert proto.description == "note"
+        assert list(proto.channel_names) == ["chan_a", "chan_b"]
+        proto_keys = {m.key.name for m in proto.metadata}
+        assert proto_keys == {"pn", "count", "flag"}
+
+    def test_measurement_from_proto_round_trips_new_fields(self):
+        """A proto with the new fields populated round-trips into TestMeasurement."""
+        ts = datetime.now(timezone.utc)
+        source = TestMeasurementCreate(
+            test_step_id="step_123",
+            name="m",
+            passed=True,
+            timestamp=ts,
+            numeric_value=1.0,
+            description="note",
+            metadata={"pn": "PN-001", "count": 3},
+            channel_names=["chan_a"],
+        ).to_proto()
+        source.measurement_id = "measurement_456"
+        source.test_report_id = "report_789"
+
+        measurement = TestMeasurement._from_proto(source)
+
+        assert measurement.description == "note"
+        assert measurement.metadata == {"pn": "PN-001", "count": 3}
+        assert measurement.channel_names == ["chan_a"]
+
+    def test_measurement_from_proto_handles_absent_new_fields(self):
+        """Proto with unset description/metadata/channel_names yields None on the model."""
+        proto = TestMeasurementProto(
+            measurement_id="measurement_abc",
+            measurement_type=TestMeasurementType.DOUBLE.value,
+            name="m",
+            test_step_id="step_123",
+            test_report_id="report_789",
+            passed=True,
+        )
+        proto.timestamp.FromDatetime(datetime.now(timezone.utc))
+        measurement = TestMeasurement._from_proto(proto)
+        assert measurement.description is None
+        assert measurement.metadata is None
+        assert measurement.channel_names is None

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

@@ -137,10 +137,25 @@ def test_new_step(self, report_context):
         assert test_step.status == TestStatus.PASSED
 
     def test_measurement_update(self, report_context):
+        expected_description = "Round-trip check for the new fields on TestMeasurement."
+        expected_metadata = {
+            "ctx_test_serial": "SN-CTX-001",
+            "ctx_test_value": 42.5,
+            "ctx_test_pass": True,
+        }
+        expected_channel_names = ["temperature_celsius"]
+
         test_step = None
         with report_context.new_step("Test Measure", "Test Measure Description") as new_step:
             test_step = new_step.current_step
-            new_step.measure(name="Test Measurement", value=10, bounds={"min": 0, "max": 10})
+            new_step.measure(
+                name="Test Measurement",
+                value=10,
+                bounds={"min": 0, "max": 10},
+                description=expected_description,
+                metadata=expected_metadata,
+                channel_names=expected_channel_names,
+            )
             new_step.measure(name="Test Measurement 2", value="string value", bounds="string value")
             new_step.measure(name="Test Measurement 3", value=True, bounds="true")
 
@@ -149,6 +164,9 @@ def test_measurement_update(self, report_context):
         assert measurements[0].name == "Test Measurement"
         assert measurements[0].numeric_value == 10
         assert measurements[0].measurement_type == TestMeasurementType.DOUBLE
+        assert measurements[0].description == expected_description
+        assert measurements[0].metadata == expected_metadata
+        assert measurements[0].channel_names == expected_channel_names
         assert measurements[1].name == "Test Measurement 2"
         assert measurements[1].string_value == "string value"
         assert measurements[1].measurement_type == TestMeasurementType.STRING