diff --git a/README.md b/README.md index 01bed2c7..febf3e81 100644 --- a/README.md +++ b/README.md @@ -32,6 +32,7 @@ The following test suites are currently supported. | IS-09-01 | IS-09 System API | | (X) | | System Parameters Server | | IS-09-02 | IS-09 System API Discovery | X | | | | | IS-10-01 | IS-10 Authorization API | | | | Authorization Server | +| IS-13-01 | IS-13 Annotation API | X | | | | | - | BCP-002-01 Natural Grouping | X | | | Included in IS-04 Node API suite | | - | BCP-002-02 Asset Distinguishing Information | X | | | Included in IS-04 Node API suite | | BCP-003-01 | BCP-003-01 Secure Communication | X | X | | See [Testing TLS](docs/2.2.%20Usage%20-%20Testing%20BCP-003-01%20TLS.md) | diff --git a/nmostesting/Config.py b/nmostesting/Config.py index c663ec91..0a2240b3 100644 --- a/nmostesting/Config.py +++ b/nmostesting/Config.py @@ -280,6 +280,17 @@ } } }, + "is-13": { + "repo": "is-13", + "versions": ["v1.0"], + "default_version": "v1.0", + "apis": { + "annotation": { + "name": "Annotation API", + "raml": "AnnotationAPI.raml" + } + } + }, "bcp-002-01": { "repo": "bcp-002-01", "versions": ["v1.0"], diff --git a/nmostesting/GenericTest.py b/nmostesting/GenericTest.py index 951eb1fa..ecb73131 100644 --- a/nmostesting/GenericTest.py +++ b/nmostesting/GenericTest.py @@ -671,7 +671,7 @@ def check_api_resource(self, test, resource, response_code, api, path): schema = self.get_schema(api, resource[1]["method"], resource[0], response.status_code) if not schema: - raise NMOSTestException(test.MANUAL("Test suite unable to locate schema")) + raise NMOSTestException(test.MANUAL(f"Test suite unable to locate schema for resource:{resource}")) return self.check_response(schema, resource[1]["method"], response) diff --git a/nmostesting/NMOSTesting.py b/nmostesting/NMOSTesting.py index e4860827..bce92215 100644 --- a/nmostesting/NMOSTesting.py +++ b/nmostesting/NMOSTesting.py @@ -49,7 +49,7 @@ from .DNS import DNS from .GenericTest import NMOSInitException from . import ControllerTest -from .TestResult import TestStates +from .TestResult import TestStates, TestResult from .TestHelper import get_default_ip from .NMOSUtils import DEFAULT_ARGS from .CRL import CRL, CRL_API @@ -82,6 +82,7 @@ from .suites import IS0901Test from .suites import IS0902Test # from .suites import IS1001Test +from .suites import IS1301Test from .suites import BCP00301Test from .suites import BCP0060101Test from .suites import BCP0060102Test @@ -340,6 +341,18 @@ # }], # "class": IS1001Test.IS1001Test # }, + "IS-13-01": { + "name": "IS-13 Annotation API", + "specs": [{ + "spec_key": "is-13", + "api_key": "annotation" + }, { + "spec_key": "is-04", + "api_key": "node", + "disable_fields": ["host", "port"] + }], + "class": IS1301Test.IS1301Test, + }, "BCP-003-01": { "name": "BCP-003-01 Secure Communication", "specs": [{ @@ -623,7 +636,7 @@ def run_tests(test, endpoints, test_selection=["all"]): try: result = test_obj.run_tests(test_selection) except Exception as ex: - print(" * ERROR: {}".format(ex)) + print(" * ERROR while running {}: {}".format(test_selection, ex)) raise ex finally: core_app.config['TEST_ACTIVE'] = False @@ -961,7 +974,8 @@ def run_noninteractive_tests(args): else: exit_code = print_test_results(results, endpoints, args) except Exception as e: - print(" * ERROR: {}".format(str(e))) + print(" * ERROR raw: {}".format(e.args)) + print(" * ERROR in non-interactive tests: {}".format(str(e) if not isinstance(e.args[0], TestResult) else e.args[0].detail)) exit_code = ExitCodes.ERROR return exit_code diff --git a/nmostesting/suites/IS1301Test.py b/nmostesting/suites/IS1301Test.py new file mode 100644 index 00000000..7eec8467 --- /dev/null +++ b/nmostesting/suites/IS1301Test.py @@ -0,0 +1,298 @@ +# Copyright (C) 2023 Advanced Media Workflow Association +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + + +""" +The script implements the IS-13 test suite according to the AMWA IS-13 NMOS Annotation +Specification (https://specs.amwa.tv/is-13/). At the end of the test, the initial state +of the tested unit is supposed to be restored but this cannot be garanteed. + +Not covered yet: +* Persistency: https://specs.amwa.tv/is-13/branches/v1.0-dev/docs/Behaviour.html#persistence-of-updates +* Read only Tags: https://specs.amwa.tv/is-13/branches/v1.0-dev/docs/Behaviour.html#read-only-tags +* Individual Tag reset +* 500 reponses are ignored + +Terminology: +* `resource` refers to self, devices, senders or receivers endpoints +* `annotation_property` refers to annotable objects: label, description or tags +""" + +from ..GenericTest import GenericTest, NMOSTestException + +from ..import Config as CONFIG +from ..import TestHelper +import re +import copy +import time + +IS13_SPEC_VERSION = "v1.0-dev" +IS13_SPEC_URL = f"https://specs.amwa.tv/is-13/branches/{IS13_SPEC_VERSION}/docs" + +ANNOTATION_API_KEY = "annotation" +NODE_API_KEY = "node" + +# Constants for label and description related tests +STRING_LENGTH_OVER_MAX_VALUE = ''.join(['X' for i in range(100)]) +STRING_LENGTH_MAX_VALUE = STRING_LENGTH_OVER_MAX_VALUE[:64] # this is the max length tolerated + +# Constants for tags related tests +TAGS_LENGTH_OVER_MAX_VALUE = {'location': ['underground'], 'studio': ['42'], 'tech': ['John', 'Mike']} +TAGS_LENGTH_MAX_VALUE = TAGS_LENGTH_OVER_MAX_VALUE.copy() +TAGS_LENGTH_MAX_VALUE.pop('tech') # must have a max of 5 +TAGS_TO_BE_STRIPPED = 'urn:x-nmos:tag:grouphint/v1.0' + + +def get_ts_from_version(version): + """ Convert the 'version' object (string) into float """ + return float(re.sub(':', '.', version)) + + +def strip_tags(tags): + if TAGS_TO_BE_STRIPPED in list(tags.keys()): + tags.pop(TAGS_TO_BE_STRIPPED) + return tags + + +class IS1301Test(GenericTest): + """ + Runs IS-13-Test + """ + + def __init__(self, apis, **kwargs): + GenericTest.__init__(self, apis, **kwargs) + self.annotation_url = self.apis[ANNOTATION_API_KEY]["url"] + self.node_url = f"{self.apis[ANNOTATION_API_KEY]['base_url']}/x-nmos/node/{self.apis[NODE_API_KEY]['version']}/" + + def get_resource(self, url): + """ Get a resource """ + + valid, r = self.do_request("GET", url) + if valid and r.status_code == 200: + try: + return r.json() + except Exception as e: + raise(f"Can't parse response as json {e.msg}") + else: + return None + + def compare_resource(self, annotation_property, value1, value2): + """ Compare strings (or dict for 'tags') """ + + if value2[annotation_property] is None: # this is a reset, value2 is null, skip + return True + + if annotation_property == "tags": # tags needs to be stripped + value2[annotation_property] = strip_tags(value2[annotation_property]) + return TestHelper.compare_json(value2[annotation_property], value1[annotation_property]) + + return value2[annotation_property] == value1[annotation_property] + + def set_resource(self, test, url, node_url, value, prev, expected, msg, link): + """ Patch a resource with one ore several object values """ + + self.log(f" {msg}") + + annotation_property = list(value.keys())[0] + valid, resp = self.do_request("PATCH", url, json=value) + if not valid: + raise NMOSTestException(test.FAIL("PATCH Request FAIL", link=f"{IS13_SPEC_URL}/Behaviour.html#setting-values")) + # TODO: if put_response.status_code == 500: + + # pause to accomodate update propagation + time.sleep(CONFIG.API_PROCESSING_TIMEOUT) + + # re-GET + resp = self.get_resource(url) + if not resp: + raise NMOSTestException(test.FAIL(f"GET /{ANNOTATION_API_KEY} FAIL")) + # check PATCH == GET + if not self.compare_resource(annotation_property, resp, expected): + raise NMOSTestException(test.FAIL(f"Compare req vs expect FAIL - {msg}", link=link)) + # check that the version (timestamp) has increased + if get_ts_from_version(prev['version']) >= get_ts_from_version(resp['version']): + raise NMOSTestException(test.FAIL(f"Version update FAIL \ + ({get_ts_from_version(prev['version'])} !>= {get_ts_from_version(resp['version'])})", \ + link=f"{IS13_SPEC_URL}/Behaviour.html#successful-response>")) + + # validate that it is reflected in IS04 + node_resp = self.get_resource(node_url) + if not node_resp: + raise NMOSTestException(test.FAIL(f"GET /{NODE_API_KEY} FAIL")) + # check PATCH == GET + if not self.compare_resource(annotation_property, resp, expected): + raise NMOSTestException(test.FAIL(f"Compare /annotation vs /node FAIL {msg}", + link=f"{IS13_SPEC_URL}/Interoperability_-_IS-04.html#consistent-resources")) + # check that the version (timestamp) has increased + if get_ts_from_version(node_resp['version']) != get_ts_from_version(resp['version']): + raise NMOSTestException(test.FAIL(f"Compare /annotation Version vs /node FAIL {msg}", + link=f"{IS13_SPEC_URL}/Interoperability_-_IS-04.html#version-increments")) + + return resp + + def log(self, msg): + """ + Enable for quick debug only + """ + # print(msg) + return + + def create_url(self, test, base_url, resource): + """ + Build the url for both annotation and node APIs which behaves differently. + For iterables resources (devices, senders, receivers), return the 1st element. + """ + + url = f"{base_url}{resource}" + r = self.get_resource(url) + if r: + if resource != "self": + if isinstance(r[0], str): # in annotation api + index = r[0] + elif isinstance(r[0], dict): # in node api + index = r[0]['id'] + else: + raise NMOSTestException(test.FAIL(f"Unexpected resource found @ {url}")) + url = f"{url}/{index}" + else: + raise NMOSTestException(test.FAIL(f"No resource found @ {url}")) + + return url + + def copy_resource(self, resource): + """ Strip and copy resource """ + + r = copy.copy(resource) + r.pop('id') + r.pop('version') + r['tags'] = strip_tags(r['tags']) + + return r + + def do_test_sequence(self, test, resource, annotation_property): + """ + In addition to the basic annotation API tests, this suite includes the test sequence: + For each resource: + For each annotation_property: + - Read initial value + - store + - Reset default value by sending null + - check value + timestamp + is-14/value + is-04 timestamp + - store + - Write max-length + - check value + timestamp + is-14/value + is-04 timestamp + - Write >max-length + - check value + timestamp + is-14/value + is-04 timestamp + - Reset default value again + - check value + timestamp + is-14/value + is-04 timestamp + - compare with 1st reset + - Restore initial value + """ + + url = self.create_url(test, f"{self.annotation_url}node/", resource) + node_url = self.create_url(test, self.node_url, resource) + + # Save initial resource value + resp = self.get_resource(url) + initial = self.copy_resource(resp) + + msg = "Reset to default and save." + link = f"{IS13_SPEC_URL}/Behaviour.html#resetting-values" + default = resp = self.set_resource(test, url, node_url, {annotation_property: None}, resp, + {annotation_property: None}, msg, link) + + msg = "Set max-length value and return complete response." + link = f"{IS13_SPEC_URL}/Behaviour.html#setting-values" + value = TAGS_LENGTH_MAX_VALUE if annotation_property == "tags" else STRING_LENGTH_MAX_VALUE + resp = self.set_resource(test, url, node_url, {annotation_property: value}, resp, + {annotation_property: value}, msg, link) + + msg = "Exceed max-length value and return truncated response" + link = f"{IS13_SPEC_URL}/Behaviour.html#additional-limitations" + value = TAGS_LENGTH_OVER_MAX_VALUE if annotation_property == "tags" else STRING_LENGTH_OVER_MAX_VALUE + expected = TAGS_LENGTH_MAX_VALUE if annotation_property == "tags" else STRING_LENGTH_MAX_VALUE + resp = self.set_resource(test, url, node_url, {annotation_property: value}, resp, + {annotation_property: expected}, msg, link) + + msg = "Reset again and compare with default." + link = f"{IS13_SPEC_URL}/Behaviour.html#resetting-values" + resp = self.set_resource(test, url, node_url, {annotation_property: None}, resp, default, + msg, link) + + msg = "Restore initial values." + link = "{IS13_SPEC_URL}/Behaviour.html#setting-values" + self.set_resource(test, url, node_url, initial, resp, msg, link) + + return test.PASS() + + def test_00_01(self, test): + """ Annotation service must be announced """ + r = self. get_resource(f"{self.node_url}self/") + print(self.annotation_url) + try: + for s in r['services']: + if 'urn:x-nmos:service:annotation' in s['type'] and s['href'] + '/' == self.annotation_url: + return test.PASS() + except Exception as e: + return test.FAIL(f"Could't parse services in '/node/self' {str(e)}") + return test.FAIL("Could't found 'annotation' as a service in '/node/self'", + link=f"{IS13_SPEC_URL}/Interoperability_-_IS-04.html#discovery") + + def test_01_01(self, test): + """ Annotation test: self/label (reset to default, set 64-byte value, set >64-byte, check IS04+version)""" + return self.do_test_sequence(test, "self", "label") + + def test_01_02(self, test): + """ Annotation test: self/description (reset to default, set 64-byte value, set >64-byte, check IS04+version)""" + return self.do_test_sequence(test, "self", "description") + + def test_01_03(self, test): + """ Annotation test: self/tags (reset to default, set 5 tags, set >5 tags, check IS04+version)""" + return self.do_test_sequence(test, "self", "tags") + + def test_02_01(self, test): + """ Annotation test: devices/../label (reset to default, set 64-byte value, set >64-byte, check IS04+version)""" + return self.do_test_sequence(test, "devices", "label") + + def test_02_02(self, test): + """Annotation test: devices/../description (reset to default, set 5 tags, set >5 tags, check IS04+version)""" + return self.do_test_sequence(test, "devices", "description") + + def test_02_03(self, test): + """Annotation test: devices/../tags (reset to default, set 5 tags, set >5 tags, check IS04+version)""" + return self.do_test_sequence(test, "devices", "tags") + + def test_03_01(self, test): + """ Annotation test: senders/../label (reset to default, set 64-byte value, set >64-byte, check IS04+version)""" + return self.do_test_sequence(test, "senders", "label") + + def test_03_02(self, test): + """Annotation test: senders/../description (reset to default, set 5 tags, set >5 tags, check IS04+version)""" + return self.do_test_sequence(test, "senders", "description") + + def test_03_03(self, test): + """Annotation test: sender/sevices/../tags (reset to default, set 5 tags, set >5 tags, check IS04+version)""" + return self.do_test_sequence(test, "senders", "tags") + + def test_04_01(self, test): + """ Annotation test: receivers/../label (reset to default, set 64-byte value, set >64-byte, check IS04+version)""" + return self.do_test_sequence(test, "receivers", "label") + + def test_04_02(self, test): + """Annotation test: receivers/../description (reset to default, set 5 tags, set >5 tags, check IS04+version)""" + return self.do_test_sequence(test, "receivers", "description") + + def test_04_03(self, test): + """Annotation test: receivers/sevices/../tags (reset to default, set 5 tags, set >5 tags, check IS04+version)""" + return self.do_test_sequence(test, "receivers", "tags")