From ca216ab48fb6822980e9b7c52558a8a9ab509220 Mon Sep 17 00:00:00 2001 From: Sushama Shroff Date: Thu, 8 Oct 2026 12:24:06 -0700 Subject: [PATCH 1/4] updates to support per resource type testdata --- test/conformance/.env.example | 8 + test/conformance/README.md | 25 +- test/conformance/source/conftest.py | 38 ++ test/conformance/source/request_config.py | 93 +++++ test/conformance/source/test_source.py | 333 ++++++++-------- test/conformance/switch.json | 129 +++++++ test/conformance/switch/conftest.py | 38 ++ test/conformance/switch/request_config.py | 81 ++++ test/conformance/switch/test_switch.py | 354 +++++++++--------- test/conformance/testdata/source.json | 44 +++ test/conformance/testdata/switch.json | 166 ++++++++ test/conformance/testdata/timetagger.json | 41 ++ test/conformance/timetagger/conftest.py | 38 ++ test/conformance/timetagger/request_config.py | 83 ++++ .../conformance/timetagger/test_timetagger.py | 353 +++++++++-------- 15 files changed, 1277 insertions(+), 547 deletions(-) create mode 100644 test/conformance/source/conftest.py create mode 100644 test/conformance/source/request_config.py create mode 100644 test/conformance/switch.json create mode 100644 test/conformance/switch/conftest.py create mode 100644 test/conformance/switch/request_config.py create mode 100644 test/conformance/testdata/source.json create mode 100644 test/conformance/testdata/switch.json create mode 100644 test/conformance/testdata/timetagger.json create mode 100644 test/conformance/timetagger/conftest.py create mode 100644 test/conformance/timetagger/request_config.py diff --git a/test/conformance/.env.example b/test/conformance/.env.example index d355ccf..140e78b 100644 --- a/test/conformance/.env.example +++ b/test/conformance/.env.example @@ -5,3 +5,11 @@ SOURCE_ADAPTER_ADDRESS=localhost:50051 SWITCH_ADAPTER_ADDRESS=localhost:50052 TIMETAGGER_ADAPTER_ADDRESS=localhost:50053 + +# Optional gRPC `reqType` header metadata. Omit if the adapter does not require it. +# GRPC_HEADER_METADATA_REQTYPE=mock + +# Supplies gRPC request bodies for each adapter RPC. Unset uses testdata/.json. +# SWITCH_CONFORMANCE_CONFIG=/path/to/your-switch.json +# SOURCE_CONFORMANCE_CONFIG=/path/to/your-source.json +# TIMETAGGER_CONFORMANCE_CONFIG=/path/to/your-timetagger.json diff --git a/test/conformance/README.md b/test/conformance/README.md index 2e7ef54..70154e7 100644 --- a/test/conformance/README.md +++ b/test/conformance/README.md @@ -47,12 +47,18 @@ TIMETAGGER_ADAPTER_ADDRESS=adapter.cloud.com:9000 ## Setup +Setup virtual env and run the below commands in it. +``` +python -m venv .venv +source .venv/bin/activate +``` + ### 1. Generate Python stubs Install grpcio-tools (if not already installed): ```bash -pip install `grpcio-tools>=1.84.0` +pip install 'grpcio-tools>=1.84.0' ``` Generate stubs from repository root: @@ -76,16 +82,19 @@ python -m grpc_tools.protoc \ ```bash cd test/conformance cp .env.example .env -# Edit .env to set adapter addresses ``` -Adapter addresses should be set in `.env` : +Edit these files: + +| File | What to set | +|---|---| +| `.env` | Adapter host/port and optional `GRPC_HEADER_METADATA_REQTYPE` | +| `testdata/switch.json` | Per-switch request fields. Copy and point the env `SWITCH_CONFORMANCE_CONFIG` at your file; default is this path. | +| `testdata/source.json` | Per-source request fields. Copy and point the env `SOURCE_CONFORMANCE_CONFIG` at your file; default is this path. | +| `testdata/timetagger.json` | Per-time-tagger request fields. Copy and point the env `TIMETAGGER_CONFORMANCE_CONFIG` at your file; default is this path. | + +Each JSON file supplies the gRPC request body for every adapter RPC. Top-level `product_id` and `resource_type` are copied onto every RPC except `HealthCheck`. Edit values so they match what the server the test is being run against accepts. -```bash -SOURCE_ADAPTER_ADDRESS=localhost:50051 -SWITCH_ADAPTER_ADDRESS=localhost:50052 -TIMETAGGER_ADAPTER_ADDRESS=localhost:50053 -``` ### 3. Install dependencies diff --git a/test/conformance/source/conftest.py b/test/conformance/source/conftest.py new file mode 100644 index 0000000..cfd5d38 --- /dev/null +++ b/test/conformance/source/conftest.py @@ -0,0 +1,38 @@ +# Copyright 2026 Cisco Systems, Inc. and its affiliates +# +# SPDX-License-Identifier: Apache-2.0 + +"""Parametrize source tests over whatever kinds the JSON config lists.""" +import pytest + +from .request_config import ( + CONFIG_ENV, + config_path, + kind_ids, + load_source_kinds, +) + +_SOURCE_KINDS = load_source_kinds() + + +if _SOURCE_KINDS: + + @pytest.fixture( + scope="class", + params=_SOURCE_KINDS, + ids=kind_ids(_SOURCE_KINDS), + ) + def source_kind(request): + """One JSON source object; class tests run all RPCs for this kind.""" + return request.param + +else: + + @pytest.fixture + def source_kind(): + """Skip device-scoped tests when the JSON lists no source kinds.""" + pytest.skip( + "No source kinds in {0} (set {1})".format( + config_path(), CONFIG_ENV + ) + ) diff --git a/test/conformance/source/request_config.py b/test/conformance/source/request_config.py new file mode 100644 index 0000000..caa5500 --- /dev/null +++ b/test/conformance/source/request_config.py @@ -0,0 +1,93 @@ +# Copyright 2026 Cisco Systems, Inc. and its affiliates +# +# SPDX-License-Identifier: Apache-2.0 + +"""Load source request testdata. Address and metadata stay in .env.""" +import json +import os + +from hal.adapters.source.v1 import source_pb2 + +CONFIG_ENV = "SOURCE_CONFORMANCE_CONFIG" +_DEFAULT_CONFIG = os.path.join( + os.path.dirname(__file__), "..", "testdata", "source.json" +) + + +def config_path(): + """Resolve the source JSON path from env or the default testdata file.""" + return os.path.abspath(os.getenv(CONFIG_ENV, _DEFAULT_CONFIG)) + + +def load_source_kinds(): + """Load the source array from JSON; missing file yields [].""" + path = config_path() + if not os.path.isfile(path): + return [] + with open(path, encoding="utf-8") as handle: + data = json.load(handle) + kinds = data.get("source", []) + if kinds is None: + return [] + if not isinstance(kinds, list): + raise ValueError( + "{0}: 'source' must be a JSON array".format(path) + ) + for index, kind in enumerate(kinds): + if not isinstance(kind, dict): + raise ValueError( + "{0}: source[{1}] must be an object".format(path, index) + ) + return kinds + + +def kind_ids(kinds): + """Pytest ids from resource_type, with suffixes if names repeat.""" + ids = [] + seen = {} + for kind in kinds: + base = str(kind.get("resource_type") or "source") + count = seen.get(base, 0) + seen[base] = count + 1 + ids.append(base if count == 0 else "{0}-{1}".format(base, count)) + return ids + + +def _normalize_fields(fields): + """json.dumps *_json values that were written as objects in testdata.""" + normalized = dict(fields) + for key, value in list(normalized.items()): + if key.endswith("_json") and isinstance(value, (dict, list)): + normalized[key] = json.dumps(value) + elif key == "config" and isinstance(value, dict): + cfg = dict(value) + config_json = cfg.get("config_json") + if isinstance(config_json, (dict, list)): + cfg["config_json"] = json.dumps(config_json) + normalized["config"] = cfg + return normalized + + +def rpc_request(kind, rpc_name): + """Build one RPC's request kwargs; copy product_id/resource_type from the kind.""" + rpcs = kind.get("rpcs") or {} + rpc_fields = rpcs.get(rpc_name) or {} + if not isinstance(rpc_fields, dict): + raise TypeError( + "rpcs.{0} must be an object, got {1}".format( + rpc_name, type(rpc_fields).__name__ + ) + ) + if rpc_name == "HealthCheck": + return _normalize_fields(rpc_fields) + # Kind-level product_id and resource_type are used by every RPC. + fields = { + "product_id": kind.get("product_id", ""), + "resource_type": kind.get("resource_type", ""), + } + fields.update(rpc_fields) + fields = _normalize_fields(fields) + config = fields.get("config") + if isinstance(config, dict): + fields["config"] = source_pb2.SourceConfig(**config) + return fields diff --git a/test/conformance/source/test_source.py b/test/conformance/source/test_source.py index e681437..b14a0c3 100644 --- a/test/conformance/source/test_source.py +++ b/test/conformance/source/test_source.py @@ -9,18 +9,17 @@ from hal.adapters.common.v1 import common_pb2 from hal.adapters.source.v1 import source_pb2, source_pb2_grpc -from utils import validate_response_field, validate_response_fields, check_optional_rpc +from utils import ( + validate_response_field, + validate_response_fields, + check_optional_rpc, +) -# Import proto constants using relative import from ..proto_constants import ( VALID_HEALTH_CHECK_STATUSES, - VALID_RESOURCE_HEALTH_STATUSES + VALID_RESOURCE_HEALTH_STATUSES, ) - - -TEST_PRODUCT_ID = "test-source-001" -TEST_RESOURCE_TYPE = "test-source-type" -TEST_RUN_ID = "test-run-123" +from .request_config import rpc_request def get_adapter_address(): @@ -29,7 +28,8 @@ def get_adapter_address(): if not address: raise ValueError( "SOURCE_ADAPTER_ADDRESS environment variable not set. " - "Copy test/.env.example to test/.env and configure adapter addresses." + "Copy test/.env.example to test/.env and configure " + "adapter addresses." ) return address @@ -75,185 +75,170 @@ def intercept_unary_unary( return source_pb2_grpc.AdapterSourceServiceStub(grpc_channel) -# Common RPC Tests (from hal/adapters/common/v1/common.proto) - def test_health_check(client): """Test HealthCheck RPC""" request = common_pb2.HealthCheckRequest() response = client.HealthCheck(request) assert response is not None - assert response.message in VALID_HEALTH_CHECK_STATUSES, \ - f"HealthCheck message must be one of {VALID_HEALTH_CHECK_STATUSES}, got '{response.message}'" - - -def test_resource_health_check(client): - """Test ResourceHealthCheck RPC""" - request = common_pb2.ResourceHealthCheckRequest( - product_id=TEST_PRODUCT_ID, - resource_type=TEST_RESOURCE_TYPE + assert response.message in VALID_HEALTH_CHECK_STATUSES, ( + "HealthCheck message must be one of " + "{0}, got '{1}'".format( + VALID_HEALTH_CHECK_STATUSES, response.message + ) ) - response = client.ResourceHealthCheck(request) - assert response is not None - assert response.message in VALID_RESOURCE_HEALTH_STATUSES, \ - f"ResourceHealthCheck message must be one of {VALID_RESOURCE_HEALTH_STATUSES}, got '{response.message}'" -def test_auto_recover(client): - """Test AutoRecover RPC (optional)""" - request = common_pb2.AutoRecoverRequest( - product_id=TEST_PRODUCT_ID, - resource_type=TEST_RESOURCE_TYPE - ) - implemented, response = check_optional_rpc(lambda: client.AutoRecover(request)) - if implemented: - assert response is not None - # Validate message field (must be non-empty when implemented) - validate_response_field(response, "message", str, allow_empty=False) +class TestSource: + """Device-scoped RPCs; class-scoped source_kind runs one kind at a time.""" - -def test_tune(client): - """Test Tune RPC (optional)""" - request = common_pb2.TuneRequest( - product_id=TEST_PRODUCT_ID, - resource_type=TEST_RESOURCE_TYPE, - config_json='{"test": "config"}' - ) - implemented, response = check_optional_rpc(lambda: client.Tune(request)) - if implemented: + def test_resource_health_check(self, client, source_kind): + """Test ResourceHealthCheck RPC""" + request = common_pb2.ResourceHealthCheckRequest( + **rpc_request(source_kind, "ResourceHealthCheck") + ) + response = client.ResourceHealthCheck(request) assert response is not None - # Validate message field (must be non-empty when implemented) - validate_response_field(response, "message", str, allow_empty=False) - -def test_get_product_info(client): - """Test GetProductInfo RPC""" - request = source_pb2.GetProductInfoRequest( - product_id=TEST_PRODUCT_ID, - resource_type=TEST_RESOURCE_TYPE - ) - response = client.GetProductInfo(request) - assert response is not None - - # Validate all fields (allow empty as they may not be set for all devices) - validate_response_fields(response, { - "serial_number": (str, True), - "product_code": (str, True), - "software_version": (str, True), - }) - - -def test_initialize(client): - """Test Initialize RPC""" - request = source_pb2.InitializeRequest( - product_id=TEST_PRODUCT_ID, - resource_type=TEST_RESOURCE_TYPE, - config=source_pb2.SourceConfig(config_json='{"test": "config"}') - ) - response = client.Initialize(request) - assert response is not None - - # Validate message field (must be non-empty) - validate_response_field(response, "message", str, allow_empty=False) - - -def test_deinitialize(client): - """Test Deinitialize RPC""" - request = source_pb2.DeinitializeRequest( - product_id=TEST_PRODUCT_ID, - resource_type=TEST_RESOURCE_TYPE - ) - response = client.Deinitialize(request) - assert response is not None - - # Validate message field (must be non-empty) - validate_response_field(response, "message", str, allow_empty=False) - - -def test_get_status(client): - """Test GetStatus RPC""" - request = source_pb2.GetStatusRequest( - product_id=TEST_PRODUCT_ID, - resource_type=TEST_RESOURCE_TYPE - ) - response = client.GetStatus(request) - assert response is not None - - # Validate all fields - validate_response_fields(response, { - "run_id": (str, True), # May be empty if not in a run - "uptime_seconds": (float, True), # Numeric fields can be 0 - "status_json": (str, True), # May be empty - }) - - # Validate state is a valid SourceState enum value (from proto definition) - valid_states = [ - source_pb2.SOURCE_STATE_UNSPECIFIED, - source_pb2.SOURCE_STATE_IDLE, - source_pb2.SOURCE_STATE_INITIALIZING, - source_pb2.SOURCE_STATE_LOCKING, - source_pb2.SOURCE_STATE_READY, - source_pb2.SOURCE_STATE_EMITTING, - source_pb2.SOURCE_STATE_FAULT, - source_pb2.SOURCE_STATE_DEINITIALIZING, - source_pb2.SOURCE_STATE_DEGRADED, - ] - assert response.state in valid_states, \ - f"GetStatus state must be a valid SourceState enum value, got {response.state}" - - -def test_set_config(client): - """Test SetConfig RPC""" - request = source_pb2.SetConfigRequest( - product_id=TEST_PRODUCT_ID, - resource_type=TEST_RESOURCE_TYPE, - config=source_pb2.SourceConfig(config_json='{"test": "config"}') - ) - response = client.SetConfig(request) - assert response is not None - - # Validate message field (must be non-empty) - validate_response_field(response, "message", str, allow_empty=False) + assert response.message in VALID_RESOURCE_HEALTH_STATUSES, ( + "ResourceHealthCheck message must be one of " + "{0}, got '{1}'".format( + VALID_RESOURCE_HEALTH_STATUSES, response.message + ) + ) + def test_auto_recover(self, client, source_kind): + """Test AutoRecover RPC (optional)""" + request = common_pb2.AutoRecoverRequest( + **rpc_request(source_kind, "AutoRecover") + ) + implemented, response = check_optional_rpc( + lambda: client.AutoRecover(request) + ) + if implemented: + assert response is not None + validate_response_field( + response, "message", str, allow_empty=False + ) + + def test_tune(self, client, source_kind): + """Test Tune RPC (optional)""" + request = common_pb2.TuneRequest( + **rpc_request(source_kind, "Tune") + ) + implemented, response = check_optional_rpc( + lambda: client.Tune(request) + ) + if implemented: + assert response is not None + validate_response_field( + response, "message", str, allow_empty=False + ) + + def test_get_product_info(self, client, source_kind): + """Test GetProductInfo RPC""" + request = source_pb2.GetProductInfoRequest( + **rpc_request(source_kind, "GetProductInfo") + ) + response = client.GetProductInfo(request) + assert response is not None + validate_response_fields(response, { + "serial_number": (str, True), + "product_code": (str, True), + "software_version": (str, True), + }) + + def test_initialize(self, client, source_kind): + """Test Initialize RPC""" + request = source_pb2.InitializeRequest( + **rpc_request(source_kind, "Initialize") + ) + response = client.Initialize(request) + assert response is not None + validate_response_field( + response, "message", str, allow_empty=False + ) -def test_start_emission(client): - """Test StartEmission RPC""" - request = source_pb2.StartEmissionRequest( - product_id=TEST_PRODUCT_ID, - resource_type=TEST_RESOURCE_TYPE, - run_id=TEST_RUN_ID, - config=source_pb2.SourceConfig(config_json='{"test": "config"}') - ) - response = client.StartEmission(request) - assert response is not None - - # Validate message field (must be non-empty) - validate_response_field(response, "message", str, allow_empty=False) + def test_deinitialize(self, client, source_kind): + """Test Deinitialize RPC""" + request = source_pb2.DeinitializeRequest( + **rpc_request(source_kind, "Deinitialize") + ) + response = client.Deinitialize(request) + assert response is not None + validate_response_field( + response, "message", str, allow_empty=False + ) + def test_get_status(self, client, source_kind): + """Test GetStatus RPC""" + request = source_pb2.GetStatusRequest( + **rpc_request(source_kind, "GetStatus") + ) + response = client.GetStatus(request) + assert response is not None + validate_response_fields(response, { + "run_id": (str, True), + "uptime_seconds": (float, True), + "status_json": (str, True), + }) + valid_states = [ + source_pb2.SOURCE_STATE_UNSPECIFIED, + source_pb2.SOURCE_STATE_IDLE, + source_pb2.SOURCE_STATE_INITIALIZING, + source_pb2.SOURCE_STATE_LOCKING, + source_pb2.SOURCE_STATE_READY, + source_pb2.SOURCE_STATE_EMITTING, + source_pb2.SOURCE_STATE_FAULT, + source_pb2.SOURCE_STATE_DEINITIALIZING, + source_pb2.SOURCE_STATE_DEGRADED, + ] + assert response.state in valid_states, ( + "GetStatus state must be a valid SourceState enum value, " + "got {0}".format(response.state) + ) -def test_stop_emission(client): - """Test StopEmission RPC""" - request = source_pb2.StopEmissionRequest( - product_id=TEST_PRODUCT_ID, - resource_type=TEST_RESOURCE_TYPE, - run_id=TEST_RUN_ID - ) - response = client.StopEmission(request) - assert response is not None - - # Validate message field (must be non-empty) - validate_response_field(response, "message", str, allow_empty=False) + def test_set_config(self, client, source_kind): + """Test SetConfig RPC""" + request = source_pb2.SetConfigRequest( + **rpc_request(source_kind, "SetConfig") + ) + response = client.SetConfig(request) + assert response is not None + validate_response_field( + response, "message", str, allow_empty=False + ) + def test_start_emission(self, client, source_kind): + """Test StartEmission RPC""" + request = source_pb2.StartEmissionRequest( + **rpc_request(source_kind, "StartEmission") + ) + response = client.StartEmission(request) + assert response is not None + validate_response_field( + response, "message", str, allow_empty=False + ) -def test_trigger_event(client): - """Test TriggerEvent RPC (optional)""" - request = source_pb2.TriggerEventRequest( - product_id=TEST_PRODUCT_ID, - resource_type=TEST_RESOURCE_TYPE, - run_id=TEST_RUN_ID, - event_type="test-event", - payload_json='{"test": "payload"}' - ) - implemented, response = check_optional_rpc(lambda: client.TriggerEvent(request)) - if implemented: + def test_stop_emission(self, client, source_kind): + """Test StopEmission RPC""" + request = source_pb2.StopEmissionRequest( + **rpc_request(source_kind, "StopEmission") + ) + response = client.StopEmission(request) assert response is not None - # Validate message field (must be non-empty when implemented) - validate_response_field(response, "message", str, allow_empty=False) + validate_response_field( + response, "message", str, allow_empty=False + ) + + def test_trigger_event(self, client, source_kind): + """Test TriggerEvent RPC (optional)""" + request = source_pb2.TriggerEventRequest( + **rpc_request(source_kind, "TriggerEvent") + ) + implemented, response = check_optional_rpc( + lambda: client.TriggerEvent(request) + ) + if implemented: + assert response is not None + validate_response_field( + response, "message", str, allow_empty=False + ) diff --git a/test/conformance/switch.json b/test/conformance/switch.json new file mode 100644 index 0000000..af5deb7 --- /dev/null +++ b/test/conformance/switch.json @@ -0,0 +1,129 @@ +{ + "_comment": "resource_type and product_id on each switch entry are copied onto every RPC except HealthCheck.", + "switch": [ + { + "resource_type": "POLATIS", + "product_id": "12345", + "rpcs": { + "HealthCheck": {}, + "ResourceHealthCheck": {}, + "GetProductInfo": {}, + "Initialize": { + "config_json": {} + }, + "Deinitialize": {}, + "GetStatus": {}, + "SetConfig": { + "config_json": {"key": "value"} + }, + "AutoRecover": {}, + "Tune": { + "config_json": {"setpoint": 1} + }, + "CommitConnection": { + "input_port": 1, + "output_port": 48, + "routing_case": 1 + }, + "ClearConnections": { + "ports": [1, 2] + }, + "ExecuteCommand": { + "command_json": {"command": "oxcshowconn"} + }, + "TriggerEvent": { + "run_id": "run-1", + "event_type": "fault", + "payload_json": {} + } + } + }, + { + "resource_type": "JW", + "product_id": "JW2302000", + "rpcs": { + "HealthCheck": {}, + "ResourceHealthCheck": {}, + "GetProductInfo": {}, + "Initialize": { + "config_json": {} + }, + "Deinitialize": {}, + "GetStatus": {}, + "SetConfig": { + "config_json": {"key": "value"} + }, + "AutoRecover": {}, + "Tune": { + "config_json": {"setpoint": 1} + }, + "CommitConnection": { + "input_port": 1, + "output_port": 2, + "routing_case": 1 + }, + "ClearConnections": { + "ports": [1, 2] + }, + "ExecuteCommand": { + "command_json": { + "connection_type": "serial", + "serial_port": "/dev/ttyUSB0", + "baudrate": 9600, + "rtscts": true, + "payload": [1, 18, 0, 1], + "timeout": 3 + } + }, + "TriggerEvent": { + "run_id": "run-1", + "event_type": "fault", + "payload_json": {} + } + } + }, + { + "resource_type": "OSW", + "product_id": "OSW_DEVICE", + "rpcs": { + "HealthCheck": {}, + "ResourceHealthCheck": {}, + "GetProductInfo": {}, + "Initialize": { + "config_json": {} + }, + "Deinitialize": {}, + "GetStatus": {}, + "SetConfig": { + "config_json": {"key": "value"} + }, + "AutoRecover": {}, + "Tune": { + "config_json": {"setpoint": 1} + }, + "CommitConnection": { + "input_port": 1, + "output_port": 2, + "routing_case": 1 + }, + "ClearConnections": { + "ports": [1, 2] + }, + "ExecuteCommand": { + "command_json": { + "connection_type": "tcp", + "host": "192.168.1.100", + "port": 23, + "payload": "S 1\n", + "timeout": 5 + } + }, + "TriggerEvent": { + "run_id": "run-1", + "event_type": "fault", + "payload_json": {} + } + } + } + ] +} diff --git a/test/conformance/switch/conftest.py b/test/conformance/switch/conftest.py new file mode 100644 index 0000000..f86c568 --- /dev/null +++ b/test/conformance/switch/conftest.py @@ -0,0 +1,38 @@ +# Copyright 2026 Cisco Systems, Inc. and its affiliates +# +# SPDX-License-Identifier: Apache-2.0 + +"""Parametrize switch tests over whatever kinds the JSON config lists.""" +import pytest + +from .request_config import ( + CONFIG_ENV, + config_path, + kind_ids, + load_switch_kinds, +) + +_SWITCH_KINDS = load_switch_kinds() + + +if _SWITCH_KINDS: + + @pytest.fixture( + scope="class", + params=_SWITCH_KINDS, + ids=kind_ids(_SWITCH_KINDS), + ) + def switch_kind(request): + """One JSON switch object; class tests run all RPCs for this kind.""" + return request.param + +else: + + @pytest.fixture + def switch_kind(): + """Skip device-scoped tests when the JSON lists no switch kinds.""" + pytest.skip( + "No switch kinds in {0} (set {1})".format( + config_path(), CONFIG_ENV + ) + ) diff --git a/test/conformance/switch/request_config.py b/test/conformance/switch/request_config.py new file mode 100644 index 0000000..8876052 --- /dev/null +++ b/test/conformance/switch/request_config.py @@ -0,0 +1,81 @@ +# Copyright 2026 Cisco Systems, Inc. and its affiliates +# +# SPDX-License-Identifier: Apache-2.0 + +"""Load switch request testdata. Address and metadata stay in .env.""" +import json +import os + +CONFIG_ENV = "SWITCH_CONFORMANCE_CONFIG" +_DEFAULT_CONFIG = os.path.join( + os.path.dirname(__file__), "..", "testdata", "switch.json" +) + + +def config_path(): + """Resolve the switch JSON path from env or the default testdata file.""" + return os.path.abspath(os.getenv(CONFIG_ENV, _DEFAULT_CONFIG)) + + +def load_switch_kinds(): + """Load the switch array from JSON; missing file yields [].""" + path = config_path() + if not os.path.isfile(path): + return [] + with open(path, encoding="utf-8") as handle: + data = json.load(handle) + kinds = data.get("switch", []) + if kinds is None: + return [] + if not isinstance(kinds, list): + raise ValueError( + "{0}: 'switch' must be a JSON array".format(path) + ) + for index, kind in enumerate(kinds): + if not isinstance(kind, dict): + raise ValueError( + "{0}: switch[{1}] must be an object".format(path, index) + ) + return kinds + + +def kind_ids(kinds): + """Pytest ids from resource_type, with suffixes if names repeat.""" + ids = [] + seen = {} + for kind in kinds: + base = str(kind.get("resource_type") or "switch") + count = seen.get(base, 0) + seen[base] = count + 1 + ids.append(base if count == 0 else "{0}-{1}".format(base, count)) + return ids + + +def _normalize_fields(fields): + """json.dumps *_json values that were written as objects in testdata.""" + normalized = dict(fields) + for key, value in list(normalized.items()): + if key.endswith("_json") and isinstance(value, (dict, list)): + normalized[key] = json.dumps(value) + return normalized + + +def rpc_request(kind, rpc_name): + """Build one RPC's request kwargs; copy product_id/resource_type from the kind.""" + rpcs = kind.get("rpcs") or {} + rpc_fields = rpcs.get(rpc_name) or {} + if not isinstance(rpc_fields, dict): + raise TypeError( + "rpcs.{0} must be an object, got {1}".format( + rpc_name, type(rpc_fields).__name__ + ) + ) + if rpc_name == "HealthCheck": + return _normalize_fields(rpc_fields) + # Kind-level product_id and resource_type are used by every RPC. + fields = { + "product_id": kind.get("product_id", ""), + "resource_type": kind.get("resource_type", ""), + } + fields.update(rpc_fields) + return _normalize_fields(fields) diff --git a/test/conformance/switch/test_switch.py b/test/conformance/switch/test_switch.py index c381c0f..4735f90 100644 --- a/test/conformance/switch/test_switch.py +++ b/test/conformance/switch/test_switch.py @@ -9,18 +9,17 @@ from hal.adapters.common.v1 import common_pb2 from hal.adapters.switch.v1 import switch_pb2, switch_pb2_grpc -from utils import validate_response_field, validate_response_fields, check_optional_rpc +from utils import ( + validate_response_field, + validate_response_fields, + check_optional_rpc, +) -# Import proto constants using relative import from ..proto_constants import ( VALID_HEALTH_CHECK_STATUSES, - VALID_RESOURCE_HEALTH_STATUSES + VALID_RESOURCE_HEALTH_STATUSES, ) - - -TEST_PRODUCT_ID = "test-switch-001" -TEST_RESOURCE_TYPE = "QSWITCH" -TEST_RUN_ID = "test-run-123" +from .request_config import rpc_request def get_adapter_address(): @@ -29,7 +28,8 @@ def get_adapter_address(): if not address: raise ValueError( "SWITCH_ADAPTER_ADDRESS environment variable not set. " - "Copy test/.env.example to test/.env and configure adapter addresses." + "Copy test/.env.example to test/.env and configure " + "adapter addresses." ) return address @@ -75,198 +75,184 @@ def intercept_unary_unary( return switch_pb2_grpc.AdapterSwitchServiceStub(grpc_channel) -# Common RPC Tests (from hal/adapters/common/v1/common.proto) - def test_health_check(client): """Test HealthCheck RPC""" request = common_pb2.HealthCheckRequest() response = client.HealthCheck(request) assert response is not None - assert response.message in VALID_HEALTH_CHECK_STATUSES, \ - f"HealthCheck message must be one of {VALID_HEALTH_CHECK_STATUSES}, got '{response.message}'" - - -def test_resource_health_check(client): - """Test ResourceHealthCheck RPC""" - request = common_pb2.ResourceHealthCheckRequest( - product_id=TEST_PRODUCT_ID, - resource_type=TEST_RESOURCE_TYPE + assert response.message in VALID_HEALTH_CHECK_STATUSES, ( + "HealthCheck message must be one of " + "{0}, got '{1}'".format( + VALID_HEALTH_CHECK_STATUSES, response.message + ) ) - response = client.ResourceHealthCheck(request) - assert response is not None - assert response.message in VALID_RESOURCE_HEALTH_STATUSES, \ - f"ResourceHealthCheck message must be one of {VALID_RESOURCE_HEALTH_STATUSES}, got '{response.message}'" -def test_auto_recover(client): - """Test AutoRecover RPC (optional)""" - request = common_pb2.AutoRecoverRequest( - product_id=TEST_PRODUCT_ID, - resource_type=TEST_RESOURCE_TYPE - ) - implemented, response = check_optional_rpc(lambda: client.AutoRecover(request)) - if implemented: - assert response is not None - # Validate message field (must be non-empty when implemented) - validate_response_field(response, "message", str, allow_empty=False) +class TestSwitch: + """Device-scoped RPCs; class-scoped switch_kind runs one kind at a time.""" - -def test_tune(client): - """Test Tune RPC (optional)""" - request = common_pb2.TuneRequest( - product_id=TEST_PRODUCT_ID, - resource_type=TEST_RESOURCE_TYPE, - config_json='{"test": "config"}' - ) - implemented, response = check_optional_rpc(lambda: client.Tune(request)) - if implemented: + def test_resource_health_check(self, client, switch_kind): + """Test ResourceHealthCheck RPC""" + request = common_pb2.ResourceHealthCheckRequest( + **rpc_request(switch_kind, "ResourceHealthCheck") + ) + response = client.ResourceHealthCheck(request) assert response is not None - # Validate message field (must be non-empty when implemented) - validate_response_field(response, "message", str, allow_empty=False) - - -def test_get_product_info(client): - """Test GetProductInfo RPC""" - request = switch_pb2.GetProductInfoRequest( - product_id=TEST_PRODUCT_ID, - resource_type=TEST_RESOURCE_TYPE - ) - response = client.GetProductInfo(request) - assert response is not None - - # Validate all fields (allow empty as they may not be set for all devices) - validate_response_fields(response, { - "serial_number": (str, True), - "product_code": (str, True), - "software_version": (str, True), - }) - - -def test_initialize(client): - """Test Initialize RPC""" - request = switch_pb2.InitializeRequest( - product_id=TEST_PRODUCT_ID, - resource_type=TEST_RESOURCE_TYPE, - config_json='{"test": "config"}' - ) - response = client.Initialize(request) - assert response is not None - - # Validate message field (must be non-empty) - validate_response_field(response, "message", str, allow_empty=False) + assert response.message in VALID_RESOURCE_HEALTH_STATUSES, ( + "ResourceHealthCheck message must be one of " + "{0}, got '{1}'".format( + VALID_RESOURCE_HEALTH_STATUSES, response.message + ) + ) + def test_auto_recover(self, client, switch_kind): + """Test AutoRecover RPC (optional)""" + request = common_pb2.AutoRecoverRequest( + **rpc_request(switch_kind, "AutoRecover") + ) + implemented, response = check_optional_rpc( + lambda: client.AutoRecover(request) + ) + if implemented: + assert response is not None + validate_response_field( + response, "message", str, allow_empty=False + ) + + def test_tune(self, client, switch_kind): + """Test Tune RPC (optional)""" + request = common_pb2.TuneRequest( + **rpc_request(switch_kind, "Tune") + ) + implemented, response = check_optional_rpc( + lambda: client.Tune(request) + ) + if implemented: + assert response is not None + validate_response_field( + response, "message", str, allow_empty=False + ) + + def test_get_product_info(self, client, switch_kind): + """Test GetProductInfo RPC""" + request = switch_pb2.GetProductInfoRequest( + **rpc_request(switch_kind, "GetProductInfo") + ) + response = client.GetProductInfo(request) + assert response is not None -def test_deinitialize(client): - """Test Deinitialize RPC""" - request = switch_pb2.DeinitializeRequest( - product_id=TEST_PRODUCT_ID, - resource_type=TEST_RESOURCE_TYPE - ) - response = client.Deinitialize(request) - assert response is not None - - # Validate message field (must be non-empty) - validate_response_field(response, "message", str, allow_empty=False) + validate_response_fields(response, { + "serial_number": (str, True), + "product_code": (str, True), + "software_version": (str, True), + }) + def test_initialize(self, client, switch_kind): + """Test Initialize RPC""" + request = switch_pb2.InitializeRequest( + **rpc_request(switch_kind, "Initialize") + ) + response = client.Initialize(request) + assert response is not None + validate_response_field( + response, "message", str, allow_empty=False + ) -def test_get_status(client): - """Test GetStatus RPC""" - request = switch_pb2.GetStatusRequest( - product_id=TEST_PRODUCT_ID, - resource_type=TEST_RESOURCE_TYPE - ) - response = client.GetStatus(request) - assert response is not None - - # Validate all fields (switch doesn't have run_id, only uptime and status_json) - validate_response_fields(response, { - "uptime_seconds": (float, True), # Numeric fields can be 0 - "status_json": (str, True), # May be empty - }) - - # Validate state is a valid SwitchState enum value (from proto definition) - valid_states = [ - switch_pb2.SWITCH_STATE_UNSPECIFIED, - switch_pb2.SWITCH_STATE_IDLE, - switch_pb2.SWITCH_STATE_INITIALIZING, - switch_pb2.SWITCH_STATE_READY, - switch_pb2.SWITCH_STATE_FAULT, - switch_pb2.SWITCH_STATE_DEINITIALIZING, - switch_pb2.SWITCH_STATE_DEGRADED, - ] - assert response.state in valid_states, \ - f"GetStatus state must be a valid SwitchState enum value, got {response.state}" - - -def test_set_config(client): - """Test SetConfig RPC""" - request = switch_pb2.SetConfigRequest( - product_id=TEST_PRODUCT_ID, - resource_type=TEST_RESOURCE_TYPE, - config_json='{"test": "config"}' - ) - response = client.SetConfig(request) - assert response is not None - - # Validate message field (must be non-empty) - validate_response_field(response, "message", str, allow_empty=False) - - -def test_commit_connection(client): - """Test CommitConnection RPC""" - request = switch_pb2.CommitConnectionRequest( - product_id=TEST_PRODUCT_ID, - resource_type=TEST_RESOURCE_TYPE, - input_port=1, - output_port=2, - routing_case=1 - ) - response = client.CommitConnection(request) - assert response is not None - - # Validate message field (must be non-empty) - validate_response_field(response, "message", str, allow_empty=False) + def test_deinitialize(self, client, switch_kind): + """Test Deinitialize RPC""" + request = switch_pb2.DeinitializeRequest( + **rpc_request(switch_kind, "Deinitialize") + ) + response = client.Deinitialize(request) + assert response is not None + validate_response_field( + response, "message", str, allow_empty=False + ) + def test_get_status(self, client, switch_kind): + """Test GetStatus RPC""" + request = switch_pb2.GetStatusRequest( + **rpc_request(switch_kind, "GetStatus") + ) + response = client.GetStatus(request) + assert response is not None -def test_clear_connections(client): - """Test ClearConnections RPC""" - request = switch_pb2.ClearConnectionsRequest( - product_id=TEST_PRODUCT_ID, - resource_type=TEST_RESOURCE_TYPE, - ports=[1, 2] - ) - response = client.ClearConnections(request) - assert response is not None - - # Validate message field (must be non-empty) - validate_response_field(response, "message", str, allow_empty=False) + validate_response_fields(response, { + "uptime_seconds": (float, True), + "status_json": (str, True), + }) + + valid_states = [ + switch_pb2.SWITCH_STATE_UNSPECIFIED, + switch_pb2.SWITCH_STATE_IDLE, + switch_pb2.SWITCH_STATE_INITIALIZING, + switch_pb2.SWITCH_STATE_READY, + switch_pb2.SWITCH_STATE_FAULT, + switch_pb2.SWITCH_STATE_DEINITIALIZING, + switch_pb2.SWITCH_STATE_DEGRADED, + ] + assert response.state in valid_states, ( + "GetStatus state must be a valid SwitchState enum value, " + "got {0}".format(response.state) + ) + def test_set_config(self, client, switch_kind): + """Test SetConfig RPC""" + request = switch_pb2.SetConfigRequest( + **rpc_request(switch_kind, "SetConfig") + ) + response = client.SetConfig(request) + assert response is not None + validate_response_field( + response, "message", str, allow_empty=False + ) -def test_execute_command(client): - """Test ExecuteCommand RPC (optional)""" - request = switch_pb2.ExecuteCommandRequest( - product_id=TEST_PRODUCT_ID, - resource_type=TEST_RESOURCE_TYPE, - command_json='{"command": "test"}' - ) - implemented, response = check_optional_rpc(lambda: client.ExecuteCommand(request)) - if implemented: + def test_commit_connection(self, client, switch_kind): + """Test CommitConnection RPC""" + request = switch_pb2.CommitConnectionRequest( + **rpc_request(switch_kind, "CommitConnection") + ) + response = client.CommitConnection(request) assert response is not None - # Validate response_json field (must be non-empty when implemented) - validate_response_field(response, "response_json", str, allow_empty=False) - - -def test_trigger_event(client): - """Test TriggerEvent RPC (optional)""" - request = switch_pb2.TriggerEventRequest( - product_id=TEST_PRODUCT_ID, - resource_type=TEST_RESOURCE_TYPE, - run_id=TEST_RUN_ID, - event_type="test-event", - payload_json='{"test": "payload"}' - ) - implemented, response = check_optional_rpc(lambda: client.TriggerEvent(request)) - if implemented: + validate_response_field( + response, "message", str, allow_empty=False + ) + + def test_clear_connections(self, client, switch_kind): + """Test ClearConnections RPC""" + request = switch_pb2.ClearConnectionsRequest( + **rpc_request(switch_kind, "ClearConnections") + ) + response = client.ClearConnections(request) assert response is not None - # Validate message field (must be non-empty when implemented) - validate_response_field(response, "message", str, allow_empty=False) + validate_response_field( + response, "message", str, allow_empty=False + ) + + def test_execute_command(self, client, switch_kind): + """Test ExecuteCommand RPC (optional)""" + request = switch_pb2.ExecuteCommandRequest( + **rpc_request(switch_kind, "ExecuteCommand") + ) + implemented, response = check_optional_rpc( + lambda: client.ExecuteCommand(request) + ) + if implemented: + assert response is not None + validate_response_field( + response, "response_json", str, allow_empty=False + ) + + def test_trigger_event(self, client, switch_kind): + """Test TriggerEvent RPC (optional)""" + request = switch_pb2.TriggerEventRequest( + **rpc_request(switch_kind, "TriggerEvent") + ) + implemented, response = check_optional_rpc( + lambda: client.TriggerEvent(request) + ) + if implemented: + assert response is not None + validate_response_field( + response, "message", str, allow_empty=False + ) diff --git a/test/conformance/testdata/source.json b/test/conformance/testdata/source.json new file mode 100644 index 0000000..14d9393 --- /dev/null +++ b/test/conformance/testdata/source.json @@ -0,0 +1,44 @@ +{ + "_comment": "request bodies for source tests. The `source` array is the matrix: one object per kind the adapter should exercise. Top-level `product_id` and `resource_type` are copied onto every RPC.", + "source": [ + { + "resource_type": "CISCO", + "product_id": "cisco-source-1", + "rpcs": { + "HealthCheck": {}, + "ResourceHealthCheck": {}, + "GetProductInfo": {}, + "Initialize": { + "config": { + "config_json": {"pump_power_mw": 50.0} + } + }, + "Deinitialize": {}, + "GetStatus": {}, + "SetConfig": { + "config": { + "config_json": {"pump_power_mw": 50.0} + } + }, + "AutoRecover": {}, + "Tune": { + "config_json": {} + }, + "StartEmission": { + "run_id": "mock-run-1", + "config": { + "config_json": {} + } + }, + "StopEmission": { + "run_id": "mock-run-1" + }, + "TriggerEvent": { + "run_id": "mock-run-1", + "event_type": "fault", + "payload_json": {} + } + } + } + ] +} diff --git a/test/conformance/testdata/switch.json b/test/conformance/testdata/switch.json new file mode 100644 index 0000000..2896d6d --- /dev/null +++ b/test/conformance/testdata/switch.json @@ -0,0 +1,166 @@ +{ + "_comment": "request bodies for switch tests. The `switch` array is the matrix: one object per kind the adapter should exercise. Top-level `product_id` and `resource_type` are copied onto every RPC.", + "switch": [ + { + "resource_type": "POLATIS", + "product_id": "12345", + "rpcs": { + "HealthCheck": {}, + "ResourceHealthCheck": {}, + "GetProductInfo": {}, + "Initialize": { + "config_json": {} + }, + "Deinitialize": {}, + "GetStatus": {}, + "SetConfig": { + "config_json": {} + }, + "AutoRecover": {}, + "Tune": { + "config_json": {} + }, + "CommitConnection": { + "input_port": 1, + "output_port": 2, + "routing_case": 0 + }, + "ClearConnections": { + "ports": [1, 2] + }, + "ExecuteCommand": { + "command_json": {"command": "oxcshowconn"} + }, + "TriggerEvent": { + "run_id": "run-1", + "event_type": "fault", + "payload_json": {} + } + } + }, + { + "resource_type": "JW", + "product_id": "JW2302000", + "rpcs": { + "HealthCheck": {}, + "ResourceHealthCheck": {}, + "GetProductInfo": {}, + "Initialize": { + "config_json": {} + }, + "Deinitialize": {}, + "GetStatus": {}, + "SetConfig": { + "config_json": {} + }, + "AutoRecover": {}, + "Tune": { + "config_json": {} + }, + "CommitConnection": { + "input_port": 1, + "output_port": 2, + "routing_case": 1 + }, + "ClearConnections": { + "ports": [1, 2] + }, + "ExecuteCommand": { + "command_json": { + "connection_type": "serial", + "serial_port": "/dev/ttyUSB0", + "baudrate": 9600, + "rtscts": true, + "payload": [1, 18, 0, 1], + "timeout": 3 + } + }, + "TriggerEvent": { + "run_id": "run-1", + "event_type": "fault", + "payload_json": {} + } + } + }, + { + "resource_type": "OSW", + "product_id": "OSW_DEVICE", + "rpcs": { + "HealthCheck": {}, + "ResourceHealthCheck": {}, + "GetProductInfo": {}, + "Initialize": { + "config_json": {} + }, + "Deinitialize": {}, + "GetStatus": {}, + "SetConfig": { + "config_json": {} + }, + "AutoRecover": {}, + "Tune": { + "config_json": {} + }, + "CommitConnection": { + "input_port": 1, + "output_port": 2, + "routing_case": 1 + }, + "ClearConnections": { + "ports": [1, 2] + }, + "ExecuteCommand": { + "command_json": { + "connection_type": "tcp", + "host": "192.168.1.100", + "port": 23, + "payload": "S 1\n", + "timeout": 5 + } + }, + "TriggerEvent": { + "run_id": "run-1", + "event_type": "fault", + "payload_json": {} + } + } + }, + { + "resource_type": "QSWITCH", + "product_id": "QSWITCH_DEVICE", + "rpcs": { + "HealthCheck": {}, + "ResourceHealthCheck": {}, + "GetProductInfo": {}, + "Initialize": { + "config_json": {} + }, + "Deinitialize": {}, + "GetStatus": {}, + "SetConfig": { + "config_json": {} + }, + "AutoRecover": {}, + "Tune": { + "config_json": {} + }, + "CommitConnection": { + "input_port": 1, + "output_port": 2, + "routing_case": 1 + }, + "ClearConnections": { + "ports": [1, 2] + }, + "ExecuteCommand": { + "command_json": {} + }, + "TriggerEvent": { + "run_id": "run-1", + "event_type": "fault", + "payload_json": {} + } + } + } + ] +} diff --git a/test/conformance/testdata/timetagger.json b/test/conformance/testdata/timetagger.json new file mode 100644 index 0000000..3160926 --- /dev/null +++ b/test/conformance/testdata/timetagger.json @@ -0,0 +1,41 @@ +{ + "_comment": "request bodies for time tagger tests. The `timetagger` array is the matrix: one object per kind the adapter should exercise. Top-level `product_id` and `resource_type` are copied onto every RPC.", + "timetagger": [ + { + "resource_type": "SWABIAN", + "product_id": "swabian-timetagger-x", + "rpcs": { + "HealthCheck": {}, + "ResourceHealthCheck": {}, + "GetProductInfo": {}, + "Initialize": { + "config_json": {} + }, + "Deinitialize": {}, + "GetStatus": {}, + "SetConfig": { + "config_json": {} + }, + "AutoRecover": {}, + "Tune": { + "config_json": {} + }, + "StartDataCollection": { + "run_id": "mock-run-1", + "config_json": { + "measurement_types": ["coincidence_rate_static_default"], + "influx_bucket_name": "qentra-data-mock" + } + }, + "StopDataCollection": { + "run_id": "mock-run-1" + }, + "TriggerEvent": { + "run_id": "mock-run-1", + "event_type": "eavesdropper", + "payload_json": {} + } + } + } + ] +} diff --git a/test/conformance/timetagger/conftest.py b/test/conformance/timetagger/conftest.py new file mode 100644 index 0000000..ef9d387 --- /dev/null +++ b/test/conformance/timetagger/conftest.py @@ -0,0 +1,38 @@ +# Copyright 2026 Cisco Systems, Inc. and its affiliates +# +# SPDX-License-Identifier: Apache-2.0 + +"""Parametrize time tagger tests over whatever kinds the JSON config lists.""" +import pytest + +from .request_config import ( + CONFIG_ENV, + config_path, + kind_ids, + load_timetagger_kinds, +) + +_TIMETAGGER_KINDS = load_timetagger_kinds() + + +if _TIMETAGGER_KINDS: + + @pytest.fixture( + scope="class", + params=_TIMETAGGER_KINDS, + ids=kind_ids(_TIMETAGGER_KINDS), + ) + def timetagger_kind(request): + """One JSON time tagger object; class tests run all RPCs for this kind.""" + return request.param + +else: + + @pytest.fixture + def timetagger_kind(): + """Skip device-scoped tests when the JSON lists no time tagger kinds.""" + pytest.skip( + "No time tagger kinds in {0} (set {1})".format( + config_path(), CONFIG_ENV + ) + ) diff --git a/test/conformance/timetagger/request_config.py b/test/conformance/timetagger/request_config.py new file mode 100644 index 0000000..040e99e --- /dev/null +++ b/test/conformance/timetagger/request_config.py @@ -0,0 +1,83 @@ +# Copyright 2026 Cisco Systems, Inc. and its affiliates +# +# SPDX-License-Identifier: Apache-2.0 + +"""Load time tagger request testdata. Address and metadata stay in .env.""" +import json +import os + +CONFIG_ENV = "TIMETAGGER_CONFORMANCE_CONFIG" +_DEFAULT_CONFIG = os.path.join( + os.path.dirname(__file__), "..", "testdata", "timetagger.json" +) + + +def config_path(): + """Resolve the time tagger JSON path from env or the default testdata file.""" + return os.path.abspath(os.getenv(CONFIG_ENV, _DEFAULT_CONFIG)) + + +def load_timetagger_kinds(): + """Load the timetagger array from JSON; missing file yields [].""" + path = config_path() + if not os.path.isfile(path): + return [] + with open(path, encoding="utf-8") as handle: + data = json.load(handle) + kinds = data.get("timetagger", []) + if kinds is None: + return [] + if not isinstance(kinds, list): + raise ValueError( + "{0}: 'timetagger' must be a JSON array".format(path) + ) + for index, kind in enumerate(kinds): + if not isinstance(kind, dict): + raise ValueError( + "{0}: timetagger[{1}] must be an object".format( + path, index + ) + ) + return kinds + + +def kind_ids(kinds): + """Pytest ids from resource_type, with suffixes if names repeat.""" + ids = [] + seen = {} + for kind in kinds: + base = str(kind.get("resource_type") or "timetagger") + count = seen.get(base, 0) + seen[base] = count + 1 + ids.append(base if count == 0 else "{0}-{1}".format(base, count)) + return ids + + +def _normalize_fields(fields): + """json.dumps *_json values that were written as objects in testdata.""" + normalized = dict(fields) + for key, value in list(normalized.items()): + if key.endswith("_json") and isinstance(value, (dict, list)): + normalized[key] = json.dumps(value) + return normalized + + +def rpc_request(kind, rpc_name): + """Build one RPC's request kwargs; copy product_id/resource_type from the kind.""" + rpcs = kind.get("rpcs") or {} + rpc_fields = rpcs.get(rpc_name) or {} + if not isinstance(rpc_fields, dict): + raise TypeError( + "rpcs.{0} must be an object, got {1}".format( + rpc_name, type(rpc_fields).__name__ + ) + ) + if rpc_name == "HealthCheck": + return _normalize_fields(rpc_fields) + # Kind-level product_id and resource_type are used by every RPC. + fields = { + "product_id": kind.get("product_id", ""), + "resource_type": kind.get("resource_type", ""), + } + fields.update(rpc_fields) + return _normalize_fields(fields) diff --git a/test/conformance/timetagger/test_timetagger.py b/test/conformance/timetagger/test_timetagger.py index bf1adb1..0d65a28 100644 --- a/test/conformance/timetagger/test_timetagger.py +++ b/test/conformance/timetagger/test_timetagger.py @@ -9,18 +9,17 @@ from hal.adapters.common.v1 import common_pb2 from hal.adapters.timetagger.v1 import timetagger_pb2, timetagger_pb2_grpc -from utils import validate_response_field, validate_response_fields, check_optional_rpc +from utils import ( + validate_response_field, + validate_response_fields, + check_optional_rpc, +) -# Import proto constants using relative import from ..proto_constants import ( VALID_HEALTH_CHECK_STATUSES, - VALID_RESOURCE_HEALTH_STATUSES + VALID_RESOURCE_HEALTH_STATUSES, ) - - -TEST_PRODUCT_ID = "test-timetagger-001" -TEST_RESOURCE_TYPE = "test-timetagger-type" -TEST_RUN_ID = "test-run-123" +from .request_config import rpc_request def get_adapter_address(): @@ -29,7 +28,8 @@ def get_adapter_address(): if not address: raise ValueError( "TIMETAGGER_ADAPTER_ADDRESS environment variable not set. " - "Copy test/.env.example to test/.env and configure adapter addresses." + "Copy test/.env.example to test/.env and configure " + "adapter addresses." ) return address @@ -44,32 +44,34 @@ def grpc_channel(): @pytest.fixture(scope="module") def metadata(): - """Get gRPC metadata from environment if specified""" - # Get reqType header if specified + """Get gRPC metadata from environment if specified.""" reqtype = os.getenv("GRPC_HEADER_METADATA_REQTYPE", "") if not reqtype: return None - return [("reqtype", reqtype.strip())] @pytest.fixture(scope="module") def client(grpc_channel, metadata): - """Create gRPC client stub with optional metadata interceptor""" + """Create gRPC client stub with optional metadata interceptor.""" if metadata: - # Create interceptor that adds metadata to all calls class MetadataInterceptor(grpc.UnaryUnaryClientInterceptor): def __init__(self, metadata): self._metadata = metadata - - def intercept_unary_unary(self, continuation, client_call_details, request): - new_details = client_call_details._replace(metadata=self._metadata) + + def intercept_unary_unary( + self, continuation, client_call_details, request + ): + new_details = client_call_details._replace( + metadata=self._metadata + ) return continuation(new_details, request) - - interceptor = MetadataInterceptor(metadata) - channel = grpc.intercept_channel(grpc_channel, interceptor) + + channel = grpc.intercept_channel( + grpc_channel, MetadataInterceptor(metadata) + ) return timetagger_pb2_grpc.AdapterTimeTaggerServiceStub(channel) - + return timetagger_pb2_grpc.AdapterTimeTaggerServiceStub(grpc_channel) @@ -78,175 +80,164 @@ def test_health_check(client): request = common_pb2.HealthCheckRequest() response = client.HealthCheck(request) assert response is not None - assert response.message in VALID_HEALTH_CHECK_STATUSES, \ - f"HealthCheck message must be one of {VALID_HEALTH_CHECK_STATUSES}, got '{response.message}'" - - -def test_resource_health_check(client): - """Test ResourceHealthCheck RPC""" - request = common_pb2.ResourceHealthCheckRequest( - product_id=TEST_PRODUCT_ID, - resource_type=TEST_RESOURCE_TYPE + assert response.message in VALID_HEALTH_CHECK_STATUSES, ( + "HealthCheck message must be one of " + "{0}, got '{1}'".format( + VALID_HEALTH_CHECK_STATUSES, response.message + ) ) - response = client.ResourceHealthCheck(request) - assert response is not None - assert response.message in VALID_RESOURCE_HEALTH_STATUSES, \ - f"ResourceHealthCheck message must be one of {VALID_RESOURCE_HEALTH_STATUSES}, got '{response.message}'" -def test_auto_recover(client): - """Test AutoRecover RPC (optional)""" - request = common_pb2.AutoRecoverRequest( - product_id=TEST_PRODUCT_ID, - resource_type=TEST_RESOURCE_TYPE - ) - implemented, response = check_optional_rpc(lambda: client.AutoRecover(request)) - if implemented: +class TestTimeTagger: + """Device-scoped RPCs; class-scoped timetagger_kind runs one kind at a time.""" + + def test_resource_health_check(self, client, timetagger_kind): + """Test ResourceHealthCheck RPC""" + request = common_pb2.ResourceHealthCheckRequest( + **rpc_request(timetagger_kind, "ResourceHealthCheck") + ) + response = client.ResourceHealthCheck(request) assert response is not None - # Validate message field (must be non-empty when implemented) - validate_response_field(response, "message", str, allow_empty=False) + assert response.message in VALID_RESOURCE_HEALTH_STATUSES, ( + "ResourceHealthCheck message must be one of " + "{0}, got '{1}'".format( + VALID_RESOURCE_HEALTH_STATUSES, response.message + ) + ) + def test_auto_recover(self, client, timetagger_kind): + """Test AutoRecover RPC (optional)""" + request = common_pb2.AutoRecoverRequest( + **rpc_request(timetagger_kind, "AutoRecover") + ) + implemented, response = check_optional_rpc( + lambda: client.AutoRecover(request) + ) + if implemented: + assert response is not None + validate_response_field( + response, "message", str, allow_empty=False + ) + + def test_tune(self, client, timetagger_kind): + """Test Tune RPC (optional)""" + request = common_pb2.TuneRequest( + **rpc_request(timetagger_kind, "Tune") + ) + implemented, response = check_optional_rpc( + lambda: client.Tune(request) + ) + if implemented: + assert response is not None + validate_response_field( + response, "message", str, allow_empty=False + ) + + def test_get_product_info(self, client, timetagger_kind): + """Test GetProductInfo RPC""" + request = timetagger_pb2.GetProductInfoRequest( + **rpc_request(timetagger_kind, "GetProductInfo") + ) + response = client.GetProductInfo(request) + assert response is not None + validate_response_fields(response, { + "serial_number": (str, True), + "product_code": (str, True), + "software_version": (str, True), + }) + + def test_initialize(self, client, timetagger_kind): + """Test Initialize RPC""" + request = timetagger_pb2.InitializeRequest( + **rpc_request(timetagger_kind, "Initialize") + ) + response = client.Initialize(request) + assert response is not None + validate_response_field( + response, "message", str, allow_empty=False + ) -def test_tune(client): - """Test Tune RPC (optional)""" - request = common_pb2.TuneRequest( - product_id=TEST_PRODUCT_ID, - resource_type=TEST_RESOURCE_TYPE, - config_json='{"test": "config"}' - ) - implemented, response = check_optional_rpc(lambda: client.Tune(request)) - if implemented: + def test_deinitialize(self, client, timetagger_kind): + """Test Deinitialize RPC""" + request = timetagger_pb2.DeinitializeRequest( + **rpc_request(timetagger_kind, "Deinitialize") + ) + response = client.Deinitialize(request) assert response is not None - # Validate message field (must be non-empty when implemented) - validate_response_field(response, "message", str, allow_empty=False) - -def test_get_product_info(client): - """Test GetProductInfo RPC""" - request = timetagger_pb2.GetProductInfoRequest( - product_id=TEST_PRODUCT_ID, - resource_type=TEST_RESOURCE_TYPE - ) - response = client.GetProductInfo(request) - assert response is not None - - # Validate all fields (allow empty as they may not be set for all devices) - validate_response_fields(response, { - "serial_number": (str, True), - "product_code": (str, True), - "software_version": (str, True), - }) - - -def test_initialize(client): - """Test Initialize RPC""" - request = timetagger_pb2.InitializeRequest( - product_id=TEST_PRODUCT_ID, - resource_type=TEST_RESOURCE_TYPE, - config_json='{"test": "config"}' - ) - response = client.Initialize(request) - assert response is not None - - # Validate message field (must be non-empty) - validate_response_field(response, "message", str, allow_empty=False) + validate_response_field( + response, "message", str, allow_empty=False + ) + def test_get_status(self, client, timetagger_kind): + """Test GetStatus RPC""" + request = timetagger_pb2.GetStatusRequest( + **rpc_request(timetagger_kind, "GetStatus") + ) + response = client.GetStatus(request) + assert response is not None + validate_response_fields(response, { + "run_id": (str, True), + "uptime_seconds": (float, True), + "status_json": (str, True), + }) + valid_states = [ + timetagger_pb2.TIME_TAGGER_STATE_UNSPECIFIED, + timetagger_pb2.TIME_TAGGER_STATE_IDLE, + timetagger_pb2.TIME_TAGGER_STATE_INITIALIZING, + timetagger_pb2.TIME_TAGGER_STATE_READY, + timetagger_pb2.TIME_TAGGER_STATE_COLLECTING, + timetagger_pb2.TIME_TAGGER_STATE_FAULT, + timetagger_pb2.TIME_TAGGER_STATE_DEINITIALIZING, + timetagger_pb2.TIME_TAGGER_STATE_DEGRADED, + ] + assert response.state in valid_states, ( + "GetStatus state must be a valid TimeTaggerState enum value, " + "got {0}".format(response.state) + ) -def test_deinitialize(client): - """Test Deinitialize RPC""" - request = timetagger_pb2.DeinitializeRequest( - product_id=TEST_PRODUCT_ID, - resource_type=TEST_RESOURCE_TYPE - ) - response = client.Deinitialize(request) - assert response is not None - - # Validate message field (must be non-empty) - validate_response_field(response, "message", str, allow_empty=False) + def test_set_config(self, client, timetagger_kind): + """Test SetConfig RPC""" + request = timetagger_pb2.SetConfigRequest( + **rpc_request(timetagger_kind, "SetConfig") + ) + response = client.SetConfig(request) + assert response is not None + validate_response_field( + response, "message", str, allow_empty=False + ) + def test_start_data_collection(self, client, timetagger_kind): + """Test StartDataCollection RPC""" + request = timetagger_pb2.StartDataCollectionRequest( + **rpc_request(timetagger_kind, "StartDataCollection") + ) + response = client.StartDataCollection(request) + assert response is not None + validate_response_field( + response, "message", str, allow_empty=False + ) -def test_get_status(client): - """Test GetStatus RPC""" - request = timetagger_pb2.GetStatusRequest( - product_id=TEST_PRODUCT_ID, - resource_type=TEST_RESOURCE_TYPE - ) - response = client.GetStatus(request) - assert response is not None - - # Validate all fields - validate_response_fields(response, { - "run_id": (str, True), # May be empty if not in a run - "uptime_seconds": (float, True), # Numeric fields can be 0 - "status_json": (str, True), # May be empty - }) - - # Validate state is a valid TimeTaggerState enum value (from proto definition) - valid_states = [ - timetagger_pb2.TIME_TAGGER_STATE_UNSPECIFIED, - timetagger_pb2.TIME_TAGGER_STATE_IDLE, - timetagger_pb2.TIME_TAGGER_STATE_INITIALIZING, - timetagger_pb2.TIME_TAGGER_STATE_READY, - timetagger_pb2.TIME_TAGGER_STATE_COLLECTING, - timetagger_pb2.TIME_TAGGER_STATE_FAULT, - timetagger_pb2.TIME_TAGGER_STATE_DEINITIALIZING, - timetagger_pb2.TIME_TAGGER_STATE_DEGRADED, - ] - assert response.state in valid_states, \ - f"GetStatus state must be a valid TimeTaggerState enum value, got {response.state}" - - -def test_set_config(client): - """Test SetConfig RPC""" - request = timetagger_pb2.SetConfigRequest( - product_id=TEST_PRODUCT_ID, - resource_type=TEST_RESOURCE_TYPE, - config_json='{"test": "config"}' - ) - response = client.SetConfig(request) - assert response is not None - - # Validate message field (must be non-empty) - validate_response_field(response, "message", str, allow_empty=False) - - -def test_start_data_collection(client): - """Test StartDataCollection RPC""" - request = timetagger_pb2.StartDataCollectionRequest( - product_id=TEST_PRODUCT_ID, - resource_type=TEST_RESOURCE_TYPE, - run_id=TEST_RUN_ID, - config_json='{"test": "config"}' - ) - response = client.StartDataCollection(request) - assert response is not None - - # Validate message field (must be non-empty) - validate_response_field(response, "message", str, allow_empty=False) - -def test_stop_data_collection(client): - """Test StopDataCollection RPC""" - request = timetagger_pb2.StopDataCollectionRequest( - product_id=TEST_PRODUCT_ID, - resource_type=TEST_RESOURCE_TYPE, - run_id=TEST_RUN_ID - ) - response = client.StopDataCollection(request) - assert response is not None - - # Validate message field (must be non-empty) - validate_response_field(response, "message", str, allow_empty=False) - -def test_trigger_event(client): - """Test TriggerEvent RPC (optional)""" - request = timetagger_pb2.TriggerEventRequest( - product_id=TEST_PRODUCT_ID, - resource_type=TEST_RESOURCE_TYPE, - run_id=TEST_RUN_ID, - event_type="test-event", - payload_json='{"test": "payload"}' - ) - implemented, response = check_optional_rpc(lambda: client.TriggerEvent(request)) - if implemented: + def test_stop_data_collection(self, client, timetagger_kind): + """Test StopDataCollection RPC""" + request = timetagger_pb2.StopDataCollectionRequest( + **rpc_request(timetagger_kind, "StopDataCollection") + ) + response = client.StopDataCollection(request) assert response is not None - # Validate message field (must be non-empty when implemented) - validate_response_field(response, "message", str, allow_empty=False) + validate_response_field( + response, "message", str, allow_empty=False + ) + + def test_trigger_event(self, client, timetagger_kind): + """Test TriggerEvent RPC (optional)""" + request = timetagger_pb2.TriggerEventRequest( + **rpc_request(timetagger_kind, "TriggerEvent") + ) + implemented, response = check_optional_rpc( + lambda: client.TriggerEvent(request) + ) + if implemented: + assert response is not None + validate_response_field( + response, "message", str, allow_empty=False + ) From 00ea79505ddd4f0acfcdb183167bb95df4b3d916 Mon Sep 17 00:00:00 2001 From: Sushama Shroff Date: Fri, 9 Oct 2026 11:57:51 -0700 Subject: [PATCH 2/4] testdata validations for mandatory fields --- test/conformance/source/request_config.py | 8 ++++++++ test/conformance/switch/request_config.py | 8 ++++++++ test/conformance/testdata/source.json | 3 ++- test/conformance/testdata/switch.json | 3 ++- test/conformance/testdata/timetagger.json | 3 ++- test/conformance/timetagger/request_config.py | 8 ++++++++ 6 files changed, 30 insertions(+), 3 deletions(-) diff --git a/test/conformance/source/request_config.py b/test/conformance/source/request_config.py index caa5500..be647b4 100644 --- a/test/conformance/source/request_config.py +++ b/test/conformance/source/request_config.py @@ -38,6 +38,14 @@ def load_source_kinds(): raise ValueError( "{0}: source[{1}] must be an object".format(path, index) ) + for field in ("product_id", "resource_type"): + value = kind.get(field) + if not isinstance(value, str) or not value.strip(): + raise ValueError( + "{0}: source[{1}] requires non-empty {2}".format( + path, index, field + ) + ) return kinds diff --git a/test/conformance/switch/request_config.py b/test/conformance/switch/request_config.py index 8876052..9aa6410 100644 --- a/test/conformance/switch/request_config.py +++ b/test/conformance/switch/request_config.py @@ -36,6 +36,14 @@ def load_switch_kinds(): raise ValueError( "{0}: switch[{1}] must be an object".format(path, index) ) + for field in ("product_id", "resource_type"): + value = kind.get(field) + if not isinstance(value, str) or not value.strip(): + raise ValueError( + "{0}: switch[{1}] requires non-empty {2}".format( + path, index, field + ) + ) return kinds diff --git a/test/conformance/testdata/source.json b/test/conformance/testdata/source.json index 14d9393..d140e2d 100644 --- a/test/conformance/testdata/source.json +++ b/test/conformance/testdata/source.json @@ -1,5 +1,6 @@ { - "_comment": "request bodies for source tests. The `source` array is the matrix: one object per kind the adapter should exercise. Top-level `product_id` and `resource_type` are copied onto every RPC.", + "_comment": "Request bodies for source tests. The tests in `test_source.py` will run against every entry in the `source` array.", + "_comment_required_fields": "Each array entry requires non-empty `product_id` and `resource_type`. Top-level `product_id` and `resource_type` are copied onto every RPC except HealthCheck.", "source": [ { "resource_type": "CISCO", diff --git a/test/conformance/testdata/switch.json b/test/conformance/testdata/switch.json index 2896d6d..08bbccc 100644 --- a/test/conformance/testdata/switch.json +++ b/test/conformance/testdata/switch.json @@ -1,5 +1,6 @@ { - "_comment": "request bodies for switch tests. The `switch` array is the matrix: one object per kind the adapter should exercise. Top-level `product_id` and `resource_type` are copied onto every RPC.", + "_comment": "Request bodies for switch tests. The tests in `test_switch.py` will run against every entry in the `switch` array.", + "_comment_required_fields": "Each array entry requires non-empty `product_id` and `resource_type`. Top-level `product_id` and `resource_type` are copied onto every RPC except HealthCheck.", "switch": [ { "resource_type": "POLATIS", diff --git a/test/conformance/testdata/timetagger.json b/test/conformance/testdata/timetagger.json index 3160926..d8e5def 100644 --- a/test/conformance/testdata/timetagger.json +++ b/test/conformance/testdata/timetagger.json @@ -1,5 +1,6 @@ { - "_comment": "request bodies for time tagger tests. The `timetagger` array is the matrix: one object per kind the adapter should exercise. Top-level `product_id` and `resource_type` are copied onto every RPC.", + "_comment": "Request bodies for time tagger tests. The tests in `test_timetagger.py` will run against every entry in the `timetagger` array.", + "_comment_required_fields": "Each array entry requires non-empty `product_id` and `resource_type`. Top-level `product_id` and `resource_type` are copied onto every RPC except HealthCheck.", "timetagger": [ { "resource_type": "SWABIAN", diff --git a/test/conformance/timetagger/request_config.py b/test/conformance/timetagger/request_config.py index 040e99e..ccc3085 100644 --- a/test/conformance/timetagger/request_config.py +++ b/test/conformance/timetagger/request_config.py @@ -38,6 +38,14 @@ def load_timetagger_kinds(): path, index ) ) + for field in ("product_id", "resource_type"): + value = kind.get(field) + if not isinstance(value, str) or not value.strip(): + raise ValueError( + "{0}: timetagger[{1}] requires non-empty {2}".format( + path, index, field + ) + ) return kinds From e64229f3256c87ebc8fd700505ae9b21861f5427 Mon Sep 17 00:00:00 2001 From: Sushama Shroff Date: Fri, 9 Oct 2026 18:05:48 -0700 Subject: [PATCH 3/4] upates README.md for testing specific sdk versions with the tests --- test/conformance/README.md | 19 +++++++++++++++++++ 1 file changed, 19 insertions(+) diff --git a/test/conformance/README.md b/test/conformance/README.md index 70154e7..dcb82ee 100644 --- a/test/conformance/README.md +++ b/test/conformance/README.md @@ -121,3 +121,22 @@ python -m pytest timetagger/test_timetagger.py -v ```bash python -m pytest source/test_source.py::test_health_check -v ``` + +## Testing an adapter built for a specific HAL version + +If you are writing your own adapter against a released HAL version (for example `v1.2.0`), run the conformance tests **from that release's tag**. Each tag contains the `.proto` files, the stubs, the `testdata/*.json` and the tests for that version, so they all agree with the contract your adapter implements. Use the tests from the tag, not from `main`, because `main` may test newer RPCs or fields that your version does not have. + +```bash +git clone https://github.com/outshift-open/quantum-hal.git +cd quantum-hal +git checkout v1.2.0 # the HAL version your adapter implements +``` + +Then follow [Setup](#setup) from the checked-out tag: + +1. Generate the stubs from this checkout (step 1). Re-run it after switching tags and delete any old `gen/python/`, because `conftest.py` puts `gen/python/` ahead of any installed `quantum-hal-sdk`. +2. Copy `.env.example` to `.env` and set your adapter's address (step 2). +3. Edit the `testdata/*.json` files at that tag so they hold values your adapter accepts. +4. Install `requirements.txt` and run `python -m pytest -v`. + +To check a different version, check out its tag and repeat the steps. From 4c23cc41804715bb349eb626145b2dbc1e3013db Mon Sep 17 00:00:00 2001 From: Sushama Shroff Date: Mon, 12 Oct 2026 09:26:49 -0700 Subject: [PATCH 4/4] nit --- .gitignore | 3 +++ 1 file changed, 3 insertions(+) diff --git a/.gitignore b/.gitignore index a230a78..1565c7b 100644 --- a/.gitignore +++ b/.gitignore @@ -1,2 +1,5 @@ .venv/ __pycache__/ + +# Stubs generated locally for the conformance tests (see test/conformance/README.md). +gen/