diff --git a/nmostesting/Config.py b/nmostesting/Config.py index 9c6dbe8f..b91d32b3 100644 --- a/nmostesting/Config.py +++ b/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"], diff --git a/nmostesting/NMOSTesting.py b/nmostesting/NMOSTesting.py index d5340423..a3737fa0 100644 --- a/nmostesting/NMOSTesting.py +++ b/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"] == "": diff --git a/nmostesting/suites/IS0401Test.py b/nmostesting/suites/IS0401Test.py index 33502f43..ec5622fe 100644 --- a/nmostesting/suites/IS0401Test.py +++ b/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:"): diff --git a/nmostesting/suites/IS1201Test.py b/nmostesting/suites/IS1201Test.py new file mode 100644 index 00000000..282b43ec --- /dev/null +++ b/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")