Initial submit for tests general (0), Senders (2), Outputs (3), Recei…

nmostesting/Config.py

@@ -186,7 +186,6 @@
 
 # Bash shell to use for running testssl.sh
 TEST_SSL_BASH = "bash"
-
 # Definition of each API specification and its versions.
 SPECIFICATIONS = {
     "is-04": {

nmostesting/GenericTest.py

@@ -666,10 +666,8 @@ def check_api_resource(self, test, resource, response_code, api, path):
         elif resource[1]['method'].upper() in ["HEAD", "OPTIONS"]:
             # For methods which don't return a payload, return immediately after the CORS header check
             return True, ""
-
         # For all other methods proceed to check the response against the schema
         schema = self.get_schema(api, resource[1]["method"], resource[0], response.status_code)
-
         if not schema:
             raise NMOSTestException(test.MANUAL("Test suite unable to locate schema"))
 

nmostesting/IS04Utils.py

@@ -11,11 +11,11 @@
 # 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 re
+from copy import deepcopy
+from fractions import Fraction
 from . import TestHelper
-
 from .NMOSUtils import NMOSUtils
-from copy import deepcopy
 
 
 class IS04Utils(NMOSUtils):
@@ -71,9 +71,106 @@ def get_resources(self, resource, url=None):
         return toReturn
 
     @staticmethod
+    def comparable_parameter_constraint_value(value):
+        if isinstance(value, dict):
+            return Fraction(value["numerator"], value.get("denominator", 1))
+        return value
+
+    @staticmethod
+    def comparable_parameter_constraint(param_constraint):
+        result = deepcopy(param_constraint)
+        if "minimum" in result:
+            result["minimum"] = IS04Utils.comparable_parameter_constraint_value(
+                result["minimum"]
+            )
+        if "maximum" in result:
+            result["maximum"] = IS04Utils.comparable_parameter_constraint_value(
+                result["maximum"]
+            )
+        if "enum" in result:
+            for i, value in enumerate(result["enum"]):
+                result["enum"][i] = IS04Utils.comparable_parameter_constraint_value(
+                    value
+                )
+        return result
+
+    @staticmethod
+    def comparable_constraint_set(constraint_set):
+        result = {
+            k: IS04Utils.comparable_parameter_constraint(v)
+            if re.match("^urn:x-nmos:cap:(?!meta:)", k)
+            else v
+            for k, v in constraint_set.items()
+        }
+        # could also remove urn:x-nmos:cap:meta:label?
+        if "urn:x-nmos:cap:meta:preference" not in result:
+            result["urn:x-nmos:cap:meta:preference"] = 0
+        if "urn:x-nmos:cap:meta:enabled" not in result:
+            result["urn:x-nmos:cap:meta:enabled"] = True
+        return result
+
+    @staticmethod
+    def comparable_constraint_sets(constraint_sets):
+        return [IS04Utils.comparable_constraint_set(_) for _ in constraint_sets]
+
+    @staticmethod
+    def compare_constraint_sets(lhs, rhs):
+        """Check that two Constraint Sets arrays are closely equivalent"""
+        return TestHelper.compare_json(
+            IS04Utils.comparable_constraint_sets(lhs),
+            IS04Utils.comparable_constraint_sets(rhs),
+        )
+
+    @staticmethod
+    def make_sampling(flow_components):
+        samplers = {
+            # Red-Green-Blue-Alpha
+            "RGBA": {"R": (1, 1), "G": (1, 1), "B": (1, 1), "A": (1, 1)},
+            # Red-Green-Blue
+            "RGB": {"R": (1, 1), "G": (1, 1), "B": (1, 1)},
+            # Non-constant luminance YCbCr
+            "YCbCr-4:4:4": {"Y": (1, 1), "Cb": (1, 1), "Cr": (1, 1)},
+            "YCbCr-4:2:2": {"Y": (1, 1), "Cb": (2, 1), "Cr": (2, 1)},
+            "YCbCr-4:2:0": {"Y": (1, 1), "Cb": (2, 2), "Cr": (2, 2)},
+            "YCbCr-4:1:1": {"Y": (1, 1), "Cb": (4, 1), "Cr": (4, 1)},
+            # Constant luminance YCbCr
+            "CLYCbCr-4:4:4": {"Yc": (1, 1), "Cbc": (1, 1), "Crc": (1, 1)},
+            "CLYCbCr-4:2:2": {"Yc": (1, 1), "Cbc": (2, 1), "Crc": (2, 1)},
+            "CLYCbCr-4:2:0": {"Yc": (1, 1), "Cbc": (2, 2), "Crc": (2, 2)},
+            # Constant intensity ICtCp
+            "ICtCp-4:4:4": {"I": (1, 1), "Ct": (1, 1), "Cp": (1, 1)},
+            "ICtCp-4:2:2": {"I": (1, 1), "Ct": (2, 1), "Cp": (2, 1)},
+            "ICtCp-4:2:0": {"I": (1, 1), "Ct": (2, 2), "Cp": (2, 2)},
+            # XYZ
+            "XYZ": {"X": (1, 1), "Y": (1, 1), "Z": (1, 1)},
+            # Key signal represented as a single component
+            "KEY": {"Key": (1, 1)},
+            # Sampling signaled by the payload
+            "UNSPECIFIED": {},
+        }
+
+        max_w, max_h = 0, 0
+        for component in flow_components:
+            w, h = component["width"], component["height"]
+            if w > max_w:
+                max_w = w
+            if h > max_h:
+                max_h = h
+
+        components_sampler = {}
+        for component in flow_components:
+            w, h = component["width"], component["height"]
+            components_sampler[component["name"]] = (max_w / w, max_h / h)
+
+        for sampling, sampler in samplers.items():
+            if sampler == components_sampler:
+                return sampling
+
     def downgrade_resource(resource_type, resource_data, requested_version):
         """Downgrades given resource data to requested version"""
-        version_major, version_minor = [int(x) for x in requested_version[1:].split(".")]
+        version_major, version_minor = [
+            int(x) for x in requested_version[1:].split(".")
+        ]
 
         data = deepcopy(resource_data)
 
@@ -94,19 +191,12 @@ def downgrade_resource(resource_type, resource_data, requested_version):
                             if key in endpoint:
                                 del endpoint[key]
                 if version_minor <= 1:
-                    keys_to_remove = [
-                        "interfaces"
-                    ]
+                    keys_to_remove = ["interfaces"]
                     for key in keys_to_remove:
                         if key in data:
                             del data[key]
                 if version_minor == 0:
-                    keys_to_remove = [
-                        "api",
-                        "clocks",
-                        "description",
-                        "tags"
-                    ]
+                    keys_to_remove = ["api", "clocks", "description", "tags"]
                     for key in keys_to_remove:
                         if key in data:
                             del data[key]
@@ -122,11 +212,7 @@ def downgrade_resource(resource_type, resource_data, requested_version):
                 if version_minor <= 1:
                     pass
                 if version_minor == 0:
-                    keys_to_remove = [
-                        "controls",
-                        "description",
-                        "tags"
-                    ]
+                    keys_to_remove = ["controls", "description", "tags"]
                     for key in keys_to_remove:
                         if key in data:
                             del data[key]
@@ -136,11 +222,7 @@ def downgrade_resource(resource_type, resource_data, requested_version):
                 if version_minor <= 2:
                     pass
                 if version_minor <= 1:
-                    keys_to_remove = [
-                        "caps",
-                        "interface_bindings",
-                        "subscription"
-                    ]
+                    keys_to_remove = ["caps", "interface_bindings", "subscription"]
                     for key in keys_to_remove:
                         if key in data:
                             del data[key]
@@ -152,9 +234,7 @@ def downgrade_resource(resource_type, resource_data, requested_version):
                 if version_minor <= 2:
                     pass
                 if version_minor <= 1:
-                    keys_to_remove = [
-                        "interface_bindings"
-                    ]
+                    keys_to_remove = ["interface_bindings"]
                     for key in keys_to_remove:
                         if key in data:
                             del data[key]
@@ -166,30 +246,22 @@ def downgrade_resource(resource_type, resource_data, requested_version):
 
             elif resource_type == "source":
                 if version_minor <= 2:
-                    keys_to_remove = [
-                        "event_type"
-                    ]
+                    keys_to_remove = ["event_type"]
                     for key in keys_to_remove:
                         if key in data:
                             del data[key]
                 if version_minor <= 1:
                     pass
                 if version_minor == 0:
-                    keys_to_remove = [
-                        "channels",
-                        "clock_name",
-                        "grain_rate"
-                    ]
+                    keys_to_remove = ["channels", "clock_name", "grain_rate"]
                     for key in keys_to_remove:
                         if key in data:
                             del data[key]
                 return data
 
             elif resource_type == "flow":
                 if version_minor <= 2:
-                    keys_to_remove = [
-                        "event_type"
-                    ]
+                    keys_to_remove = ["event_type"]
                     for key in keys_to_remove:
                         if key in data:
                             del data[key]
@@ -208,7 +280,7 @@ def downgrade_resource(resource_type, resource_data, requested_version):
                         "interlace_mode",
                         "media_type",
                         "sample_rate",
-                        "transfer_characteristic"
+                        "transfer_characteristic",
                     ]
                     for key in keys_to_remove:
                         if key in data:
@@ -217,18 +289,14 @@ def downgrade_resource(resource_type, resource_data, requested_version):
 
             elif resource_type == "subscription":
                 if version_minor <= 2:
-                    keys_to_remove = [
-                        "authorization"
-                    ]
+                    keys_to_remove = ["authorization"]
                     for key in keys_to_remove:
                         if key in data:
                             del data[key]
                 if version_minor <= 1:
                     pass
                 if version_minor == 0:
-                    keys_to_remove = [
-                        "secure"
-                    ]
+                    keys_to_remove = ["secure"]
                     for key in keys_to_remove:
                         if key in data:
                             del data[key]

nmostesting/NMOSTesting.py

@@ -83,7 +83,6 @@
 from .suites import IS0902Test
 # from .suites import IS1001Test
 from .suites import IS1101Test
-from .suites import IS1102Test
 from .suites import BCP00301Test
 from .suites import BCP0050101Test
 from .suites import BCP0060101Test
@@ -342,19 +341,14 @@
         "specs": [{
             "spec_key": "is-11",
             "api_key": "streamcompatibility"
-        }],
-        "class": IS1101Test.IS1101Test
-    },
-    "IS-11-02": {
-        "name": "IS-11 Interaction with IS-04",
-        "specs": [{
+        }, {
             "spec_key": "is-04",
             "api_key": "node"
         }, {
-            "spec_key": "is-11",
-            "api_key": "streamcompatibility"
+            "spec_key": "is-05",
+            "api_key": "connection"
         }],
-        "class": IS1102Test.IS1102Test
+        "class": IS1101Test.IS1101Test
     },
     "BCP-003-01": {
         "name": "BCP-003-01 Secure Communication",

nmostesting/TestHelper.py

@@ -264,17 +264,17 @@ def loader(uri):
     return schema
 
 
-def check_content_type(headers, expected_type="application/json"):
+def check_content_type(headers, expected_type="application/json"):
     """Check the Content-Type header of an API request or response"""
     if "Content-Type" not in headers:
         return False, "API failed to signal a Content-Type."
     else:
         ctype = headers["Content-Type"]
         ctype_params = ctype.split(";")
-        if ctype_params[0] != expected_type:
+        if ctype_params[0] != expected_type:
             return False, "API signalled a Content-Type of {} rather than {}." \
                           .format(ctype, expected_type)
-        elif ctype_params[0] in ["application/json", "application/sdp"]:
+        elif ctype_params[0] in ["application/json", "application/sdp"]:
             if len(ctype_params) == 2 and ctype_params[1].strip().lower() == "charset=utf-8":
                 return True, "API signalled an unnecessary 'charset' in its Content-Type: {}" \
                              .format(ctype)

nmostesting/suites/IS0401Test.py

@@ -1410,7 +1410,7 @@ def test_20_01(self, test):
                     headers = None
                 valid, response = self.do_request("GET", href, headers=headers)
                 if valid and response.status_code == 200:
-                    valid, message = check_content_type(response.headers, "application/sdp")
+                    valid, message = check_content_type(response.headers, "application/sdp")
                     if not content_type_warn and (not valid or message != ""):
                         content_type_warn = message
                 elif valid and response.status_code == 404:

nmostesting/suites/IS0501Test.py

@@ -1026,7 +1026,7 @@ def test_42(self, test):
                 url = self.url + "single/senders/{}/transportfile".format(sender)
                 valid, response = self.do_request("GET", url)
                 if valid and response.status_code == 200:
-                    valid, message = check_content_type(response.headers, "application/sdp")
+                    valid, message = check_content_type(response.headers, "application/sdp")
                     if valid and message != "":
                         return test.FAIL(message)
                     elif not valid: