diff --git a/nmostesting/NMOSTesting.py b/nmostesting/NMOSTesting.py index 77f363bd..9933141b 100644 --- a/nmostesting/NMOSTesting.py +++ b/nmostesting/NMOSTesting.py @@ -324,6 +324,12 @@ "specs": [{ "spec_key": "is-11", "api_key": "streamcompatibility" + }, { + "spec_key": "is-04", + "api_key": "node" + }, { + "spec_key": "is-05", + "api_key": "connection" }], "class": IS1101Test.IS1101Test }, diff --git a/nmostesting/nmosTestingComplexCompare.py b/nmostesting/nmosTestingComplexCompare.py new file mode 100644 index 00000000..42b46df3 --- /dev/null +++ b/nmostesting/nmosTestingComplexCompare.py @@ -0,0 +1,262 @@ +# constraint_set schema +# +# { +# "$schema": "http://json-schema.org/draft-04/schema#", +# "description": "Describes a Constraint Set", +# "title": "Constraint Set", +# "type": "object", +# "minProperties": 1, +# "properties": { +# "urn:x-nmos:cap:meta:label": { +# "description": "Freeform string label for the Constraint Set", +# "type": "string" +# }, +# "urn:x-nmos:cap:meta:preference": { +# "description": "This value expresses the relative 'weight' that the Receiver assigns to +# its preference for the streams satisfied by the associated Constraint Set. +# The weight is an integer in the range -100 through 100, +# where -100 is least preferred and 100 is most preferred. +# When the attribute is omitted, the effective value for the associated Constraint Set is 0.", +# "type": "integer", +# "default": 0, +# "maximum": 100, +# "minimum": -100 +# }, +# "urn:x-nmos:cap:meta:enabled": { +# "description": "This value indicates whether a Constraint Set is available to use immediately (true) +# or whether this is an offline capability which can be activated via +# some unspecified configuration mechanism (false). +# When the attribute is omitted its value is assumed to be true.", +# "type": "boolean", +# "default": true +# } +# }, +# "patternProperties": { +# "^urn:x-nmos:cap:(?!meta:)": { +# "$ref": "param_constraint.json" +# } +# } +# } +# +# We want to compare that two constraint sets are equal based on the properties of teh schema that allow default +# values for properties not defined. For example the "preference" property defined to true or undefined is the +# same such that comparing an object A heving the property set to true and an object B not having the property +# defined will indicate equality because the schema defines a default value. +# +# This function verifies two constraint sets where at most one constraint set is expected and each constraint +# set must have at most one "sample_rate" paremeter constraint. We then compare the two objects based on their +# respective schemas: constraint_set and rational. We expect the constraint to be defined using the "enum" +# keyword with a single array entry. +# +# return true if equal, false otherwise + + +def compare_complex_sample_rate_constraint( + response_constraints, sample_rate_constraints +): + + # NOTE: We already know that both response_constraints, sample_rate_constraints are valid + # and have been each independently been validated against the schemas. We only check equality. + + # Each constraint_sets array must have a single entry + if len(response_constraints) != 1 or len(sample_rate_constraints) != 1: + return False + + # If the sample_rate property is not defined, objects are not equivalent + try: + response_constraints_enum = response_constraints["constraint_sets"][0][ + "urn:x-nmos:cap:format:sample_rate" + ]["enum"] + except Exception: + return False + + # If the sample_rate property is not defined, objects are not equivalent + try: + sample_rate_constraints_enum = sample_rate_constraints["constraint_sets"][0][ + "urn:x-nmos:cap:format:sample_rate" + ]["enum"] + except Exception: + return False + + # There must be a single entry in the enum array + if len(response_constraints_enum) != 1 or len(sample_rate_constraints_enum) != 1: + return False + + try: + response_numerator = response_constraints_enum[0]["numerator"] + response_denominator = 1 + + if "denominator" in response_constraints_enum[0]: + response_denominator = response_constraints_enum[0]["denominator"] + + sample_rate_numerator = sample_rate_constraints_enum[0]["numerator"] + sample_rate_denominator = 1 + + if "denominator" in sample_rate_constraints_enum[0]: + sample_rate_denominator = sample_rate_constraints_enum[0]["denominator"] + + if ( + response_numerator != sample_rate_numerator + or response_denominator != sample_rate_denominator + ): + return False + except Exception: + return False + + # There must be no other patternProperties + for prop in sample_rate_constraints["constraint_sets"][0]: + if ( + prop != "urn:x-nmos:cap:format:sample_rate" + and prop != "urn:x-nmos:cap:meta:enabled" + and prop != "urn:x-nmos:cap:meta:preference" + ): + return False + + for prop in response_constraints["constraint_sets"][0]: + if ( + prop != "urn:x-nmos:cap:format:sample_rate" + and prop != "urn:x-nmos:cap:meta:enabled" + and prop != "urn:x-nmos:cap:meta:preference" + ): + return False + + # Check meta:enabled considering default values + response_enabled = True + if "urn:x-nmos:cap:meta:enabled" in response_constraints["constraint_sets"][0]: + response_enabled = response_constraints["constraint_sets"][0][ + "urn:x-nmos:cap:meta:enabled" + ] + + sample_rate_enabled = True + if "urn:x-nmos:cap:meta:enabled" in sample_rate_constraints["constraint_sets"][0]: + sample_rate_enabled = sample_rate_constraints["constraint_sets"][0][ + "urn:x-nmos:cap:meta:enabled" + ] + + if response_enabled != sample_rate_enabled: + return False + + # Check meta:preference considering default values + response_preference = 0 + if "urn:x-nmos:cap:meta:preference" in response_constraints["constraint_sets"][0]: + response_preference = response_constraints["constraint_sets"][0][ + "urn:x-nmos:cap:meta:preference" + ] + + sample_rate_preference = 0 + if ( + "urn:x-nmos:cap:meta:preference" + in sample_rate_constraints["constraint_sets"][0] + ): + sample_rate_preference = sample_rate_constraints["constraint_sets"][0][ + "urn:x-nmos:cap:meta:preference" + ] + + if response_preference != sample_rate_preference: + return False + + # If we get here it is because the two objects are equal + return True + + +def compare_complex_grain_rate_constraint(response_constraints, grain_rate_constraints): + + # NOTE: We already know that both response_constraints, grain_rate_constraints are valid + # and have been each independently been validated against the schemas. We only check equality. + + # Each constraint_sets array must have a single entry + if len(response_constraints) != 1 or len(grain_rate_constraints) != 1: + return False + + # If the grain_rate property is not defined, objects are not equivalent + try: + response_constraints_enum = response_constraints["constraint_sets"][0][ + "urn:x-nmos:cap:format:grain_rate" + ]["enum"] + except Exception: + return False + + # If the grain_rate property is not defined, objects are not equivalent + try: + grain_rate_constraints_enum = grain_rate_constraints["constraint_sets"][0][ + "urn:x-nmos:cap:format:grain_rate" + ]["enum"] + except Exception: + return False + + # There must be a single entry in the enum array + if len(response_constraints_enum) != 1 or len(grain_rate_constraints_enum) != 1: + return False + + try: + response_numerator = response_constraints_enum[0]["numerator"] + response_denominator = 1 + + if "denominator" in response_constraints_enum[0]: + response_denominator = response_constraints_enum[0]["denominator"] + + grain_rate_numerator = grain_rate_constraints_enum[0]["numerator"] + grain_rate_denominator = 1 + + if "denominator" in grain_rate_constraints_enum[0]: + grain_rate_denominator = grain_rate_constraints_enum[0]["denominator"] + + if ( + response_numerator != grain_rate_numerator + or response_denominator != grain_rate_denominator + ): + return False + except Exception: + return False + + # There must be no other patternProperties + for prop in grain_rate_constraints["constraint_sets"][0]: + if ( + prop != "urn:x-nmos:cap:format:grain_rate" + and prop != "urn:x-nmos:cap:meta:enabled" + and prop != "urn:x-nmos:cap:meta:preference" + ): + return False + + for prop in response_constraints["constraint_sets"][0]: + if ( + prop != "urn:x-nmos:cap:format:grain_rate" + and prop != "urn:x-nmos:cap:meta:enabled" + and prop != "urn:x-nmos:cap:meta:preference" + ): + return False + + # Check meta:enabled considering default values + response_enabled = True + if "urn:x-nmos:cap:meta:enabled" in response_constraints["constraint_sets"][0]: + response_enabled = response_constraints["constraint_sets"][0][ + "urn:x-nmos:cap:meta:enabled" + ] + + grain_rate_enabled = True + if "urn:x-nmos:cap:meta:enabled" in grain_rate_constraints["constraint_sets"][0]: + grain_rate_enabled = grain_rate_constraints["constraint_sets"][0][ + "urn:x-nmos:cap:meta:enabled" + ] + + if response_enabled != grain_rate_enabled: + return False + + # Check meta:preference considering default values + response_preference = 0 + if "urn:x-nmos:cap:meta:preference" in response_constraints["constraint_sets"][0]: + response_preference = response_constraints["constraint_sets"][0][ + "urn:x-nmos:cap:meta:preference" + ] + + grain_rate_preference = 0 + if "urn:x-nmos:cap:meta:preference" in grain_rate_constraints["constraint_sets"][0]: + grain_rate_preference = grain_rate_constraints["constraint_sets"][0][ + "urn:x-nmos:cap:meta:preference" + ] + + if response_preference != grain_rate_preference: + return False + + # If we get here it is because the two objects are equal + return True diff --git a/nmostesting/suites/IS1101Test.py b/nmostesting/suites/IS1101Test.py index 637d3149..f3077ec6 100644 --- a/nmostesting/suites/IS1101Test.py +++ b/nmostesting/suites/IS1101Test.py @@ -17,6 +17,8 @@ COMPAT_API_KEY = "streamcompatibility" CONTROLS = "controls" +NODE_API_KEY = "node" +CONN_API_KEY = "connection" class IS1101Test(GenericTest): @@ -30,11 +32,12 @@ def __init__(self, apis): "/inputs/{inputId}/edid", "/inputs/{inputId}/edid/base", "/inputs/{inputId}/edid/effective", - "/outputs/{outputId}/edid", + "/outputs/{outputId}/edid" ] GenericTest.__init__(self, apis, omit_paths) self.compat_url = self.apis[COMPAT_API_KEY]["url"] - self.base_url = self.apis[COMPAT_API_KEY]["base_url"] + self.node_url = self.apis[NODE_API_KEY]["url"] + self.conn_url = self.apis[CONN_API_KEY]["url"] self.connected_outputs = [] self.edid_connected_outputs = [] self.not_edid_connected_outputs = [] @@ -45,34 +48,33 @@ def __init__(self, apis): self.caps = "" # GENERAL TESTS - def test_00_01(self, test): + def test_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/ """ - valid_res, response = TestHelper.do_request( - "GET", self.base_url + "/x-nmos/node/v1.3/devices/" + _, response = TestHelper.do_request( + "GET", self.node_url + "devices/" ) - if valid_res: - response_json = response.json() - controls = response_json[0][CONTROLS] - control_href = "" + if response.status_code != 200: + return test.FAIL("The request has not succeeded.", response) + controls = response.json()[0][CONTROLS] + control_href = "" + if len(controls) > 0: for control in controls: - if control["type"] == "urn:x-nmos:control:stream-compat/v1.0": + if control["type"] == "urn:x-nmos:control:stream-compat/" + self.apis[COMPAT_API_KEY]["version"]: control_href = control["href"] break - if len(control) == 0: - return test.WARNING("IS-11 API is not available") if not control_href.endswith(self.compat_url): return test.FAIL("IS-11 URL is invalid") return test.PASS() - return test.FAIL(response) + return test.FAIL("IS-11 API is not available") - def test_00_02(self, test): + def test_02(self, test): "Put all senders into inactive state" - senders_url = self.base_url + "/x-nmos/connection/v1.1/single/senders/" + senders_url = self.conn_url + "single/senders/" _, response = TestHelper.do_request("GET", senders_url) if response.status_code != 200: - return test.FAIL(response.json()) + return test.FAIL("The request has not succeeded.", response.json()) senders = response.json() if len(senders) > 0: for sender in senders: @@ -88,16 +90,16 @@ def test_00_02(self, test): or response.json()["master_enable"] or response.json()["activation"]["mode"] != "activate_immediate" ): - return test.FAIL(response.json()) + return test.FAIL("The request has not succeeded.", response.json()) return test.PASS() - return test.UNCLEAR("Could not find any IS-04 senders to test") + return test.UNCLEAR("Could not find any senders to test") - def test_00_03(self, test): + def test_03(self, test): "Put all the receivers into inactive state" - receivers_url = self.base_url + "/x-nmos/connection/v1.1/single/receivers/" + receivers_url = self.conn_url + "single/receivers/" _, response = TestHelper.do_request("GET", receivers_url) if response.status_code != 200: - return test.FAIL(response.json()) + return test.FAIL("The request has not succeeded.", response.json()) receivers = response.json() if len(receivers) > 0: for receiver in receivers: @@ -112,11 +114,11 @@ def test_00_03(self, test): or response.json()["master_enable"] or response.json()["activation"]["mode"] != "activate_immediate" ): - return test.FAIL(response.json()) + return test.FAIL("The request has not succeeded.", response.json()) return test.PASS() - return test.UNCLEAR("Could not find any IS-04 receivers to test") + return test.UNCLEAR("Could not find any receivers to test") # OUTPUTS TESTS def test_03_01(self, test): @@ -131,14 +133,14 @@ def test_03_01(self, test): if len(self.outputs) == 0: return test.UNCLEAR("No outputs") return test.PASS() - return test.FAIL(response.json()) + return test.FAIL("The request has not succeeded.", response.json()) def test_03_02(self, test): """ Verify that some of the outputs of the device are connected. """ if len(self.outputs) == 0: - return test.UNCLEAR("No IS11 receivers outputs") + return test.UNCLEAR("No IS11 outputs") for output in self.outputs: _, response = TestHelper.do_request( "GET", self.compat_url + "outputs/" + output + "properties/" @@ -150,9 +152,9 @@ def test_03_02(self, test): if output["connected"]: self.connected_outputs.append(output["id"]) else: - return test.FAIL(response.json()) + return test.FAIL("The request has not succeeded.", response.json()) if len(self.connected_outputs) == 0: - return test.UNCLEAR("None Outputs support edid.") + return test.UNCLEAR("No connected outputs.") return test.PASS() def test_03_03(self, test): @@ -161,7 +163,7 @@ def test_03_03(self, test): a signal as test 0 put all of the receivers inactive. """ if len(self.connected_outputs) == 0: - return test.UNCLEAR("no connected outputs") + return test.UNCLEAR("No connected outputs.") for output_id in self.connected_outputs: _, response = TestHelper.do_request( "GET", self.compat_url + "outputs/" + output_id + "/properties/" @@ -170,10 +172,10 @@ def test_03_03(self, test): if response.json()["status"]["state"] == "signal_present": self.active_connected_outputs.append(response.json()) else: - return test.FAIL(response.json()) + return test.FAIL("The request has not succeeded.", response.json()) if len(self.active_connected_outputs) != 0: return test.UNCLEAR( - "Connected output have a signal while all receivers are inactive" + "Connected output have a signal while all receivers are inactive." ) return test.PASS() @@ -182,7 +184,7 @@ def test_03_04(self, test): Verify that connected outputs supporting EDID behave according to the RAML file. """ if len(self.connected_outputs) == 0: - return test.UNCLEAR("no connected outputs") + return test.UNCLEAR("No connected outputs.") for output_id in self.connected_outputs: _, response = TestHelper.do_request( "GET", self.compat_url + "outputs/" + output_id + "/properties/" @@ -191,9 +193,9 @@ def test_03_04(self, test): if response.json()["edid_support"]: self.edid_connected_outputs.append(response.json()["id"]) else: - return test.FAIL(response.json()) + return test.FAIL("The request has not succeeded.", response.json()) if self.edid_connected_outputs == 0: - return test.UNCLEAR("Outputs not supporting edid") + return test.UNCLEAR("Outputs not supporting edid.") return test.PASS() def test_03_04_01(self, test): @@ -201,13 +203,13 @@ def test_03_04_01(self, test): Verify that an output indicating EDID support behaves according to the RAML file. """ if len(self.edid_connected_outputs) == 0: - return test.UNCLEAR("no edid connected outputs") + return test.UNCLEAR("No edid connected outputs.") for output_id in self.edid_connected_outputs: _, response = TestHelper.do_request( "GET", self.compat_url + "outputs/" + output_id ) if response.status_code != 200: - return test.FAIL(response.json()) + return test.FAIL("The request has not succeeded.", response.json()) return test.PASS() def test_03_04_02(self, test): @@ -217,27 +219,27 @@ def test_03_04_02(self, test): """ is_valid_response = True if len(self.edid_connected_outputs) == 0: - return test.UNCLEAR("no edid connected outputs") + return test.UNCLEAR("No edid connected outputs.") for output_id in self.edid_connected_outputs: _, response = TestHelper.do_request( "GET", self.compat_url + "outputs/" + output_id + "/edid/" ) if ( response.status_code != 200 - and response.headers["Content-Type"] != "application/octet-stream" + or response.headers["Content-Type"] != "application/octet-stream" ): is_valid_response = False break if is_valid_response: return test.PASS() - return test.FAIL(response.json()) + return test.FAIL("The request has not succeeded.", response.json()) def test_03_05(self, test): """ Verify that connected outputs not supporting EDID behave according to the RAML file. """ if len(self.connected_outputs) == 0: - return test.UNCLEAR("no connected outputs") + return test.UNCLEAR("No connected outputs.") for output_id in self.connected_outputs: _, response = TestHelper.do_request( "GET", self.compat_url + "outputs/" + output_id + "/properties/" @@ -246,9 +248,9 @@ def test_03_05(self, test): if not response.json()["edid_support"]: self.not_edid_connected_outputs.append(response.json()["id"]) else: - return test.FAIL(response.json()) + return test.FAIL("The request has not succeeded.", response.json()) if len(self.not_edid_connected_outputs) == 0: - return test.UNCLEAR("Outputs supporting edid") + return test.UNCLEAR("Outputs supporting edid.") return test.PASS() def test_03_05_01(self, test): @@ -256,13 +258,13 @@ def test_03_05_01(self, test): Verify that there is no EDID support. """ if len(self.not_edid_connected_outputs) == 0: - return test.UNCLEAR("none of not edid connected outputs") + return test.UNCLEAR("None of not edid connected outputs.") for output_id in self.not_edid_connected_outputs: _, response = TestHelper.do_request( "GET", self.compat_url + "outputs/" + output_id + "/edid/" ) if response.status_code != 204: - return test.UNCLEAR("status code should be 204") + return test.UNCLEAR("Status code should be 204.") return test.PASS() # RECEIVERS TESTS @@ -273,33 +275,29 @@ def test_04_01(self, test): _, response = TestHelper.do_request("GET", self.compat_url + "receivers/") if response.status_code != 200: - return test.FAIL(response.json()) + return test.FAIL("The request has not succeeded.", response.json()) self.receivers = response.json() - return ( - test.PASS() - if len(self.receivers) != 0 - else test.UNCLEAR("No IS_11 receivers") - ) + if len(self.receivers) == 0: + return test.UNCLEAR("No IS_11 receivers.") + return test.PASS() def test_04_01_01(self, test): """ Verify that IS-11 Receivers exist on the Node API as Receivers. """ + if len(self.receivers) == 0: + return test.UNCLEAR("No IS_11 receivers.") for receiver_id in self.receivers: _, response = TestHelper.do_request( - "GET", self.base_url + "/x-nmos/node/v1.3/receivers/" + receiver_id + "GET", self.node_url + "receivers/" + receiver_id ) if response.status_code != 200: - return test.FAIL(response.json()) + return test.FAIL("The request has not succeeded.", response.json()) if response.json()["id"] != receiver_id[:-1]: return test.UNCLEAR( - "The IS-11 Receiver doesn't exist on the Node API as receiver" + "The IS-11 Receiver doesn't exist on the Node API as receiver." ) - return ( - test.PASS() - if len(self.receivers) != 0 - else test.UNCLEAR("No IS_11 receivers") - ) + return test.PASS() def test_04_02(self, test): """ @@ -307,50 +305,44 @@ def test_04_02(self, test): """ _, response = TestHelper.do_request("GET", self.compat_url + "receivers/") if response.status_code != 200: - return test.FAIL(response.json()) + return test.FAIL("The request has not succeeded.", response.json()) self.receivers = response.json() - return ( - test.PASS() - if len(self.receivers) != 0 - else test.UNCLEAR("No IS_11 receivers") - ) + if len(self.receivers) == 0: + return test.UNCLEAR("No IS_11 receivers.") + return test.PASS() def test_04_02_01(self, test): """ Verify that the status is "unknown" or "non_compliant_stream" as per our pre-conditions of not being master_enabled. """ + if len(self.receivers) == 0: + return test.UNCLEAR("No IS_11 receivers.") for receiver_id in self.receivers: _, response = TestHelper.do_request( "GET", self.compat_url + "receivers/" + receiver_id + "status/" ) if response.status_code != 200: - return test.FAIL(response.json()) + return test.FAIL("The request has not succeeded.", response.json()) if response.json()["state"] not in ["unknown", "non compliant stream"]: - return test.FAIL("the state is not unknown or non compliant stream") - return ( - test.PASS() - if len(self.receivers) != 0 - else test.UNCLEAR("No IS_11 receivers") - ) + return test.FAIL("The state is not unknown or non compliant stream.") + return test.PASS() def test_04_02_02(self, test): """ Verify that the Receiver supports Receiver Capabilities. """ + if len(self.receivers) == 0: + return test.UNCLEAR("No IS_11 receivers.") for receiver_id in self.receivers: _, response = TestHelper.do_request( - "GET", self.base_url + "/x-nmos/node/v1.3/receivers/" + receiver_id + "GET", self.node_url + "receivers/" + receiver_id ) if response.status_code != 200: - return test.FAIL(response.json()) + return test.FAIL("The request has not succeeded.", response.json()) self.caps = response.json()["caps"] if "constraint_sets" not in self.caps: - return test.UNCLEAR(" The receiver does not have constraint_sets in caps") + return test.UNCLEAR("The receiver does not have constraint_sets in caps.") if len(self.caps["constraint_sets"]) == 0: - return test.UNCLEAR(" The receiver does not support BCP-004-01") - return ( - test.PASS() - if len(self.receivers) != 0 - else test.UNCLEAR("No IS_11 receivers") - ) + return test.UNCLEAR("The receiver does not support BCP-004-01.") + return test.PASS() diff --git a/nmostesting/suites/IS1102Test.py b/nmostesting/suites/IS1102Test.py index e51a0fb7..d5da7726 100644 --- a/nmostesting/suites/IS1102Test.py +++ b/nmostesting/suites/IS1102Test.py @@ -16,6 +16,7 @@ from .. import TestHelper import time import re +from .. import nmosTestingComplexCompare COMPAT_API_KEY = "streamcompatibility" NODE_API_KEY = "node" @@ -48,6 +49,7 @@ class IS1102Test(GenericTest): """ Runs Node Tests covering IS-11 """ + def __init__(self, apis): # Don't auto-test paths responding with an EDID binary as they don't have a JSON Schema omit_paths = [ @@ -59,7 +61,6 @@ def __init__(self, apis): GenericTest.__init__(self, apis, omit_paths) self.node_url = self.apis[NODE_API_KEY]["url"] self.compat_url = self.apis[COMPAT_API_KEY]["url"] - self.base_url = self.apis[COMPAT_API_KEY]["base_url"] self.senders = "" self.senders_2 = "" self.flow_format = {} @@ -80,54 +81,6 @@ def __init__(self, apis): Runs Node Tests covering IS-11 for Senders """ - def compare_complex(self, response_constraints, sample_rate_constraints): - - response_constraints_enum = response_constraints["constraint_sets"][0][ - "urn:x-nmos:cap:format:sample_rate" - ]["enum"] - sample_rate_constraints_enum = sample_rate_constraints["constraint_sets"][0][ - "urn:x-nmos:cap:format:sample_rate" - ]["enum"] - - if len(response_constraints_enum) > 0 and len(sample_rate_constraints_enum) > 0: - if ( - "numerator" in response_constraints_enum[0] - and "numerator" in sample_rate_constraints_enum[0] - and "denominator" in response_constraints_enum[0] - and "denominator" in sample_rate_constraints_enum[0] - ): - return ( - response_constraints_enum[0]["numerator"] - == sample_rate_constraints_enum[0]["numerator"] - and response_constraints_enum[0]["denominator"] - == sample_rate_constraints_enum[0]["denominator"] - ) - - if ( - "numerator" in response_constraints_enum[0] - and "numerator" in sample_rate_constraints_enum[0] - and "denominator" in response_constraints_enum[0] - and "denominator" not in sample_rate_constraints_enum[0] - ): - return ( - response_constraints_enum[0]["numerator"] - == sample_rate_constraints_enum[0]["numerator"] - and response_constraints_enum[0]["denominator"] == 1 - ) - - if ( - "numerator" in response_constraints_enum[0] - and "numerator" in sample_rate_constraints_enum[0] - and "denominator" not in response_constraints_enum[0] - and "denominator" in sample_rate_constraints_enum[0] - ): - return ( - response_constraints_enum[0]["numerator"] - == sample_rate_constraints_enum[0]["numerator"] - and 1 == sample_rate_constraints_enum[0]["denominator"] - ) - return False - def getSdpColorSampling(self, flow_components): """ getColorSamplingFromComponents supports RGB, @@ -297,6 +250,8 @@ def get_another_grain_rate(self, grain_rate): def get_another_sample_rate(self, sample_rate): numerator = sample_rate["numerator"] + if numerator == 0: + return {"numerator": 48000} if numerator == 48000: return {"numerator": 44100} if numerator == 44100: @@ -310,8 +265,10 @@ def get_another_sample_rate(self, sample_rate): def test_02_00(self, test): "Reset active constraints of all senders" _, response = TestHelper.do_request("GET", self.compat_url + "senders/") - if response.status_code == 200: - self.senders = response.json() + if response.status_code != 200: + return test.FAIL("The request has not succeeded.", response.json()) + self.senders = response.json() + if len(self.senders) != 0: for sender in self.senders: _, response = TestHelper.do_request( "DELETE", @@ -320,8 +277,7 @@ def test_02_00(self, test): if response.status_code != 200: return test.FAIL("senders constraints cannot be deleted") return test.PASS() - - return test.FAIL(response.json()) + return test.UNCLEAR("There is no IS-11 senders.") def test_02_01(self, test): "Verify that the device supports the concept of IS-11 Sender" @@ -330,7 +286,7 @@ def test_02_01(self, test): return test.FAIL(response.json) self.senders = response.json() if len(self.senders) == 0: - return test.UNCLEAR("there is no IS-11 senders") + return test.UNCLEAR("There is no IS-11 senders.") return test.PASS() def test_02_01_01(self, test): @@ -341,19 +297,21 @@ def test_02_01_01(self, test): "GET", self.node_url + "senders/" + sender_id ) if response.status_code != 200: - return test.FAIL() + return test.FAIL("The request has not succeeded.", response.json()) sender_node = response.json()["id"] if sender_id[:-1] != sender_node: - return test.FAIL("") + return test.FAIL("Senders are different") return test.PASS() - return test.UNCLEAR("there is no IS-11 senders") + return test.UNCLEAR("There is no IS-11 senders.") def test_02_02(self, test): "Verify senders (generic with/without inputs)" _, response = TestHelper.do_request("GET", self.compat_url + "senders/") if response.status_code != 200: - return test.FAIL(response.json()) + return test.FAIL("The request has not succeeded.", response.json()) self.senders_2 = response.json() + if len(self.senders_2) == 0: + return test.UNCLEAR("There is no IS-11 senders.") return test.PASS() def test_02_02_01(self, test): @@ -364,7 +322,7 @@ def test_02_02_01(self, test): "GET", self.compat_url + "senders/" + sender_id + "status/" ) if response.status_code != 200: - return test.FAIL(response.json()) + return test.FAIL("The request has not succeeded.", response.json()) state = response.json()["state"] if state in ["awating_essence", "no_essence"]: for i in range(0, 5): @@ -379,7 +337,7 @@ def test_02_02_01(self, test): if state != "unconstrained": return test.FAIL("inputs are unstable.") return test.PASS() - return test.UNCLEAR("there is no IS-11 senders") + return test.UNCLEAR("There is no IS-11 senders.") def test_02_02_03(self, test): """ @@ -395,7 +353,7 @@ def test_02_02_03(self, test): return test.FAIL(response.json) sender_node = response.json()["id"] if sender_id[:-1] != sender_node: - return test.FAIL("") + return test.FAIL("Senders are different") sender_flow_id = response.json()["flow_id"] if sender_flow_id is None: return test.FAIL("the sender must have a flow") @@ -406,7 +364,7 @@ def test_02_02_03(self, test): "GET", self.node_url + "flows/" + sender_flow_id ) if response.status_code != 200: - return test.FAIL(response.json()) + return test.FAIL("The request has not succeeded.", response.json()) flow_format = response.json()["format"] self.flow_format[sender_id] = flow_format if flow_format == "urn:x-nmos:format:video": @@ -427,7 +385,7 @@ def test_02_02_03(self, test): ): print("only audio and video senders are tested at this time.") return test.PASS() - return test.UNCLEAR("there is no IS-11 senders") + return test.UNCLEAR("There is no IS-11 senders.") def test_02_02_03_01(self, test): "Verify that the video sender supports the minimum set of video constraints" @@ -435,15 +393,15 @@ def test_02_02_03_01(self, test): pattern = "^urn:x-nmos:cap:" if len(self.flow_format_video) == 0: - return test.UNCLEAR("There is no video format") + return test.UNCLEAR("There is no video format.") for sender_id in self.flow_format_video: _, response = TestHelper.do_request( "GET", - self.compat_url + "senders/" + sender_id + "constraints/supported/ ", + self.compat_url + "senders/" + sender_id + "constraints/supported/", ) if response.status_code != 200: - return test.FAIL(response.json()) + return test.FAIL("The request has not succeeded.", response.json()) supportedConstraints = response.json()["parameter_constraints"] for item in supportedConstraints: if not re.search(pattern, item): @@ -459,15 +417,15 @@ def test_02_02_03_02(self, test): pattern = "^urn:x-nmos:cap:" if len(self.flow_format_audio) == 0: - return test.UNCLEAR("There is no video format") + return test.UNCLEAR("There is no video format.") for sender_id in self.flow_format_audio: _, response = TestHelper.do_request( "GET", - self.compat_url + "senders/" + sender_id + "constraints/supported/ ", + self.compat_url + "senders/" + sender_id + "constraints/supported/", ) if response.status_code != 200: - return test.FAIL(response.json()) + return test.FAIL("The request has not succeeded.", response.json()) supportedConstraints = response.json()["parameter_constraints"] for item in supportedConstraints: if not re.search(pattern, item): @@ -484,14 +442,14 @@ def test_02_02_04_01(self, test): the associated IS-04 sender. """ if len(self.flow_format_video) == 0: - return test.UNCLEAR("There is no video format") + return test.UNCLEAR("There is no video format.") for sender_id in self.flow_format_video: _, response = TestHelper.do_request( "GET", self.node_url + "senders/" + sender_id ) if response.status_code != 200: - return test.FAIL(response.json()) + return test.FAIL("The request has not succeeded.", response.json()) version = response.json()["version"] self.version[sender_id] = version self.grain_rate_constraints[sender_id] = { @@ -510,47 +468,49 @@ def test_02_02_04_01(self, test): json=self.grain_rate_constraints[sender_id], ) if response.status_code != 200: - return test.FAIL(response.json()) + return test.FAIL("The request has not succeeded.", response.json()) _, response = TestHelper.do_request( "GET", self.node_url + "senders/" + sender_id ) if response.status_code != 200: - return test.FAIL(response.json()) + return test.FAIL("The request has not succeeded.", response.json()) version = response.json()["version"] if version == self.version[sender_id]: - return test.FAIL() + return test.FAIL("Version are different") self.version[sender_id] = version _, response = TestHelper.do_request( - "GET", self.compat_url + "senders/" + sender_id + "constraints/active/ " + "GET", self.compat_url + "senders/" + sender_id + "constraints/active/" ) if response.status_code != 200: - return test.FAIL(response.json()) + return test.FAIL("The request has not succeeded.", response.json()) constraints = response.json() - if constraints != self.grain_rate_constraints[sender_id]: - return test.FAIL() + if not nmosTestingComplexCompare.compare_complex_grain_rate_constraint( + constraints, self.grain_rate_constraints[sender_id] + ): + return test.FAIL("Contraints are different") _, response = TestHelper.do_request( "DELETE", self.compat_url + "senders/" + sender_id + "constraints/active/", ) if response.status_code != 200: - return test.FAIL(response.json()) + return test.FAIL("The request has not succeeded.", response.json()) _, response = TestHelper.do_request( "GET", self.node_url + "senders/" + sender_id ) if response.status_code != 200: - return test.FAIL(response.json()) + return test.FAIL("The request has not succeeded.", response.json()) version = response.json()["version"] if version == self.version[sender_id]: - return test.FAIL() + return test.FAIL("Version are different") self.version[sender_id] = version _, response = TestHelper.do_request( "GET", self.compat_url + "senders/" + sender_id + "constraints/active/" ) if response.status_code != 200: - return test.FAIL(response.json()) + return test.FAIL("The request has not succeeded.", response.json()) constraints = response.json() if constraints != self.empty_constraints[sender_id]: - return test.FAIL("Constraints doesn't match") + return test.FAIL("Contraints are different") return test.PASS() def test_02_02_04_02(self, test): @@ -559,13 +519,13 @@ def test_02_02_04_02(self, test): sender(audio) changes the version of the associated IS-04 sender. """ if len(self.flow_format_audio) == 0: - return test.UNCLEAR("There is no audio format") + return test.UNCLEAR("There is no audio format.") for sender_id in self.flow_format_audio: _, response = TestHelper.do_request( "GET", self.node_url + "senders/" + sender_id ) if response.status_code != 200: - return test.FAIL(response.json()) + return test.FAIL("The request has not succeeded.", response.json()) version = response.json()["version"] self.version[sender_id] = version self.sample_rate_constraints[sender_id] = { @@ -584,25 +544,25 @@ def test_02_02_04_02(self, test): json=self.sample_rate_constraints[sender_id], ) if response.status_code != 200: - return test.FAIL(response.json()) + return test.FAIL("The request has not succeeded.", response.json()) _, response = TestHelper.do_request( "GET", self.node_url + "senders/" + sender_id ) if response.status_code != 200: - return test.FAIL(response.json()) + return test.FAIL("The request has not succeeded.", response.json()) version = response.json()["version"] if version == self.version[sender_id]: - return test.FAIL() + return test.FAIL("Version are different") self.version[sender_id] = version _, response = TestHelper.do_request( "GET", self.compat_url + "senders/" + sender_id + "constraints/active/" ) if response.status_code != 200: - return test.FAIL(response.json()) + return test.FAIL("The request has not succeeded.", response.json()) constraints = response.json() - if not self.compare_complex( + if not nmosTestingComplexCompare.compare_complex_sample_rate_constraint( constraints, self.sample_rate_constraints[sender_id] ): return test.FAIL( @@ -616,43 +576,42 @@ def test_02_02_04_02(self, test): self.compat_url + "senders/" + sender_id + "constraints/active/", ) if response.status_code != 200: - return test.FAIL(response.json()) + return test.FAIL("The request has not succeeded.", response.json()) _, response = TestHelper.do_request( "GET", self.node_url + "senders/" + sender_id ) if response.status_code != 200: - return test.FAIL(response.json()) + return test.FAIL("The request has not succeeded.", response.json()) version = response.json()["version"] if version == self.version[sender_id]: - return test.FAIL() + return test.FAIL("Version are different") self.version[sender_id] = version _, response = TestHelper.do_request( - "GET", self.compat_url + "senders/" + sender_id + "constraints/active/ " + "GET", self.compat_url + "senders/" + sender_id + "constraints/active/" ) if response.status_code != 200: - return test.FAIL(response.json()) + return test.FAIL("The request has not succeeded.", response.json()) constraints = response.json() if constraints != self.empty_constraints[sender_id]: - return test.FAIL("Constraints doesn't match") + return test.FAIL("Contraints are different") return test.PASS() def test_02_02_05_01(self, test): - """Verify that setting NOP constraints for frame_width, - frame_height and grain_rate does not change the flow of - a sender (video) and that the state goes from \"unconstrained\" - to \"constrained\" + """ + Verify that setting NOP constraints for frame(width,height), + grain_rate doesn't change the flow of a sender(video). """ if len(self.flow_format_video) == 0: - return test.UNCLEAR("There is no audio format ") + return test.UNCLEAR("There is no video format.") for sender_id in self.flow_format_video: _, response = TestHelper.do_request( "GET", self.compat_url + "senders/" + sender_id + "status/" ) if response.status_code != 200: - return test.FAIL(response.json()) + return test.FAIL("The request has not succeeded.", response.json()) state = response.json()["state"] if state in ["awating_essence", "no_essence"]: for i in range(0, 5): @@ -693,13 +652,13 @@ def test_02_02_05_01(self, test): json=self.constraints[sender_id], ) if response.status_code != 200: - return test.FAIL(response.json()) + return test.FAIL("The request has not succeeded.", response.json()) _, response = TestHelper.do_request( "GET", self.compat_url + "senders/" + sender_id + "status/" ) if response.status_code != 200: - return test.FAIL(response.json()) + return test.FAIL("The request has not succeeded.", response.json()) state = response.json()["state"] if state in ["awating_essence", "no_essence"]: @@ -728,7 +687,7 @@ def test_02_02_05_01(self, test): "GET", self.node_url + "flows/" + sender_flow_id ) if response.status_code != 200: - return test.FAIL(response.json()) + return test.FAIL("The request has not succeeded.", response.json()) if ( self.flow_grain_rate[sender_id] != response.json()["grain_rate"] @@ -742,23 +701,24 @@ def test_02_02_05_01(self, test): self.compat_url + "senders/" + sender_id + "constraints/active/", ) if response.status_code != 200: - return test.FAIL(response.json()) + return test.FAIL("The request has not succeeded.", response.json()) return test.PASS() def test_02_02_05_02(self, test): - """Verify that setting NOP constraints for sample_rate does not change the flow of a sender (audio) and \ - that the state goes from \"unconstrained\" to \"constrained\"""" + """ + Verify that setting NOP constraints for sample_rate doesn't change the flow of a sender(audio). + """ if len(self.flow_format_audio) == 0: - return test.UNCLEAR("There is no audio format ") + return test.UNCLEAR("There is no audio format.") for sender_id in self.flow_format_audio: _, response = TestHelper.do_request( "GET", self.compat_url + "senders/" + sender_id + "status/" ) if response.status_code != 200: - return test.FAIL(response.json()) + return test.FAIL("The request has not succeeded.", response.json()) state = response.json()["state"] if state in ["awating_essence", "no_essence"]: @@ -789,13 +749,13 @@ def test_02_02_05_02(self, test): json=self.constraints[sender_id], ) if response.status_code != 200: - return test.FAIL(response.json()) + return test.FAIL("The request has not succeeded.", response.json()) _, response = TestHelper.do_request( "GET", self.compat_url + "senders/" + sender_id + "status/" ) if response.status_code != 200: - return test.FAIL(response.json()) + return test.FAIL("The request has not succeeded.", response.json()) state = response.json()["state"] if state in ["awating_essence", "no_essence"]: @@ -824,7 +784,7 @@ def test_02_02_05_02(self, test): "GET", self.node_url + "flows/" + sender_flow_id ) if response.status_code != 200: - return test.FAIL(response.json()) + return test.FAIL("The request has not succeeded.", response.json()) flow_sample_rate = response.json()["sample_rate"] if self.flow_sample_rate[sender_id] != flow_sample_rate: return test.FAIL("Different sample rate") @@ -834,21 +794,23 @@ def test_02_02_05_02(self, test): self.compat_url + "senders/" + sender_id + "constraints/active/", ) if response.status_code != 200: - return test.FAIL(response.json()) + return test.FAIL("The request has not succeeded.", response.json()) return test.PASS() def test_02_02_06_01(self, test): - """Verify that setting NOP constraints for supported constraints does not change the flow of a sender (video) \ - and that the state goes from \"unconstrained\" to \"constrained\"""" + """ + Verify that setting NOP constraints for supported constraints + doesn't change the flow of a sender(video). + """ if len(self.flow_format_video) == 0: - return test.UNCLEAR("There is no audio format ") + return test.UNCLEAR("There is no video format.") for sender_id in self.flow_format_video: _, response = TestHelper.do_request( "GET", self.compat_url + "senders/" + sender_id + "status/" ) if response.status_code != 200: - return test.FAIL(response.json()) + return test.FAIL("The request has not succeeded.", response.json()) state = response.json()["state"] if state in ["awating_essence", "no_essence"]: for i in range(0, 5): @@ -874,7 +836,7 @@ def test_02_02_06_01(self, test): "GET", self.node_url + "flows/" + sender["flow_id"] ) if response.status_code != 200: - return test.FAIL(response.json()) + return test.FAIL("The request has not succeeded.", response.json()) flow = response.json() color_sampling = self.getSdpColorSampling(flow["components"]) if color_sampling == "invalid array of video components": @@ -882,41 +844,44 @@ def test_02_02_06_01(self, test): constraint_set = {} for item in REF_SUPPORTED_CONSTRAINTS_VIDEO: + try: + if item == "urn:x-nmos:cap:meta:label": + constraint_set["urn:x-nmos:cap:meta:label"] = "video constraint" + if item == "urn:x-nmos:cap:meta:preference": + constraint_set["urn:x-nmos:cap:meta:preference"] = 0 + if item == "urn:x-nmos:cap:meta:enabled": + constraint_set["urn:x-nmos:cap:meta:enabled"] = True + if item == "urn:x-nmos:cap:format:media_type": + constraint_set["urn:x-nmos:cap:format:media_type"] = { + "enum": [flow["media_type"]] + } + if item == "urn:x-nmos:cap:format:grain_rate": + constraint_set["urn:x-nmos:cap:format:grain_rate"] = { + "enum": [flow["grain_rate"]] + } + if item == "urn:x-nmos:cap:format:frame_width": + constraint_set["urn:x-nmos:cap:format:frame_width"] = { + "enum": [flow["frame_width"]] + } + if item == "urn:x-nmos:cap:format:frame_height": + constraint_set["urn:x-nmos:cap:format:frame_height"] = { + "enum": [flow["frame_height"]] + } + if item == "urn:x-nmos:cap:format:interlace_mode": + constraint_set["urn:x-nmos:cap:format:interlace_mode"] = { + "enum": [flow["interlace_mode"]] + } + if item == "urn:x-nmos:cap:format:color_sampling": + constraint_set["urn:x-nmos:cap:format:color_sampling"] = { + "enum": [color_sampling] + } + if item == "urn:x-nmos:cap:format:component_depth": + constraint_set["urn:x-nmos:cap:format:component_depth"] = { + "enum": [flow["components"][0]["bit_depth"]] + } + except Exception: + pass - if item == "urn:x-nmos:cap:meta:label": - constraint_set["urn:x-nmos:cap:meta:label"] = "video constraint" - if item == "urn:x-nmos:cap:meta:preference": - constraint_set["urn:x-nmos:cap:meta:preference"] = 0 - if item == "urn:x-nmos:cap:meta:enabled": - constraint_set["urn:x-nmos:cap:meta:enabled"] = True - if item == "urn:x-nmos:cap:format:media_type": - constraint_set["urn:x-nmos:cap:format:media_type"] = { - "enum": [flow["media_type"]] - } - if item == "urn:x-nmos:cap:format:grain_rate": - constraint_set["urn:x-nmos:cap:format:grain_rate"] = { - "enum": [flow["grain_rate"]] - } - if item == "urn:x-nmos:cap:format:frame_width": - constraint_set["urn:x-nmos:cap:format:frame_width"] = { - "enum": [flow["frame_width"]] - } - if item == "urn:x-nmos:cap:format:frame_height": - constraint_set["urn:x-nmos:cap:format:frame_height"] = { - "enum": [flow["frame_height"]] - } - if item == "urn:x-nmos:cap:format:interlace_mode": - constraint_set["urn:x-nmos:cap:format:interlace_mode"] = { - "enum": [flow["interlace_mode"]] - } - if item == "urn:x-nmos:cap:format:color_sampling": - constraint_set["urn:x-nmos:cap:format:color_sampling"] = { - "enum": [color_sampling] - } - if item == "urn:x-nmos:cap:format:component_depth": - constraint_set["urn:x-nmos:cap:format:component_depth"] = { - "enum": [flow["components"][0]["bit_depth"]] - } self.constraints[sender_id] = {"constraint_sets": [constraint_set]} _, response = TestHelper.do_request( @@ -925,13 +890,13 @@ def test_02_02_06_01(self, test): json=self.constraints[sender_id], ) if response.status_code != 200: - return test.FAIL(response.json()) + return test.FAIL("The request has not succeeded.", response.json()) _, response = TestHelper.do_request( "GET", self.node_url + "flows/" + sender["flow_id"] ) if response.status_code != 200: - return test.FAIL(response.json()) + return test.FAIL("The request has not succeeded.", response.json()) new_flow = response.json() new_color_sampling = self.getSdpColorSampling(new_flow["components"]) @@ -939,53 +904,54 @@ def test_02_02_06_01(self, test): return test.FAIL("invalid array of video components") for item in REF_SUPPORTED_CONSTRAINTS_VIDEO: - - if item == "urn:x-nmos:cap:format:media_type": - if flow["media_type"] != new_flow["media_type"]: - return test.FAIL("different media_type") - if item == "urn:x-nmos:cap:format:grain_rate": - if flow["grain_rate"] != new_flow["grain_rate"]: - return test.FAIL("different grain_rate") - if item == "urn:x-nmos:cap:format:frame_width": - if flow["frame_width"] != new_flow["frame_width"]: - return test.FAIL("different frame_width") - if item == "urn:x-nmos:cap:format:frame_height": - if flow["frame_height"] != new_flow["frame_height"]: - return test.FAIL("different frame_height") - if item == "urn:x-nmos:cap:format:interlace_mode": - if flow["interlace_mode"] != new_flow["interlace_mode"]: - return test.FAIL("different interlace_mode") - if item == "urn:x-nmos:cap:format:color_sampling": - if color_sampling != new_color_sampling: - return test.FAIL("different color_sampling") - if item == "urn:x-nmos:cap:format:component_depth": - if ( - flow["components"][0]["bit_depth"] - != new_flow["components"][0]["bit_depth"] - ): - return test.FAIL("different component_depth") - + try: + if item == "urn:x-nmos:cap:format:media_type": + if flow["media_type"] != new_flow["media_type"]: + return test.FAIL("different media_type") + if item == "urn:x-nmos:cap:format:grain_rate": + if flow["grain_rate"] != new_flow["grain_rate"]: + return test.FAIL("different grain_rate") + if item == "urn:x-nmos:cap:format:frame_width": + if flow["frame_width"] != new_flow["frame_width"]: + return test.FAIL("different frame_width") + if item == "urn:x-nmos:cap:format:frame_height": + if flow["frame_height"] != new_flow["frame_height"]: + return test.FAIL("different frame_height") + if item == "urn:x-nmos:cap:format:interlace_mode": + if flow["interlace_mode"] != new_flow["interlace_mode"]: + return test.FAIL("different interlace_mode") + if item == "urn:x-nmos:cap:format:color_sampling": + if color_sampling != new_color_sampling: + return test.FAIL("different color_sampling") + if item == "urn:x-nmos:cap:format:component_depth": + if ( + flow["components"][0]["bit_depth"] + != new_flow["components"][0]["bit_depth"] + ): + return test.FAIL("different component_depth") + except Exception: + pass _, response = TestHelper.do_request( "DELETE", self.compat_url + "senders/" + sender_id + "constraints/active/", ) if response.status_code != 200: - return test.FAIL(response.json()) + return test.FAIL("The request has not succeeded.", response.json()) return test.PASS() def test_02_02_06_02(self, test): - """Verify that setting NOP constraints for supported - constraints does not change the flow of a sender (audio) - and that the state goes from \"unconstrained\" to \"constrained\" + """ + Verify that setting NOP constraints for supported + constraints doesn't change the flow of a sender(audio). """ if len(self.flow_format_audio) == 0: - return test.UNCLEAR("There is no audio format") + return test.UNCLEAR("There is no audio format.") for sender_id in self.flow_format_audio: _, response = TestHelper.do_request( "GET", self.compat_url + "senders/" + sender_id + "status/" ) if response.status_code != 200: - test.FAIL(response.json()) + test.FAIL("The request has not succeeded.", response.json()) state = response.json()["state"] if state in ["awating_essence", "no_essence"]: for i in range(0, 5): @@ -1010,7 +976,7 @@ def test_02_02_06_02(self, test): "GET", self.node_url + "flows/" + sender["flow_id"] ) if response.status_code != 200: - return test.FAIL(response.json()) + return test.FAIL("The request has not succeeded.", response.json()) flow = response.json() constraint_set = {} @@ -1018,34 +984,36 @@ def test_02_02_06_02(self, test): "GET", self.node_url + "sources/" + flow["source_id"] ) if response.status_code != 200: - return test.FAIL(response.json()) + return test.FAIL("The request has not succeeded.", response.json()) source = response.json() for item in REF_SUPPORTED_CONSTRAINTS_AUDIO: - - if item == "urn:x-nmos:cap:meta:label": - constraint_set["urn:x-nmos:cap:meta:label"] = "video constraint" - if item == "urn:x-nmos:cap:meta:preference": - constraint_set["urn:x-nmos:cap:meta:preference"] = 0 - if item == "urn:x-nmos:cap:meta:enabled": - constraint_set["urn:x-nmos:cap:meta:enabled"] = True - if item == "urn:x-nmos:cap:format:media_type": - constraint_set["urn:x-nmos:cap:format:media_type"] = { - "enum": [flow["media_type"]] - } - if item == "urn:x-nmos:cap:format:sample_rate": - constraint_set["urn:x-nmos:cap:format:sample_rate"] = { - "enum": [flow["sample_rate"]] - } - if item == "urn:x-nmos:cap:format:channel_count": - constraint_set["urn:x-nmos:cap:format:channel_count"] = { - "enum": [len(source["channels"])] - } - if item == "urn:x-nmos:cap:format:sample_depth": - constraint_set["urn:x-nmos:cap:format:sample_depth"] = { - "enum": [flow["bit_depth"]] - } - + try: + + if item == "urn:x-nmos:cap:meta:label": + constraint_set["urn:x-nmos:cap:meta:label"] = "audio constraint" + if item == "urn:x-nmos:cap:meta:preference": + constraint_set["urn:x-nmos:cap:meta:preference"] = 0 + if item == "urn:x-nmos:cap:meta:enabled": + constraint_set["urn:x-nmos:cap:meta:enabled"] = True + if item == "urn:x-nmos:cap:format:media_type": + constraint_set["urn:x-nmos:cap:format:media_type"] = { + "enum": [flow["media_type"]] + } + if item == "urn:x-nmos:cap:format:sample_rate": + constraint_set["urn:x-nmos:cap:format:sample_rate"] = { + "enum": [flow["sample_rate"]] + } + if item == "urn:x-nmos:cap:format:channel_count": + constraint_set["urn:x-nmos:cap:format:channel_count"] = { + "enum": [len(source["channels"])] + } + if item == "urn:x-nmos:cap:format:sample_depth": + constraint_set["urn:x-nmos:cap:format:sample_depth"] = { + "enum": [flow["bit_depth"]] + } + except Exception: + pass self.constraints[sender_id] = {"constraint_sets": [constraint_set]} _, response = TestHelper.do_request( "PUT", @@ -1053,56 +1021,57 @@ def test_02_02_06_02(self, test): json=self.constraints[sender_id], ) if response.status_code != 200: - return test.FAIL(response.json()) + return test.FAIL("The request has not succeeded.", response.json()) _, response = TestHelper.do_request( "GET", self.node_url + "flows/" + sender["flow_id"] ) if response.status_code != 200: - return test.FAIL(response.json()) + return test.FAIL("The request has not succeeded.", response.json()) new_flow = response.json() _, response = TestHelper.do_request( "GET", self.node_url + "sources/" + flow["source_id"] ) if response.status_code != 200: - return test.FAIL(response.json()) + return test.FAIL("The request has not succeeded.", response.json()) new_source = response.json() for item in REF_SUPPORTED_CONSTRAINTS_AUDIO: - - if item == "urn:x-nmos:cap:format:media_type": - if flow["media_type"] != new_flow["media_type"]: - return test.FAIL("different media_type") - if item == "urn:x-nmos:cap:format:sample_rate": - if flow["sample_rate"] != new_flow["sample_rate"]: - return test.FAIL("different sample_rate") - if item == "urn:x-nmos:cap:format:channel_count": - if len(source["channels"]) != len(new_source["channels"]): - return test.FAIL("different channel_count") - if item == "urn:x-nmos:cap:format:sample_depth": - if flow["bit_depth"] != new_flow["bit_depth"]: - return test.FAIL("different sample_depth") - + try: + if item == "urn:x-nmos:cap:format:media_type": + if flow["media_type"] != new_flow["media_type"]: + return test.FAIL("different media_type") + if item == "urn:x-nmos:cap:format:sample_rate": + if flow["sample_rate"] != new_flow["sample_rate"]: + return test.FAIL("different sample_rate") + if item == "urn:x-nmos:cap:format:channel_count": + if len(source["channels"]) != len(new_source["channels"]): + return test.FAIL("different channel_count") + if item == "urn:x-nmos:cap:format:sample_depth": + if flow["bit_depth"] != new_flow["bit_depth"]: + return test.FAIL("different sample_depth") + except Exception: + pass _, response = TestHelper.do_request( "DELETE", self.compat_url + "senders/" + sender_id + "constraints/active/", ) if response.status_code != 200: - return test.FAIL(response.json()) + return test.FAIL("The request has not succeeded.", response.json()) return test.PASS() def test_02_02_07_01(self, test): "Verify that the device adhere to the preference of the constraint_set." if len(self.flow_format_video) == 0: - return test.UNCLEAR("There is no audio format ") + return test.UNCLEAR("There is no video format.") for sender_id in self.flow_format_video: _, response = TestHelper.do_request( "GET", self.compat_url + "senders/" + sender_id + "status/" ) if response.status_code != 200: - return test.FAIL(response.json()) + return test.FAIL("The request has not succeeded.", response.json()) state = response.json()["state"] if state in ["awating_essence", "no_essence"]: for i in range(0, 5): @@ -1128,7 +1097,7 @@ def test_02_02_07_01(self, test): "GET", self.node_url + "flows/" + sender["flow_id"] ) if response.status_code != 200: - return test.FAIL(response.json()) + return test.FAIL("The request has not succeeded.", response.json()) flow = response.json() color_sampling = self.getSdpColorSampling(flow["components"]) if color_sampling == "invalid array of video components": @@ -1137,78 +1106,87 @@ def test_02_02_07_01(self, test): constraint_set1 = {} for item in REF_SUPPORTED_CONSTRAINTS_VIDEO: - - if item == "urn:x-nmos:cap:meta:label": - constraint_set0["urn:x-nmos:cap:meta:label"] = "video constraint" - if item == "urn:x-nmos:cap:meta:preference": - constraint_set0["urn:x-nmos:cap:meta:preference"] = 0 - if item == "urn:x-nmos:cap:meta:enabled": - constraint_set0["urn:x-nmos:cap:meta:enabled"] = True - if item == "urn:x-nmos:cap:format:media_type": - constraint_set0["urn:x-nmos:cap:format:media_type"] = { - "enum": [flow["media_type"]] - } - if item == "urn:x-nmos:cap:format:grain_rate": - constraint_set0["urn:x-nmos:cap:format:grain_rate"] = { - "enum": [flow["grain_rate"]] - } - if item == "urn:x-nmos:cap:format:frame_width": - constraint_set0["urn:x-nmos:cap:format:frame_width"] = { - "enum": [flow["frame_width"]] - } - if item == "urn:x-nmos:cap:format:frame_height": - constraint_set0["urn:x-nmos:cap:format:frame_height"] = { - "enum": [flow["frame_height"]] - } - if item == "urn:x-nmos:cap:format:interlace_mode": - constraint_set0["urn:x-nmos:cap:format:interlace_mode"] = { - "enum": [flow["interlace_mode"]] - } - if item == "urn:x-nmos:cap:format:color_sampling": - constraint_set0["urn:x-nmos:cap:format:color_sampling"] = { - "enum": [color_sampling] - } - if item == "urn:x-nmos:cap:format:component_depth": - constraint_set0["urn:x-nmos:cap:format:component_depth"] = { - "enum": [flow["components"][0]["bit_depth"]] - } + try: + + if item == "urn:x-nmos:cap:meta:label": + constraint_set0[ + "urn:x-nmos:cap:meta:label" + ] = "video constraint" + if item == "urn:x-nmos:cap:meta:preference": + constraint_set0["urn:x-nmos:cap:meta:preference"] = 0 + if item == "urn:x-nmos:cap:meta:enabled": + constraint_set0["urn:x-nmos:cap:meta:enabled"] = True + if item == "urn:x-nmos:cap:format:media_type": + constraint_set0["urn:x-nmos:cap:format:media_type"] = { + "enum": [flow["media_type"]] + } + if item == "urn:x-nmos:cap:format:grain_rate": + constraint_set0["urn:x-nmos:cap:format:grain_rate"] = { + "enum": [flow["grain_rate"]] + } + if item == "urn:x-nmos:cap:format:frame_width": + constraint_set0["urn:x-nmos:cap:format:frame_width"] = { + "enum": [flow["frame_width"]] + } + if item == "urn:x-nmos:cap:format:frame_height": + constraint_set0["urn:x-nmos:cap:format:frame_height"] = { + "enum": [flow["frame_height"]] + } + if item == "urn:x-nmos:cap:format:interlace_mode": + constraint_set0["urn:x-nmos:cap:format:interlace_mode"] = { + "enum": [flow["interlace_mode"]] + } + if item == "urn:x-nmos:cap:format:color_sampling": + constraint_set0["urn:x-nmos:cap:format:color_sampling"] = { + "enum": [color_sampling] + } + if item == "urn:x-nmos:cap:format:component_depth": + constraint_set0["urn:x-nmos:cap:format:component_depth"] = { + "enum": [flow["components"][0]["bit_depth"]] + } + except Exception: + pass for item in REF_SUPPORTED_CONSTRAINTS_VIDEO: - - if item == "urn:x-nmos:cap:meta:label": - constraint_set1["urn:x-nmos:cap:meta:label"] = "video constraint" - if item == "urn:x-nmos:cap:meta:preference": - constraint_set1["urn:x-nmos:cap:meta:preference"] = -100 - if item == "urn:x-nmos:cap:meta:enabled": - constraint_set1["urn:x-nmos:cap:meta:enabled"] = True - if item == "urn:x-nmos:cap:format:media_type": - constraint_set1["urn:x-nmos:cap:format:media_type"] = { - "enum": [flow["media_type"]] - } - if item == "urn:x-nmos:cap:format:grain_rate": - constraint_set1["urn:x-nmos:cap:format:grain_rate"] = { - "enum": [self.get_another_grain_rate(flow["grain_rate"])] - } - if item == "urn:x-nmos:cap:format:frame_width": - constraint_set1["urn:x-nmos:cap:format:frame_width"] = { - "enum": [flow["frame_width"]] - } - if item == "urn:x-nmos:cap:format:frame_height": - constraint_set1["urn:x-nmos:cap:format:frame_height"] = { - "enum": [flow["frame_height"]] - } - if item == "urn:x-nmos:cap:format:interlace_mode": - constraint_set1["urn:x-nmos:cap:format:interlace_mode"] = { - "enum": [flow["interlace_mode"]] - } - if item == "urn:x-nmos:cap:format:color_sampling": - constraint_set1["urn:x-nmos:cap:format:color_sampling"] = { - "enum": [color_sampling] - } - if item == "urn:x-nmos:cap:format:component_depth": - constraint_set1["urn:x-nmos:cap:format:component_depth"] = { - "enum": [flow["components"][0]["bit_depth"]] - } + try: + if item == "urn:x-nmos:cap:meta:label": + constraint_set1[ + "urn:x-nmos:cap:meta:label" + ] = "video constraint" + if item == "urn:x-nmos:cap:meta:preference": + constraint_set1["urn:x-nmos:cap:meta:preference"] = -100 + if item == "urn:x-nmos:cap:meta:enabled": + constraint_set1["urn:x-nmos:cap:meta:enabled"] = True + if item == "urn:x-nmos:cap:format:media_type": + constraint_set1["urn:x-nmos:cap:format:media_type"] = { + "enum": [flow["media_type"]] + } + if item == "urn:x-nmos:cap:format:grain_rate": + constraint_set1["urn:x-nmos:cap:format:grain_rate"] = { + "enum": [self.get_another_grain_rate(flow["grain_rate"])] + } + if item == "urn:x-nmos:cap:format:frame_width": + constraint_set1["urn:x-nmos:cap:format:frame_width"] = { + "enum": [flow["frame_width"]] + } + if item == "urn:x-nmos:cap:format:frame_height": + constraint_set1["urn:x-nmos:cap:format:frame_height"] = { + "enum": [flow["frame_height"]] + } + if item == "urn:x-nmos:cap:format:interlace_mode": + constraint_set1["urn:x-nmos:cap:format:interlace_mode"] = { + "enum": [flow["interlace_mode"]] + } + if item == "urn:x-nmos:cap:format:color_sampling": + constraint_set1["urn:x-nmos:cap:format:color_sampling"] = { + "enum": [color_sampling] + } + if item == "urn:x-nmos:cap:format:component_depth": + constraint_set1["urn:x-nmos:cap:format:component_depth"] = { + "enum": [flow["components"][0]["bit_depth"]] + } + except Exception: + pass self.constraints[sender_id] = { "constraint_sets": [constraint_set0, constraint_set0] @@ -1219,13 +1197,13 @@ def test_02_02_07_01(self, test): json=self.constraints[sender_id], ) if response.status_code != 200: - return test.FAIL(response.json()) + return test.FAIL("The request has not succeeded.", response.json()) _, response = TestHelper.do_request( "GET", self.node_url + "flows/" + sender["flow_id"] ) if response.status_code != 200: - return test.FAIL(response.json()) + return test.FAIL("The request has not succeeded.", response.json()) new_flow = response.json() new_color_sampling = self.getSdpColorSampling(new_flow["components"]) @@ -1233,51 +1211,53 @@ def test_02_02_07_01(self, test): return test.FAIL("invalid array of video components") for item in REF_SUPPORTED_CONSTRAINTS_VIDEO: - - if item == "urn:x-nmos:cap:format:media_type": - if flow["media_type"] != new_flow["media_type"]: - return test.FAIL("different media_type") - if item == "urn:x-nmos:cap:format:grain_rate": - if flow["grain_rate"] != new_flow["grain_rate"]: - return test.FAIL("different grain_rate") - if item == "urn:x-nmos:cap:format:frame_width": - if flow["frame_width"] != new_flow["frame_width"]: - return test.FAIL("different frame_width") - if item == "urn:x-nmos:cap:format:frame_height": - if flow["frame_height"] != new_flow["frame_height"]: - return test.FAIL("different frame_height") - if item == "urn:x-nmos:cap:format:interlace_mode": - if flow["interlace_mode"] != new_flow["interlace_mode"]: - return test.FAIL("different interlace_mode") - if item == "urn:x-nmos:cap:format:color_sampling": - if color_sampling != new_color_sampling: - return test.FAIL("different color_sampling") - if item == "urn:x-nmos:cap:format:component_depth": - if ( - flow["components"][0]["bit_depth"] - != new_flow["components"][0]["bit_depth"] - ): - return test.FAIL("different component_depth") + try: + if item == "urn:x-nmos:cap:format:media_type": + if flow["media_type"] != new_flow["media_type"]: + return test.FAIL("different media_type") + if item == "urn:x-nmos:cap:format:grain_rate": + if flow["grain_rate"] != new_flow["grain_rate"]: + return test.FAIL("different grain_rate") + if item == "urn:x-nmos:cap:format:frame_width": + if flow["frame_width"] != new_flow["frame_width"]: + return test.FAIL("different frame_width") + if item == "urn:x-nmos:cap:format:frame_height": + if flow["frame_height"] != new_flow["frame_height"]: + return test.FAIL("different frame_height") + if item == "urn:x-nmos:cap:format:interlace_mode": + if flow["interlace_mode"] != new_flow["interlace_mode"]: + return test.FAIL("different interlace_mode") + if item == "urn:x-nmos:cap:format:color_sampling": + if color_sampling != new_color_sampling: + return test.FAIL("different color_sampling") + if item == "urn:x-nmos:cap:format:component_depth": + if ( + flow["components"][0]["bit_depth"] + != new_flow["components"][0]["bit_depth"] + ): + return test.FAIL("different component_depth") + except Exception: + pass _, response = TestHelper.do_request( "DELETE", self.compat_url + "senders/" + sender_id + "constraints/active/", ) if response.status_code != 200: - return test.FAIL(response.json()) + return test.FAIL("The request has not succeeded.", response.json()) return test.PASS() def test_02_02_07_02(self, test): "Verify that the device adhere to the preference of the constraint_set." if len(self.flow_format_audio) == 0: - return test.UNCLEAR("There is no audio format") + return test.UNCLEAR("There is no audio format.") for sender_id in self.flow_format_audio: _, response = TestHelper.do_request( "GET", self.compat_url + "senders/" + sender_id + "status/" ) if response.status_code != 200: - test.FAIL(response.json()) + test.FAIL("The request has not succeeded.", response.json()) state = response.json()["state"] if state in ["awating_essence", "no_essence"]: for i in range(0, 5): @@ -1303,67 +1283,75 @@ def test_02_02_07_02(self, test): "GET", self.node_url + "flows/" + sender["flow_id"] ) if response.status_code != 200: - return test.FAIL(response.json()) + return test.FAIL("The request has not succeeded.", response.json()) flow = response.json() _, response = TestHelper.do_request( "GET", self.node_url + "sources/" + flow["source_id"] ) if response.status_code != 200: - return test.FAIL(response.json()) + return test.FAIL("The request has not succeeded.", response.json()) source = response.json() constraint_set0 = {} constraint_set1 = {} for item in REF_SUPPORTED_CONSTRAINTS_AUDIO: - - if item == "urn:x-nmos:cap:meta:label": - constraint_set0["urn:x-nmos:cap:meta:label"] = "video constraint" - if item == "urn:x-nmos:cap:meta:preference": - constraint_set0["urn:x-nmos:cap:meta:preference"] = 0 - if item == "urn:x-nmos:cap:meta:enabled": - constraint_set0["urn:x-nmos:cap:meta:enabled"] = True - if item == "urn:x-nmos:cap:format:media_type": - constraint_set0["urn:x-nmos:cap:format:media_type"] = { - "enum": [flow["media_type"]] - } - if item == "urn:x-nmos:cap:format:sample_rate": - constraint_set0["urn:x-nmos:cap:format:sample_rate"] = { - "enum": [flow["sample_rate"]] - } - if item == "urn:x-nmos:cap:format:channel_count": - constraint_set0["urn:x-nmos:cap:format:channel_count"] = { - "enum": [len(source["channels"])] - } - if item == "urn:x-nmos:cap:format:sample_depth": - constraint_set0["urn:x-nmos:cap:format:sample_depth"] = { - "enum": [flow["bit_depth"]] - } + try: + if item == "urn:x-nmos:cap:meta:label": + constraint_set0[ + "urn:x-nmos:cap:meta:label" + ] = "audio constraint" + if item == "urn:x-nmos:cap:meta:preference": + constraint_set0["urn:x-nmos:cap:meta:preference"] = 0 + if item == "urn:x-nmos:cap:meta:enabled": + constraint_set0["urn:x-nmos:cap:meta:enabled"] = True + if item == "urn:x-nmos:cap:format:media_type": + constraint_set0["urn:x-nmos:cap:format:media_type"] = { + "enum": [flow["media_type"]] + } + if item == "urn:x-nmos:cap:format:sample_rate": + constraint_set0["urn:x-nmos:cap:format:sample_rate"] = { + "enum": [flow["sample_rate"]] + } + if item == "urn:x-nmos:cap:format:channel_count": + constraint_set0["urn:x-nmos:cap:format:channel_count"] = { + "enum": [len(source["channels"])] + } + if item == "urn:x-nmos:cap:format:sample_depth": + constraint_set0["urn:x-nmos:cap:format:sample_depth"] = { + "enum": [flow["bit_depth"]] + } + except Exception: + pass for item in REF_SUPPORTED_CONSTRAINTS_AUDIO: - - if item == "urn:x-nmos:cap:meta:label": - constraint_set1["urn:x-nmos:cap:meta:label"] = "video constraint" - if item == "urn:x-nmos:cap:meta:preference": - constraint_set1["urn:x-nmos:cap:meta:preference"] = -100 - if item == "urn:x-nmos:cap:meta:enabled": - constraint_set1["urn:x-nmos:cap:meta:enabled"] = True - if item == "urn:x-nmos:cap:format:media_type": - constraint_set1["urn:x-nmos:cap:format:media_type"] = { - "enum": [flow["media_type"]] - } - if item == "urn:x-nmos:cap:format:sample_rate": - constraint_set1["urn:x-nmos:cap:format:sample_rate"] = { - "enum": [self.get_another_sample_rate(flow["sample_rate"])] - } - if item == "urn:x-nmos:cap:format:channel_count": - constraint_set1["urn:x-nmos:cap:format:channel_count"] = { - "enum": [len(source["channels"])] - } - if item == "urn:x-nmos:cap:format:sample_depth": - constraint_set1["urn:x-nmos:cap:format:sample_depth"] = { - "enum": [flow["bit_depth"]] - } + try: + if item == "urn:x-nmos:cap:meta:label": + constraint_set1[ + "urn:x-nmos:cap:meta:label" + ] = "video constraint" + if item == "urn:x-nmos:cap:meta:preference": + constraint_set1["urn:x-nmos:cap:meta:preference"] = -100 + if item == "urn:x-nmos:cap:meta:enabled": + constraint_set1["urn:x-nmos:cap:meta:enabled"] = True + if item == "urn:x-nmos:cap:format:media_type": + constraint_set1["urn:x-nmos:cap:format:media_type"] = { + "enum": [flow["media_type"]] + } + if item == "urn:x-nmos:cap:format:sample_rate": + constraint_set1["urn:x-nmos:cap:format:sample_rate"] = { + "enum": [self.get_another_sample_rate(flow["sample_rate"])] + } + if item == "urn:x-nmos:cap:format:channel_count": + constraint_set1["urn:x-nmos:cap:format:channel_count"] = { + "enum": [len(source["channels"])] + } + if item == "urn:x-nmos:cap:format:sample_depth": + constraint_set1["urn:x-nmos:cap:format:sample_depth"] = { + "enum": [flow["bit_depth"]] + } + except Exception: + pass self.constraints[sender_id] = { "constraint_sets": [constraint_set0, constraint_set1] @@ -1375,40 +1363,41 @@ def test_02_02_07_02(self, test): json=self.constraints[sender_id], ) if response.status_code != 200: - return test.FAIL(response.json()) + return test.FAIL("The request has not succeeded.", response.json()) _, response = TestHelper.do_request( "GET", self.node_url + "flows/" + sender["flow_id"] ) if response.status_code != 200: - return test.FAIL(response.json()) + return test.FAIL("The request has not succeeded.", response.json()) new_flow = response.json() _, response = TestHelper.do_request( "GET", self.node_url + "sources/" + flow["source_id"] ) if response.status_code != 200: - return test.FAIL(response.json()) + return test.FAIL("The request has not succeeded.", response.json()) new_source = response.json() for item in REF_SUPPORTED_CONSTRAINTS_AUDIO: - - if item == "urn:x-nmos:cap:format:media_type": - if flow["media_type"] != new_flow["media_type"]: - return test.FAIL("different media_type") - if item == "urn:x-nmos:cap:format:sample_rate": - if flow["sample_rate"] != new_flow["sample_rate"]: - return test.FAIL("different sample_rate") - if item == "urn:x-nmos:cap:format:channel_count": - if len(source["channels"]) != len(new_source["channels"]): - return test.FAIL("different channel_count") - if item == "urn:x-nmos:cap:format:sample_depth": - if flow["bit_depth"] != new_flow["bit_depth"]: - return test.FAIL("different sample_depth") - + try: + if item == "urn:x-nmos:cap:format:media_type": + if flow["media_type"] != new_flow["media_type"]: + return test.FAIL("different media_type") + if item == "urn:x-nmos:cap:format:sample_rate": + if flow["sample_rate"] != new_flow["sample_rate"]: + return test.FAIL("different sample_rate") + if item == "urn:x-nmos:cap:format:channel_count": + if len(source["channels"]) != len(new_source["channels"]): + return test.FAIL("different channel_count") + if item == "urn:x-nmos:cap:format:sample_depth": + if flow["bit_depth"] != new_flow["bit_depth"]: + return test.FAIL("different sample_depth") + except Exception: + pass _, response = TestHelper.do_request( "DELETE", self.compat_url + "senders/" + sender_id + "constraints/active/", ) if response.status_code != 200: - return test.FAIL(response.json()) + return test.FAIL("The request has not succeeded.", response.json()) return test.PASS()