From bc7e9971908157bd76c328b2d932d069ca43631d Mon Sep 17 00:00:00 2001 From: Gareth Sylvester-Bradley Date: Mon, 14 Aug 2023 15:23:57 +0100 Subject: [PATCH 01/15] python-zeroconf 0.75.0 strict=False for _nmos-registration._tcp --- nmostesting/suites/IS0401Test.py | 20 ++++++++++---------- requirements.txt | 2 +- 2 files changed, 11 insertions(+), 11 deletions(-) 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/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 From b10ff7157f3e0e97e736e5a820c57a6a4cc0383f Mon Sep 17 00:00:00 2001 From: Gareth Sylvester-Bradley Date: Fri, 18 Aug 2023 14:00:25 +0100 Subject: [PATCH 02/15] Python 3.6 (and 3.7!) is End-Of-Life, and so is Node.js 14.x Libraries like python-zeroconf are starting to depend on more recent Python features --- Dockerfile | 10 +++++----- docs/1.1. Installation - Local.md | 2 +- 2 files changed, 6 insertions(+), 6 deletions(-) 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) From 3f10a8d478b9148c10f12ddef7af485002e4c9c2 Mon Sep 17 00:00:00 2001 From: Gareth Sylvester-Bradley <31761158+garethsb@users.noreply.github.com> Date: Thu, 5 Oct 2023 09:42:50 +0100 Subject: [PATCH 03/15] Pass IP address of API under test to testssl.sh So that same one is used by Python tests and testssl.sh tests Co-authored-by: Simon Lo --- nmostesting/suites/BCP00301Test.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) 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"])] ) From 796a11507c979ea6343717ce0a7fc3cdfe0332ea Mon Sep 17 00:00:00 2001 From: Gareth Sylvester-Bradley <31761158+garethsb@users.noreply.github.com> Date: Thu, 5 Oct 2023 10:15:19 +0100 Subject: [PATCH 04/15] Add MOCK_SERVICES_WARM_UP_DELAY --- nmostesting/Config.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/nmostesting/Config.py b/nmostesting/Config.py index 9c6dbe8f..c663ec91 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`. From 43a1f764284eb4da960e9f5d75463ae91ead52d9 Mon Sep 17 00:00:00 2001 From: Gareth Sylvester-Bradley <31761158+garethsb@users.noreply.github.com> Date: Thu, 5 Oct 2023 10:22:42 +0100 Subject: [PATCH 05/15] Delay running tests until after MOCK_SERVICES_WARM_UP_DELAY Co-authored-by: Simon Lo --- nmostesting/NMOSTesting.py | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/nmostesting/NMOSTesting.py b/nmostesting/NMOSTesting.py index a4d320f9..e5f2176b 100644 --- a/nmostesting/NMOSTesting.py +++ b/nmostesting/NMOSTesting.py @@ -1139,6 +1139,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(" * Warm up your engines, testing begins in {} seconds..." + .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. From 215f736469e5260b7629592e5adf8769c50555ca Mon Sep 17 00:00:00 2001 From: Gareth Sylvester-Bradley <31761158+garethsb@users.noreply.github.com> Date: Thu, 5 Oct 2023 12:43:06 +0100 Subject: [PATCH 06/15] Tweak message Co-authored-by: jonathan-r-thorpe <64410119+jonathan-r-thorpe@users.noreply.github.com> --- nmostesting/NMOSTesting.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/nmostesting/NMOSTesting.py b/nmostesting/NMOSTesting.py index e5f2176b..e4860827 100644 --- a/nmostesting/NMOSTesting.py +++ b/nmostesting/NMOSTesting.py @@ -1142,7 +1142,7 @@ def main(args): # 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(" * Warm up your engines, testing begins in {} seconds..." + 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) From d5691326bff4c8d6e0ca9d1af0314732fe942deb Mon Sep 17 00:00:00 2001 From: N-Nagorny Date: Wed, 11 Oct 2023 22:22:02 +0400 Subject: [PATCH 07/15] Rewrite tests of Inputs in the independent manner --- nmostesting/IS11Utils.py | 32 +- nmostesting/suites/IS1101Test.py | 1168 ++++++++++------------------- test_data/IS1101/invalid_edid.bin | 16 - test_data/IS1101/valid_edid.bin | Bin 128 -> 0 bytes 4 files changed, 407 insertions(+), 809 deletions(-) delete mode 100644 test_data/IS1101/invalid_edid.bin delete mode 100644 test_data/IS1101/valid_edid.bin diff --git a/nmostesting/IS11Utils.py b/nmostesting/IS11Utils.py index 935250e5..e1dc1c53 100644 --- a/nmostesting/IS11Utils.py +++ b/nmostesting/IS11Utils.py @@ -12,60 +12,36 @@ # See the License for the specific language governing permissions and # limitations under the License. -from enum import Enum - from . import TestHelper from .NMOSUtils import NMOSUtils -SND_RCV_SUBSET = Enum('SndRcvSubset', ['ALL', 'WITH_I_O', 'WITHOUT_I_O']) - class IS11Utils(NMOSUtils): def __init__(self, url): NMOSUtils.__init__(self, 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/suites/IS1101Test.py b/nmostesting/suites/IS1101Test.py index b9a9cb68..671ce0e4 100644 --- a/nmostesting/suites/IS1101Test.py +++ b/nmostesting/suites/IS1101Test.py @@ -17,21 +17,19 @@ 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" REF_SUPPORTED_CONSTRAINTS_VIDEO = [ "urn:x-nmos:cap:meta:label", @@ -55,6 +53,44 @@ "urn:x-nmos:cap:format:sample_depth" ] +VALID_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 +]) + +INVALID_EDID = bytearray([ + 0x00, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, + 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0xde +]) + class IS1101Test(GenericTest): """ @@ -64,7 +100,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 +112,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 +128,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 = [] @@ -127,16 +154,35 @@ def build_output_properties_url(self, id): def set_up_tests(self): 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.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() + # GENERAL TESTS - def test_00_01(self, test): + def test_1(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 +194,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 +229,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 +238,236 @@ 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_2(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.") - 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") + for inputId in self.is11_utils.sampled_list(self.edid_inputs): + self.get_effective_edid(test, inputId) - 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 - """ + return test.PASS() - if len(self.connected_inputs) != 0 and len(self.edid_connected_inputs) != 0: + def test_3(self, test): + """Inputs with Base EDID support handles PUTting and DELETing the Base 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=VALID_EDID) + if not valid or response.status_code != 204: + return test.FAIL("Unexpected response from " + "the Stream Compatibility Management API: {}".format(response)) - 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() + # 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 != VALID_EDID: + return test.FAIL("The Base EDID of Input {}" + "doesn't match the Base EDID that has been put".format(inputId)) + + # Verify that /edid/effective returns the last Base EDID put + if self.get_effective_edid(test, inputId) != VALID_EDID: + return test.FAIL("The Effective EDID of Input {}" + "doesn't match the Base EDID that has been put".format(inputId)) + + # 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_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)) + # Verify that /edid/effective returned to its defaults + if self.get_effective_edid(test, inputId) != default_edid: + return test.FAIL("The Effective EDID of Input {}" + "doesn't match its initial value".format(inputId)) - return test.PASS() - return test.UNCLEAR("No resources found to perform this test") + return test.PASS() - 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_4(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=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() - 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_5(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=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_04_06(self, test): + def test_6(self, test): """ - Verify that the last PUT base EDID can be retrieved + 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: - 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") + if len(self.base_edid_inputs) == 0: + return test.UNCLEAR("Not tested. No inputs with Base EDID support found.") - 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") + for inputId in self.is11_utils.sampled_list(self.base_edid_inputs): + sender_ids = self.get_inputs_senders(test, inputId) - 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") + in_version_1 = "" + in_version_2 = "" + in_version_3 = "" - 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") + snd_versions_1 = {} + snd_versions_2 = {} + snd_versions_3 = {} - 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") + for sender_id in sender_ids: + snd_versions_1[sender_id] = "" + snd_versions_2[sender_id] = "" + snd_versions_3[sender_id] = "" - 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") + 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)) - 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())) + 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()["edid_support"]: - self.not_edid_connected_inputs.append(response.json()["id"]) + 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)) - 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() - - return test.UNCLEAR("No resources found to perform this test") - - 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: - - 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() - - return test.UNCLEAR("No resources found to perform this test") - - 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: - - 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() - - return test.UNCLEAR("No resources found to perform this test") - - 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") - - 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: - - 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() - return test.UNCLEAR("No resources found to perform this test") + # 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=VALID_EDID) + if not valid or response.status_code != 204: + return test.FAIL("Unexpected response from " + "the Stream Compatibility Management API: {}".format(response)) - def test_01_06_01(self, test): - """ - Verify that the input supports changing the base EDID which is optional from the specification. - """ - 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") + 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)) - 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)) + 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)) - self.version[input_id] = version - 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 + 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)) - 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") - - 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") + # 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)) - 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_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)) - 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.") - 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[device["id"]] = input["version"] - - 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())) try: - version = response.json()["version"] + 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 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_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)) - 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())) - try: - if not response.json()["connected"]: - self.disconnected_input.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.disconnected_input) == 0: - return test.UNCLEAR("All inputs are connected") - return test.PASS() - return test.UNCLEAR("No resources found to perform this test") + return test.PASS() # SENDERS TESTS """ @@ -2344,7 +1930,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 +2046,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 +2095,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=VALID_EDID, headers={"Content-Type": "application/octet-stream"}, ) - file.close() time.sleep(CONFIG.STABLE_STATE_DELAY) valid, response = TestHelper.do_request( @@ -2590,7 +2174,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 +2238,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 +2288,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 +2525,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: @@ -3049,7 +2635,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( @@ -4297,127 +3883,179 @@ def test_06_03(self, test): 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 = [] - - 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)) + 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): try: - input = response.json() - if not input["edid_support"]: - continue - - 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)) - - 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 - ]) + 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=base_edid) + data=VALID_EDID) if not valid or response.status_code != 204: return test.FAIL("Unexpected response from " "the Stream Compatibility Management API: {}".format(response)) time.sleep(CONFIG.STABLE_STATE_DELAY) - valid, response = self.do_request("GET", self.compat_url + "inputs/" + inputId + "/edid/effective") - if not valid or response.status_code != 200: + if self.get_effective_edid(test, inputId) == effective_edid_before: + 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)) - if response.content == effective_edid_before: - return test.FAIL("Effective EDID doesn't change when Base EDID changes") + time.sleep(CONFIG.STABLE_STATE_DELAY) - inputs_tested.append(inputId) + if self.get_effective_edid(test, inputId) != effective_edid_before: + 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() - if len(inputs_tested) > 0: - return test.PASS() + def deactivate_connection_resources(self, port): + url = self.conn_url + "single/" + port + "s/" + valid, response = TestHelper.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 myPort in response.json(): + if myPort == "9dd466cb-36c3-5a4e-82d0-b422b119ba69/": + continue + staged_url = url + myPort + "staged/" + deactivate_json = { + "master_enable": False, + "activation": {"mode": "activate_immediate"}, + } + + valid_patch, response = TestHelper.do_request("PATCH", staged_url, json=deactivate_json) + if not valid_patch: + raise NMOSInitException("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" + ): + raise NMOSInitException("The patch request to {} has failed: {}" + .format(staged_url, response)) + + 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 = TestHelper.do_request("GET", url) + if valid and r.status_code == 200: + return len(r.json()) > 0 else: - return test.UNCLEAR("Not tested. No inputs with EDID support found.") + raise NMOSInitException("The request {} has failed: {}".format(url, r)) - def test_06_05(self, test): - """Effective EDID updates if Base EDID removed""" + def receiver_has_i_o(self, id): + return self.has_i_o(id, "receiver") - if len(self.inputs) == 0: - return test.UNCLEAR("Not tested. No inputs found.") + 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/" - inputs_tested = [] + valid, r = TestHelper.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)) - 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)) + def is_input_connected(self, id): + return self.has_boolean_property_true(id, "input", "connected") - try: - input = response.json() - if not input["edid_support"]: - continue + def has_input_edid_support(self, id): + return self.has_boolean_property_true(id, "input", "edid_support") - 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_input_base_edid_support(self, id): + return self.has_boolean_property_true(id, "input", "base_edid_support") - effective_edid_before = response.content + def is_output_connected(self, id): + return self.has_boolean_property_true(id, "output", "connected") - 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 delete_active_constraints(self): + """ + Reset the active constraints of all the senders such that the base EDID is the effective EDID + """ - time.sleep(CONFIG.STABLE_STATE_DELAY) + for sender_id in self.senders: + url = self.compat_url + "senders/" + sender_id + "/constraints/active/" + valid, response = TestHelper.do_request("DELETE", 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 NMOSInitException("Unexpected response from the Stream Compatibility API: {}".format(response)) + if response.status_code != 200: + raise NMOSInitException("The request {} has failed: {}".format(url, response)) - if response.content == effective_edid_before: - return test.FAIL("Effective EDID doesn't change when Base EDID changes") + def delete_base_edid(self): + for id in self.base_edid_inputs: + url = self.compat_url + "inputs/" + id + "/edid/base/" + valid, response = TestHelper.do_request("DELETE", url) - inputs_tested.append(inputId) + 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)) + def get_inputs_senders(self, test, input_id): + sender_ids = [] + + for sender_id in self.senders: + url = self.compat_url + "senders/" + sender_id + "/inputs" + valid, response = TestHelper.do_request("GET", url) + + 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: - return test.FAIL("Non-JSON response returned from Node API") - except KeyError as e: - return test.FAIL("Unable to find expected key: {}".format(e)) + raise NMOSTestException( + test.FAIL("Non-JSON response returned from the Stream Compatibility Management API") + ) - if len(inputs_tested) > 0: - return test.PASS() - else: - return test.UNCLEAR("Not tested. No inputs with EDID support found.") + return sender_ids + + 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") + ) + + 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 diff --git a/test_data/IS1101/invalid_edid.bin b/test_data/IS1101/invalid_edid.bin deleted file mode 100644 index f78f276b..00000000 --- a/test_data/IS1101/invalid_edid.bin +++ /dev/null @@ -1,16 +0,0 @@ -0x00, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0x00, -0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, -0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, -0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, -0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, -0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, -0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, -0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, -0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, -0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, -0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, -0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, -0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, -0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, -0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, -0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0xde \ No newline at end of file diff --git a/test_data/IS1101/valid_edid.bin b/test_data/IS1101/valid_edid.bin deleted file mode 100644 index 8811f5ca5d1c8eb26ac391318c9f654e2bc33e26..0000000000000000000000000000000000000000 GIT binary patch literal 0 HcmV?d00001 literal 128 zcmZSh4+acAx{nyl4Lx1dco~@+l%y-P-c4N`;xki?KOkiP`-aAbLVrdiV4%e?BalJM z!Jt5xA&5_qfkBRef#ENMg|}QF9|M;H7yw27Fa%fTCHp28m!uW}MS&v!89a;v!z~RA K42{h|B3=N%ydePq From 4b9debbe635e04abc022a95f7d734504dc61cc49 Mon Sep 17 00:00:00 2001 From: N-Nagorny Date: Thu, 12 Oct 2023 18:36:46 +0400 Subject: [PATCH 08/15] Add missed default Effective EDID get --- nmostesting/suites/IS1101Test.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/nmostesting/suites/IS1101Test.py b/nmostesting/suites/IS1101Test.py index 671ce0e4..b91614d8 100644 --- a/nmostesting/suites/IS1101Test.py +++ b/nmostesting/suites/IS1101Test.py @@ -2587,6 +2587,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": [ { From d3bb28816892b46493dcb5c6925a77f0de77461b Mon Sep 17 00:00:00 2001 From: N-Nagorny Date: Fri, 13 Oct 2023 01:05:10 +0400 Subject: [PATCH 09/15] Rewrite tests of Outputs in the independent manner --- nmostesting/suites/IS1101Test.py | 194 +++++++------------------------ 1 file changed, 42 insertions(+), 152 deletions(-) diff --git a/nmostesting/suites/IS1101Test.py b/nmostesting/suites/IS1101Test.py index b91614d8..0ce7ab47 100644 --- a/nmostesting/suites/IS1101Test.py +++ b/nmostesting/suites/IS1101Test.py @@ -172,6 +172,15 @@ def set_up_tests(self): self.base_edid_inputs = list(filter(self.has_input_base_edid_support, self.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" @@ -2935,167 +2944,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_7(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_8(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 @@ -3988,6 +3862,9 @@ def has_input_base_edid_support(self, id): def is_output_connected(self, id): return self.has_boolean_property_true(id, "output", "connected") + def has_output_edid_support(self, id): + return self.has_boolean_property_true(id, "output", "edid_support") + def delete_active_constraints(self): """ Reset the active constraints of all the senders such that the base EDID is the effective EDID @@ -4061,3 +3938,16 @@ def get_effective_edid(self, test, input_id): "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 From e9948017fba756857460f10672b23a99f43d782d Mon Sep 17 00:00:00 2001 From: N-Nagorny Date: Fri, 27 Oct 2023 03:31:51 +0400 Subject: [PATCH 10/15] Return the valid and the invalid EDID examples to the filesystem --- nmostesting/suites/IS1101Test.py | 62 ++++++++---------------------- test_data/IS1101/invalid_edid.bin | 16 ++++++++ test_data/IS1101/valid_edid.bin | Bin 0 -> 128 bytes 3 files changed, 32 insertions(+), 46 deletions(-) create mode 100644 test_data/IS1101/invalid_edid.bin create mode 100644 test_data/IS1101/valid_edid.bin diff --git a/nmostesting/suites/IS1101Test.py b/nmostesting/suites/IS1101Test.py index 0ce7ab47..51ff6753 100644 --- a/nmostesting/suites/IS1101Test.py +++ b/nmostesting/suites/IS1101Test.py @@ -30,6 +30,8 @@ CONTROLS = "controls" NODE_API_KEY = "node" CONN_API_KEY = "connection" +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", @@ -53,44 +55,6 @@ "urn:x-nmos:cap:format:sample_depth" ] -VALID_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 -]) - -INVALID_EDID = bytearray([ - 0x00, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0x00, - 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, - 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, - 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, - 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, - 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, - 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, - 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, - 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, - 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, - 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, - 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, - 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, - 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, - 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, - 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0xde -]) - class IS1101Test(GenericTest): """ @@ -152,6 +116,12 @@ 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() @@ -270,7 +240,7 @@ def test_3(self, test): valid, response = self.do_request("PUT", self.compat_url + "inputs/" + inputId + "/edid/base", headers={"Content-Type": "application/octet-stream"}, - data=VALID_EDID) + 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)) @@ -284,12 +254,12 @@ def test_3(self, test): ): return test.FAIL("Unexpected response from " "the Stream Compatibility Management API: {}".format(response)) - if response.content != VALID_EDID: + 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)) # Verify that /edid/effective returns the last Base EDID put - if self.get_effective_edid(test, inputId) != VALID_EDID: + if self.get_effective_edid(test, inputId) != self.valid_edid: return test.FAIL("The Effective EDID of Input {}" "doesn't match the Base EDID that has been put".format(inputId)) @@ -321,7 +291,7 @@ def test_4(self, test): valid, response = self.do_request("PUT", self.compat_url + "inputs/" + inputId + "/edid/base", headers={"Content-Type": "application/octet-stream"}, - data=INVALID_EDID) + 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)) @@ -351,7 +321,7 @@ def test_5(self, test): valid, response = self.do_request("PUT", self.compat_url + "inputs/" + inputId + "/edid/base", headers={"Content-Type": "application/octet-stream"}, - data=VALID_EDID) + data=self.valid_edid) if not valid or response.status_code != 405: return test.FAIL("Unexpected response " "for PUT /edid/base: {}".format(response)) @@ -407,7 +377,7 @@ def test_6(self, test): valid, response = self.do_request("PUT", self.compat_url + "inputs/" + inputId + "/edid/base", headers={"Content-Type": "application/octet-stream"}, - data=VALID_EDID) + 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)) @@ -2106,7 +2076,7 @@ def test_02_03_04(self, test): response = requests.put( self.compat_url + "inputs/" + input_id + "/edid/base/", - data=VALID_EDID, + data=self.valid_edid, headers={"Content-Type": "application/octet-stream"}, ) time.sleep(CONFIG.STABLE_STATE_DELAY) @@ -3769,7 +3739,7 @@ def test_06_04(self, test): valid, response = self.do_request("PUT", self.compat_url + "inputs/" + inputId + "/edid/base", headers={"Content-Type": "application/octet-stream"}, - data=VALID_EDID) + 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)) diff --git a/test_data/IS1101/invalid_edid.bin b/test_data/IS1101/invalid_edid.bin new file mode 100644 index 00000000..f78f276b --- /dev/null +++ b/test_data/IS1101/invalid_edid.bin @@ -0,0 +1,16 @@ +0x00, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0x00, +0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, +0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, +0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, +0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, +0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, +0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, +0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, +0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, +0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, +0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, +0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, +0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, +0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, +0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, +0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0xde \ No newline at end of file diff --git a/test_data/IS1101/valid_edid.bin b/test_data/IS1101/valid_edid.bin new file mode 100644 index 0000000000000000000000000000000000000000..8811f5ca5d1c8eb26ac391318c9f654e2bc33e26 GIT binary patch literal 128 zcmZSh4+acAx{nyl4Lx1dco~@+l%y-P-c4N`;xki?KOkiP`-aAbLVrdiV4%e?BalJM z!Jt5xA&5_qfkBRef#ENMg|}QF9|M;H7yw27Fa%fTCHp28m!uW}MS&v!89a;v!z~RA K42{h|B3=N%ydePq literal 0 HcmV?d00001 From cf451f05b33e70bc391e578c9aa35f48c69d781e Mon Sep 17 00:00:00 2001 From: N-Nagorny Date: Fri, 27 Oct 2023 03:33:01 +0400 Subject: [PATCH 11/15] Clean IS-11 helper functions --- nmostesting/suites/IS1101Test.py | 49 ++++++++++++++++---------------- 1 file changed, 25 insertions(+), 24 deletions(-) diff --git a/nmostesting/suites/IS1101Test.py b/nmostesting/suites/IS1101Test.py index 51ff6753..c022130e 100644 --- a/nmostesting/suites/IS1101Test.py +++ b/nmostesting/suites/IS1101Test.py @@ -3767,38 +3767,39 @@ def test_06_04(self, test): def deactivate_connection_resources(self, port): url = self.conn_url + "single/" + port + "s/" - valid, response = TestHelper.do_request("GET", url) + 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 myPort in response.json(): - if myPort == "9dd466cb-36c3-5a4e-82d0-b422b119ba69/": - continue - staged_url = url + myPort + "staged/" - deactivate_json = { - "master_enable": False, - "activation": {"mode": "activate_immediate"}, - } + try: + for myPort in response.json(): + staged_url = url + myPort + "staged/" + deactivate_json = { + "master_enable": False, + "activation": {"mode": "activate_immediate"}, + } - valid_patch, response = TestHelper.do_request("PATCH", staged_url, json=deactivate_json) - if not valid_patch: - raise NMOSInitException("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" - ): - raise NMOSInitException("The patch request to {} has failed: {}" - .format(staged_url, response)) + 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 = TestHelper.do_request("GET", url) + valid, r = self.do_request("GET", url) if valid and r.status_code == 200: return len(r.json()) > 0 else: @@ -3811,7 +3812,7 @@ 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/" - valid, r = TestHelper.do_request("GET", url) + valid, r = self.do_request("GET", url) if valid and r.status_code == 200: if r.json()[property]: return True @@ -3842,7 +3843,7 @@ def delete_active_constraints(self): for sender_id in self.senders: url = self.compat_url + "senders/" + sender_id + "/constraints/active/" - valid, response = TestHelper.do_request("DELETE", url) + valid, response = self.do_request("DELETE", url) if not valid: raise NMOSInitException("Unexpected response from the Stream Compatibility API: {}".format(response)) @@ -3852,7 +3853,7 @@ def delete_active_constraints(self): def delete_base_edid(self): for id in self.base_edid_inputs: url = self.compat_url + "inputs/" + id + "/edid/base/" - valid, response = TestHelper.do_request("DELETE", url) + valid, response = self.do_request("DELETE", url) if not valid: raise NMOSInitException("Unexpected response from the Stream Compatibility API: {}".format(response)) @@ -3864,7 +3865,7 @@ def get_inputs_senders(self, test, input_id): for sender_id in self.senders: url = self.compat_url + "senders/" + sender_id + "/inputs" - valid, response = TestHelper.do_request("GET", url) + valid, response = self.do_request("GET", url) if not valid: raise NMOSTestException( From 5a64c550daadb617e070e7fcc21052d1f45800a2 Mon Sep 17 00:00:00 2001 From: N-Nagorny Date: Tue, 31 Oct 2023 09:16:49 -0400 Subject: [PATCH 12/15] Turn back the top-level sectioning --- nmostesting/suites/IS1101Test.py | 18 +++++++++--------- 1 file changed, 9 insertions(+), 9 deletions(-) diff --git a/nmostesting/suites/IS1101Test.py b/nmostesting/suites/IS1101Test.py index c022130e..5dae8972 100644 --- a/nmostesting/suites/IS1101Test.py +++ b/nmostesting/suites/IS1101Test.py @@ -161,7 +161,7 @@ def set_up_tests(self): self.delete_base_edid() # GENERAL TESTS - def test_1(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"] @@ -174,7 +174,7 @@ def test_1(self, test): ) # INPUTS TESTS - def test_01_03(self, test): + def test_01_00(self, test): """ Verify that all connected inputs have a signal """ @@ -217,7 +217,7 @@ 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_2(self, test): + 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.") @@ -227,7 +227,7 @@ def test_2(self, test): return test.PASS() - def test_3(self, test): + def test_01_02(self, test): """Inputs with Base EDID support handles PUTting and DELETing the Base EDID""" if len(self.base_edid_inputs) == 0: return test.UNCLEAR("Not tested. No inputs with Base EDID support found.") @@ -282,7 +282,7 @@ def test_3(self, test): return test.PASS() - def test_4(self, test): + 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.") @@ -297,7 +297,7 @@ def test_4(self, test): "the Stream Compatibility Management API: {}".format(response)) return test.PASS() - def test_5(self, 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.") @@ -327,7 +327,7 @@ def test_5(self, test): "for PUT /edid/base: {}".format(response)) return test.PASS() - def test_6(self, test): + def test_01_05(self, test): """ Inputs with Base EDID increment their version and versions of associated Senders after the Base EDID gets modified @@ -2914,7 +2914,7 @@ def test_02_04_01(self, test): return test.PASS() # OUTPUTS TESTS - def test_7(self, test): + 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.") @@ -2924,7 +2924,7 @@ def test_7(self, test): return test.PASS() - def test_8(self, test): + def test_03_01(self, test): """ Disconnected Outputs with EDID support and Outputs without EDID support return the EDID """ From 1f836762f1b45530a0ec681868882845bb3015dc Mon Sep 17 00:00:00 2001 From: N-Nagorny Date: Tue, 31 Oct 2023 14:19:25 -0400 Subject: [PATCH 13/15] Insert waits between modifying Base EDID and GETting Effective EDID --- nmostesting/suites/IS1101Test.py | 41 +++++++++++++++++++++++++------- 1 file changed, 32 insertions(+), 9 deletions(-) diff --git a/nmostesting/suites/IS1101Test.py b/nmostesting/suites/IS1101Test.py index 5dae8972..26c6506b 100644 --- a/nmostesting/suites/IS1101Test.py +++ b/nmostesting/suites/IS1101Test.py @@ -12,6 +12,7 @@ # See the License for the specific language governing permissions and # limitations under the License. +from functools import partial import time import re @@ -228,7 +229,10 @@ def test_01_01(self, test): return test.PASS() def test_01_02(self, test): - """Inputs with Base EDID support handles PUTting and DELETing the Base EDID""" + """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.") @@ -259,7 +263,10 @@ def test_01_02(self, test): "doesn't match the Base EDID that has been put".format(inputId)) # Verify that /edid/effective returns the last Base EDID put - if self.get_effective_edid(test, inputId) != self.valid_edid: + 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)) @@ -276,7 +283,8 @@ def test_01_02(self, test): "the Stream Compatibility Management API: {}".format(response)) # Verify that /edid/effective returned to its defaults - if self.get_effective_edid(test, inputId) != default_edid: + 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)) @@ -3729,6 +3737,12 @@ def test_06_03(self, test): def test_06_04(self, test): """Effective EDID updates if Base EDID changes""" + 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.base_edid_inputs) == 0: return test.UNCLEAR("Not tested. No inputs with Base EDID support found.") @@ -3744,9 +3758,10 @@ def test_06_04(self, test): return test.FAIL("Unexpected response from " "the Stream Compatibility Management API: {}".format(response)) - time.sleep(CONFIG.STABLE_STATE_DELAY) - - if self.get_effective_edid(test, inputId) == effective_edid_before: + 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") @@ -3754,9 +3769,10 @@ def test_06_04(self, test): return test.FAIL("Unexpected response from " "the Stream Compatibility Management API: {}".format(response)) - time.sleep(CONFIG.STABLE_STATE_DELAY) - - if self.get_effective_edid(test, inputId) != effective_edid_before: + 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: @@ -3922,3 +3938,10 @@ def get_outputs_edid(self, test, output_id): "the Stream Compatibility Management API: {}".format(response)) ) return response.content + + 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 From bfbd5dfa5d8e528e6686c561c8491dd4b7b42bdf Mon Sep 17 00:00:00 2001 From: N-Nagorny Date: Wed, 1 Nov 2023 11:51:17 -0400 Subject: [PATCH 14/15] Add "adjust_to_caps" into test 06.04 --- nmostesting/suites/IS1101Test.py | 112 ++++++++++++++++++------------- 1 file changed, 65 insertions(+), 47 deletions(-) diff --git a/nmostesting/suites/IS1101Test.py b/nmostesting/suites/IS1101Test.py index 26c6506b..34f1fffa 100644 --- a/nmostesting/suites/IS1101Test.py +++ b/nmostesting/suites/IS1101Test.py @@ -142,6 +142,7 @@ def set_up_tests(self): 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)) @@ -456,6 +457,54 @@ def test_01_05(self, test): 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 """ Runs Node Tests covering IS-11 for Senders @@ -3734,53 +3783,6 @@ def test_06_03(self, test): return test.PASS() - def test_06_04(self, test): - """Effective EDID updates if Base EDID changes""" - - 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.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): - 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) - 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() - def deactivate_connection_resources(self, port): url = self.conn_url + "single/" + port + "s/" valid, response = self.do_request("GET", url) @@ -3824,6 +3826,22 @@ def has_i_o(self, id, type): def receiver_has_i_o(self, id): return self.has_i_o(id, "receiver") + def is_input_adjust_to_caps(self, id): + return self.has_property(id, "input", "adjust_to_caps") + + def has_property(self, id, type, property): + i_o = "inputs/" if type == "input" else "outputs/" + url = self.compat_url + i_o + id + "/properties/" + + 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)) + 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/" From d783ba2c4ab446e734f4cfe59df9e75477c52b22 Mon Sep 17 00:00:00 2001 From: N-Nagorny Date: Wed, 1 Nov 2023 16:23:30 -0400 Subject: [PATCH 15/15] Add tear_down_tests --- nmostesting/suites/IS1101Test.py | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) diff --git a/nmostesting/suites/IS1101Test.py b/nmostesting/suites/IS1101Test.py index 34f1fffa..0ba16c15 100644 --- a/nmostesting/suites/IS1101Test.py +++ b/nmostesting/suites/IS1101Test.py @@ -162,6 +162,11 @@ def set_up_tests(self): 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_00(self, test): """At least one Device is showing an IS-11 control advertisement matching the API under test""" @@ -260,7 +265,7 @@ def is_edid_equal_to_effective_edid(self, test, inputId, edid): 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 {}" + return test.FAIL("The Base EDID of Input {} " "doesn't match the Base EDID that has been put".format(inputId)) # Verify that /edid/effective returns the last Base EDID put @@ -268,7 +273,7 @@ def is_edid_equal_to_effective_edid(self, test, inputId, edid): partial(is_edid_equal_to_effective_edid, self, test, inputId, self.valid_edid) ) if not result: - return test.FAIL("The Effective EDID of Input {}" + return test.FAIL("The Effective EDID of Input {} " "doesn't match the Base EDID that has been put".format(inputId)) # Delete the Base EDID @@ -286,7 +291,7 @@ def is_edid_equal_to_effective_edid(self, test, inputId, edid): # 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 {}" + return test.FAIL("The Effective EDID of Input {} " "doesn't match its initial value".format(inputId)) return test.PASS()