BCP-007-03 Controller Test Suite

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': <unique label>, 'description': '',
-        #   'registered': <is registered with mock Registry>}
+        #   'registered': <is registered with mock Registry>,
+        #   'transport': <optional; defaults to RTP template 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': <unique label>, 'description': '',
-        #   'connectable': <has IS-05 connection API>, 'registered': <is registered with mock Registry>}
+        #   'connectable': <has IS-05 connection API>, 'registered': <is registered with mock Registry>,
+        #   'transport': <optional; defaults to RTP template 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,15 +316,16 @@ 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
         node_data["version"] = NMOSUtils.get_TAI_time()
         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

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": [{