[IS-12] Initial Commit for IS-12 Test Suite

nmostesting/Config.py

@@ -277,6 +277,16 @@
             }
         }
     },
+    "is-12": {
+        "repo": "is-12",
+        "versions": ["v1.0"],
+        "default_version": "v1.0",
+        "apis": {
+            "ncp": {
+                "name": "Control Protocol"
+            },
+        }
+    },
     "bcp-002-01": {
         "repo": "bcp-002-01",
         "versions": ["v1.0"],

nmostesting/NMOSTesting.py

@@ -82,6 +82,7 @@
 from .suites import IS0901Test
 from .suites import IS0902Test
 # from .suites import IS1001Test
+from .suites import IS1201Test
 from .suites import BCP00301Test
 from .suites import BCP0060101Test
 from .suites import BCP0060102Test
@@ -340,6 +341,20 @@
     #     }],
     #     "class": IS1001Test.IS1001Test
     # },
+    "IS-12-01": {
+        "name": "IS-12 NMOS Control Protocol",
+        "specs": [{
+            "spec_key": "is-04",
+            "api_key": "node",
+            "disable_fields": ["selector"]
+        }, {
+            "spec_key": "is-12",
+            "api_key": "ncp",
+            "websocket": True,
+        }],
+        "class": IS1201Test.IS1201Test,
+        "selector": True
+    },
     "BCP-003-01": {
         "name": "BCP-003-01 Secure Communication",
         "specs": [{
@@ -540,12 +555,13 @@ def index_page():
 def run_tests(test, endpoints, test_selection=["all"]):
     if test in TEST_DEFINITIONS:
         test_def = TEST_DEFINITIONS[test]
-        protocol = "http"
-        if CONFIG.ENABLE_HTTPS:
-            protocol = "https"
         apis = {}
         tested_urls = []
         for index, spec in enumerate(test_def["specs"]):
+            if spec.get("websocket"):
+                protocol = "wss" if CONFIG.ENABLE_HTTPS else "ws"
+            else:
+                protocol = "https" if CONFIG.ENABLE_HTTPS else "http"
             spec_key = spec["spec_key"]
             api_key = spec["api_key"]
             if endpoints[index]["host"] == "" or endpoints[index]["port"] == "":

nmostesting/suites/IS0401Test.py

@@ -1302,11 +1302,14 @@ def test_20(self, test):
                         api_endpoint_host_warn = True
             for service in node_self["services"]:
                 href = service["href"]
+                # Only warn about these at the end so that more major failures are flagged first
+                # Protocols other than HTTP and WebSocket may be used, so don't incorrectly flag those too
                 if href.startswith("http") and not href.startswith(self.protocol + "://"):
-                    # Only warn about these at the end so that more major failures are flagged first
-                    # Protocols other than HTTP may be used, so don't incorrectly flag those too
                     service_href_scheme_warn = True
-                if href.startswith("https://") and is_ip_address(urlparse(href).hostname):
+                if href.startswith("ws") and not href.startswith(self.ws_protocol + "://"):
+                    service_href_scheme_warn = True
+                if (href.startswith("https://") or href.startswith("wss://")) and \
+                        is_ip_address(urlparse(href).hostname):
                     service_href_hostname_warn = True
                 if self.is04_utils.compare_api_version(api["version"], "v1.3") >= 0 and \
                         service["type"].startswith("urn:x-nmos:"):
@@ -1332,11 +1335,14 @@ def test_20(self, test):
                 for device in node_devices:
                     for control in device["controls"]:
                         href = control["href"]
+                        # Only warn about these at the end so that more major failures are flagged first
+                        # Protocols other than HTTP and WebSocket may be used, so don't incorrectly flag those too
                         if href.startswith("http") and not href.startswith(self.protocol + "://"):
-                            # Only warn about these at the end so that more major failures are flagged first
-                            # Protocols other than HTTP may be used, so don't incorrectly flag those too
                             control_href_scheme_warn = True
-                        if href.startswith("https://") and is_ip_address(urlparse(href).hostname):
+                        if href.startswith("ws") and not href.startswith(self.ws_protocol + "://"):
+                            control_href_scheme_warn = True
+                        if (href.startswith("https://") or href.startswith("wss://")) and \
+                                is_ip_address(urlparse(href).hostname):
                             control_href_hostname_warn = True
                         if self.is04_utils.compare_api_version(api["version"], "v1.3") >= 0 and \
                                 control["type"].startswith("urn:x-nmos:"):

nmostesting/suites/IS1201Test.py

@@ -0,0 +1,87 @@
+# Copyright (C) 2023 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 time
+
+from .. import Config as CONFIG
+from ..GenericTest import GenericTest, NMOSTestException
+from ..IS04Utils import IS04Utils
+from ..TestHelper import WebsocketWorker
+
+NODE_API_KEY = "node"
+CONTROL_API_KEY = "ncp"
+
+
+class IS1201Test(GenericTest):
+
+    def __init__(self, apis, **kwargs):
+        # Remove the RAML key to prevent this test suite from auto-testing IS-04 API
+        apis[NODE_API_KEY].pop("raml", None)
+        GenericTest.__init__(self, apis, **kwargs)
+        self.node_url = self.apis[NODE_API_KEY]["url"]
+        self.ncp_url = self.apis[CONTROL_API_KEY]["url"]
+        self.is04_utils = IS04Utils(self.node_url)
+        self.ncp_websocket = None
+
+    def set_up_tests(self):
+        # Do nothing
+        pass
+
+    def tear_down_tests(self):
+        # Clean up Websocket resources
+        if self.ncp_websocket:
+            self.ncp_websocket.close()
+
+    def test_01(self, test):
+        """At least one Device is showing an IS-12 control advertisement matching the API under test"""
+
+        control_type = "urn:x-nmos:control:ncp/" + self.apis[CONTROL_API_KEY]["version"]
+        return self.is04_utils.do_test_device_control(
+            test,
+            self.node_url,
+            control_type,
+            self.ncp_url,
+            self.authorization
+        )
+
+    def create_ncp_socket(self, test):
+        # Reuse socket if connection already established
+        if self.ncp_websocket:
+            return True
+
+        # Create a WebSocket connection to NMOS Control Protocol endpoint
+        self.ncp_websocket = WebsocketWorker(self.apis[CONTROL_API_KEY]["url"])
+        self.ncp_websocket.start()
+
+        # Give WebSocket client a chance to start and open its connection
+        start_time = time.time()
+        while time.time() < start_time + CONFIG.WS_MESSAGE_TIMEOUT:
+            if self.ncp_websocket.is_open():
+                break
+            time.sleep(0.2)
+
+        if self.ncp_websocket.did_error_occur():
+            raise NMOSTestException(test.FAIL("Error opening WebSocket connection to {}: {}"
+                                              .format(self.apis[CONTROL_API_KEY]["url"],
+                                                      self.ncp_websocket.get_error_message())))
+        else:
+            return self.ncp_websocket.is_open()
+
+    def test_02(self, test):
+        """WebSocket successfully opened on advertised urn:x-nmos:control:ncp endpoint"""
+
+        if not self.create_ncp_socket(test):
+            return test.FAIL("Failed to open WebSocket successfully")
+
+        return test.PASS("WebSocket successfully opened")