Check IS-11 control advertisement like other test suites do

README.md

@@ -38,6 +38,8 @@ The following test suites are currently supported.
 | BCP-003-01 | BCP-003-01 Secure Communication | X | X | | See [Testing TLS](docs/2.2.%20Usage%20-%20Testing%20BCP-003-01%20TLS.md) |
 | - | BCP-003-02 Authorization | X | X | | See [Testing Authorization](docs/2.3.%20Usage%20-%20Testing%20IS-10%20Authorization.md) |
 | - | BCP-004-01 Receiver Capabilities | X | | | Included in IS-04 Node API and IS-05 Interaction with IS-04 suites |
+| BCP-006-01-01 | BCP-006-01 NMOS With JPEG XS | X | | | |
+| BCP-006-01-02 | BCP-006-01 Controller | | | X | See [Testing Controllers](docs/2.8.%20Usage%20-%20Testing%20Controllers.md) |
 
 When testing any of the above APIs it is important that they contain representative data. The test results will generate 'Could Not Test' results if no testable entities can be located. In addition, if devices support many modes of operation (including multiple video/audio formats) it is strongly recommended to re-test them in multiple modes.
 

nmostesting/NMOSUtils.py

@@ -16,9 +16,11 @@
 import functools
 import random
 from urllib.parse import urlparse
+from requests.compat import json
 
 
 from . import Config as CONFIG
+from . import TestHelper
 
 # The UTC leap seconds table below was extracted from the information provided at
 # http://www.ietf.org/timezones/data/leap-seconds.list
@@ -179,3 +181,35 @@ def sampled_list(resource_list):
     @staticmethod
     def sort_versions(versions_list):
         return sorted(versions_list, key=functools.cmp_to_key(NMOSUtils.compare_api_version))
+
+    @staticmethod
+    def do_test_device_control(test, node_url, type, href, authorization):
+        """At least one Device is showing the given control advertisement matching the API under test"""
+
+        valid, devices = TestHelper.do_request("GET", node_url + "devices")
+        if not valid:
+            return test.FAIL("Node API did not respond as expected: {}".format(devices))
+
+        found_type = False
+        found_api = False
+        try:
+            for device in devices.json():
+                controls = device["controls"]
+                for control in controls:
+                    if control["type"] == type:
+                        found_type = True
+                        if NMOSUtils.compare_urls(href, control["href"]) and \
+                                authorization is control.get("authorization", False):
+                            found_api = True
+        except json.JSONDecodeError:
+            return test.FAIL("Non-JSON response returned from Node API")
+        except KeyError:
+            return test.FAIL("One or more Devices were missing the 'controls' attribute")
+
+        if found_api:
+            return test.PASS()
+        elif found_type:
+            return test.FAIL("Found one or more Device controls, but no href and authorization mode matched the "
+                             "API under test")
+        else:
+            return test.FAIL("Unable to find any Devices which expose the control type '{}'".format(type))

nmostesting/suites/IS0502Test.py

@@ -305,34 +305,14 @@ def test_01(self, test):
     def test_02(self, test):
         """At least one Device is showing an IS-05 control advertisement matching the API under test"""
 
-        valid, devices = self.do_request("GET", self.node_url + "devices")
-        if not valid:
-            return test.FAIL("Node API did not respond as expected: {}".format(devices))
-
-        is05_devices = []
-        found_api_match = False
-        try:
-            device_type = "urn:x-nmos:control:sr-ctrl/" + self.apis[CONN_API_KEY]["version"]
-            for device in devices.json():
-                controls = device["controls"]
-                for control in controls:
-                    if control["type"] == device_type:
-                        is05_devices.append(control["href"])
-                        if self.is05_utils.compare_urls(self.connection_url, control["href"]) and \
-                                self.authorization is control.get("authorization", False):
-                            found_api_match = True
-        except json.JSONDecodeError:
-            return test.FAIL("Non-JSON response returned from Node API")
-        except KeyError:
-            return test.FAIL("One or more Devices were missing the 'controls' attribute")
-
-        if len(is05_devices) > 0 and found_api_match:
-            return test.PASS()
-        elif len(is05_devices) > 0:
-            return test.FAIL("Found one or more Device controls, but no href and authorization mode matched the "
-                             "Connection API under test")
-        else:
-            return test.FAIL("Unable to find any Devices which expose the control type '{}'".format(device_type))
+        control_type = "urn:x-nmos:control:sr-ctrl/" + self.apis[CONN_API_KEY]["version"]
+        return self.is05_utils.do_test_device_control(
+            test,
+            self.node_url,
+            control_type,
+            self.connection_url,
+            self.authorization
+        )
 
     def test_03(self, test):
         """Receivers shown in Connection API matches those shown in Node API"""

nmostesting/suites/IS0702Test.py

@@ -637,6 +637,18 @@ def test_06(self, test):
         else:
             return test.UNCLEAR("Not tested. No resources found.")
 
+    def test_07(self, test):
+        """At least one Device is showing an IS-07 control advertisement matching the API under test"""
+
+        control_type = "urn:x-nmos:control:events/" + self.apis[EVENTS_API_KEY]["version"]
+        return self.is04_utils.do_test_device_control(
+            test,
+            self.node_url,
+            control_type,
+            self.events_url,
+            self.authorization
+        )
+
     def get_websocket_connection_sources(self, test):
         """Returns a dictionary of WebSocket sources available for connection"""
         connection_sources = {}

nmostesting/suites/IS0802Test.py

@@ -160,6 +160,7 @@ def refresh_is04_resources(self, resource_type):
 
         return self.get_is04_resources(resource_type)
 
+    # hm, see NMOSUtils.do_test_device_control
     def find_device_advertisement(self):
         test = globalConfig.test
 

nmostesting/suites/IS1101Test.py

@@ -12,11 +12,12 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import time
+import re
+
 from ..NMOSUtils import NMOSUtils
 from ..GenericTest import GenericTest
 from .. import TestHelper
-import time
-import re
 from ..IS04Utils import IS04Utils
 
 COMPAT_API_KEY = "streamcompatibility"
@@ -91,25 +92,16 @@ def __init__(self, apis, **kwargs):
 
     # GENERAL TESTS
     def test_00_01(self, test):
-        """Verify that IS-11 is exposed in the Node API as \
-        urn:x-nmos:control:stream-compat/v1.0 at url /x-nmos/streamcompatibility/v1.0/
-        """
-        _, response = TestHelper.do_request(
-            "GET", self.node_url + "devices/"
+        """At least one Device is showing an IS-11 control advertisement matching the API under test"""
+
+        control_type = "urn:x-nmos:control:stream-compat/" + self.apis[COMPAT_API_KEY]["version"]
+        return NMOSUtils.do_test_device_control(
+            test,
+            self.node_url,
+            control_type,
+            self.compat_url,
+            self.authorization
         )
-        if response.status_code != 200:
-            return test.FAIL("The request has not succeeded: {}".format(response.json()))
-        controls = response.json()[0][CONTROLS]
-        control_href = ""
-        if len(controls) > 0:
-            for control in controls:
-                if control["type"] == "urn:x-nmos:control:stream-compat/" + self.apis[COMPAT_API_KEY]["version"]:
-                    control_href = control["href"]
-                    break
-            if not NMOSUtils.compare_urls(control_href, self.compat_url):
-                return test.FAIL("IS-11 URL is invalid")
-            return test.PASS()
-        return test.WARNING("IS-11 API is not available")
 
     def test_00_02(self, test):
         "Put all senders into inactive state"