Merge pull request #5 from AMWA-TV/master

Sync branch to master

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
@@ -148,9 +150,8 @@ def compare_urls(url1, url2):
         url2_parsed = urlparse(url2.rstrip("/"))
 
         for attr in ["scheme", "path"]:
-            if attr == "hostname":
-                if getattr(url1_parsed, attr) != getattr(url2_parsed, attr):
-                    return False
+            if getattr(url1_parsed, attr) != getattr(url2_parsed, attr):
+                return False
         for attr in ["hostname"]:
             if getattr(url1_parsed, attr).lower().rstrip('.') != getattr(url2_parsed, attr).lower().rstrip('.'):
                 return False
@@ -179,3 +180,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