Conversation
- fix flake issues - add comments in BCP0050201Test
| "url": "https://github.com/alabou/", | ||
| "branch": "bcp-005-02", | ||
| "versions": ["bcp-005-02"], | ||
| "default_version": "bcp-005-02", |
Co-authored-by: jonathan-r-thorpe <64410119+jonathan-r-thorpe@users.noreply.github.com>
| @@ -0,0 +1,474 @@ | |||
| # Copyright (C) 2025 Matrox Graphics Inc. | |||
There was a problem hiding this comment.
| # Copyright (C) 2025 Matrox Graphics Inc. | |
| # Copyright (C) 2022 Advanced Media Workflow Association |
There was a problem hiding this comment.
There are other test files with "Copyright 2018 British Broadcasting Corporation" and "Copyright (C) 2018 Riedel Communications GmbH & Co. KG". It would be nice to know the rules and have them applied to all the files. For now adding AMWA Copyright in addition to Matrox.
There was a problem hiding this comment.
Yes good point - maybe this is something @peterbrightwell might be able to answer?
| self.is05_resources = {"senders": [], "receivers": [], | ||
| "_requested": [], "transport_types": {}, "transport_files": {}} |
There was a problem hiding this comment.
Only used by get_is05_partial_resources which is itself unused, so can be removed
| self.is05_resources = {"senders": [], "receivers": [], | |
| "_requested": [], "transport_types": {}, "transport_files": {}} |
| self.is05_resources = {"senders": [], "receivers": [], | ||
| "_requested": [], "transport_types": {}, "transport_files": {}} | ||
| self.is04_utils = IS04Utils(self.node_url) | ||
| self.is05_utils = IS05Utils(self.connection_url) |
There was a problem hiding this comment.
Only used by get_is05_partial_resources which is itself unused, so can be removed
| self.is05_utils = IS05Utils(self.connection_url) |
| def get_is05_partial_resources(self, resource_type): | ||
| """Retrieve all Senders or Receivers from a Connection API, keeping hold of the returned IDs""" | ||
| assert resource_type in ["senders", "receivers"] | ||
|
|
||
| # Prevent this being executed twice in one test run | ||
| if resource_type in self.is05_resources["_requested"]: | ||
| return True, "" | ||
|
|
||
| path_url = "single/" + resource_type | ||
| full_url = self.connection_url + path_url | ||
| valid, resources = self.do_request("GET", full_url) | ||
| if not valid: | ||
| return False, "Connection API did not respond as expected: {}".format(resources) | ||
|
|
||
| schema = self.get_schema(CONNECTION_API_KEY, "GET", "/" + path_url, resources.status_code) | ||
| valid, message = self.check_response(schema, "GET", resources) | ||
| if not valid: | ||
| raise NMOSTestException(message) | ||
|
|
||
| # The following call to is05_utils.get_transporttype does not validate against the IS-05 schemas, | ||
| # which is good for allowing extended transport. The transporttype-response-schema.json schema is | ||
| # broken as it does not allow additional transport, nor x-nmos ones, nor vendor specific ones. | ||
| try: | ||
| for resource in resources.json(): | ||
| resource_id = resource.rstrip("/") | ||
| self.is05_resources[resource_type].append(resource_id) | ||
| if self.is05_utils.compare_api_version(self.apis[CONNECTION_API_KEY]["version"], "v1.1") >= 0: | ||
| transport_type = self.is05_utils.get_transporttype(resource_id, resource_type.rstrip("s")) | ||
| self.is05_resources["transport_types"][resource_id] = transport_type | ||
| else: | ||
| self.is05_resources["transport_types"][resource_id] = "urn:x-nmos:transport:rtp" | ||
| if resource_type == "senders": | ||
| transport_file = self.is05_utils.get_transportfile(resource_id) | ||
| self.is05_resources["transport_files"][resource_id] = transport_file | ||
| self.is05_resources["_requested"].append(resource_type) | ||
| except json.JSONDecodeError: | ||
| return False, "Non-JSON response returned from Node API" | ||
|
|
||
| return True, "" | ||
|
|
There was a problem hiding this comment.
This function is unused by the tests, so can be removed.
| def get_is05_partial_resources(self, resource_type): | |
| """Retrieve all Senders or Receivers from a Connection API, keeping hold of the returned IDs""" | |
| assert resource_type in ["senders", "receivers"] | |
| # Prevent this being executed twice in one test run | |
| if resource_type in self.is05_resources["_requested"]: | |
| return True, "" | |
| path_url = "single/" + resource_type | |
| full_url = self.connection_url + path_url | |
| valid, resources = self.do_request("GET", full_url) | |
| if not valid: | |
| return False, "Connection API did not respond as expected: {}".format(resources) | |
| schema = self.get_schema(CONNECTION_API_KEY, "GET", "/" + path_url, resources.status_code) | |
| valid, message = self.check_response(schema, "GET", resources) | |
| if not valid: | |
| raise NMOSTestException(message) | |
| # The following call to is05_utils.get_transporttype does not validate against the IS-05 schemas, | |
| # which is good for allowing extended transport. The transporttype-response-schema.json schema is | |
| # broken as it does not allow additional transport, nor x-nmos ones, nor vendor specific ones. | |
| try: | |
| for resource in resources.json(): | |
| resource_id = resource.rstrip("/") | |
| self.is05_resources[resource_type].append(resource_id) | |
| if self.is05_utils.compare_api_version(self.apis[CONNECTION_API_KEY]["version"], "v1.1") >= 0: | |
| transport_type = self.is05_utils.get_transporttype(resource_id, resource_type.rstrip("s")) | |
| self.is05_resources["transport_types"][resource_id] = transport_type | |
| else: | |
| self.is05_resources["transport_types"][resource_id] = "urn:x-nmos:transport:rtp" | |
| if resource_type == "senders": | |
| transport_file = self.is05_utils.get_transportfile(resource_id) | |
| self.is05_resources["transport_files"][resource_id] = transport_file | |
| self.is05_resources["_requested"].append(resource_type) | |
| except json.JSONDecodeError: | |
| return False, "Non-JSON response returned from Node API" | |
| return True, "" |
| def check_response_without_transport_params(self, schema, method, response): | ||
| """Confirm that a given Requests response conforms to the expected schema and has any expected headers without | ||
| considering the 'transport_params' attribute""" | ||
| ctype_valid, ctype_message = check_content_type(response.headers) | ||
| if not ctype_valid: | ||
| return False, ctype_message | ||
|
|
||
| cors_valid, cors_message = self.check_CORS(method, response.headers) | ||
| if not cors_valid: | ||
| return False, cors_message | ||
|
|
||
| fields_to_ignore = ["transport_params"] | ||
|
|
||
| data = response.json() | ||
|
|
||
| filtered_data = {k: v for k, v in data.items() if k not in fields_to_ignore} | ||
|
|
||
| filtered_data["transport_params"] = [] | ||
|
|
||
| try: | ||
| self.validate_schema(filtered_data, schema) | ||
| except ValidationError as e: | ||
| return False, "Response schema validation error {}".format(e) | ||
| except json.JSONDecodeError: | ||
| return False, "Invalid JSON received" | ||
|
|
||
| return True, ctype_message | ||
|
|
There was a problem hiding this comment.
This function is unused by the tests, so can be removed.
| def check_response_without_transport_params(self, schema, method, response): | |
| """Confirm that a given Requests response conforms to the expected schema and has any expected headers without | |
| considering the 'transport_params' attribute""" | |
| ctype_valid, ctype_message = check_content_type(response.headers) | |
| if not ctype_valid: | |
| return False, ctype_message | |
| cors_valid, cors_message = self.check_CORS(method, response.headers) | |
| if not cors_valid: | |
| return False, cors_message | |
| fields_to_ignore = ["transport_params"] | |
| data = response.json() | |
| filtered_data = {k: v for k, v in data.items() if k not in fields_to_ignore} | |
| filtered_data["transport_params"] = [] | |
| try: | |
| self.validate_schema(filtered_data, schema) | |
| except ValidationError as e: | |
| return False, "Response schema validation error {}".format(e) | |
| except json.JSONDecodeError: | |
| return False, "Invalid JSON received" | |
| return True, ctype_message |
Co-authored-by: jonathan-r-thorpe <64410119+jonathan-r-thorpe@users.noreply.github.com>
Co-authored-by: jonathan-r-thorpe <64410119+jonathan-r-thorpe@users.noreply.github.com>
- fix test suite name to add -01
Test Suite for BCP-005-02
Notes: