diff --git a/nmostesting/ControllerTest.py b/nmostesting/ControllerTest.py index 99f34beff..d995b3f34 100644 --- a/nmostesting/ControllerTest.py +++ b/nmostesting/ControllerTest.py @@ -33,11 +33,17 @@ CONN_API_KEY = "connection" REG_API_KEY = "registration" +RTP_RESOURCE_DATA_DIR = "test_data/controller" +MXL_RESOURCE_DATA_DIR = "test_data/controller/mxl" +MXL_TRANSPORT = "urn:x-nmos:transport:mxl" +MXL_CONNECTION_API_VERSION = "v1.2" + class ControllerTest(GenericTest): """ Testing initial set up of new test suite for controller testing """ + def __init__(self, apis, registries, node, dns_server, auths, disable_auto=True, **kwargs): # Remove the Testing Facade spec_path as there are no corresponding GitHub repos for the Testing Facade API apis[TESTING_FACADE_API_KEY].pop("spec_path", None) @@ -52,7 +58,9 @@ def __init__(self, apis, registries, node, dns_server, auths, disable_auto=True, self.dns_server = dns_server self.mock_registry_base_url = '' self.mock_node_base_url = '' - self.test_data = self.load_resource_data() + self.rtp_test_data = self._load_resource_templates(RTP_RESOURCE_DATA_DIR) + self.rtp_test_data['node']['id'] = self.node.id + self.mxl_test_data = self._load_resource_templates(MXL_RESOURCE_DATA_DIR) self.senders = [] self.sender_ip_addresses = {} self.receivers = [] @@ -137,26 +145,72 @@ def _format_device_metadata(self, label, description, id): """ Used to format answers based on device metadata """ return label + ' (' + description + ', ' + id + ')' + def _sender_transport(self, sender): + return sender.get("transport", self.rtp_test_data["sender"]["transport"]) + + def _receiver_transport(self, receiver): + return receiver.get("transport", self.rtp_test_data["receiver"]["transport"]) + + def _uses_mxl_transport(self, transport): + return transport == MXL_TRANSPORT + + def _connection_api_version_for_transport(self, transport): + if self._uses_mxl_transport(transport): + return MXL_CONNECTION_API_VERSION + return self.connection_api_version + + def _device_controls(self, transport): + connection_api_version = self._connection_api_version_for_transport(transport) + return { + "href": self.mock_node_base_url + "x-nmos/connection/" + connection_api_version + "/", + "type": "urn:x-nmos:control:sr-ctrl/" + connection_api_version + } + + def _sender_uses_rtp_transport(self, sender): + return self._sender_transport(sender).startswith("urn:x-nmos:transport:rtp") + + def _sender_uses_mxl_transport(self, sender): + return self._uses_mxl_transport(self._sender_transport(sender)) + + def _receiver_uses_rtp_transport(self, receiver): + return self._receiver_transport(receiver).startswith("urn:x-nmos:transport:rtp") + + def _receiver_uses_mxl_transport(self, receiver): + return self._uses_mxl_transport(self._receiver_transport(receiver)) + + def _resource_templates_for_sender(self, sender): + if self._sender_uses_mxl_transport(sender): + return self.mxl_test_data + return self.rtp_test_data + + def _resource_templates_for_receiver(self, receiver): + if self._receiver_uses_mxl_transport(receiver): + return self.mxl_test_data + return self.rtp_test_data + def _populate_registry(self, test): """Populate registry and mock node with mock senders and receivers""" self.node.reset() # Ensure previously added senders and receivers are removed if self.primary_registry: self.primary_registry.common.reset() # Ensure any previously registered senders and receivers are removed sender_ip_final_octet = 159 + mxl_domain_id = str(uuid.uuid4()) # Register node self._register_node(test, self.node.id, "AMWA Test Suite Node", "AMWA Test Suite Node") # self.senders should be initialized in the set_up_tests() override of derived test # each mock sender defined as: {'label': , 'description': '', - # 'registered': } + # 'registered': , + # 'transport': } for sender in self.senders: sender["id"] = str(uuid.uuid4()) sender["device_id"] = str(uuid.uuid4()) sender["flow_id"] = str(uuid.uuid4()) sender["source_id"] = str(uuid.uuid4()) - sender["manifest_href"] = self.mock_node_base_url + "x-nmos/connection/" + self.connection_api_version \ - + "/single/senders/" + sender["id"] + "/transportfile" + if self._sender_uses_rtp_transport(sender): + sender["manifest_href"] = self.mock_node_base_url + "x-nmos/connection/" \ + + self.connection_api_version + "/single/senders/" + sender["id"] + "/transportfile" sender["version"] = NMOSUtils.get_TAI_time() sender["display_answer"] = self._format_device_metadata(sender['label'], sender['description'], sender['id']) @@ -165,22 +219,26 @@ def _populate_registry(self, test): time.sleep(0.1) if sender["registered"]: self._register_sender(test, sender) - # Add sender to mock node - sender_json = self._create_sender_json(sender) - sender_ip_address = self.senders_ip_base + str(sender_ip_final_octet) - self.node.add_sender(sender_json, sender_ip_address, sender.get("sdp_params", {})) - self.sender_ip_addresses[sender["id"]] = sender_ip_address - sender_ip_final_octet += 1 + # Add RTP senders to mock node (IS-05 Connection API) + if self._sender_uses_rtp_transport(sender): + sender_json = self._create_sender_json(sender) + sender_ip_address = self.senders_ip_base + str(sender_ip_final_octet) + self.node.add_sender(sender_json, sender_ip_address, sender.get("sdp_params", {})) + self.sender_ip_addresses[sender["id"]] = sender_ip_address + sender_ip_final_octet += 1 + elif self._sender_uses_mxl_transport(sender): + self.node.add_mxl_sender(self._create_sender_json(sender), mxl_domain_id) # self.receivers should be initialized in the set_up_tests() override of derived test # each mock receiver defined as: {'label': , 'description': '', - # 'connectable': , 'registered': } + # 'connectable': , 'registered': , + # 'transport': } for receiver in self.receivers: receiver["id"] = str(uuid.uuid4()) receiver["device_id"] = str(uuid.uuid4()) - receiver["controls_href"] = self.mock_node_base_url + "x-nmos/connection/" \ - + self.connection_api_version + "/" - receiver["controls_type"] = "urn:x-nmos:control:sr-ctrl/" + self.connection_api_version + receiver_controls = self._device_controls(self._receiver_transport(receiver)) + receiver["controls_href"] = receiver_controls["href"] + receiver["controls_type"] = receiver_controls["type"] receiver["version"] = NMOSUtils.get_TAI_time() receiver["display_answer"] = self._format_device_metadata( receiver['label'], receiver['description'], receiver['id']) @@ -192,20 +250,20 @@ def _populate_registry(self, test): # Add receiver to mock node # Note: mock node is currently only a mock Connection API # so only add 'connectable' receivers - if receiver["connectable"]: + if receiver["connectable"] and self._receiver_uses_rtp_transport(receiver): receiver_json = self._create_receiver_json(receiver) self.node.add_receiver(receiver_json) + elif receiver["connectable"] and self._receiver_uses_mxl_transport(receiver): + self.node.add_mxl_receiver(self._create_receiver_json(receiver), mxl_domain_id) - def load_resource_data(self): - """Loads test data from files""" - result_data = dict() + def _load_resource_templates(self, directory): + """Loads IS-04 resource templates from a controller test data directory.""" + templates = {} resources = ["node", "device", "source", "flow", "sender", "receiver"] for resource in resources: - with open("test_data/controller/v1.3_{}.json".format(resource)) as resource_data: - resource_json = json.load(resource_data) - result_data[resource] = resource_json - result_data['node']['id'] = self.node.id - return result_data + with open("{}/v1.3_{}.json".format(directory, resource)) as resource_data: + templates[resource] = json.load(resource_data) + return templates def post_resource(self, test, type, data=None, reg_url=None, codes=None, fail=Test.FAIL, headers=None): """ @@ -214,7 +272,7 @@ def post_resource(self, test, type, data=None, reg_url=None, codes=None, fail=Te Otherwise, on success, returns values of the Location header and X-Paging-Timestamp debugging header. """ if not data: - data = self.test_data[type] + data = self.rtp_test_data[type] if not reg_url: reg_url = self.mock_registry_base_url + 'x-nmos/registration/v1.3/' @@ -258,7 +316,7 @@ def _register_node(self, test, node_id, label, description): """ Perform POST requests on the Registration API to create node registration """ - node_data = deepcopy(self.test_data["node"]) + node_data = deepcopy(self.rtp_test_data["node"]) node_data["id"] = node_id node_data["label"] = label node_data["description"] = description @@ -266,7 +324,8 @@ def _register_node(self, test, node_id, label, description): self.post_resource(test, "node", node_data, codes=[201]) def _create_sender_json(self, sender): - sender_data = deepcopy(self.test_data["sender"]) + templates = self._resource_templates_for_sender(sender) + sender_data = deepcopy(templates["sender"]) if "sdp_params" in sender: # Mapping sdp_params names to sender names @@ -283,6 +342,7 @@ def _create_sender_json(self, sender): "device_id", "flow_id", "manifest_href", + "transport", "version"] for property in overridden_properties: @@ -298,23 +358,24 @@ def _register_sender(self, test, sender, codes=[201], fail=Test.FAIL): Use to create sender [code=201] or to update existing sender [code=200] """ # use the test data as a template for creating new resources + templates = self._resource_templates_for_sender(sender) # Register device - device_data = deepcopy(self.test_data["device"]) + device_data = deepcopy(templates["device"]) device_data["id"] = sender["device_id"] device_data["label"] = "AMWA Test Device" device_data["description"] = "AMWA Test Device" device_data["node_id"] = self.node.id - device_data["controls"][0]["href"] = self.mock_node_base_url + "x-nmos/connection/" \ - + self.connection_api_version + "/" - device_data["controls"][0]["type"] = "urn:x-nmos:control:sr-ctrl/" + self.connection_api_version + sender_controls = self._device_controls(self._sender_transport(sender)) + device_data["controls"][0]["href"] = sender_controls["href"] + device_data["controls"][0]["type"] = sender_controls["type"] device_data["senders"] = [sender["id"]] device_data["receivers"] = [] device_data["version"] = sender["version"] self.post_resource(test, "device", device_data, codes=codes, fail=fail) # Register source - source_data = deepcopy(self.test_data["source"]) + source_data = deepcopy(templates["source"]) source_data["id"] = sender["source_id"] source_data["label"] = "AMWA Test Source" source_data["description"] = "AMWA Test Source" @@ -323,7 +384,7 @@ def _register_sender(self, test, sender, codes=[201], fail=Test.FAIL): self.post_resource(test, "source", source_data, codes=codes, fail=fail) # Register flow - flow_data = deepcopy(self.test_data["flow"]) + flow_data = deepcopy(templates["flow"]) flow_data["id"] = sender["flow_id"] flow_data["label"] = "AMWA Test Flow" flow_data["description"] = "AMWA Test Flow" @@ -351,10 +412,10 @@ def delete_sender(self, test, sender): raise NMOSTestException(test.FAIL("Registration API returned an unexpected response: {}".format(r))) def _create_receiver_json(self, receiver): - # Register receiver - receiver_data = deepcopy(self.test_data["receiver"]) + templates = self._resource_templates_for_receiver(receiver) + receiver_data = deepcopy(templates["receiver"]) - overriden_properties = ["id", "label", "description", "device_id", "version", "caps"] + overriden_properties = ["id", "label", "description", "device_id", "transport", "version", "caps"] for property in overriden_properties: if property in receiver: @@ -369,9 +430,10 @@ def _register_receiver(self, test, receiver, codes=[201], fail=Test.FAIL): Use to create receiver [code=201] or to update existing receiver [code=200] """ # use the test data as a template for creating new resources + templates = self._resource_templates_for_receiver(receiver) # Register device - device_data = deepcopy(self.test_data["device"]) + device_data = deepcopy(templates["device"]) device_data["id"] = receiver["device_id"] device_data["label"] = "AMWA Test Device" device_data["description"] = "AMWA Test Device" @@ -448,7 +510,9 @@ def pre_tests_message(self): """) try: - self.testing_facade_utils.invoke_testing_facade(question, [], test_type="action") + metadata = {'test_class': self.__class__.__name__} + self.testing_facade_utils.invoke_testing_facade( + question, [], test_type="action", metadata=metadata) except TestingFacadeException: # pre_test_introducton timed out diff --git a/nmostesting/NMOSTesting.py b/nmostesting/NMOSTesting.py index 6f53136c4..53a2d9ba3 100644 --- a/nmostesting/NMOSTesting.py +++ b/nmostesting/NMOSTesting.py @@ -89,6 +89,7 @@ from .suites import BCP0050101Test from .suites import BCP0060101Test from .suites import BCP0060102Test +from .suites import BCP0070302Test from .suites import BCP00604Test from .suites import BCP0070301Test from .suites import BCP0080101Test @@ -488,6 +489,22 @@ }], "class": BCP0070301Test.BCP0070301Test }, + "BCP-007-03-02": { + "name": "BCP-007-03 Controller", + "specs": [{ + "spec_key": "testing-facade", + "api_key": "testquestion" + }, { + "spec_key": "is-04", + "api_key": "query", + "disable_fields": ["host", "port"] + }, { + "spec_key": "is-05", + "api_key": "connection", + "disable_fields": ["host", "port"] + }], + "class": BCP0070302Test.BCP0070302Test + }, "BCP-008-01-01": { "name": "BCP-008-01 Receiver Status Monitoring", "specs": [{ diff --git a/nmostesting/RQLUtils.py b/nmostesting/RQLUtils.py new file mode 100644 index 000000000..037277575 --- /dev/null +++ b/nmostesting/RQLUtils.py @@ -0,0 +1,440 @@ +# Copyright (C) 2026 Advanced Media Workflow Association +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Minimal Resource Query Language (RQL) support for the mock registry Query API.""" + +import re + +SUPPORTED_OPERATORS = frozenset({ + 'and', 'or', 'not', + 'eq', 'ne', 'gt', 'ge', 'lt', 'le', + 'in', 'out', + 'matches', 'rel', 'sub', +}) + +TYPED_VALUE_TYPES = frozenset({ + 'string', 'number', 'integer', 'boolean', + 'version', 'api_version', 'rational', 'sampling', +}) + +SUPPORTED_QUERY_PARAMS = frozenset({ + 'query.rql', 'query.downgrade', 'query.strip', 'query.match_type', +}) + + +class RQLParseError(Exception): + """Raised when an RQL query string cannot be parsed.""" + + +class UnsupportedRQLOperator(RQLParseError): + """Raised when an RQL operator is not supported by this implementation.""" + + +class RQLParser(object): + def __init__(self, query): + self.query = query + self.length = len(query) + self.position = 0 + + def parse(self): + self._skip_whitespace() + if self.position >= self.length: + raise RQLParseError("Empty RQL query") + expression = self._parse_value() + self._skip_whitespace() + if self.position < self.length: + raise RQLParseError("Unexpected trailing input in RQL query") + return expression + + def _peek(self): + if self.position >= self.length: + return '' + return self.query[self.position] + + def _advance(self, count=1): + self.position += count + + def _skip_whitespace(self): + while self.position < self.length and self.query[self.position].isspace(): + self.position += 1 + + def _expect(self, character): + self._skip_whitespace() + if self._peek() != character: + raise RQLParseError("Expected '{}' in RQL query".format(character)) + self._advance() + + def _parse_identifier(self): + self._skip_whitespace() + start = self.position + while self.position < self.length: + character = self.query[self.position] + if character.isalnum() or character in '._:-/': + self.position += 1 + else: + break + if start == self.position: + raise RQLParseError("Expected identifier in RQL query") + return self.query[start:self.position] + + def _parse_quoted_string(self): + quote = self._peek() + self._advance() + start = self.position + while self.position < self.length and self.query[self.position] != quote: + if self.query[self.position] == '\\' and self.position + 1 < self.length: + self.position += 2 + else: + self.position += 1 + if self.position >= self.length: + raise RQLParseError("Unterminated string in RQL query") + value = self.query[start:self.position] + self._advance() + return value + + def _parse_parenthesized_list(self, parse_item): + self._expect('(') + items = [] + self._skip_whitespace() + if self._peek() != ')': + items.append(parse_item()) + while True: + self._skip_whitespace() + if self._peek() != ',': + break + self._advance() + items.append(parse_item()) + self._expect(')') + return items + + def _parse_tuple(self): + return self._parse_parenthesized_list(self._parse_atom) + + def _parse_atom(self): + self._skip_whitespace() + character = self._peek() + if character in '"\'': + return self._parse_quoted_string() + if character == '(': + return self._parse_tuple() + return _literal_from_identifier(self._parse_identifier()) + + def _parse_call(self): + operator_name = self._parse_identifier() + if operator_name not in SUPPORTED_OPERATORS: + raise UnsupportedRQLOperator(operator_name) + arguments = self._parse_parenthesized_list(self._parse_value) + return {'name': operator_name, 'args': arguments} + + def _parse_value(self): + self._skip_whitespace() + character = self._peek() + if character in '"\'': + return self._parse_quoted_string() + if character == '(': + return self._parse_tuple() + + identifier_start = self.position + identifier = self._parse_identifier() + self._skip_whitespace() + if self._peek() == '(': + self.position = identifier_start + return self._parse_call() + return _literal_from_identifier(identifier) + + +def _typed_value_from_identifier(identifier): + colon_index = identifier.find(':') + if colon_index <= 0: + return None + type_name = identifier[:colon_index] + if type_name not in TYPED_VALUE_TYPES: + return None + return {'type': type_name, 'value': identifier[colon_index + 1:]} + + +def _literal_from_identifier(identifier): + if identifier == 'true': + return True + if identifier == 'false': + return False + if identifier == 'null': + return None + typed_value = _typed_value_from_identifier(identifier) + if typed_value is not None: + return typed_value + try: + if '.' in identifier: + return float(identifier) + return int(identifier) + except ValueError: + return identifier + + +def _is_typed_value(value): + return isinstance(value, dict) and 'type' in value and 'value' in value + + +def _unwrap_value(value): + if _is_typed_value(value): + return value['value'] + return value + + +def _resolve_related_resource(relation_name, relation_id, all_resources): + if all_resources is None or relation_id is None: + return None + + relation_property_name = relation_name.rsplit('.', 1)[-1] + if not relation_property_name.endswith('_id'): + return None + + return all_resources.get(relation_property_name[:-3], {}).get(str(relation_id)) + + +def parse_query(query_string): + return RQLParser(query_string).parse() + + +def extract_property_values(resource, property_path): + """Return all values reachable via a dot-separated property path.""" + if resource is None: + return [] + + current_values = [resource] + for part in property_path.split('.'): + next_values = [] + for value in current_values: + if isinstance(value, dict): + if part in value: + next_values.append(value[part]) + elif isinstance(value, list): + for element in value: + if isinstance(element, dict) and part in element: + next_values.append(element[part]) + elif not isinstance(element, dict) and part == '': + next_values.append(element) + current_values = next_values + + flattened_values = [] + for value in current_values: + if isinstance(value, list): + flattened_values.extend(value) + else: + flattened_values.append(value) + return flattened_values + + +def _grain_rate_from_rational_string(rational_string): + if '/' in rational_string: + numerator_string, denominator_string = rational_string.split('/', 1) + return { + 'numerator': int(numerator_string), + 'denominator': int(denominator_string), + } + return {'numerator': int(rational_string), 'denominator': 1} + + +def _normalize_grain_rate(value): + if value is None: + return None + if _is_typed_value(value) and value['type'] == 'rational': + return _grain_rate_from_rational_string(value['value']) + if isinstance(value, dict) and 'numerator' in value and 'denominator' in value: + return value + if isinstance(value, str): + try: + return _grain_rate_from_rational_string(value) + except ValueError: + return None + return None + + +def _values_equal(left, right): + if left is None or right is None: + return left is right + if isinstance(left, bool) or isinstance(right, bool): + return left is right + + left_grain_rate = _normalize_grain_rate(left) + right_grain_rate = _normalize_grain_rate(right) + if left_grain_rate is not None and right_grain_rate is not None: + if (left_grain_rate['numerator'] * right_grain_rate['denominator'] + == right_grain_rate['numerator'] * left_grain_rate['denominator']): + return True + + if type(left) is not type(right): + if isinstance(left, (int, float)) and isinstance(right, (int, float)): + return left == right + return str(left) == str(right) + return left == right + + +def _string_matches_pattern(target, pattern, ignore_case): + if not isinstance(target, str): + return False + flags = re.IGNORECASE if ignore_case else 0 + try: + return re.search(pattern, target, flags) is not None + except re.error as error: + raise RQLParseError("Invalid regular expression in matches(): {}".format(error)) + + +def evaluate_query(expression, resource, all_resources=None): + if isinstance(expression, dict): + operator_name = expression['name'] + arguments = expression['args'] + + if operator_name == 'and': + return all(evaluate_query(argument, resource, all_resources) for argument in arguments) + if operator_name == 'or': + return any(evaluate_query(argument, resource, all_resources) for argument in arguments) + if operator_name == 'not': + return not evaluate_query(arguments[0], resource, all_resources) + + if operator_name == 'matches': + property_path = arguments[0] + if not isinstance(property_path, str): + raise RQLParseError("Property path must be a string") + pattern = str(_unwrap_value(arguments[1])) + ignore_case = len(arguments) > 2 and _unwrap_value(arguments[2]) == 'i' + property_values = extract_property_values(resource, property_path) + return any( + _string_matches_pattern(str(property_value), pattern, ignore_case) + for property_value in property_values + ) + + if operator_name == 'rel': + relation_name = arguments[0] + if not isinstance(relation_name, str): + raise RQLParseError("Relation name must be a string") + subquery = arguments[1] + relation_values = extract_property_values(resource, relation_name) + for relation_value in relation_values: + related_resource = _resolve_related_resource( + relation_name, _unwrap_value(relation_value), all_resources) + if related_resource and evaluate_query(subquery, related_resource, all_resources): + return True + return False + + if operator_name == 'sub': + property_path = arguments[0] + if not isinstance(property_path, str): + raise RQLParseError("Property path must be a string") + subquery = arguments[1] + sub_resources = extract_property_values(resource, property_path) + if not sub_resources: + return False + return any( + evaluate_query(subquery, sub_resource, all_resources) + for sub_resource in sub_resources + if isinstance(sub_resource, dict) + ) + + if operator_name in ('eq', 'ne', 'gt', 'ge', 'lt', 'le', 'in', 'out'): + property_path = arguments[0] + if not isinstance(property_path, str): + raise RQLParseError("Property path must be a string") + comparison_value = arguments[1] if _is_typed_value(arguments[1]) else _unwrap_value(arguments[1]) + property_values = extract_property_values(resource, property_path) + if not property_values: + if operator_name == 'eq' and comparison_value is None: + return True + if operator_name == 'in' and isinstance(arguments[1], list): + return any(candidate is None for candidate in arguments[1]) + return False + + if operator_name in ('in', 'out'): + if not isinstance(arguments[1], list): + raise RQLParseError("{}() requires a tuple argument".format(operator_name)) + if operator_name == 'in': + return any( + _values_equal(property_value, candidate) + for property_value in property_values + for candidate in arguments[1] + ) + return all( + not any(_values_equal(property_value, candidate) for candidate in arguments[1]) + for property_value in property_values + ) + + return any( + _evaluate_relation(operator_name, property_value, comparison_value) + for property_value in property_values + ) + + raise RQLParseError("Invalid RQL expression") + + +def _evaluate_relation(operator_name, left, right): + if operator_name == 'eq': + return _values_equal(left, right) + if operator_name == 'ne': + return not _values_equal(left, right) + + if type(left) is not type(right): + try: + left = float(left) + right = float(right) + except (TypeError, ValueError): + left = str(left) + right = str(right) + + if operator_name == 'gt': + return left > right + if operator_name == 'ge': + return left >= right + if operator_name == 'lt': + return left < right + if operator_name == 'le': + return left <= right + return False + + +def resource_matches_query_params(resource, query_params, all_resources=None, rql_expression=None): + """Return True when resource satisfies basic and/or RQL query parameters.""" + if resource is None: + return False + + for parameter_name, parameter_value in query_params.items(): + if parameter_name.startswith('paging.'): + continue + if parameter_name == 'id' or parameter_name == 'transport': + continue + if parameter_name.startswith('query.'): + if parameter_name == 'query.rql': + expression = rql_expression if rql_expression is not None else parse_query(parameter_value) + if not evaluate_query(expression, resource, all_resources): + return False + elif parameter_name in SUPPORTED_QUERY_PARAMS: + continue + else: + raise UnsupportedRQLOperator(parameter_name) + else: + property_values = extract_property_values(resource, parameter_name) + if not property_values or not any( + _values_equal(property_value, parameter_value) for property_value in property_values): + return False + + return True + + +def has_unsupported_query_params(query_params): + for parameter_name in query_params: + if not parameter_name.startswith('query.'): + continue + if parameter_name in SUPPORTED_QUERY_PARAMS: + continue + return True + return False diff --git a/nmostesting/mocks/Node.py b/nmostesting/mocks/Node.py index 42aa0fdf7..cd229c9ff 100644 --- a/nmostesting/mocks/Node.py +++ b/nmostesting/mocks/Node.py @@ -27,6 +27,30 @@ from ..IS10Utils import IS10Utils from .Auth import PRIMARY_AUTH +MXL_TRANSPORT = "urn:x-nmos:transport:mxl" +MXL_TRANSPORT_PARAM_KEYS = ("mxl_domain_id", "mxl_flow_id") + + +def _resource_transport(resource, resource_id): + if resource == 'senders': + return NODE.senders[resource_id]['sender'].get('transport', 'urn:x-nmos:transport:rtp') + return NODE.receivers[resource_id]['receiver'].get('transport', 'urn:x-nmos:transport:rtp') + + +def _mxl_activation_block(): + return { + "activation_time": None, + "mode": None, + "requested_time": None + } + + +def _initial_mxl_transport_params(): + return [{ + "mxl_domain_id": None, + "mxl_flow_id": None + }] + class Node(object): def __init__(self, port_increment): @@ -174,6 +198,158 @@ def add_receiver(self, receiver): 'receiver': receiver } + def add_mxl_sender(self, sender, mxl_domain_id): + """ + Adds IS-05 Connection API state for an MXL Sender registered in the mock Registry. + """ + transport_params = [{ + "mxl_domain_id": mxl_domain_id, + "mxl_flow_id": sender['flow_id'] + }] + connection_state = { + 'transport_params': deepcopy(transport_params), + 'staged': { + "activation": _mxl_activation_block(), + "master_enable": True, + "receiver_id": None, + 'transport_params': deepcopy(transport_params) + }, + 'active': { + "activation": _mxl_activation_block(), + "master_enable": True, + "receiver_id": None, + 'transport_params': deepcopy(transport_params) + } + } + + self.senders[sender['id']] = { + 'sender': sender, + 'activations': connection_state, + 'mxl_domain_id': mxl_domain_id + } + + def add_mxl_receiver(self, receiver, mxl_domain_id): + """ + Adds IS-05 Connection API state for an MXL Receiver registered in the mock Registry. + """ + transport_params = _initial_mxl_transport_params() + activations = { + 'transport_params': deepcopy(transport_params), + 'staged': { + "activation": _mxl_activation_block(), + "master_enable": False, + "sender_id": None, + 'transport_params': deepcopy(transport_params) + }, + 'active': { + "activation": _mxl_activation_block(), + "master_enable": False, + "sender_id": None, + 'transport_params': deepcopy(transport_params) + } + } + + self.receivers[receiver['id']] = { + 'activations': activations, + 'receiver': receiver, + 'mxl_domain_id': mxl_domain_id + } + + def _resolve_mxl_flow_id(self, resource, resource_data, response_data): + transport_params = response_data['transport_params'][0] + flow_id = transport_params.get('mxl_flow_id') + + if flow_id != 'auto': + return flow_id + + if resource == 'senders': + return resource_data['sender']['flow_id'] + + sender_id = response_data.get('sender_id') + if sender_id and sender_id in self.senders: + return self.senders[sender_id]['sender']['flow_id'] + + return resource_data.get('mxl_flow_id') + + def _resolve_mxl_transport_params(self, resource, resource_data, response_data): + transport_params = response_data['transport_params'][0] + resolved_params = {} + + for param_name in MXL_TRANSPORT_PARAM_KEYS: + param_value = transport_params.get(param_name) + + if resource == 'senders' and param_value in (None, 'auto'): + if param_name == 'mxl_domain_id': + resolved_params[param_name] = resource_data['mxl_domain_id'] + else: + resolved_params[param_name] = resource_data['sender']['flow_id'] + elif param_value == 'auto': + if param_name == 'mxl_domain_id': + resolved_params[param_name] = resource_data['mxl_domain_id'] + else: + resolved_params[param_name] = self._resolve_mxl_flow_id(resource, resource_data, response_data) + else: + resolved_params[param_name] = param_value + + return resolved_params + + def _patch_staged_mxl(self, resource, resource_id, request_json): + resource_data = self.senders[resource_id] if resource == 'senders' else self.receivers[resource_id] + activations = resource_data['activations'] + response_data = deepcopy(activations['staged']) + response_code = 200 + + if resource == 'senders': + resource_type = 'sender' + connected_resource_id = 'receiver_id' + else: + resource_type = 'receiver' + connected_resource_id = 'sender_id' + + if 'transport_params' in request_json: + for key, value in request_json['transport_params'][0].items(): + response_data['transport_params'][0][key] = value + + for item in (connected_resource_id, 'activation', 'master_enable'): + if item in request_json: + response_data[item] = request_json[item] + + if resource == 'receivers' and response_data['transport_params'][0].get('mxl_flow_id') == 'auto': + return {'code': 400, 'debug': None, + 'error': 'Transport param mxl_flow_id does not satisfy constraints.'}, 400 + + if request_json.get('activation'): + response_data['transport_params'][0].update( + self._resolve_mxl_transport_params(resource, resource_data, response_data)) + + response_data['activation']['activation_time'] = IS04Utils.get_TAI_time() + + if response_data['activation']['mode'] == 'activate_immediate': + response_data['activation']['requested_time'] = None + else: + response_code = 202 + + subscription_update = resource_data[resource_type] + subscription_update['subscription']['active'] = response_data['master_enable'] + subscription_update['version'] = IS04Utils.get_TAI_time() + + if subscription_update['subscription']['active'] is True: + subscription_update['subscription'][connected_resource_id] = response_data[connected_resource_id] + else: + subscription_update['subscription'][connected_resource_id] = None + + do_request('POST', self.registry_url + 'x-nmos/registration/' + self.registry_version + + '/resource', json={'type': resource_type, 'data': subscription_update}) + + activations['active'] = response_data + activations['transport_params'] = response_data['transport_params'] + + staged_data = deepcopy(response_data) + staged_data['activation'] = _mxl_activation_block() + activations['staged'] = staged_data + + return response_data, response_code + def clear_staged_requests(self): self.staged_requests = [] @@ -247,6 +423,9 @@ def patch_staged(self, resource, resource_id, request_json): Updates mock Registry subscription in cases of activation/deactivation """ # Get current staged and active details for resource + if _resource_transport(resource, resource_id) == MXL_TRANSPORT: + return self._patch_staged_mxl(resource, resource_id, request_json) + resource_data = self.senders[resource_id] if resource == 'senders' else self.receivers[resource_id] activations = resource_data['activations'] response_data = deepcopy(activations['staged']) @@ -283,7 +462,7 @@ def patch_staged(self, resource, resource_id, request_json): response_data['transport_params'][0][key] = transport_params[key] # Check response transport params against constraints - constraints = _get_constraints(resource) + constraints = _get_constraints(resource, resource_id) for key, value in constraints.items(): if key in response_data['transport_params'][0]: @@ -416,7 +595,7 @@ def x_nmos_root(): @NODE_API.route('/x-nmos/connection', methods=['GET'], strict_slashes=False) @check_authorization def connection_root(): - base_data = ['v1.0/', 'v1.1/'] + base_data = ['v1.0/', 'v1.1/', 'v1.2/'] return make_response(Response(json.dumps(base_data), mimetype='application/json')) @@ -465,10 +644,13 @@ def connection(version, resource, resource_id): return make_response(Response(json.dumps(base_data), mimetype='application/json')) -def _get_constraints(resource): +def _get_constraints(resource, resource_id): """ Returns basic constraint set for senders or receivers """ + if _resource_transport(resource, resource_id) == MXL_TRANSPORT: + return {"mxl_domain_id": {}, "mxl_flow_id": {}} + constraints = {"destination_port": {}, "rtp_enabled": {}} if resource == 'receivers': @@ -488,7 +670,7 @@ def _get_constraints(resource): methods=["GET"], strict_slashes=False) @check_authorization def constraints(version, resource, resource_id): - base_data = [_get_constraints(resource)] + base_data = [_get_constraints(resource, resource_id)] return make_response(Response(json.dumps(base_data), mimetype='application/json')) @@ -580,8 +762,13 @@ def active(version, resource, resource_id): methods=["GET"], strict_slashes=False) @check_authorization def transport_type(version, resource, resource_id): - # TODO fetch from resource info - base_data = "urn:x-nmos:transport:rtp" + try: + if _resource_transport(resource, resource_id) == MXL_TRANSPORT: + base_data = MXL_TRANSPORT + else: + base_data = "urn:x-nmos:transport:rtp" + except KeyError: + abort(404) return make_response(Response(json.dumps(base_data), mimetype='application/json')) @@ -638,6 +825,9 @@ def transport_file(version, resource, resource_id): # GET should either redirect to the location of the transport file or return it directly try: if resource == 'senders': + if _resource_transport(resource, resource_id) == MXL_TRANSPORT: + abort(404) + sender = NODE.senders[resource_id] sdp_params = {**CONFIG.SDP_PREFERENCES, **sender.get('sdp_params', {})} diff --git a/nmostesting/mocks/Registry.py b/nmostesting/mocks/Registry.py index 401f5db6e..7b67f6cc6 100644 --- a/nmostesting/mocks/Registry.py +++ b/nmostesting/mocks/Registry.py @@ -18,6 +18,7 @@ import uuid import functools +from urllib.parse import quote from flask import request, jsonify, abort, Blueprint, Response from threading import Event, Lock @@ -26,6 +27,9 @@ WEBSOCKET_PORT_BASE, ENABLE_HTTPS, SPECIFICATIONS from authlib.jose import jwt from ..IS04Utils import IS04Utils +from ..RQLUtils import ( + RQLParseError, UnsupportedRQLOperator, has_unsupported_query_params, parse_query, resource_matches_query_params +) from ..TestHelper import SubscriptionWebsocketWorker, get_default_ip, get_mocks_hostname from .Auth import PRIMARY_AUTH @@ -205,7 +209,8 @@ def subscribe_to_query_api(self, version, subscription_request, secure=False): if self._get_resource_type(subscription['resource_path']) == resource_type and subscription['max_update_rate_ms'] == subscription_request['max_update_rate_ms'] and subscription['persist'] == subscription_request['persist'] - and subscription['secure'] == subscription_request['secure']]), None) + and subscription['secure'] == subscription_request['secure'] + and subscription['params'] == subscription_request['params']]), None) if subscription: return subscription, False @@ -265,12 +270,45 @@ def _create_and_queue_data_grains(self, resource_type, resource_ids, pre_resourc # Guard against concurrent subscription creation self.subscription_lock.acquire() - subscription_ids = [id for id, subscription in self.get_resources()['subscription'].items() - if self._get_resource_type(subscription['resource_path']) == resource_type] + subscriptions = {subscription_id: subscription + for subscription_id, subscription in self.get_resources()['subscription'].items() + if self._get_resource_type(subscription['resource_path']) == resource_type} timestamp = IS04Utils.get_TAI_time() - for subscription_id in subscription_ids: + for subscription_id, subscription in subscriptions.items(): + query_params = subscription.get('params', {}) + rql_query_string = query_params.get('query.rql') + rql_expression = parse_query(rql_query_string) if rql_query_string else None + grain_entries = [] + api_version = self.subscription_websockets[subscription_id]['api_version'] + all_resources = self.get_resources() + + for resource_id in resource_ids: + pre_resource = pre_resources.get(resource_id) + post_resource = post_resources.get(resource_id) + pre_match = pre_resource and resource_matches_query_params( + pre_resource, query_params, all_resources, rql_expression) + post_match = post_resource and resource_matches_query_params( + post_resource, query_params, all_resources, rql_expression) + + if not pre_match and not post_match: + continue + + data = {'path': resource_id} + if pre_match: + data['pre'] = IS04Utils.downgrade_resource(resource_type, + pre_resource, + api_version) + if post_match: + data['post'] = IS04Utils.downgrade_resource(resource_type, + post_resource, + api_version) + grain_entries.append(data) + + if not grain_entries: + continue + data_grain = {'grain_type': 'event', 'source_id': self.query_api_id, 'flow_id': subscription_id, @@ -280,21 +318,7 @@ def _create_and_queue_data_grains(self, resource_type, resource_ids, pre_resourc 'rate': {'denominator': 1, 'numerator': 0}, 'duration': {'denominator': 1, 'numerator': 0}, 'grain': {'type': 'urn:x-nmos:format:data.event', - 'topic': '/' + resource_type + 's/', 'data': []}} - - api_version = self.subscription_websockets[subscription_id]['api_version'] - - for resource_id in resource_ids: - data = {'path': resource_id} - if pre_resources.get(resource_id): - data['pre'] = IS04Utils.downgrade_resource(resource_type, - pre_resources[resource_id], - api_version) - if post_resources.get(resource_id): - data['post'] = IS04Utils.downgrade_resource(resource_type, - post_resources[resource_id], - api_version) - data_grain["grain"]["data"].append(data) + 'topic': '/' + resource_type + 's/', 'data': grain_entries}} self.subscription_websockets[subscription_id]['server'].queue_message(json.dumps(data_grain)) @@ -473,13 +497,8 @@ def query_resource(version, resource): registry = REGISTRIES[flask.current_app.config["REGISTRY_INSTANCE"]] registry.requested_query_api_version = version - # NOTE: Advanced Query Syntax (RQL) is not currently supported - # Only paging and id parameters have been implemented in the Basic Query Syntax - # All other Basic Query Syntax parameters will be ignored, such that this endpoint will currently either: - # * return all resources of a specified type subject to paging constraints - # * e.g. http://:/x-nmos/query//nodes will return all registered nodes - # * or return a specific resource according to the resource id - # * e.g. http://:/x-nmos/query//nodes?id= will return a single registered node + # Basic Query Syntax: paging, id, transport, basic field matching, and query.rql are implemented. + # Unsupported query.* parameters return 501. Malformed RQL returns 400. MIN_SINCE = "0:0" MAX_UNTIL = IS04Utils.get_TAI_time() @@ -509,19 +528,54 @@ def query_resource(version, resource): registry.query_api_called = True - # Reject RQL queries - for param in request.args: - if param.startswith('query.rql'): - abort(501) + query_params = request.args.to_dict(flat=False) + flat_query_params = {key: values[0] if len(values) == 1 else values + for key, values in query_params.items()} + + if has_unsupported_query_params(flat_query_params): + abort(501) + + rql_expression = None + try: + if flat_query_params.get('query.rql'): + rql_expression = parse_query(flat_query_params['query.rql']) + except UnsupportedRQLOperator: + abort(501) + except RQLParseError: + abort(400) + + transport_filter = request.args.get('transport') + transport_query = "" + if transport_filter is not None and resource_type in ('sender', 'receiver'): + transport_query = "&transport=" + quote(transport_filter, safe='') + + rql_filter = flat_query_params.get('query.rql') + if rql_filter is not None: + transport_query += "&query.rql=" + quote(rql_filter, safe='') + + def resource_matches_transport_filter(resource): + if transport_filter is None or resource_type not in ('sender', 'receiver'): + return True + return resource.get('transport', '').startswith(transport_filter) + + def resource_matches_filters(resource): + if not resource_matches_transport_filter(resource): + return False + try: + return resource_matches_query_params( + resource, flat_query_params, registry.get_resources(), rql_expression) + except UnsupportedRQLOperator: + return False # Check to see if resource is being requested as a query if request.args.get('id'): resource_id = request.args.get('id') - base_data.append(IS04Utils.downgrade_resource(resource_type, - registry.get_resources()[resource_type][resource_id], - version)) + resource_data = registry.get_resources()[resource_type][resource_id] + if resource_matches_filters(resource_data): + base_data.append(IS04Utils.downgrade_resource(resource_type, resource_data, version)) else: - data = registry.get_resources()[resource_type] + data = {resource_id: resource for resource_id, resource in registry.get_resources()[resource_type].items() + if resource_matches_filters(resource)} # only paginate for version v1.1 and up if IS04Utils.compare_api_version("v1.1", version) > 0: @@ -569,15 +623,15 @@ def query_resource(version, resource): link = "<" + protocol + "://" + host + ":" + port \ + "/x-nmos/query/" + version + "/" + resource_type + "s/?paging.since=" + until \ - + "&paging.limit=" + str(limit) + ">; rel=\"next\"" + + "&paging.limit=" + str(limit) + transport_query + ">; rel=\"next\"" link += ",<" + protocol + "://" + host + ":" + port \ + "/x-nmos/query/" + version + "/" + resource_type + "s/?paging.until=" + since \ - + "&paging.limit=" + str(limit) + ">; rel=\"prev\"" + + "&paging.limit=" + str(limit) + transport_query + ">; rel=\"prev\"" link += ",<" + protocol + "://" + host + ":" + port \ + "/x-nmos/query/" + version + "/" + resource_type + "s/?paging.since=0:0&paging.limit=" \ - + str(limit) + ">; rel=\"first\"" + + str(limit) + transport_query + ">; rel=\"first\"" response.headers["Link"] = link response.headers["X-Paging-Limit"] = limit @@ -621,9 +675,11 @@ def post_subscription(version): # Note: 'secure' not required in request, but is required in response secure = subscription_request['secure'] if 'secure' in subscription_request else ENABLE_HTTPS - # The current implementation of WebSockets in this mock Registry does not support query parameters in request - if len(subscription_request['params']) > 0: + subscription_params = subscription_request.get('params', {}) + if has_unsupported_query_params(subscription_params): abort(501) + if subscription_params.get('query.rql'): + parse_query(subscription_params['query.rql']) subscription, created = registry.subscribe_to_query_api(version, subscription_request, secure) @@ -637,6 +693,10 @@ def post_subscription(version): except SubscriptionException: abort(400) + except UnsupportedRQLOperator: + abort(501) + except RQLParseError: + abort(400) status_code = 201 if created else 200 diff --git a/nmostesting/suites/BCP0070302Test.py b/nmostesting/suites/BCP0070302Test.py new file mode 100644 index 000000000..11917a5dd --- /dev/null +++ b/nmostesting/suites/BCP0070302Test.py @@ -0,0 +1,572 @@ +# Copyright (C) 2026 Advanced Media Workflow Association +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import textwrap +from operator import itemgetter + +from .. import Config as CONFIG +from ..ControllerTest import ControllerTest, MXL_TRANSPORT, TestingFacadeException +from ..NMOSUtils import NMOSUtils + +MXL_ENDPOINT_COUNT = 4 +MXL_PROFILE_COUNT_PER_TYPE = 2 +TEST_EXAMPLE_COUNT = 3 + +MXL_RECEIVER_DEACTIVATE_JSON = { + 'master_enable': False, + 'sender_id': None, + 'activation': {'mode': 'activate_immediate'}, +} + + +def _mxl_components(frame_width, frame_height, bit_depth=10): + return [ + {'name': 'Y', 'width': frame_width, 'height': frame_height, 'bit_depth': bit_depth}, + {'name': 'Cb', 'width': frame_width // 2, 'height': frame_height, 'bit_depth': bit_depth}, + {'name': 'Cr', 'width': frame_width // 2, 'height': frame_height, 'bit_depth': bit_depth}, + ] + + +MXL_FLOW_PROFILE_1080P25 = { + 'profile_id': '1080p25', + 'media_type': 'video/v210', + 'frame_width': 1920, + 'frame_height': 1080, + 'grain_rate': {'numerator': 25, 'denominator': 1}, + 'interlace_mode': 'progressive', + 'colorspace': 'BT709', + 'transfer_characteristic': 'SDR', + 'components': _mxl_components(1920, 1080), +} + +MXL_FLOW_PROFILE_720P50 = { + 'profile_id': '720p50', + 'media_type': 'video/v210', + 'frame_width': 1280, + 'frame_height': 720, + 'grain_rate': {'numerator': 50, 'denominator': 1}, + 'interlace_mode': 'progressive', + 'colorspace': 'BT709', + 'transfer_characteristic': 'SDR', + 'components': _mxl_components(1280, 720), +} + +CAP_URI_TO_FLOW_FIELD = { + 'urn:x-nmos:cap:format:media_type': 'media_type', + 'urn:x-nmos:cap:format:frame_width': 'frame_width', + 'urn:x-nmos:cap:format:frame_height': 'frame_height', + 'urn:x-nmos:cap:format:grain_rate': 'grain_rate', + 'urn:x-nmos:cap:format:interlace_mode': 'interlace_mode', + 'urn:x-nmos:cap:format:colorspace': 'colorspace', + 'urn:x-nmos:cap:format:transfer_characteristic': 'transfer_characteristic', +} + + +class BCP0070302Test(ControllerTest): + """ + Controller tests for AMWA BCP-007-03 MXL. + """ + + def __init__(self, apis, registries, node, dns_server, **kwargs): + ControllerTest.__init__(self, apis, registries, node, dns_server, **kwargs) + + def _apply_flow_profile_to_sender(self, sender, profile): + sender['profile_id'] = profile['profile_id'] + sender['flow_params'] = {key: value for key, value in profile.items() if key != 'profile_id'} + + def _apply_caps_profile_to_receiver(self, receiver, profile): + receiver['caps'] = self._generate_caps([profile]) + + def _generate_constraint_set(self, profile): + return { + 'urn:x-nmos:cap:format:media_type': { + 'enum': [profile['media_type']] + }, + 'urn:x-nmos:cap:format:frame_width': { + 'enum': [profile['frame_width']] + }, + 'urn:x-nmos:cap:format:frame_height': { + 'enum': [profile['frame_height']] + }, + 'urn:x-nmos:cap:format:grain_rate': { + 'enum': [profile['grain_rate']] + }, + 'urn:x-nmos:cap:format:interlace_mode': { + 'enum': [profile['interlace_mode']] + }, + 'urn:x-nmos:cap:format:colorspace': { + 'enum': [profile['colorspace']] + }, + 'urn:x-nmos:cap:format:transfer_characteristic': { + 'enum': [profile['transfer_characteristic']] + }, + 'urn:x-nmos:cap:format:component_depth': { + 'enum': [max(component['bit_depth'] for component in profile['components'])] + }, + } + + def _generate_caps(self, accepted_profiles): + caps = { + 'media_types': [], + 'constraint_sets': [], + } + + for profile in accepted_profiles: + media_type = profile['media_type'] + if media_type not in caps['media_types']: + caps['media_types'].append(media_type) + caps['constraint_sets'].append(self._generate_constraint_set(profile)) + + return caps + + def _flow_value_for_cap(self, flow_params, cap_uri): + if cap_uri == 'urn:x-nmos:cap:format:component_depth': + return max(component['bit_depth'] for component in flow_params['components']) + + flow_field = CAP_URI_TO_FLOW_FIELD[cap_uri] + return flow_params[flow_field] + + def _constraint_satisfied(self, flow_value, constraint): + for constraint_key, constraint_value in constraint.items(): + if constraint_key == 'enum' and flow_value not in constraint_value: + return False + if constraint_key == 'minimum' and flow_value < constraint_value: + return False + if constraint_key == 'maximum' and flow_value > constraint_value: + return False + return True + + def _flow_matches_constraint_set(self, flow_params, constraint_set): + for cap_uri, constraint in constraint_set.items(): + if cap_uri not in CAP_URI_TO_FLOW_FIELD and cap_uri != 'urn:x-nmos:cap:format:component_depth': + continue + flow_value = self._flow_value_for_cap(flow_params, cap_uri) + if not self._constraint_satisfied(flow_value, constraint): + return False + return True + + def _is_compatible(self, sender, receiver): + if not self._sender_uses_mxl_transport(sender) or not self._receiver_uses_mxl_transport(receiver): + return False + + flow_params = sender.get('flow_params', {}) + receiver_caps = receiver.get('caps', {}) + media_type = flow_params.get('media_type') + + if media_type not in receiver_caps.get('media_types', []): + return False + + constraint_sets = receiver_caps.get('constraint_sets', []) + if not constraint_sets: + return False + + return any(self._flow_matches_constraint_set(flow_params, constraint_set) + for constraint_set in constraint_sets) + + def _assign_mxl_transport_and_profiles(self, endpoints, apply_profile): + mxl_indices = NMOSUtils.RANDOM.sample(range(len(endpoints)), MXL_ENDPOINT_COUNT) + profiles = ( + [MXL_FLOW_PROFILE_1080P25] * MXL_PROFILE_COUNT_PER_TYPE + + [MXL_FLOW_PROFILE_720P50] * MXL_PROFILE_COUNT_PER_TYPE + ) + NMOSUtils.RANDOM.shuffle(profiles) + + for endpoint_index, profile in zip(mxl_indices, profiles): + endpoints[endpoint_index]['transport'] = MXL_TRANSPORT + apply_profile(endpoints[endpoint_index], profile) + + def _select_test_mxl_senders(self): + connectable_mxl_receivers = self._registered_connectable_mxl_receivers() + testable_senders = [ + sender for sender in self.senders + if sender['registered'] and self._sender_uses_mxl_transport(sender) + and any(self._is_compatible(sender, receiver) for receiver in connectable_mxl_receivers) + ] + + example_count = min(CONFIG.MAX_TEST_ITERATIONS or TEST_EXAMPLE_COUNT, len(testable_senders)) + if example_count == 0: + return [] + + return NMOSUtils.RANDOM.sample(testable_senders, example_count) + + def _resource_facade_metadata(self, resource): + return { + 'id': resource['id'], + 'label': resource['label'], + 'description': resource['description'], + } + + def _build_possible_answers(self, candidates): + return [ + { + 'answer_id': 'answer_' + str(index), + 'display_answer': candidate['display_answer'], + 'resource': self._resource_facade_metadata(candidate), + } + for index, candidate in enumerate(candidates) + ] + + def _multi_choice_mismatch(self, test, actual_answers, expected_answers, + missing_fail_message, extra_fail_message): + actual = set(actual_answers) + expected = set(expected_answers) + if expected - actual: + return test.FAIL(missing_fail_message) + if actual - expected: + return test.FAIL(extra_fail_message) + + def _select_candidate_resources(self, primary_resources, other_resources, + max_primary_count, candidate_count): + primary_sample_count = NMOSUtils.RANDOM.randint(1, min(max_primary_count, len(primary_resources))) + candidates = NMOSUtils.RANDOM.sample(primary_resources, primary_sample_count) + + if len(candidates) < candidate_count: + other_sample_count = min(candidate_count - len(candidates), len(other_resources)) + candidates.extend(NMOSUtils.RANDOM.sample(other_resources, other_sample_count)) + + candidates.sort(key=itemgetter('label')) + return candidates + + def _run_mxl_discovery_test(self, test, question, resources, is_mxl, is_non_mxl, + max_mxl_count, candidate_count, + missing_fail_message, extra_fail_message, pass_message): + mxl_resources = [resource for resource in resources if is_mxl(resource)] + non_mxl_resources = [resource for resource in resources if is_non_mxl(resource)] + candidates = self._select_candidate_resources( + mxl_resources, non_mxl_resources, max_mxl_count, candidate_count) + + possible_answers = self._build_possible_answers(candidates) + expected_answers = [ + 'answer_' + str(index) for index, resource in enumerate(candidates) if is_mxl(resource) + ] + + actual_answers = self.testing_facade_utils.invoke_testing_facade( + question, possible_answers, test_type='multi_choice', + test_method_name=test.name)['answer_response'] + + mismatch = self._multi_choice_mismatch( + test, actual_answers, expected_answers, missing_fail_message, extra_fail_message) + if mismatch: + return mismatch + + return test.PASS(pass_message) + + def _compatible_receivers_for_sender(self, sender): + return [ + receiver for receiver in self._registered_connectable_mxl_receivers() + if self._is_compatible(sender, receiver) + ] + + def _deactivate_mxl_receiver(self, receiver): + self.node.patch_staged('receivers', receiver['id'], MXL_RECEIVER_DEACTIVATE_JSON) + + def set_up_tests(self): + NMOSUtils.RANDOM.seed(a=CONFIG.RANDOM_SEED) + + self.senders = [ + {'label': 's1/connery', 'description': 'Mock sender 1', 'registered': True}, + {'label': 's2/niven', 'description': 'Mock sender 2', 'registered': True}, + {'label': 's3/lazenby', 'description': 'Mock sender 3', 'registered': True}, + {'label': 's4/moore', 'description': 'Mock sender 4', 'registered': True}, + {'label': 's5/dalton', 'description': 'Mock sender 5', 'registered': True}, + {'label': 's6/brosnan', 'description': 'Mock sender 6', 'registered': True}, + {'label': 's7/craig', 'description': 'Mock sender 7', 'registered': True}, + ] + self._assign_mxl_transport_and_profiles(self.senders, self._apply_flow_profile_to_sender) + + self.receivers = [ + {'label': 'r1/dr_no', 'description': 'Mock receiver 1', + 'connectable': True, 'registered': True}, + {'label': 'r2/blofeld', 'description': 'Mock receiver 2', + 'connectable': True, 'registered': True}, + {'label': 'r3/goldfinger', 'description': 'Mock receiver 3', + 'connectable': True, 'registered': True}, + {'label': 'r4/scaramanga', 'description': 'Mock receiver 4', + 'connectable': True, 'registered': True}, + {'label': 'r5/le_chiffre', 'description': 'Mock receiver 5', + 'connectable': True, 'registered': True}, + {'label': 'r6/silva', 'description': 'Mock receiver 6', + 'connectable': True, 'registered': True}, + {'label': 'r7/oberhauser', 'description': 'Mock receiver 7', + 'connectable': True, 'registered': True}, + ] + self._assign_mxl_transport_and_profiles(self.receivers, self._apply_caps_profile_to_receiver) + + ControllerTest.set_up_tests(self) + + def _reset_mxl_receivers(self): + for receiver in self._registered_connectable_mxl_receivers(): + self._deactivate_mxl_receiver(receiver) + + def _registered_connectable_mxl_receivers(self): + return [ + receiver for receiver in self.receivers + if receiver['registered'] and receiver['connectable'] + and self._receiver_uses_mxl_transport(receiver) + ] + + def _transport_file_acceptable(self, patch_data): + if 'transport_file' not in patch_data: + return True + + transport_file = patch_data['transport_file'] + if transport_file is None: + return True + + return transport_file.get('data') is None and transport_file.get('type') is None + + def test_01(self, test): + """ + Ensure NCuT can discover MXL Senders via the IS-04 Query API + """ + MAX_COMPATIBLE_SENDER_COUNT = 3 + CANDIDATE_SENDER_COUNT = 4 + + try: + question = """\ + The NCuT should be able to discover MXL Senders \ + (transport urn:x-nmos:transport:mxl) registered in the Registry. + + Refresh the NCuT's view of the Registry and select the MXL Senders \ + from the following list. + + Once MXL Senders have been identified press 'Submit'. If unable \ + to identify MXL Senders, press 'Submit' without making a selection. + """ + + return self._run_mxl_discovery_test( + test, question, self.senders, + self._sender_uses_mxl_transport, self._sender_uses_rtp_transport, + MAX_COMPATIBLE_SENDER_COUNT, CANDIDATE_SENDER_COUNT, + 'Not all MXL Senders identified', + 'Senders incorrectly identified as MXL', + 'All MXL Senders correctly identified') + + except TestingFacadeException as exception: + return test.UNCLEAR(exception.args[0]) + + def test_02(self, test): + """ + Ensure NCuT can discover MXL Receivers via the IS-04 Query API + """ + MAX_COMPATIBLE_RECEIVER_COUNT = 3 + CANDIDATE_RECEIVER_COUNT = 4 + + try: + question = """\ + The NCuT should be able to discover MXL Receivers \ + (transport urn:x-nmos:transport:mxl) registered in the Registry. + + Refresh the NCuT's view of the Registry and select the MXL Receivers \ + from the following list. + + Once MXL Receivers have been identified press 'Submit'. If unable \ + to identify MXL Receivers, press 'Submit' without making a selection. + """ + + return self._run_mxl_discovery_test( + test, question, self.receivers, + self._receiver_uses_mxl_transport, self._receiver_uses_rtp_transport, + MAX_COMPATIBLE_RECEIVER_COUNT, CANDIDATE_RECEIVER_COUNT, + 'Not all MXL Receivers identified', + 'Receivers incorrectly identified as MXL', + 'All MXL Receivers correctly identified') + + except TestingFacadeException as exception: + return test.UNCLEAR(exception.args[0]) + + def test_03(self, test): + """ + Connect an MXL Receiver to an MXL Sender via the IS-05 Connection API and ensure + the NCuT does not provide a transport_file when staging the connection + """ + try: + mxl_senders = self._select_test_mxl_senders() + + if not mxl_senders: + return test.FAIL('No registered MXL Senders available for connection test') + + tested_connection = False + + for sender in mxl_senders: + self.node.clear_staged_requests() + self._reset_mxl_receivers() + + compatible_receivers = self._compatible_receivers_for_sender(sender) + if not compatible_receivers: + continue + + receiver = NMOSUtils.RANDOM.choice(compatible_receivers) + tested_connection = True + + question = textwrap.dedent(f"""\ + It should be possible to connect available MXL Senders to compatible MXL Receivers \ + using the IS-05 Connection API. When staging the connection, the NCuT MUST NOT \ + provide a transport file in the PATCH request to the Receiver /staged endpoint. + + Use the NCuT to perform an 'immediate' activation between sender: + + {sender['display_answer']} + + and receiver: + + {receiver['display_answer']} + + Click the 'Next' button once the connection is active. + """) + + metadata = { + 'sender': self._resource_facade_metadata(sender), + 'receiver': self._resource_facade_metadata(receiver), + } + + self.testing_facade_utils.invoke_testing_facade( + question, [], test_type='action', metadata=metadata) + + patch_requests = [ + request for request in self.node.staged_requests + if request['method'] == 'PATCH' and request['resource'] == 'receivers' + ] + if len(patch_requests) < 1: + return test.FAIL('No PATCH request was received by the node') + if len(patch_requests) > 1: + return test.FAIL('Multiple PATCH requests were found') + + patch_request = patch_requests[0] + patch_data = patch_request['data'] + + if patch_request['resource_id'] != receiver['id']: + return test.FAIL('Connection request sent to incorrect receiver') + + if 'master_enable' not in patch_data: + return test.FAIL('Master enable not found in PATCH request') + if not patch_data['master_enable']: + return test.FAIL('Master_enable not set to True in PATCH request') + + if patch_data.get('sender_id') and patch_data['sender_id'] != sender['id']: + return test.FAIL('Incorrect sender found in PATCH request') + + if 'activation' not in patch_data: + return test.FAIL('No activation details in PATCH request') + if patch_data['activation'].get('mode') != 'activate_immediate': + return test.FAIL('Immediate activation not requested in PATCH request') + + if receiver['id'] in self.primary_registry.get_resources()['receiver']: + receiver_details = self.primary_registry.get_resources()['receiver'][receiver['id']] + + if not receiver_details['subscription']['active']: + return test.FAIL('Receiver does not have active subscription') + + subscription_sender_id = receiver_details['subscription'].get('sender_id') + if subscription_sender_id and subscription_sender_id != sender['id']: + return test.FAIL('Receiver did not connect to correct sender') + + if 'sender_id' not in patch_data or not patch_data['sender_id']: + return test.WARNING('Sender id SHOULD be set in patch request') + + if 'transport_params' in patch_data: + transport_params = patch_data['transport_params'][0] + mxl_flow_id = transport_params.get('mxl_flow_id') + if mxl_flow_id and mxl_flow_id not in (None, 'auto') and mxl_flow_id != sender['flow_id']: + return test.FAIL('Incorrect mxl_flow_id found in PATCH request') + + if not self._transport_file_acceptable(patch_data): + return test.FAIL( + 'transport_file attribute was provided with non-null data or type in PATCH request') + + self._deactivate_mxl_receiver(receiver) + + if not tested_connection: + return test.FAIL('No compatible MXL Sender and Receiver pairs available for connection test') + + return test.PASS('Connections successfully established without transport_file in PATCH requests') + + except TestingFacadeException as exception: + return test.UNCLEAR(exception.args[0]) + finally: + self._reset_mxl_receivers() + + def test_04(self, test): + """ + Ensure NCuT can evaluate MXL Flow compatibility using BCP-004-01 Receiver Capabilities + """ + MAX_COMPATIBLE_RECEIVER_COUNT = 4 + CANDIDATE_RECEIVER_COUNT = 6 + + try: + mxl_senders = self._select_test_mxl_senders() + + if not mxl_senders: + return test.FAIL('No registered MXL Senders available for compatibility test') + + tested_compatibility = False + + for iteration, sender in enumerate(mxl_senders): + compatible_receivers = self._compatible_receivers_for_sender(sender) + if not compatible_receivers: + continue + + tested_compatibility = True + + question = textwrap.dedent(f"""\ + The NCuT should be able to evaluate MXL Flow compatibility between MXL Senders and \ + MXL Receivers using the BCP-004-01 Receiver Capabilities mechanism. + + Refresh the NCuT's view of the Registry and carefully select the Receivers \ + that are compatible with the following MXL Sender: + + {sender['display_answer']} + + Once compatible Receivers have been identified press 'Submit'. If unable \ + to identify compatible Receivers, press 'Submit' without making a selection. + """) + + connectable_mxl_receivers = self._registered_connectable_mxl_receivers() + other_receivers = [ + receiver for receiver in connectable_mxl_receivers + if receiver not in compatible_receivers + ] + candidate_receivers = self._select_candidate_resources( + compatible_receivers, other_receivers, + MAX_COMPATIBLE_RECEIVER_COUNT, CANDIDATE_RECEIVER_COUNT) + + possible_answers = self._build_possible_answers(candidate_receivers) + expected_answers = [ + 'answer_' + str(index) for index, receiver in enumerate(candidate_receivers) + if self._is_compatible(sender, receiver) + ] + + metadata = { + 'sender': self._resource_facade_metadata(sender), + } + + actual_answers = self.testing_facade_utils.invoke_testing_facade( + question, possible_answers, test_type='multi_choice', + multipart_test=iteration, metadata=metadata)['answer_response'] + + mismatch = self._multi_choice_mismatch( + test, actual_answers, expected_answers, + 'Not all compatible Receivers identified for Sender {}'.format(sender['display_answer']), + 'Receivers incorrectly identified as compatible for Sender {}' + .format(sender['display_answer'])) + if mismatch: + return mismatch + + if not tested_compatibility: + return test.FAIL('No compatible MXL Sender and Receiver pairs available for compatibility test') + + return test.PASS('All compatible Receivers correctly identified') + + except TestingFacadeException as exception: + return test.UNCLEAR(exception.args[0]) diff --git a/test_data/controller/mxl/v1.3_device.json b/test_data/controller/mxl/v1.3_device.json new file mode 100644 index 000000000..720f66136 --- /dev/null +++ b/test_data/controller/mxl/v1.3_device.json @@ -0,0 +1,17 @@ +{ + "description": "", + "tags": {}, + "label": "AMWA Test Device", + "version": "1539094460:257325706", + "senders": [], + "receivers": [], + "controls": [ + { + "href": "http://127.0.0.1:5200/x-nmos/connection/v1.2/", + "type": "urn:x-nmos:control:sr-ctrl/v1.2" + } + ], + "type": "urn:x-nmos:device:generic", + "node_id": "00000000-0000-0000-0000-000000000000", + "id": "3f30d310-681f-4533-b919-18b000de7716" +} diff --git a/test_data/controller/mxl/v1.3_flow.json b/test_data/controller/mxl/v1.3_flow.json new file mode 100644 index 000000000..73c611fc4 --- /dev/null +++ b/test_data/controller/mxl/v1.3_flow.json @@ -0,0 +1,41 @@ +{ + "colorspace": "BT709", + "description": "", + "format": "urn:x-nmos:format:video", + "frame_height": 1080, + "tags": {}, + "transfer_characteristic": "SDR", + "frame_width": 1920, + "label": "", + "grain_rate": { + "denominator": 1, + "numerator": 25 + }, + "version": "1539094460:257325706", + "parents": [], + "interlace_mode": "progressive", + "components": [ + { + "width": 1920, + "bit_depth": 10, + "name": "Y", + "height": 1080 + }, + { + "width": 960, + "bit_depth": 10, + "name": "Cb", + "height": 1080 + }, + { + "width": 960, + "bit_depth": 10, + "name": "Cr", + "height": 1080 + } + ], + "source_id": "2ed8d1cb-8af4-3fd5-aa6e-b7f6c00c263d", + "media_type": "video/v210", + "id": "19aa36aa-0018-3e9e-8faf-56fd45a0ee1f", + "device_id": "3f30d310-681f-4533-b919-18b000de7716" +} diff --git a/test_data/controller/mxl/v1.3_node.json b/test_data/controller/mxl/v1.3_node.json new file mode 100644 index 000000000..ce7009611 --- /dev/null +++ b/test_data/controller/mxl/v1.3_node.json @@ -0,0 +1,35 @@ +{ + "description": "", + "tags": {}, + "label": "AMWA Test Suite Node", + "version": "1539094460:257325706", + "hostname": "127.0.0.1", + "href": "http://127.0.0.1:5200/", + "api": { + "endpoints": [ + { + "authorization": false, + "protocol": "http", + "version": "v1.3", + "host": "127.0.0.1", + "port": 5200, + "href": "http://127.0.0.1:5200/x-nmos/node/v1.3/" + }, + { + "authorization": false, + "protocol": "http", + "version": "v1.2", + "host": "127.0.0.1", + "port": 5200, + "href": "http://127.0.0.1:5200/x-nmos/connection/v1.2/" + } + ] + }, + "id": "00000000-0000-0000-0000-000000000000", + "clocks": [ + { + "name": "clk0", + "ref_type": "internal" + } + ] +} diff --git a/test_data/controller/mxl/v1.3_receiver.json b/test_data/controller/mxl/v1.3_receiver.json new file mode 100644 index 000000000..26ca12e99 --- /dev/null +++ b/test_data/controller/mxl/v1.3_receiver.json @@ -0,0 +1,21 @@ +{ + "description": "", + "format": "urn:x-nmos:format:video", + "tags": {}, + "label": "AMWA Test MXL Receiver", + "subscription": { + "active": false, + "sender_id": null + }, + "version": "1539094459:773471072", + "caps": { + "media_types": [ + "video/v210" + ], + "constraint_sets": [] + }, + "interface_bindings": [], + "id": "d51d7cc7-9a84-4e5b-a2a3-49d5db61a67a", + "transport": "urn:x-nmos:transport:mxl", + "device_id": "3f30d310-681f-4533-b919-18b000de7716" +} diff --git a/test_data/controller/mxl/v1.3_sender.json b/test_data/controller/mxl/v1.3_sender.json new file mode 100644 index 000000000..a1d771989 --- /dev/null +++ b/test_data/controller/mxl/v1.3_sender.json @@ -0,0 +1,16 @@ +{ + "description": "", + "tags": {}, + "label": "AMWA Test MXL Sender", + "version": "1539094460:303320729", + "manifest_href": null, + "flow_id": "19aa36aa-0018-3e9e-8faf-56fd45a0ee1f", + "subscription": { + "active": true, + "receiver_id": null + }, + "interface_bindings": [], + "id": "414b3f12-fa4d-4af4-abaa-1fb2beeed59a", + "transport": "urn:x-nmos:transport:mxl", + "device_id": "3f30d310-681f-4533-b919-18b000de7716" +} diff --git a/test_data/controller/mxl/v1.3_source.json b/test_data/controller/mxl/v1.3_source.json new file mode 100644 index 000000000..947fdae26 --- /dev/null +++ b/test_data/controller/mxl/v1.3_source.json @@ -0,0 +1,9 @@ +{ + "description": "", + "format": "urn:x-nmos:format:video", + "tags": {}, + "label": "AMWA Test MXL Source", + "version": "1539094460:257325706", + "device_id": "3f30d310-681f-4533-b919-18b000de7716", + "id": "2ed8d1cb-8af4-3fd5-aa6e-b7f6c00c263d" +} diff --git a/testingfacade/APIs/schemas/question-schema.json b/testingfacade/APIs/schemas/question-schema.json index f588773a2..0ff77b400 100644 --- a/testingfacade/APIs/schemas/question-schema.json +++ b/testingfacade/APIs/schemas/question-schema.json @@ -113,7 +113,7 @@ "description": "URL of the test suite API endpoint to send the POST request with answers" }, "metadata": { - "description": "Extra information to assist fully automated testing where an action is required on a specific resource", + "description": "Extra information to assist fully automated testing", "default": null, "anyOf": [ { @@ -122,6 +122,10 @@ { "type": "object", "properties": { + "test_class": { + "type": "string", + "description": "Python class name of the NMOS Testing controller suite, used to select automated test instructions" + }, "sender": { "$ref": "resource-details-schema.json" },