diff --git a/Dockerfile b/Dockerfile index 08fcfef8..8c000875 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,4 +1,4 @@ -FROM ubuntu:bionic +FROM ubuntu:focal WORKDIR /home/nmos-testing ADD . . @@ -7,9 +7,9 @@ ADD .git .git RUN apt-get update \ && export DEBIAN_FRONTEND=noninteractive \ && apt-get install -y wget \ - && wget https://deb.nodesource.com/setup_14.x \ - && chmod 755 setup_14.x \ - && /home/nmos-testing/setup_14.x \ + && wget https://deb.nodesource.com/setup_16.x \ + && chmod 755 setup_16.x \ + && /home/nmos-testing/setup_16.x \ && apt-get install -y --no-install-recommends \ gcc openssl libssl-dev wget ca-certificates avahi-daemon avahi-utils libnss-mdns libavahi-compat-libdnssd-dev \ python3 python3-pip python3-dev nodejs \ @@ -26,7 +26,7 @@ RUN apt-get update \ && rm v3.0.7.tar.gz \ && npm config set unsafe-perm true \ && npm install -g AMWA-TV/sdpoker#v0.3.0 \ - && rm /home/nmos-testing/setup_14.x \ + && rm /home/nmos-testing/setup_16.x \ && apt-get remove -y wget \ && apt-get clean -y --no-install-recommends \ && apt-get autoclean -y --no-install-recommends \ diff --git a/docs/1.1. Installation - Local.md b/docs/1.1. Installation - Local.md index d7249421..57fd68fa 100644 --- a/docs/1.1. Installation - Local.md +++ b/docs/1.1. Installation - Local.md @@ -4,7 +4,7 @@ Please ensure that the following dependencies are installed on your system first. -- Python 3.6 or higher, including the 'pip' package manager +- Python 3.8 or higher, including the 'pip' package manager - Git - [testssl.sh](https://testssl.sh) (required for BCP-003-01 testing, see our [README](../testssl/README.md) for instructions) - [OpenSSL](https://www.openssl.org/) (required for BCP-003-01 OCSP testing) diff --git a/nmostesting/Config.py b/nmostesting/Config.py index 3cabc7ec..1d06d829 100644 --- a/nmostesting/Config.py +++ b/nmostesting/Config.py @@ -19,6 +19,9 @@ # Please consult the documentation for instructions on how to adjust these values for common testing setups including # unicast DNS-SD and HTTPS testing. +# Number of seconds to wait after starting the mock DNS server, authorization server, etc. before running tests. +# This gives the API or client under test a chance to use these services before any test case is run. +MOCK_SERVICES_WARM_UP_DELAY = 0 # Enable or disable DNS-SD advertisements. Browsing is always permitted. # The IS-04 Node tests create a mock registry on the network unless the `ENABLE_DNS_SD` parameter is set to `False`. diff --git a/nmostesting/IS11Utils.py b/nmostesting/IS11Utils.py index 54822f4f..f4023cba 100644 --- a/nmostesting/IS11Utils.py +++ b/nmostesting/IS11Utils.py @@ -12,8 +12,6 @@ # See the License for the specific language governing permissions and # limitations under the License. -from enum import Enum - from . import TestHelper from . import Config as CONFIG from .IS04Utils import IS04Utils @@ -22,7 +20,6 @@ from .GenericTest import GenericTest NODE_API_KEY = "node" CONN_API_KEY = "connection" -SND_RCV_SUBSET = Enum('SndRcvSubset', ['ALL', 'WITH_I_O', 'WITHOUT_I_O']) class IS11Utils(NMOSUtils, GenericTest): @@ -41,47 +38,27 @@ def __init__(self, url, apis): self.reference_is05_utils = IS05Utils(CONFIG.IS11_REFERENCE_SENDER_CONNECTION_API_URL) # TODO: Remove the duplication (IS05Utils) - def get_senders(self, filter=SND_RCV_SUBSET.ALL): + def get_senders(self): """Gets a list of the available senders on the API""" toReturn = [] valid, r = TestHelper.do_request("GET", self.url + "senders/") if valid and r.status_code == 200: try: for value in r.json(): - if filter == SND_RCV_SUBSET.ALL: - toReturn.append(value[:-1]) - else: - valid_io, r_io = TestHelper.do_request("GET", self.url + "senders/" + value + "inputs/") - if valid_io and r_io.status_code == 200: - try: - if len(r_io.json()) > 0 and filter == SND_RCV_SUBSET.WITH_I_O or \ - len(r_io.json()) == 0 and filter == SND_RCV_SUBSET.WITHOUT_I_O: - toReturn.append(value[:-1]) - except ValueError: - pass + toReturn.append(value[:-1]) except ValueError: pass return toReturn # TODO: Remove the duplication (IS05Utils) - def get_receivers(self, filter=SND_RCV_SUBSET.ALL): + def get_receivers(self): """Gets a list of the available receivers on the API""" toReturn = [] valid, r = TestHelper.do_request("GET", self.url + "receivers/") if valid and r.status_code == 200: try: for value in r.json(): - if filter == SND_RCV_SUBSET.ALL: - toReturn.append(value[:-1]) - else: - valid_io, r_io = TestHelper.do_request("GET", self.url + "receivers/" + value + "outputs/") - if valid_io and r_io.status_code == 200: - try: - if len(r_io.json()) > 0 and filter == SND_RCV_SUBSET.WITH_I_O or \ - len(r_io.json()) == 0 and filter == SND_RCV_SUBSET.WITHOUT_I_O: - toReturn.append(value[:-1]) - except ValueError: - pass + toReturn.append(value[:-1]) except ValueError: pass return toReturn diff --git a/nmostesting/NMOSTesting.py b/nmostesting/NMOSTesting.py index 0a004d3e..74983135 100644 --- a/nmostesting/NMOSTesting.py +++ b/nmostesting/NMOSTesting.py @@ -1163,6 +1163,13 @@ def main(args): print(" * Testing tool running on 'http://{}:{}'. Version '{}'" .format(get_default_ip(), core_app.config['PORT'], TOOL_VERSION)) + # Give an API or client that is already running a chance to use the mock services + # before running any test cases + if CONFIG.MOCK_SERVICES_WARM_UP_DELAY: + print(" * Waiting for {} seconds to allow discovery of mock services" + .format(CONFIG.MOCK_SERVICES_WARM_UP_DELAY)) + time.sleep(CONFIG.MOCK_SERVICES_WARM_UP_DELAY) + exit_code = 0 if "suite" not in vars(CMD_ARGS): # Interactive testing mode. Await user input. diff --git a/nmostesting/suites/BCP00301Test.py b/nmostesting/suites/BCP00301Test.py index 08fa1da9..de6cf2c1 100644 --- a/nmostesting/suites/BCP00301Test.py +++ b/nmostesting/suites/BCP00301Test.py @@ -56,7 +56,9 @@ def perform_test_ssl(self, test, args=None): "--openssl-timeout", str(CONFIG.HTTP_TIMEOUT), "--add-ca", - CONFIG.CERT_TRUST_ROOT_CA + CONFIG.CERT_TRUST_ROOT_CA, + "--ip", + self.apis[SECURE_API_KEY]["ip"] ] + args + ["{}:{}".format(self.apis[SECURE_API_KEY]["hostname"], self.apis[SECURE_API_KEY]["port"])] ) diff --git a/nmostesting/suites/IS0401Test.py b/nmostesting/suites/IS0401Test.py index 64c53e67..1325b1cb 100644 --- a/nmostesting/suites/IS0401Test.py +++ b/nmostesting/suites/IS0401Test.py @@ -32,10 +32,6 @@ from ..IS04Utils import IS04Utils from ..TestHelper import get_default_ip, is_ip_address, load_resolved_schema, check_content_type -# monkey patch zeroconf to allow us to advertise "_nmos-registration._tcp" -from zeroconf import service_type_name -service_type_name.__kwdefaults__['strict'] = False - NODE_API_KEY = "node" RECEIVER_CAPS_KEY = "receiver-caps" CAPS_REGISTER_KEY = "caps-register" @@ -94,6 +90,10 @@ def tear_down_tests(self): if self.dns_server: self.dns_server.reset() + def _strict_service_name(self, info): + # avoid zeroconf._exceptions.BadTypeInNameException: Service name (nmos-registration) must be <= 15 bytes + return len(info.type[1:info.type.find('.')]) <= 15 + def _mdns_info(self, port, service_type, txt={}, api_ver=None, api_proto=None, api_auth=None, ip=None): """Get an mDNS ServiceInfo object in order to create an advertisement""" if api_ver is None: @@ -192,9 +192,9 @@ def do_registry_basics_prereqs(self): if CONFIG.DNS_SD_MODE == "multicast": # Advertise the primary registry and invalid ones at pri 0, and allow the Node to do a basic registration if self.is04_utils.compare_api_version(self.apis[NODE_API_KEY]["version"], "v1.0") != 0: - self.zc.register_service(registry_mdns[0]) - self.zc.register_service(registry_mdns[1]) - self.zc.register_service(registry_mdns[2]) + self.zc.register_service(registry_mdns[0], strict=self._strict_service_name(registry_mdns[0])) + self.zc.register_service(registry_mdns[1], strict=self._strict_service_name(registry_mdns[1])) + self.zc.register_service(registry_mdns[2], strict=self._strict_service_name(registry_mdns[2])) # Wait for n seconds after advertising the service for the first POST from a Node start_time = time.time() @@ -226,7 +226,7 @@ def do_registry_basics_prereqs(self): if CONFIG.DNS_SD_MODE == "multicast": for info in registry_mdns[3:]: - self.zc.register_service(info) + self.zc.register_service(info, strict=self._strict_service_name(info)) # Kill registries one by one to collect data around failover self.invalid_registry.disable() @@ -1455,7 +1455,7 @@ def test_21(self, test): if CONFIG.DNS_SD_MODE == "multicast": # Advertise a registry at pri 0 and allow the Node to do a basic registration - self.zc.register_service(registry_info) + self.zc.register_service(registry_info, strict=self._strict_service_name(registry_info)) # Wait for n seconds after advertising the service for the first POST and then DELETE from a Node self.primary_registry.wait_for_registration(CONFIG.DNS_SD_ADVERT_TIMEOUT) @@ -2160,7 +2160,7 @@ def collect_mdns_announcements(self): if CONFIG.DNS_SD_MODE == "multicast": # Advertise a registry at pri 0 and allow the Node to do a basic registration - self.zc.register_service(registry_info) + self.zc.register_service(registry_info, strict=self._strict_service_name(registry_info)) # Wait for n seconds after advertising the service for the first POST from a Node self.primary_registry.wait_for_registration(CONFIG.DNS_SD_ADVERT_TIMEOUT) diff --git a/nmostesting/suites/IS1101Test.py b/nmostesting/suites/IS1101Test.py index f9b625dd..be103ffb 100644 --- a/nmostesting/suites/IS1101Test.py +++ b/nmostesting/suites/IS1101Test.py @@ -12,26 +12,27 @@ # See the License for the specific language governing permissions and # limitations under the License. +from functools import partial import time import re from requests.compat import json from ..NMOSUtils import NMOSUtils -from ..GenericTest import GenericTest +from ..GenericTest import GenericTest, NMOSInitException, NMOSTestException from .. import TestHelper from .. import Config as CONFIG from ..IS04Utils import IS04Utils from ..IS05Utils import IS05Utils import requests import datetime -from ..IS11Utils import IS11Utils, SND_RCV_SUBSET +from ..IS11Utils import IS11Utils COMPAT_API_KEY = "streamcompatibility" CONTROLS = "controls" NODE_API_KEY = "node" CONN_API_KEY = "connection" -VALID_EDID = "test_data/IS1101/valid_edid.bin" -INVALID_EDID = "test_data/IS1101/invalid_edid.bin" +VALID_EDID_PATH = "test_data/IS1101/valid_edid.bin" +INVALID_EDID_PATH = "test_data/IS1101/invalid_edid.bin" REF_SUPPORTED_CONSTRAINTS_VIDEO = [ "urn:x-nmos:cap:meta:label", @@ -64,7 +65,6 @@ def __init__(self, apis, **kwargs): # Don't auto-test paths responding with an EDID binary as they don't have a JSON Schema omit_paths = [ "/single/senders/{senderId}/transportfile", - "/inputs/{inputId}/edid", "/inputs/{inputId}/edid/base", "/inputs/{inputId}/edid/effective", "/outputs/{outputId}/edid" @@ -77,8 +77,6 @@ def __init__(self, apis, **kwargs): self.not_edid_connected_outputs = [] self.edid_connected_outputs = [] self.reference_senders = {} - self.receivers_with_outputs = [] - self.receivers_without_outputs = [] self.flow = "" self.caps = "" self.flow_format = {} @@ -95,13 +93,7 @@ def __init__(self, apis, **kwargs): self.constraints = {} self.some_input = {} self.input_senders = [] - self.connected_inputs = [] - self.disconnected_input = [] self.not_active_connected_inputs = [] - self.not_edid_connected_inputs = [] - self.edid_connected_inputs = [] - self.support_base_edid = {} - self.default_edid = {} self.another_grain_rate_constraints = {} self.another_sample_rate_constraints = {} self.not_input_senders = [] @@ -125,18 +117,58 @@ def build_output_properties_url(self, id): return self.compat_url + "outputs/" + id + "/properties/" def set_up_tests(self): + with open(INVALID_EDID_PATH, "rb") as f: + self.invalid_edid = f.read() + + with open(VALID_EDID_PATH, "rb") as f: + self.valid_edid = f.read() + self.senders = self.is11_utils.get_senders() self.receivers = self.is11_utils.get_receivers() - self.receivers_with_outputs = self.is11_utils.get_receivers(SND_RCV_SUBSET.WITH_I_O) - self.receivers_without_outputs = self.is11_utils.get_receivers(SND_RCV_SUBSET.WITHOUT_I_O) + + self.receivers_with_outputs = list(filter(self.receiver_has_i_o, self.receivers)) + self.receivers_without_outputs = list(set(self.receivers) - set(self.receivers_with_outputs)) + self.inputs = self.is11_utils.get_inputs() self.outputs = self.is11_utils.get_outputs() + self.connected_inputs = list(filter(self.is_input_connected, self.inputs)) + self.disconnected_inputs = list(set(self.inputs) - set(self.connected_inputs)) + + self.edid_connected_inputs = list(filter(self.has_input_edid_support, self.connected_inputs)) + self.not_edid_connected_inputs = list(set(self.connected_inputs) - set(self.edid_connected_inputs)) + + self.edid_inputs = list(filter(self.has_input_edid_support, self.inputs)) + self.non_edid_inputs = list(set(self.inputs) - set(self.edid_inputs)) + + self.base_edid_inputs = list(filter(self.has_input_base_edid_support, self.edid_inputs)) + self.adjust_to_caps_inputs = list(filter(self.is_input_adjust_to_caps, self.base_edid_inputs)) + + self.connected_outputs = list(filter(self.is_output_connected, self.outputs)) + self.disconnected_outputs = list(set(self.outputs) - set(self.connected_outputs)) + + self.edid_outputs = list(filter(self.has_output_edid_support, self.outputs)) + self.non_edid_outputs = list(set(self.outputs) - set(self.edid_outputs)) + + self.edid_connected_outputs = list(filter(self.has_output_edid_support, self.connected_outputs)) + self.edid_disconnected_outputs = list(filter(self.has_output_edid_support, self.disconnected_outputs)) + self.state_no_essence = "no_essence" self.state_awaiting_essence = "awaiting_essence" + self.deactivate_connection_resources("sender") + self.deactivate_connection_resources("receiver") + + self.delete_active_constraints() + self.delete_base_edid() + + def tear_down_tests(self): + for inputId in self.is11_utils.sampled_list(self.base_edid_inputs): + # DELETE the Base EDID of the Input + self.do_request("DELETE", self.compat_url + "inputs/" + inputId + "/edid/base") + # GENERAL TESTS - def test_00_01(self, test): + def test_00_00(self, test): """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"] @@ -148,127 +180,16 @@ def test_00_01(self, test): self.authorization ) - def test_00_02(self, test): - "Put all senders into inactive state" - senders_url = self.conn_url + "single/senders/" - valid, response = TestHelper.do_request("GET", senders_url) - if not valid: - return test.FAIL("Unexpected response from the Connection API: {}".format(response)) - if response.status_code != 200: - return test.FAIL("The request {} has failed {}".format(senders_url, response.json())) - senders = response.json() - if len(senders) > 0: - for sender in senders: - url = senders_url + sender + "staged/" - deactivate_json = { - "master_enable": False, - "activation": {"mode": "activate_immediate"}, - } - - valid_patch, response = TestHelper.do_request("PATCH", url, json=deactivate_json) - if not valid_patch: - return test.FAIL("Unexpected response from the Connection API: {}".format(response)) - if ( - response.status_code != 200 - or response.json()["master_enable"] - or response.json()["activation"]["mode"] != "activate_immediate" - ): - return test.FAIL("The patch request to {} has failed: {}".format(url, response.json())) - return test.PASS() - return test.UNCLEAR("Could not find any senders to test") - - def test_00_03(self, test): - "Put all the receivers into inactive state" - receivers_url = self.conn_url + "single/receivers/" - valid, response = TestHelper.do_request("GET", receivers_url) - if not valid: - return test.FAIL("Unexpected response from the connection API: {}".format(response)) - if response.status_code != 200: - return test.FAIL("The connection request {} has failed: {}".format(receivers_url, response.json())) - receivers = response.json() - if len(receivers) > 0: - for receiver in receivers: - url = receivers_url + receiver + "staged/" - deactivate_json = { - "master_enable": False, - "activation": {"mode": "activate_immediate"}, - } - valid_patch, response = TestHelper.do_request("PATCH", url, json=deactivate_json) - if not valid_patch: - return test.FAIL("Unexpected response from the Connection API: {}".format(response)) - if ( - response.status_code != 200 - or response.json()["master_enable"] - or response.json()["activation"]["mode"] != "activate_immediate" - ): - return test.FAIL("The patch request to {} has failed: {}".format(url, response.json())) - - return test.PASS() - - return test.UNCLEAR("Could not find any receivers to test") - # INPUTS TESTS def test_01_00(self, test): - """ - Reset the active constraints of all the senders such that the base EDID is the effective EDID - """ - - if len(self.senders) > 0: - for sender_id in self.senders: - valid, response = TestHelper.do_request( - "DELETE", - self.build_constraints_active_url(sender_id), - ) - if not valid: - return test.FAIL("Unexpected response from the streamcompatibility API: {}".format(response)) - if response.status_code != 200: - return test.FAIL("The sender {} constraints cannot be deleted".format(sender_id)) - return test.PASS() - return test.UNCLEAR("There are no IS-11 senders") - - def test_01_01(self, test): - """ - Verify that the device supports the concept of Input. - """ - - if len(self.inputs) == 0: - return test.UNCLEAR("No inputs") - return test.PASS() - - def test_01_02(self, test): - """ - Verify that some of the inputs of the device are connected - """ - if len(self.inputs) != 0: - for input in self.inputs: - valid, response = TestHelper.do_request( - "GET", self.compat_url + "inputs/" + input + "/properties/" - ) - if not valid: - return test.FAIL("Unexpected response from the streamcompatibility API: {}".format(response)) - if response.status_code != 200: - return test.FAIL("The input {} properties streamcompatibility request has failed: {}" - .format(input, response)) - try: - if response.json()["connected"]: - self.connected_inputs.append(response.json()) - except json.JSONDecodeError: - return test.FAIL("Non-JSON response returned from Node API") - except KeyError as e: - return test.FAIL("Unable to find expected key: {}".format(e)) - if len(self.connected_inputs) == 0: - return test.UNCLEAR("inputs are not connected") - return test.PASS() - return test.UNCLEAR("No resources found to perform this test") - - def test_01_03(self, test): """ Verify that all connected inputs have a signal """ if len(self.connected_inputs) != 0: for connectedInput in self.connected_inputs: - state = connectedInput["status"]["state"] - id = connectedInput["id"] + input = self.get_json(test, self.compat_url + "inputs/" + connectedInput + "/properties/") + state = input["status"]["state"] + id = input["id"] if state == "no_signal" or state == "awaiting_signal": if state == "awaiting_signal": for i in range(0, CONFIG.STABLE_STATE_ATTEMPTS): @@ -294,7 +215,7 @@ def test_01_03(self, test): if state == "awaiting_signal": return test.FAIL("Expected state of input {} is \"awaiting_signal\", got \"{}\"" .format(id, state)) - self.not_active_connected_inputs.append(connectedInput) + self.not_active_connected_inputs.append(input) if len(self.not_active_connected_inputs) != 0: for input in self.not_active_connected_inputs: self.connected_inputs.remove(input) @@ -303,585 +224,291 @@ def test_01_03(self, test): return test.UNCLEAR("No connected input have a signal") return test.UNCLEAR("No resources found to perform this test") - def test_01_04_00(self, test): - """ - Verify that connected inputs supporting EDID behave according to the RAML file - """ - if len(self.connected_inputs) != 0: - for connectedInput in self.connected_inputs: - id = connectedInput["id"] - valid, response = TestHelper.do_request( - "GET", self.compat_url + "inputs/" + id + "/properties/" - ) - if not valid: - return test.FAIL("Unexpected response from the streamcompatibility API: {}".format(response)) - if response.status_code != 200: - return test.FAIL("The input {} properties streamcompatibility request has failed: {}" - .format(id, response)) - try: - if response.json()["edid_support"]: - self.edid_connected_inputs.append(response.json()["id"]) - except json.JSONDecodeError: - return test.FAIL("Non-JSON response returned from Node API") - except KeyError as e: - return test.FAIL("Unable to find expected key: {}".format(e)) - if len(self.edid_connected_inputs) != 0: - self.test_01_04_pass = True - return test.PASS() - return test.UNCLEAR("No resources found to perform this test") - return test.UNCLEAR("No resources found to perform this test") - - def test_01_04_01(self, test): - """ - Verify that an input indicating EDID support behaves according to the RAML file. - """ - if len(self.connected_inputs) != 0 and len(self.edid_connected_inputs) != 0: - for input_id in self.edid_connected_inputs: - valid, response = TestHelper.do_request( - "GET", self.compat_url + "inputs/" + input_id + "/edid/" - ) - if not valid: - return test.FAIL("Unexpected response from the streamcompatibility API: {}".format(response)) - if response.status_code != 200: - return test.FAIL("The input {} edid streamcompatibility request has failed: {}" - .format(input_id, response)) - return test.PASS() - return test.UNCLEAR("No resources found to perform this test") - - def test_01_04_02(self, test): - """ - Verify that a valid EDID can be retrieved from the device; - this EDID represents the default EDID of the device - """ - - if len(self.connected_inputs) != 0 and len(self.edid_connected_inputs) != 0: - - for input_id in self.edid_connected_inputs: - valid, response = TestHelper.do_request( - "GET", self.compat_url + "inputs/" + input_id + "/edid/effective/" - ) - if not valid: - return test.FAIL("Unexpected response from the streamcompatibility API: {}".format(response)) - if ( - response.status_code != 200 - and response.headers["Content-Type"] != "application/octet-stream" - ): - return test.FAIL("The input {} edid effective streamcompatibility request has failed: {}" - .format(input_id, response)) - self.default_edid[input_id] = response.content - return test.PASS() - - return test.UNCLEAR("No resources found to perform this test") - - def test_01_04_03(self, test): - """ - Verify if the device supports changing the base EDID - """ - if len(self.connected_inputs) != 0 and len(self.edid_connected_inputs) != 0: - for input_id in self.edid_connected_inputs: - valid, response = TestHelper.do_request( - "DELETE", self.compat_url + "inputs/" + input_id + "/edid/base/" - ) - if not valid: - return test.FAIL("Unexpected response from the streamcompatibility API: {}".format(response)) - if response.status_code == 405: - return test.UNCLEAR( - "device does not support changing the base EDID" - ) - if response.status_code == 204: - self.support_base_edid[input_id] = True - else: - return test.FAIL("The input {} base edid cannot be deleted".format(input_id)) - - return test.PASS() - return test.UNCLEAR("No resources found to perform this test") - - def test_01_04_04(self, test): - """ - Verify that there is no base EDID after a delete - """ - if len(self.connected_inputs) != 0 and len(self.edid_connected_inputs) != 0: - for input_id in self.edid_connected_inputs: - valid, response = TestHelper.do_request( - "GET", self.compat_url + "inputs/" + input_id + "/edid/base/" - ) - if not valid: - return test.FAIL("Unexpected response from the streamcompatibility API: {}".format(response)) - if response.status_code != 204: - return test.FAIL("The input {} edid base streamcompatibility request has failed: {}" - .format(input_id, response)) - return test.PASS() - return test.UNCLEAR("No resources found to perform this test") - - def test_01_04_05(self, test): - """ - Verify that a valid base EDID can be put - """ - if len(self.connected_inputs) != 0 and len(self.edid_connected_inputs) != 0: - for input_id in self.edid_connected_inputs: - file = open(VALID_EDID, "rb") - response = requests.put( - self.compat_url + "inputs/" + input_id + "/edid/base/", - data=file, - headers={"Content-Type": "application/octet-stream"}, - ) - file.close() - if response.status_code != 204: - return test.FAIL("The input {} edid base change has failed: {}".format(input_id, response)) - return test.PASS() - return test.UNCLEAR("No resources found to perform this test") - - def test_01_04_06(self, test): - """ - Verify that the last PUT base EDID can be retrieved - """ - if len(self.connected_inputs) != 0 and len(self.edid_connected_inputs) != 0: - for input_id in self.edid_connected_inputs: - valid, response = TestHelper.do_request( - "GET", self.compat_url + "inputs/" + input_id + "/edid/base/" - ) - if not valid: - return test.FAIL("Unexpected response from the streamcompatibility API: {}".format(response)) - if response.status_code != 200: - return test.FAIL("The input {} edid base streamcompatibility request has failed: {}" - .format(input_id, response)) - if response.content != open(VALID_EDID, "rb").read(): - return test.FAIL("Edid files does'nt match") - return test.PASS() - return test.UNCLEAR("No resources found to perform this test") - - def test_01_04_07(self, test): - """ - Verify that the base EDID without constraints is visible as the effective EDID - """ - if len(self.connected_inputs) != 0 and len(self.edid_connected_inputs) != 0: - for input_id in self.edid_connected_inputs: - time.sleep(CONFIG.STABLE_STATE_DELAY) - valid, response = TestHelper.do_request( - "GET", self.compat_url + "inputs/" + input_id + "/edid/effective/" - ) - if not valid: - return test.FAIL("Unexpected response from the streamcompatibility API: {}".format(response)) - if response.status_code != 200: - return test.FAIL("The input {} edid effective streamcompatibility request has failed: {}" - .format(input_id, response)) - if response.content != open(VALID_EDID, "rb").read(): - return test.FAIL("Edid files does'nt match") - return test.PASS() - return test.UNCLEAR("No resources found to perform this test") - - def test_01_04_08(self, test): - """ - Verify that the base EDID can be deleted - """ - if len(self.connected_inputs) != 0 and len(self.edid_connected_inputs) != 0: - for input_id in self.edid_connected_inputs: - valid, response = TestHelper.do_request( - "DELETE", self.compat_url + "inputs/" + input_id + "/edid/base/" - ) - if not valid: - return test.FAIL("Unexpected response from the streamcompatibility API: {}".format(response)) - if response.status_code != 204: - return test.FAIL("The input {} base edid cannot be deleted".format(input_id)) - return test.PASS() - return test.UNCLEAR("No resources found to perform this test") - - def test_01_04_09(self, test): - """ - Verify that the base EDID is properly deleted - """ - if len(self.connected_inputs) != 0 and len(self.edid_connected_inputs) != 0: - for input_id in self.edid_connected_inputs: - valid, response = TestHelper.do_request( - "GET", self.compat_url + "inputs/" + input_id + "/edid/base/" - ) - if not valid: - return test.FAIL("Unexpected response from the streamcompatibility API: {}".format(response)) - if response.status_code != 204: - return test.FAIL("The input {} edid base streamcompatibility request has failed: {}" - .format(input_id, response)) - return test.PASS() - return test.UNCLEAR("No resources found to perform this test") - - def test_01_04_10(self, test): - """ - Verify that the default EDID becomes visible again after deleting the base EDID - """ - if len(self.connected_inputs) != 0 and len(self.edid_connected_inputs) != 0: - for input_id in self.edid_connected_inputs: - time.sleep(CONFIG.STABLE_STATE_DELAY) - valid, response = TestHelper.do_request( - "GET", self.compat_url + "inputs/" + input_id + "/edid/effective/" - ) - if not valid: - return test.FAIL("Unexpected response from the streamcompatibility API: {}".format(response)) - if response.status_code != 200: - return test.FAIL("The input {} edid effective streamcompatibility request has failed: {}" - .format(input_id, response)) - if response.content != self.default_edid[input_id]: - return test.FAIL("Edid files does'nt match") - return test.PASS() - return test.UNCLEAR("No resources found to perform this test") - - def test_01_04_11(self, test): - """ - Verify that a put of an invalid EDID fail - """ - if len(self.connected_inputs) != 0 and len(self.edid_connected_inputs) != 0: - for input_id in self.edid_connected_inputs: - file = open(INVALID_EDID, "rb") - response = requests.put( - self.compat_url + "inputs/" + input_id + "/edid/base/", - data=file, - headers={"Content-Type": "application/octet-stream"}, - ) - file.close() - if response.status_code != 400: - return test.FAIL("The input {} edid base change has failed: {}".format(input_id, response)) - return test.PASS() - return test.UNCLEAR("No resources found to perform this test") - - def test_01_05_00(self, test): - """ - Verify that connected inputs not supporting EDID behave according to the RAML file - """ - if len(self.connected_inputs) != 0: - for connectedInput in self.connected_inputs: - id = connectedInput["id"] - valid, response = TestHelper.do_request( - "GET", self.compat_url + "inputs/" + id + "/properties/" - ) - if not valid: - return test.FAIL("Unexpected response from the streamcompatibility API: {}".format(response)) - if response.status_code != 200: - return test.FAIL("The input {} properties streamcompatibility request has failed: {}" - .format(id, response.json())) - try: - if not response.json()["edid_support"]: - self.not_edid_connected_inputs.append(response.json()["id"]) - except json.JSONDecodeError: - return test.FAIL("Non-JSON response returned from Node API") - except KeyError as e: - return test.FAIL("Unable to find expected key: {}".format(e)) - if len(self.not_edid_connected_inputs) != 0: - return test.PASS() - return test.UNCLEAR("No connected inputs not supporting EDID ") - return test.UNCLEAR("No resources found to perform this test") - - def test_01_05_01(self, test): - """ - Verify that there is no EDID support - TODO: Remove, duplicates test_01_05_02 - """ - if len(self.connected_inputs) != 0 and len(self.not_edid_connected_inputs) != 0: - - for input_id in self.not_edid_connected_inputs: - valid, response = TestHelper.do_request( - "GET", self.compat_url + "inputs/" + input_id + "/edid/effective/" - ) - if not valid: - return test.FAIL("Unexpected response from the streamcompatibility API: {}".format(response)) - if response.status_code != 204: - return test.FAIL("The input {} edid effective streamcompatibility request has failed: {}" - .format(input_id, response)) - return test.PASS() + def test_01_01(self, test): + """Inputs with EDID support return the Effective EDID""" + if len(self.edid_inputs) == 0: + return test.UNCLEAR("Not tested. No inputs with EDID support found.") - return test.UNCLEAR("No resources found to perform this test") + for inputId in self.is11_utils.sampled_list(self.edid_inputs): + self.get_effective_edid(test, inputId) - def test_01_05_02(self, test): - """ - Verify that there is no effective EDID - """ - if len(self.connected_inputs) != 0 and len(self.not_edid_connected_inputs) != 0: + return test.PASS() - for input_id in self.not_edid_connected_inputs: - valid, response = TestHelper.do_request( - "GET", self.compat_url + "inputs/" + input_id + "/edid/effective/" - ) - if not valid: - return test.FAIL("Unexpected response from the streamcompatibility API: {}".format(response)) - if response.status_code != 204: - return test.FAIL("The input {} edid effective streamcompatibility request has failed: {}" - .format(input_id, response)) - return test.PASS() + def test_01_02(self, test): + """Inputs with Base EDID support handle PUTting and DELETing the Base EDID""" + def is_edid_equal_to_effective_edid(self, test, inputId, edid): + return self.get_effective_edid(test, inputId) == edid + + if len(self.base_edid_inputs) == 0: + return test.UNCLEAR("Not tested. No inputs with Base EDID support found.") + + for inputId in self.is11_utils.sampled_list(self.base_edid_inputs): + # Save the default value of the Effective EDID + default_edid = self.get_effective_edid(test, inputId) + + # PUT the Base EDID to the Input + valid, response = self.do_request("PUT", + self.compat_url + "inputs/" + inputId + "/edid/base", + headers={"Content-Type": "application/octet-stream"}, + data=self.valid_edid) + if not valid or response.status_code != 204: + return test.FAIL("Unexpected response from " + "the Stream Compatibility Management API: {}".format(response)) - return test.UNCLEAR("No resources found to perform this test") + # Verify that /edid/base returns the last Base EDID put + valid, response = self.do_request("GET", self.compat_url + "inputs/" + inputId + "/edid/base") + if ( + not valid + or response.status_code != 200 + or response.headers["Content-Type"] != "application/octet-stream" + ): + return test.FAIL("Unexpected response from " + "the Stream Compatibility Management API: {}".format(response)) + if response.content != self.valid_edid: + return test.FAIL("The Base EDID of Input {} " + "doesn't match the Base EDID that has been put".format(inputId)) - def test_01_05_03(self, test): - """ - Verify that there is no base EDID (DELETE failure) - """ - if len(self.connected_inputs) != 0 and len(self.not_edid_connected_inputs) != 0: + # Verify that /edid/effective returns the last Base EDID put + result = self.wait_until_true( + partial(is_edid_equal_to_effective_edid, self, test, inputId, self.valid_edid) + ) + if not result: + return test.FAIL("The Effective EDID of Input {} " + "doesn't match the Base EDID that has been put".format(inputId)) - for input_id in self.not_edid_connected_inputs: - valid, response = TestHelper.do_request( - "DELETE", self.compat_url + "inputs/" + input_id + "/edid/base/" - ) - if not valid: - return test.FAIL("Unexpected response from the streamcompatibility API: {}".format(response)) - if response.status_code != 405: - return test.FAIL("The input {} edid base streamcompatibility request has failed: {}" - .format(input_id, response)) - return test.PASS() + # Delete the Base EDID + valid, response = self.do_request("DELETE", self.compat_url + "inputs/" + inputId + "/edid/base") + if not valid or response.status_code != 204: + return test.FAIL("Unexpected response from " + "the Stream Compatibility Management API: {}".format(response)) - return test.UNCLEAR("No resources found to perform this test") + # Verify that the Base EDID is properly deleted + valid, response = self.do_request("GET", self.compat_url + "inputs/" + inputId + "/edid/base") + if not valid or response.status_code != 204: + return test.FAIL("Unexpected response from " + "the Stream Compatibility Management API: {}".format(response)) - def test_01_05_04(self, test): - """ - Verify that there is no base EDID (PUT failure) - """ - if len(self.connected_inputs) != 0 and len(self.not_edid_connected_inputs) != 0: - for input_id in self.not_edid_connected_inputs: - file = open(VALID_EDID, "rb") - response = requests.put( - self.compat_url + "inputs/" + input_id + "/edid/base/", - data=file, - headers={"Content-Type": "application/octet-stream"}, - ) - file.close() - if response.status_code != 405: - return test.FAIL("The input {} edid base change has failed: {}".format(input_id, response)) - return test.PASS() - return test.UNCLEAR("No resources found to perform this test") + # Verify that /edid/effective returned to its defaults + result = self.wait_until_true(partial(is_edid_equal_to_effective_edid, self, test, inputId, default_edid)) + if not result: + return test.FAIL("The Effective EDID of Input {} " + "doesn't match its initial value".format(inputId)) - def test_01_05_05(self, test): - """ - Verify that there is no base EDID (GET failure) - """ - if len(self.connected_inputs) != 0 and len(self.not_edid_connected_inputs) != 0: + return test.PASS() - for input_id in self.not_edid_connected_inputs: - valid, response = TestHelper.do_request( - "GET", self.compat_url + "inputs/" + input_id + "/edid/base/" - ) - if not valid: - return test.FAIL("Unexpected response from the streamcompatibility API: {}".format(response)) - if response.status_code != 204: - return test.FAIL("The input {} edid base streamcompatibility request has failed: {}" - .format(input_id, response)) - return test.PASS() + def test_01_03(self, test): + """Inputs with Base EDID support reject an invalid EDID""" + if len(self.base_edid_inputs) == 0: + return test.UNCLEAR("Not tested. No inputs with Base EDID support found.") + + for inputId in self.is11_utils.sampled_list(self.base_edid_inputs): + valid, response = self.do_request("PUT", + self.compat_url + "inputs/" + inputId + "/edid/base", + headers={"Content-Type": "application/octet-stream"}, + data=self.invalid_edid) + if not valid or response.status_code != 400: + return test.FAIL("Unexpected response from " + "the Stream Compatibility Management API: {}".format(response)) + return test.PASS() - return test.UNCLEAR("No resources found to perform this test") + def test_01_04(self, test): + """Inputs without EDID support reject requests to /edid/*""" + if len(self.non_edid_inputs) == 0: + return test.UNCLEAR("Not tested. No inputs without EDID support found.") + + for inputId in self.is11_utils.sampled_list(self.non_edid_inputs): + valid, response = self.do_request("GET", self.compat_url + "inputs/" + inputId + "/edid/effective") + if not valid or response.status_code != 204: + return test.FAIL("Unexpected response " + "for GET /edid/effective: {}".format(response)) + + valid, response = self.do_request("GET", self.compat_url + "inputs/" + inputId + "/edid/base") + if not valid or response.status_code != 204: + return test.FAIL("Unexpected response " + "for GET /edid/base: {}".format(response)) + + valid, response = self.do_request("DELETE", self.compat_url + "inputs/" + inputId + "/edid/base") + if not valid or response.status_code != 405: + return test.FAIL("Unexpected response " + "for DELETE /edid/base: {}".format(response)) + + valid, response = self.do_request("PUT", + self.compat_url + "inputs/" + inputId + "/edid/base", + headers={"Content-Type": "application/octet-stream"}, + data=self.valid_edid) + if not valid or response.status_code != 405: + return test.FAIL("Unexpected response " + "for PUT /edid/base: {}".format(response)) + return test.PASS() - def test_01_06_01(self, test): + def test_01_05(self, test): """ - Verify that the input supports changing the base EDID which is optional from the specification. + Inputs with Base EDID increment their version and versions of associated Senders + after the Base EDID gets modified """ - if len(self.connected_inputs) != 0 and len(self.edid_connected_inputs) != 0: - for input_id in self.edid_connected_inputs: - if ( - input_id in self.support_base_edid - and not self.support_base_edid[input_id] - ): - continue - return test.PASS() - return test.UNCLEAR("No resources found to perform this test") + if len(self.base_edid_inputs) == 0: + return test.UNCLEAR("Not tested. No inputs with Base EDID support found.") - def test_01_06_02(self, test): - """ - Verify that the Input resource version changes when putting/deleting base EDID - """ - if len(self.connected_inputs) != 0 and len(self.edid_connected_inputs) != 0: - for input_id in self.edid_connected_inputs: - valid, response = TestHelper.do_request( - "GET", self.compat_url + "inputs/" + input_id + "/properties/" - ) - if not valid: - return test.FAIL("Unexpected response from the streamcompatibility API: {}".format(response)) - if response.status_code != 200: - return test.FAIL("The input {} properties streamcompatibility request has failed: {}" - .format(input_id, response)) - try: - version = response.json()["version"] - except json.JSONDecodeError: - return test.FAIL("Non-JSON response returned from Node API") - except KeyError as e: - return test.FAIL("Unable to find expected key: {}".format(e)) - self.version[input_id] = version + for inputId in self.is11_utils.sampled_list(self.base_edid_inputs): + sender_ids = self.get_inputs_senders(test, inputId) - file = open(VALID_EDID, "rb") - response = requests.put( - self.compat_url + "inputs/" + input_id + "/edid/base/", - data=file, - headers={"Content-Type": "application/octet-stream"}, - ) - file.close() - if response.status_code != 204: - return test.FAIL("The input {} edid base change has failed: {}".format(input_id, response.json())) - time.sleep(CONFIG.STABLE_STATE_DELAY) - valid, response = TestHelper.do_request( - "GET", self.compat_url + "inputs/" + input_id + "/properties/" - ) - if not valid: - return test.FAIL("Unexpected response from the streamcompatibility API: {}".format(response)) - if response.status_code != 200: - return test.FAIL("The input {} properties streamcompatibility request has failed: {}" - .format(input_id, response)) - try: - version = response.json()["version"] - except json.JSONDecodeError: - return test.FAIL("Non-JSON response returned from Node API") - except KeyError as e: - return test.FAIL("Unable to find expected key: {}".format(e)) - if version == self.version[input_id]: - return test.FAIL("Version doesn't change") - self.version[input_id] = version + in_version_1 = "" + in_version_2 = "" + in_version_3 = "" - valid, response = TestHelper.do_request( - "DELETE", self.compat_url + "inputs/" + input_id + "/edid/base/" - ) - if not valid: - return test.FAIL("Unexpected response from the streamcompatibility API: {}".format(response)) - if response.status_code != 204: - return test.FAIL("The input {} base edid cannot be deleted".format(input_id)) - time.sleep(CONFIG.STABLE_STATE_DELAY) - valid, response = TestHelper.do_request( - "GET", self.compat_url + "inputs/" + input_id + "/properties/" - ) - if not valid: - return test.FAIL("Unexpected response from the streamcompatibility API: {}".format(response)) - if response.status_code != 200: - return test.FAIL("The input {} properties streamcompatibility request has failed: {}" - .format(input_id, response.json())) - try: - version = response.json()["version"] - except json.JSONDecodeError: - return test.FAIL("Non-JSON response returned from Node API") - except KeyError as e: - return test.FAIL("Unable to find expected key: {}".format(e)) - if version == self.version[input_id]: - return test.FAIL("Version does'nt change") - return test.PASS() - return test.UNCLEAR("No resources found to perform this test") + snd_versions_1 = {} + snd_versions_2 = {} + snd_versions_3 = {} - def test_01_07_01(self, test): - """ - Verify that the input supports changing the base EDID - """ - if len(self.connected_inputs) != 0 and len(self.edid_connected_inputs) != 0: - for input_id in self.edid_connected_inputs: - if ( - input_id in self.support_base_edid - and not self.support_base_edid[input_id] - ): - continue - return test.PASS() - return test.UNCLEAR("No resources found to perform this test") + for sender_id in sender_ids: + snd_versions_1[sender_id] = "" + snd_versions_2[sender_id] = "" + snd_versions_3[sender_id] = "" - def test_01_07_02(self, test): - """ - Verify that the input is associated with a device - """ - if len(self.connected_inputs) != 0 and len(self.edid_connected_inputs) != 0: - for input_id in self.edid_connected_inputs: - valid, response = TestHelper.do_request( - "GET", self.compat_url + "inputs/" + input_id + "/properties/" - ) - if not valid: - return test.FAIL("Unexpected response from the streamcompatibility API: {}".format(response)) - if response.status_code != 200: - return test.FAIL("The input {} properties streamcompatibility request has failed: {}" - .format(input_id, response.json())) - try: - input = response.json() - if not input["device_id"]: - return test.FAIL("no device_id") - except json.JSONDecodeError: - return test.FAIL("Non-JSON response returned from Node API") - except KeyError as e: - return test.FAIL("Unable to find expected key: {}".format(e)) - self.version[input_id] = input["version"] + valid, response = self.do_request("GET", self.compat_url + "inputs/" + inputId + "/properties") + if not valid or response.status_code != 200: + return test.FAIL("Unexpected response from " + "the Stream Compatibility Management API: {}".format(response)) + try: + in_version_1 = response.json()["version"] + except json.JSONDecodeError: + return test.FAIL("Non-JSON response returned from the Stream Compatibility Management API") + except KeyError as e: + return test.FAIL("Unable to find expected key: {}".format(e)) - valid, response = TestHelper.do_request( - "GET", self.node_url + "devices/" + input["device_id"] - ) - if not valid: + for sender_id in sender_ids: + valid, response = self.do_request("GET", self.node_url + "senders/" + sender_id) + if not valid or response.status_code != 200: return test.FAIL("Unexpected response from the Node API: {}".format(response)) - if response.status_code != 200: - return test.FAIL("The device {} Node API request has failed: {}" - .format(input["device_id"], response.json())) try: - device = response.json() - if device["id"] != input["device_id"]: - return test.FAIL("device_id does'nt match.") + snd_versions_1[sender_id] = response.json()["version"] except json.JSONDecodeError: - return test.FAIL("Non-JSON response returned from Node API") + return test.FAIL("Non-JSON response returned from the Node API") except KeyError as e: return test.FAIL("Unable to find expected key: {}".format(e)) - self.version[device["id"]] = input["version"] + # PUT the Base EDID to the Input + valid, response = self.do_request("PUT", + self.compat_url + "inputs/" + inputId + "/edid/base", + headers={"Content-Type": "application/octet-stream"}, + data=self.valid_edid) + if not valid or response.status_code != 204: + return test.FAIL("Unexpected response from " + "the Stream Compatibility Management API: {}".format(response)) - file = open(VALID_EDID, "rb") - response = requests.put( - self.compat_url + "inputs/" + input_id + "/edid/base/", - data=file, - headers={"Content-Type": "application/octet-stream"}, - ) - file.close() - time.sleep(CONFIG.STABLE_STATE_DELAY) - valid, response = TestHelper.do_request( - "GET", self.compat_url + "inputs/" + input_id + "/properties/" - ) - if not valid: - return test.FAIL("Unexpected response from the streamcompatibility API: {}".format(response)) - if response.status_code != 200: - return test.FAIL("The input {} properties streamcompatibility request has failed: {}" - .format(input_id, response.json())) + valid, response = self.do_request("GET", self.compat_url + "inputs/" + inputId + "/properties") + if not valid or response.status_code != 200: + return test.FAIL("Unexpected response from " + "the Stream Compatibility Management API: {}".format(response)) + try: + in_version_2 = response.json()["version"] + except json.JSONDecodeError: + return test.FAIL("Non-JSON response returned from the Stream Compatibility Management API") + except KeyError as e: + return test.FAIL("Unable to find expected key: {}".format(e)) + + for sender_id in sender_ids: + valid, response = self.do_request("GET", self.node_url + "senders/" + sender_id) + if not valid or response.status_code != 200: + return test.FAIL("Unexpected response from the Node API: {}".format(response)) try: - version = response.json()["version"] + snd_versions_2[sender_id] = response.json()["version"] except json.JSONDecodeError: - return test.FAIL("Non-JSON response returned from Node API") + return test.FAIL("Non-JSON response returned from the Node API") except KeyError as e: return test.FAIL("Unable to find expected key: {}".format(e)) - if version == self.version[input_id]: - return test.FAIL("Version should not match.") - valid, response = TestHelper.do_request( - "GET", self.node_url + "devices/" + device["id"] - ) - if not valid: - return test.FAIL("Unexpected response from the Node API: {}".format(response)) - if response.status_code != 200: - return test.FAIL("The device {} Node API request has failed: {}" - .format(device["id"], response.json())) - version = response.json()["version"] - if version == self.version[device["id"]]: - return test.FAIL("Version should not match.") - valid, response = TestHelper.do_request( - "DELETE", self.compat_url + "inputs/" + input_id + "/edid/base/" - ) - if not valid: - return test.FAIL("Unexpected response from the streamcompatibility API: {}".format(response)) - if response.status_code != 204: - return test.FAIL("The input {} base edid cannot be deleted".format(input_id)) - return test.PASS() - return test.UNCLEAR("No resources found to perform this test") + if in_version_2 == in_version_1: + return test.FAIL("Input {} didn't increment its version after PUTting the Base EDID".format(inputId)) + for sender_id in sender_ids: + if snd_versions_2[sender_id] == snd_versions_1[sender_id]: + return test.FAIL("Sender {} didn't increment its version " + "after PUTting the Base EDID to Input {}".format(sender_id, inputId)) - def test_01_08(self, test): - """ - Verify that disconnected inputs have a minimum of functionality - """ - if len(self.edid_connected_inputs) != 0: - for input_id in self.edid_connected_inputs: - valid, response = TestHelper.do_request( - "GET", self.compat_url + "inputs/" + input_id + "/properties/" - ) - if not valid: - return test.FAIL("Unexpected response from the streamcompatibility API: {}".format(response)) - if response.status_code != 200: - return test.FAIL("The input {} properties streamcompatibility request has failed: {}" - .format(input_id, response.json())) + # DELETE the Base EDID of the Input + valid, response = self.do_request("DELETE", self.compat_url + "inputs/" + inputId + "/edid/base") + if not valid or response.status_code != 204: + return test.FAIL("Unexpected response from " + "the Stream Compatibility Management API: {}".format(response)) + + valid, response = self.do_request("GET", self.compat_url + "inputs/" + inputId + "/properties") + if not valid or response.status_code != 200: + return test.FAIL("Unexpected response from " + "the Stream Compatibility Management API: {}".format(response)) + try: + in_version_3 = response.json()["version"] + except json.JSONDecodeError: + return test.FAIL("Non-JSON response returned from the Stream Compatibility Management API") + except KeyError as e: + return test.FAIL("Unable to find expected key: {}".format(e)) + + for sender_id in sender_ids: + valid, response = self.do_request("GET", self.node_url + "senders/" + sender_id) + if not valid or response.status_code != 200: + return test.FAIL("Unexpected response from the Node API: {}".format(response)) try: - if not response.json()["connected"]: - self.disconnected_input.append(response.json()) + snd_versions_3[sender_id] = response.json()["version"] except json.JSONDecodeError: - return test.FAIL("Non-JSON response returned from Node API") + return test.FAIL("Non-JSON response returned from the Node API") except KeyError as e: return test.FAIL("Unable to find expected key: {}".format(e)) - if len(self.disconnected_input) == 0: - return test.UNCLEAR("All inputs are connected") - return test.PASS() - return test.UNCLEAR("No resources found to perform this test") + + if in_version_3 == in_version_2: + return test.FAIL("Input {} didn't increment its version after DELETing the Base EDID".format(inputId)) + for sender_id in sender_ids: + if snd_versions_3[sender_id] == snd_versions_2[sender_id]: + return test.FAIL("Sender {} didn't increment its version " + "after PUTting the Base EDID to Input {}".format(sender_id, inputId)) + + return test.PASS() + + def test_01_06(self, test): + """Effective EDID updates if Base EDID changes with 'adjust_to_caps'""" + + def is_edid_equal_to_effective_edid(self, test, inputId, edid): + return self.get_effective_edid(test, inputId) == edid + + def is_edid_inequal_to_effective_edid(self, test, inputId, edid): + return self.get_effective_edid(test, inputId) != edid + + if len(self.adjust_to_caps_inputs) == 0: + return test.UNCLEAR("Not tested. No inputs with 'adjust_to_caps' support found.") + + for inputId in self.is11_utils.sampled_list(self.adjust_to_caps_inputs): + try: + effective_edid_before = self.get_effective_edid(test, inputId) + + valid, response = self.do_request("PUT", + self.compat_url + "inputs/" + inputId + "/edid/base", + headers={"Content-Type": "application/octet-stream"}, + data=self.valid_edid, + params={"adjust_to_caps": "true"}) + if not valid or response.status_code != 204: + return test.FAIL("Unexpected response from " + "the Stream Compatibility Management API: {}".format(response)) + + result = self.wait_until_true( + partial(is_edid_inequal_to_effective_edid, self, test, inputId, effective_edid_before) + ) + if not result: + return test.FAIL("Effective EDID doesn't change when Base EDID changes") + + valid, response = self.do_request("DELETE", self.compat_url + "inputs/" + inputId + "/edid/base") + if not valid or response.status_code != 204: + return test.FAIL("Unexpected response from " + "the Stream Compatibility Management API: {}".format(response)) + + result = self.wait_until_true( + partial(is_edid_equal_to_effective_edid, self, test, inputId, effective_edid_before) + ) + if not result: + return test.FAIL("Effective EDID doesn't restore after Base EDID DELETion") + + except json.JSONDecodeError: + return test.FAIL("Non-JSON response returned from Node API") + except KeyError as e: + return test.FAIL("Unable to find expected key: {}".format(e)) + return test.PASS() # SENDERS TESTS """ @@ -2344,7 +1971,7 @@ def test_02_03_01(self, test): return test.PASS() return test.UNCLEAR("No resources found to perform this test") - def test_02_03_02(self, test): + def _test_02_03_02(self, test): """ Verify that the input passed its test suite """ @@ -2460,7 +2087,7 @@ def test_02_03_04(self, test): for input_id in response.json(): if ( input_id in self.edid_connected_inputs - and input_id in self.support_base_edid + and input_id in self.base_edid_inputs ): inputs.append(input_id) else: @@ -2509,13 +2136,11 @@ def test_02_03_04(self, test): return test.FAIL("Unable to find expected key: {}".format(e)) self.version[sender_id] = version - file = open(VALID_EDID, "rb") response = requests.put( self.compat_url + "inputs/" + input_id + "/edid/base/", - data=file, + data=self.valid_edid, headers={"Content-Type": "application/octet-stream"}, ) - file.close() time.sleep(CONFIG.STABLE_STATE_DELAY) valid, response = TestHelper.do_request( @@ -2590,7 +2215,7 @@ def test_02_03_05_01(self, test): for input_id in response.json(): if ( input_id in self.edid_connected_inputs - and input_id in self.support_base_edid + and input_id in self.base_edid_inputs ): inputs.append(input_id) else: @@ -2654,6 +2279,8 @@ def test_02_03_05_01(self, test): self.version[sender_id] = version + default_edid = self.get_effective_edid(test, input_id) + self.another_grain_rate_constraints[sender_id] = { "constraint_sets": [ { @@ -2702,7 +2329,7 @@ def test_02_03_05_01(self, test): input_id, response ) ) - if response.content == self.default_edid[input_id]: + if response.content == default_edid: print("Grain rate constraint are not changing effective EDID") valid, response = TestHelper.do_request( @@ -2939,7 +2566,7 @@ def test_02_03_05_02(self, test): for input_id in response.json(): if ( input_id in self.edid_connected_inputs - and input_id in self.support_base_edid + and input_id in self.base_edid_inputs ): inputs.append(input_id) else: @@ -3001,6 +2628,8 @@ def test_02_03_05_02(self, test): return test.FAIL("Unable to find expected key: {}".format(e)) self.version[sender_id] = version + default_edid = self.get_effective_edid(test, input_id) + self.another_sample_rate_constraints[sender_id] = { "constraint_sets": [ { @@ -3049,7 +2678,7 @@ def test_02_03_05_02(self, test): input_id, response ) ) - if response.content == self.default_edid[input_id]: + if response.content == default_edid: print("Grain rate constraint are not changing effective EDID") valid, response = TestHelper.do_request( @@ -3347,167 +2976,32 @@ def test_02_04_01(self, test): return test.PASS() # OUTPUTS TESTS - def test_03_01(self, test): - """ - Verify that the device supports the concept of Output. - """ - if len(self.outputs) == 0: - return test.UNCLEAR("No outputs") - return test.PASS() - - 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 IS-11 outputs") - for output in self.outputs: - valid, response = TestHelper.do_request( - "GET", self.compat_url + "outputs/" + output + "/properties/" - ) - if not valid: - return test.FAIL("Unexpected response from the streamcompatibility API: {}".format(response)) - if response.status_code == 200: - outputs_properties_json = [] - outputs_properties_json.append(response.json()) - for output in outputs_properties_json: - if output["connected"]: - self.connected_outputs.append(output["id"]) - else: - return test.FAIL("The output {} properties streamcompatibility request has failed: {}" - .format(output, response.json())) - if len(self.connected_outputs) == 0: - return test.UNCLEAR("No connected outputs") - return test.PASS() - - def test_03_03(self, test): - """ - Verify that all connected outputs do not have - a signal as test 0 put all of the receivers inactive. - """ - if len(self.connected_outputs) == 0: - return test.UNCLEAR("No connected outputs") - - active_connected_outputs = [] + def test_03_00(self, test): + """Connected Outputs with EDID support return the EDID""" + if len(self.edid_connected_outputs) == 0: + return test.UNCLEAR("Not tested. No connected Outputs with EDID support found.") - for output_id in self.connected_outputs: - valid, response = TestHelper.do_request( - "GET", self.compat_url + "outputs/" + output_id + "/properties/" - ) - if not valid: - return test.FAIL("Unexpected response from the streamcompatibility API: {}".format(response)) - if response.status_code == 200: - if response.json()["status"]["state"] == "signal_present": - active_connected_outputs.append(response.json()) - else: - return test.FAIL("The output {} properties streamcompatibility request has failed: {}" - .format(output_id, response.json())) - if len(active_connected_outputs) != 0: - return test.UNCLEAR( - "Connected output have a signal while all receivers are inactive." - ) - return test.PASS() + for id in self.is11_utils.sampled_list(self.edid_connected_outputs): + self.get_outputs_edid(test, id) - 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") - for output_id in self.connected_outputs: - valid, response = TestHelper.do_request( - "GET", self.compat_url + "outputs/" + output_id + "/properties/" - ) - if not valid: - return test.FAIL("Unexpected response from the streamcompatibility API: {}".format(response)) - if response.status_code == 200: - if response.json()["edid_support"]: - self.edid_connected_outputs.append(response.json()["id"]) - else: - return test.FAIL("The output {} properties streamcompatibility request has failed: {}" - .format(output_id, response.json())) - if self.edid_connected_outputs == 0: - return test.UNCLEAR("Outputs not supporting edid") return test.PASS() - def test_03_04_01(self, test): + def test_03_01(self, test): """ - Verify that an output indicating EDID support behaves according to the RAML file. + Disconnected Outputs with EDID support and Outputs without EDID support return the EDID """ - if len(self.edid_connected_outputs) == 0: - return test.UNCLEAR("No edid connected outputs") - for output_id in self.edid_connected_outputs: - valid, response = TestHelper.do_request( - "GET", self.compat_url + "outputs/" + output_id - ) - if not valid: - return test.FAIL("Unexpected response from the streamcompatibility API: {}".format(response)) - if response.status_code != 200: - return test.FAIL("The streamcompatibility request for output {} has failed: {}" - .format(output_id, response.json())) - return test.PASS() + target_outputs = self.non_edid_outputs + self.edid_disconnected_outputs - def test_03_04_02(self, test): - """ - Verify that a valid EDID can be retrieved from the device; - this EDID represents the default EDID of the device. - """ - is_valid_response = True - if len(self.edid_connected_outputs) == 0: - return test.UNCLEAR("No edid connected outputs") - for output_id in self.edid_connected_outputs: - valid, response = TestHelper.do_request( - "GET", self.compat_url + "outputs/" + output_id + "/edid/" - ) - if not valid: - return test.FAIL("Unexpected response from the streamcompatibility API: {}".format(response)) - if ( - response.status_code != 200 - or response.headers["Content-Type"] != "application/octet-stream" - ): - is_valid_response = False - break - if is_valid_response: - return test.PASS() - return test.FAIL("The output {} edid streamcompatibility request has failed: {}" - .format(output_id, response.json())) + if len(target_outputs) == 0: + return test.UNCLEAR("Not tested. No disconnected Outputs with EDID support " + "and Outputs without EDID support found.") - 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") - for output_id in self.connected_outputs: - valid, response = TestHelper.do_request( - "GET", self.compat_url + "outputs/" + output_id + "/properties/" - ) - if not valid: - return test.FAIL("Unexpected response from the streamcompatibility API: {}".format(response)) - if response.status_code == 200: - if not response.json()["edid_support"]: - self.not_edid_connected_outputs.append(response.json()["id"]) - else: - return test.FAIL("The output {} properties streamcompatibility request has failed: {}" - .format(output_id, response.json())) - if len(self.not_edid_connected_outputs) == 0: - return test.UNCLEAR("Outputs supporting edid") - return test.PASS() + for id in self.is11_utils.sampled_list(target_outputs): + valid, response = self.do_request("GET", self.compat_url + "outputs/" + id + "/edid") + if not valid or response.status_code != 204: + return test.FAIL("Unexpected response " + "for GET /edid: {}".format(response)) - 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") - for output_id in self.not_edid_connected_outputs: - valid, response = TestHelper.do_request( - "GET", self.compat_url + "outputs/" + output_id + "/edid/" - ) - if not valid: - return test.FAIL("Unexpected response from the streamcompatibility API: {}".format(response)) - if response.status_code != 204: - return test.FAIL("Status code should be 204") return test.PASS() # RECEIVERS TESTS @@ -4137,130 +3631,183 @@ def test_06_03(self, test): return test.PASS() - def test_06_04(self, test): - """Effective EDID updates if Base EDID changes""" - - if len(self.inputs) == 0: - return test.UNCLEAR("Not tested. No inputs found.") - - inputs_tested = [] + def deactivate_connection_resources(self, port): + url = self.conn_url + "single/" + port + "s/" + valid, response = self.do_request("GET", url) + if not valid: + raise NMOSInitException("Unexpected response from the Connection API: {}".format(response)) + if response.status_code != 200: + raise NMOSInitException("The request {} has failed: {}".format(url, response)) - for inputId in self.is11_utils.sampled_list(self.inputs): - valid, response = self.do_request("GET", self.compat_url + "inputs/" + inputId + "/properties") - if not valid or response.status_code != 200: - return test.FAIL("Unexpected response from " - "the Stream Compatibility Management API: {}".format(response)) + try: + for myPort in response.json(): + staged_url = url + myPort + "staged/" + deactivate_json = { + "master_enable": False, + "activation": {"mode": "activate_immediate"}, + } - try: - input = response.json() - if not input["edid_support"]: - continue + valid_patch, patch_response = self.do_request("PATCH", staged_url, json=deactivate_json) + if not valid_patch: + raise NMOSInitException("Unexpected response from the Connection API: {}".format(patch_response)) + if ( + patch_response.status_code != 200 + or patch_response.json()["master_enable"] + or patch_response.json()["activation"]["mode"] != "activate_immediate" + ): + raise NMOSInitException("The patch request to {} has failed: {}" + .format(staged_url, patch_response)) + except json.JSONDecodeError: + raise NMOSInitException("Non-JSON response returned from the Connection API") + + def has_i_o(self, id, type): + connector = "senders/" if type == "sender" else "receivers/" + i_o = "/inputs/" if type == "sender" else "/outputs/" + url = self.compat_url + connector + id + i_o + + valid, r = self.do_request("GET", url) + if valid and r.status_code == 200: + return len(r.json()) > 0 + else: + raise NMOSInitException("The request {} has failed: {}".format(url, r)) - valid, response = self.do_request("GET", self.compat_url + "inputs/" + inputId + "/edid/effective") - if not valid or response.status_code != 200: - return test.FAIL("Unexpected response from " - "the Stream Compatibility Management API: {}".format(response)) + def receiver_has_i_o(self, id): + return self.has_i_o(id, "receiver") - effective_edid_before = response.content - - base_edid = bytearray([ - 0x00, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0x00, - 0x04, 0x43, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, - 0x00, 0x0a, 0x01, 0x04, 0x80, 0x00, 0x00, 0x00, - 0x02, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, - 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01, 0x01, - 0x01, 0x01, 0x01, 0x01, 0x01, 0x01, 0x01, 0x01, - 0x01, 0x01, 0x01, 0x01, 0x01, 0x01, 0x00, 0x00, - 0x00, 0x10, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, - 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, - 0x00, 0x00, 0x00, 0x10, 0x00, 0x00, 0x00, 0x00, - 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, - 0x00, 0x00, 0x00, 0x00, 0x00, 0x10, 0x00, 0x00, - 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, - 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x10, - 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, - 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0xde - ]) + def is_input_adjust_to_caps(self, id): + return self.has_property(id, "input", "adjust_to_caps") - valid, response = self.do_request("PUT", - self.compat_url + "inputs/" + inputId + "/edid/base", - headers={"Content-Type": "application/octet-stream"}, - data=base_edid) - if not valid or response.status_code != 204: - return test.FAIL("Unexpected response from " - "the Stream Compatibility Management API: {}".format(response)) + def has_property(self, id, type, property): + i_o = "inputs/" if type == "input" else "outputs/" + url = self.compat_url + i_o + id + "/properties/" - time.sleep(CONFIG.STABLE_STATE_DELAY) + valid, r = self.do_request("GET", url) + if valid and r.status_code == 200: + if property in r.json(): + return True + else: + return False + else: + raise NMOSInitException("The request {} has failed: {}".format(url, r)) - valid, response = self.do_request("GET", self.compat_url + "inputs/" + inputId + "/edid/effective") - if not valid or response.status_code != 200: - return test.FAIL("Unexpected response from " - "the Stream Compatibility Management API: {}".format(response)) + def has_boolean_property_true(self, id, type, property): + i_o = "inputs/" if type == "input" else "outputs/" + url = self.compat_url + i_o + id + "/properties/" - if response.content == effective_edid_before: - return test.FAIL("Effective EDID doesn't change when Base EDID changes") + valid, r = self.do_request("GET", url) + if valid and r.status_code == 200: + if r.json()[property]: + return True + else: + return False + else: + raise NMOSInitException("The request {} has failed: {}".format(url, r)) - inputs_tested.append(inputId) + def is_input_connected(self, id): + return self.has_boolean_property_true(id, "input", "connected") - except json.JSONDecodeError: - return test.FAIL("Non-JSON response returned from Node API") - except KeyError as e: - return test.FAIL("Unable to find expected key: {}".format(e)) + def has_input_edid_support(self, id): + return self.has_boolean_property_true(id, "input", "edid_support") - if len(inputs_tested) > 0: - return test.PASS() - else: - return test.UNCLEAR("Not tested. No inputs with EDID support found.") + def has_input_base_edid_support(self, id): + return self.has_boolean_property_true(id, "input", "base_edid_support") - def test_06_05(self, test): - """Effective EDID updates if Base EDID removed""" + def is_output_connected(self, id): + return self.has_boolean_property_true(id, "output", "connected") - if len(self.inputs) == 0: - return test.UNCLEAR("Not tested. No inputs found.") + def has_output_edid_support(self, id): + return self.has_boolean_property_true(id, "output", "edid_support") - inputs_tested = [] + def delete_active_constraints(self): + """ + Reset the active constraints of all the senders such that the base EDID is the effective EDID + """ - for inputId in self.is11_utils.sampled_list(self.inputs): - valid, response = self.do_request("GET", self.compat_url + "inputs/" + inputId + "/properties") - if not valid or response.status_code != 200: - return test.FAIL("Unexpected response from " - "the Stream Compatibility Management API: {}".format(response)) + for sender_id in self.senders: + url = self.compat_url + "senders/" + sender_id + "/constraints/active/" + valid, response = self.do_request("DELETE", url) - try: - input = response.json() - if not input["edid_support"]: - continue + if not valid: + raise NMOSInitException("Unexpected response from the Stream Compatibility API: {}".format(response)) + if response.status_code != 200: + raise NMOSInitException("The request {} has failed: {}".format(url, response)) - valid, response = self.do_request("GET", self.compat_url + "inputs/" + inputId + "/edid/effective") - if not valid or response.status_code != 200: - return test.FAIL("Unexpected response from " - "the Stream Compatibility Management API: {}".format(response)) + def delete_base_edid(self): + for id in self.base_edid_inputs: + url = self.compat_url + "inputs/" + id + "/edid/base/" + valid, response = self.do_request("DELETE", url) - effective_edid_before = response.content + if not valid: + raise NMOSInitException("Unexpected response from the Stream Compatibility API: {}".format(response)) + if response.status_code != 204: + raise NMOSInitException("The request {} has failed: {}".format(url, response)) - valid, response = self.do_request("DELETE", self.compat_url + "inputs/" + inputId + "/edid/base") - if not valid or response.status_code != 204: - return test.FAIL("Unexpected response from " - "the Stream Compatibility Management API: {}".format(response)) + def get_inputs_senders(self, test, input_id): + sender_ids = [] - time.sleep(CONFIG.STABLE_STATE_DELAY) + for sender_id in self.senders: + url = self.compat_url + "senders/" + sender_id + "/inputs" + valid, response = self.do_request("GET", url) - valid, response = self.do_request("GET", self.compat_url + "inputs/" + inputId + "/edid/effective") - if not valid or response.status_code != 200: - return test.FAIL("Unexpected response from " - "the Stream Compatibility Management API: {}".format(response)) + if not valid: + raise NMOSTestException( + test.FAIL("Unexpected response from the Stream Compatibility API: {}".format(response)) + ) + if response.status_code != 200: + raise NMOSTestException(test.FAIL("The request {} has failed: {}".format(url, response))) + try: + response_json = response.json() + if input_id in response_json: + sender_ids.append(sender_id) + except json.JSONDecodeError: + raise NMOSTestException( + test.FAIL("Non-JSON response returned from the Stream Compatibility Management API") + ) - if response.content == effective_edid_before: - return test.FAIL("Effective EDID doesn't change when Base EDID changes") + return sender_ids - inputs_tested.append(inputId) + def get_json(self, test, url): + valid, response = self.do_request("GET", url) + if not valid or response.status_code != 200: + raise NMOSTestException( + test.FAIL("Unexpected response from {}: {}".format(url, response)) + ) + try: + return response.json() + except json.JSONDecodeError: + raise NMOSTestException( + test.FAIL("Non-JSON response returned from Node API") + ) - except json.JSONDecodeError: - return test.FAIL("Non-JSON response returned from Node API") - except KeyError as e: - return test.FAIL("Unable to find expected key: {}".format(e)) + def get_effective_edid(self, test, input_id): + valid, response = self.do_request("GET", self.compat_url + "inputs/" + input_id + "/edid/effective") + if ( + not valid + or response.status_code != 200 + or response.headers["Content-Type"] != "application/octet-stream" + ): + raise NMOSTestException( + test.FAIL("Unexpected response from " + "the Stream Compatibility Management API: {}".format(response)) + ) + return response.content + + def get_outputs_edid(self, test, output_id): + valid, response = self.do_request("GET", self.compat_url + "outputs/" + output_id + "/edid") + if ( + not valid + or response.status_code != 200 + or response.headers["Content-Type"] != "application/octet-stream" + ): + raise NMOSTestException( + test.FAIL("Unexpected response from " + "the Stream Compatibility Management API: {}".format(response)) + ) + return response.content - if len(inputs_tested) > 0: - return test.PASS() - else: - return test.UNCLEAR("Not tested. No inputs with EDID support found.") + def wait_until_true(self, predicate): + for i in range(0, CONFIG.STABLE_STATE_ATTEMPTS): + if predicate(): + return True + time.sleep(CONFIG.STABLE_STATE_DELAY) + return False diff --git a/requirements.txt b/requirements.txt index 5a51fa50..93bb689e 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,7 +1,7 @@ flask>=2.0.0 wtforms jsonschema -zeroconf>=0.32.0 +zeroconf>=0.75.0 requests netifaces gitpython