diff --git a/.gitignore b/.gitignore index 2cbb645a2faccf4ffa11a9aefb8e78d24a4560df..080d4b7feafa2dfbcb46f4bc64c3a5c142c6e667 100644 --- a/.gitignore +++ b/.gitignore @@ -4,4 +4,5 @@ results *.http .vscode/ .venv -.DS_Store \ No newline at end of file +.DS_Store +/tools/interoperability_visualizer/output/ diff --git a/doc/analysis/generaterobotdata.py b/doc/analysis/generaterobotdata.py index 6ab1cf7259909974b5c2928ec34e8ee3d4ed2547..0beb1569ebc42b1545de73ca456ccaea1d21040f 100644 --- a/doc/analysis/generaterobotdata.py +++ b/doc/analysis/generaterobotdata.py @@ -654,6 +654,7 @@ class GenerateRobotData: self.test_suite = { 'tp_id': tp_id, + 'name': self.generate_iop_test_name(), 'test_objective': self.suite.doc, 'reference': reference, 'config_id': str(), @@ -670,6 +671,16 @@ class GenerateRobotData: test_name=self.robot.test_case_names[0] ) + def generate_iop_test_name(self) -> str: + test_names = self.robot.test_case_names + if len(test_names) == 1: + return test_names[0] + + _, separator, name = test_names[0].partition(' ') + name = name.replace(' With Default Context', '') + name = name.replace(' With User Context', '') + return f'{self.robot.test_suite}{separator}{name}' + def _get_iop_setup(self, test_name: str): test = next((item for item in self.suite.tests if item.name == test_name), None) if test is None: @@ -971,7 +982,7 @@ class GenerateRobotData: # Get test information from the parsed robot file tags = self.robot.get_iop_test_tags(test_name) comments = self.robot.get_iop_comments(test_name) - + test_case = { 'name': test_name, 'permutation_iop_id': self.base_TP_id, diff --git a/doc/tests/test_iop_preconditions.py b/doc/tests/test_iop_preconditions.py index b0798716cdb4e38603db057b6d6dd45836929c61..72683049fb2737d8c4ba8d2590ab7d7e54501415 100644 --- a/doc/tests/test_iop_preconditions.py +++ b/doc/tests/test_iop_preconditions.py @@ -11,7 +11,13 @@ from analysis.generaterobotdata import GenerateRobotData class IopPreconditionsTest(unittest.TestCase): - def generate(self, variables: str, setup: str, test_body: str = ' No Operation'): + def generate( + self, + variables: str, + setup: str, + test_body: str = ' No Operation', + test_name: str = 'IOP_999_01 Generated Preconditions', + ): robot = f'''*** Settings *** Documentation Objective Test Setup Prepare Preconditions @@ -22,7 +28,7 @@ Test Setup Prepare Preconditions *** Test Cases *** -IOP_999_01 Generated Preconditions +{test_name} [Documentation] Pre-conditions: documentation must be ignored. [Tags] ... since_v1.6.1 @@ -70,6 +76,20 @@ Prepare Preconditions ['since_v1.6.1', 'iop', '4_3_3'] ) + def test_preserves_context_in_single_test_case_name(self): + name = 'IOP_999_01 Scenario With Default Context' + self.generate( + variables='${b1_url} ${EMPTY}', + setup=' No Operation', + test_name=name, + ) + + self.assertEqual(self.generated_info['name'], name) + self.assertEqual( + self.generated_info['test_cases'][0]['name'], + name + ) + def test_generates_mixed_context_and_data_objects(self): result = self.generate( variables='''${first_payload} fixtures/first-entity.jsonld @@ -181,12 +201,12 @@ ${b1_url} ${EMPTY} *** Test Cases *** -IOP_999_01_01 Default Context +IOP_999_01_01 Scenario With Default Context [Tags] since_v1.6.1 iop 4_3_3 default-context [Setup] Prepare Preconditions ${core_context} ${core_context} -IOP_999_01_02 User Context +IOP_999_01_02 Scenario With User Context [Tags] since_v1.6.1 iop 4_3_3 user-context [Setup] Prepare Preconditions context=${ngsild_test_suite_context} ${ngsild_test_suite_context} @@ -210,6 +230,14 @@ Prepare Preconditions user_conditions = self.generated_info['test_cases'][1]['initial_conditions'] self.assertNotIn('initial_conditions', self.generated_info) + self.assertEqual(self.generated_info['name'], 'IOP_999_01 Scenario') + self.assertEqual( + [test_case['name'] for test_case in self.generated_info['test_cases']], + [ + 'IOP_999_01_01 Scenario With Default Context', + 'IOP_999_01_02 Scenario With User Context', + ] + ) self.assertEqual( default_conditions['pre_conditions'][0], { diff --git a/tools/interoperability_visualizer/README.md b/tools/interoperability_visualizer/README.md new file mode 100644 index 0000000000000000000000000000000000000000..10f2d0bedfab4ffafa1f99ea4f6f99d5dd70e4bd --- /dev/null +++ b/tools/interoperability_visualizer/README.md @@ -0,0 +1,70 @@ +# Interoperability Architecture Visualizer + +Generates an offline interactive architecture view of the Robot Framework tests in +`IOP_TP/NGSI-LD/Interoperability`. + +## Run + +With the project environment active: + +```console +python tools/interoperability_visualizer/generate.py +``` + +Using the local `ttf3` Conda environment: + +```console +conda run -n ttf3 python tools/interoperability_visualizer/generate.py +``` + +The generated page is written to: + +```text +tools/interoperability_visualizer/output/interoperability-architecture.html +``` + +Select a different Robot file or directory with `--source`: + +```console +python tools/interoperability_visualizer/generate.py \ + --source IOP_TP/NGSI-LD/Interoperability/Provision/CreateEntity +``` + +Select another filename inside the dedicated output directory with `--output-name`: + +```console +python tools/interoperability_visualizer/generate.py --output-name provision.html +``` + +Output paths and non-HTML output names are rejected. The complete `output/` directory is ignored by Git. + +## Viewer + +- Select any concrete interoperability test permutation. +- Switch between the full-size **Preview** graph and **Expected output** workspace. +- Brokers use a left-to-right hierarchy rooted at `b1`; each registration hop advances one level. +- Hover over a broker to inspect every entity created in that broker during setup. +- Hover over a registration arc to inspect its effective creation payload. +- Browse assertion-derived expected responses by broker or in execution order. +- Timeline view treats each consecutive inline comment block as one numbered Robot step and nests every correlated + Robot keyword operation and expected HTTP response below it. +- Every numbered comment step has its own assertion panel, including `Client sends` and `Agent checks` steps. +- Expand a response step to see its prominent HTTP status, request, assertions, and expected JSON inline. +- Click a broker or registration to pin it; press Escape or use the close button to unpin it. +- Drag brokers to rearrange the architecture. +- Drag empty graph space to pan, use the mouse wheel to zoom, and use **Reset view** to restore the hierarchy and viewport. + +Registration arrows start at the broker where the registration is created and point to the broker referenced by its +endpoint. Effective payloads resolve entity IDs and contexts while keeping runtime broker URLs and generated +registration IDs symbolic. + +Expected responses are derived statically from the Robot request and assertion calls. HTTP 201 and 404 steps remain +status-only, and unsupported future assertion patterns are displayed as unresolved instead of inventing a body. + +## Test + +```console +conda run -n ttf3 python -m unittest discover \ + -s tools/interoperability_visualizer/tests \ + -p 'test_*.py' +``` diff --git a/tools/interoperability_visualizer/expected_outputs.py b/tools/interoperability_visualizer/expected_outputs.py new file mode 100644 index 0000000000000000000000000000000000000000..6572ec1cadd42a74a2a2100b77666ccf6e69e4dc --- /dev/null +++ b/tools/interoperability_visualizer/expected_outputs.py @@ -0,0 +1,554 @@ +import copy +import re +from dataclasses import dataclass +from pathlib import Path +from urllib.parse import quote, urlencode + +from robot.api import get_model + +from extractor import ( + ExtractionError, + _broker_name, + _call_arguments, + _load_payload, + _payload_path, +) + + +REQUESTS = { + "Create Entity": ("POST", "/ngsi-ld/v1/entities"), + "Retrieve Entity": ("GET", "/ngsi-ld/v1/entities/{id}"), + "Query Entities": ("GET", "/ngsi-ld/v1/entities"), + "Query Entities Via POST": ("POST", "/ngsi-ld/v1/entityOperations/query"), +} +SIGNATURES = { + "Create Entity": ("filename", "entity_id", "local", "broker_url", "context"), + "Retrieve Entity": ( + "id", "accept", "attrs", "context", "geometryProperty", "options", "format", "lang", + "join", "joinLevel", "pick", "omit", "local", "broker_url", "type", + ), + "Query Entities": ( + "entity_ids", "entity_types", "accept", "attrs", "context", "geoproperty", "options", + "limit", "entity_id_pattern", "scopeq", "georel", "coordinates", "geometry", "count", + "q", "local", "datasetId", "join", "joinLevel", "pick", "omit", "orderBy", "broker_url", + ), + "Query Entities Via POST": ( + "entities", "content_type", "accept", "context", "attrs", "geometry_property", "join", + "joinLevel", "local", "options", "broker_url", + ), +} +VARIABLE = re.compile(r"^[$@&]\{([^}]+)\}(.*)$") +VARIABLE_ANYWHERE = re.compile(r"[$@&]\{([^}]+)\}") +PATH_PART = re.compile(r"\[([^]]+)\]") +JSON_ENTITY_ID = re.compile(r"@\.id\s*==\s*['\"]([^'\"]+)['\"]") + + +@dataclass(frozen=True) +class Response: + request_id: str + + +@dataclass(frozen=True) +class Ref: + request_id: str + path: tuple = () + + +@dataclass(frozen=True) +class Unknown: + expression: str + + +def extract_expected_steps( + robot_file: Path, + suite, + test, + setup_env: dict, + root: Path, + relative_robot_file: str, +) -> tuple[list[dict], list[dict]]: + execution = next((keyword for keyword in suite.resource.keywords if keyword.name == test.template), None) + if execution is None: + raise ExtractionError( + f"{relative_robot_file}: template keyword '{test.template}' for '{test.name}' was not found" + ) + + model = get_model(str(robot_file)) + parsed = _parsed_keyword(model, str(test.template)) + if parsed is None: + raise ExtractionError( + f"{relative_robot_file}: parsed template keyword '{test.template}' for '{test.name}' was not found" + ) + + env = _initial_environment(setup_env) + env.update(_template_variables(test, execution)) + evaluator = Evaluator(env, root, relative_robot_file, test.name) + evaluator.process(parsed.body) + return evaluator.finish(), evaluator.finish_timeline() + + +class Evaluator: + def __init__(self, env: dict, root: Path, robot_file: str, test_name: str): + self.env = env + self.root = root + self.robot_file = robot_file + self.test_name = test_name + self.requests = [] + self.by_id = {} + self.bindings = {} + self.comment = "" + self.timeline_steps = [] + self.current_timeline = None + + def process(self, body) -> None: + index = 0 + while index < len(body): + item = body[index] + if type(item).__name__ == "Comment": + comments = [] + source_line = int(item.lineno) + while index < len(body) and type(body[index]).__name__ == "Comment": + comments.append(_comment_text(body[index])) + index += 1 + self.comment = "\n".join(comment for comment in comments if comment) + self.current_timeline = { + "id": f"timeline-step-{len(self.timeline_steps) + 1}", + "order": len(self.timeline_steps) + 1, + "text": self.comment, + "source_line": source_line, + "operations": [], + "request_ids": [], + } + self.timeline_steps.append(self.current_timeline) + continue + if type(item).__name__ == "KeywordCall": + name = str(item.keyword) + args = tuple(str(arg) for arg in item.args) + request_ids = self._request_ids(args) + request_count = len(self.requests) + self._call(name, args, item.assign, item.lineno) + if len(self.requests) > request_count: + request_ids.append(self.requests[-1]["id"]) + self._timeline_operation(name, item.lineno, request_ids) + index += 1 + + def _timeline_operation(self, name: str, source_line: int, request_ids: list[str]) -> None: + if self.current_timeline is None: + return + unique_ids = list(dict.fromkeys(request_ids)) + if name in REQUESTS: + kind = "request" + elif name.startswith("Check "): + kind = "assertion" + else: + kind = "preparation" + self.current_timeline["operations"].append( + { + "id": f"{self.current_timeline['id']}-operation-{len(self.current_timeline['operations']) + 1}", + "order": len(self.current_timeline["operations"]) + 1, + "keyword": name, + "kind": kind, + "source_line": int(source_line), + "request_ids": unique_ids, + } + ) + for request_id in unique_ids: + if request_id not in self.current_timeline["request_ids"]: + self.current_timeline["request_ids"].append(request_id) + + def _request_ids(self, expressions: tuple[str, ...]) -> list[str]: + request_ids = [] + for expression in expressions: + key, separator, value = expression.partition("=") + target = value if separator and re.match(r"^[A-Za-z_][A-Za-z0-9_-]*$", key) else expression + request_ids.extend(_request_ids_in_value(self.value(target))) + return list(dict.fromkeys(request_ids)) + + def _call(self, name: str, args: tuple[str, ...], assign, lineno: int) -> None: + if name in REQUESTS: + self._request(name, args, assign, lineno) + elif name == "Load Entity": + self._load_entity(args, assign) + elif name == "Create List": + self._assign(assign, [self.value(arg) for arg in args]) + elif name == "Create Dictionary": + self._assign(assign, self._dictionary(args)) + elif name == "Remove From Dictionary" and args: + target = self.value(args[0]) + if isinstance(target, dict): + for key in args[1:]: + target.pop(str(self.value(key)), None) + elif name == "Keep In Dictionary" and args: + target = self.value(args[0]) + if isinstance(target, dict): + keys = {str(self.value(key)) for key in args[1:]} + for key in tuple(target): + if key not in keys: + del target[key] + elif name == "Set To Dictionary" and args: + target = self.value(args[0]) + if isinstance(target, dict): + target.update(self._dictionary(args[1:])) + elif name == "Get Value From JSON" and len(args) >= 2: + self._select_json_entity(args, assign) + elif name == "Check Response Status Code" and len(args) >= 2: + request = self._request_from(args[1]) + if request: + status = self.value(args[0]) + try: + request["status"] = int(status) + except (TypeError, ValueError): + request["status"] = None + self._expectation(request) + elif name == "Check Response Body Containing Entities URIS set to" and len(args) >= 2: + request = self._request_from(args[1]) + if request: + ids = self.value(args[0]) + request["expected_ids"] = ids if isinstance(ids, list) else Unknown(args[0]) + self._expectation(request) + elif name == "Check Resource Set To" and len(args) >= 2: + self._resource_equality(args[0], args[1]) + + def _request(self, name: str, args: tuple[str, ...], assign, lineno: int) -> None: + values = _call_arguments(args, SIGNATURES[name], name) + if "broker_url" not in values: + raise ExtractionError(f"{self.robot_file}: {name} lacks broker_url in '{self.test_name}'") + broker = _broker_name( + values["broker_url"], + _broker_environment(self.env), + f"{self.robot_file}: {name} broker in '{self.test_name}'", + ) + request_id = f"step-{len(self.requests) + 1}" + method, base_endpoint = REQUESTS[name] + endpoint = self._endpoint(name, base_endpoint, values) + request = { + "id": request_id, + "order": len(self.requests) + 1, + "method": method, + "endpoint": endpoint, + "broker": broker, + "status": None, + "request_comment": self.comment, + "expectation_comments": [], + "expected_body": None, + "resolution": "unresolved", + "source_line": int(lineno), + "operation": name, + "expected_ids": None, + } + self.requests.append(request) + self.by_id[request_id] = request + self._assign(assign, Response(request_id)) + + def _endpoint(self, name: str, base_endpoint: str, values: dict[str, str]) -> str: + if name == "Retrieve Entity": + entity_id = self.value(values.get("id", "")) + if isinstance(entity_id, Unknown): + entity_id = values.get("id", "") + base_endpoint = base_endpoint.format(id=quote(str(entity_id), safe=":-._~")) + params = [] + type_value = values.get("type") if name == "Retrieve Entity" else values.get("entity_types") + for label, raw in (("type", type_value), ("attrs", values.get("attrs")), ("local", values.get("local"))): + if raw is None: + continue + value = self.value(raw) + if not isinstance(value, Unknown) and value != "": + params.append((label, str(value))) + return f"{base_endpoint}?{urlencode(params)}" if params else base_endpoint + + def _load_entity(self, args: tuple[str, ...], assign) -> None: + if len(args) < 2: + self._assign(assign, Unknown("Load Entity")) + return + filename = self.value(args[0]) + entity_id = self.value(args[1]) + if isinstance(filename, Unknown) or isinstance(entity_id, Unknown): + self._assign(assign, Unknown("Load Entity")) + return + path = _payload_path(self.root / "data" / "entities", str(filename), self.root, self.robot_file) + payload = _load_payload(path, self.robot_file) + payload["id"] = entity_id + self._assign(assign, payload) + + def _select_json_entity(self, args: tuple[str, ...], assign) -> None: + request = self._request_from(args[0]) + path = self.value(args[1]) + if not request or isinstance(path, Unknown): + self._assign(assign, Unknown(args[1])) + return + match = JSON_ENTITY_ID.search(str(path)) + if match is None: + self._assign(assign, Unknown(args[1])) + return + self._assign(assign, [Ref(request["id"], ("entity", match.group(1)))]) + + def _resource_equality(self, expected_expression: str, actual_expression: str) -> None: + expected = self.value(expected_expression) + actual = self.value(actual_expression) + request = self._request_from(actual_expression) or self._request_from(expected_expression) + if request: + self._expectation(request) + if isinstance(actual, Ref): + self.bindings[actual] = copy.deepcopy(expected) + elif isinstance(expected, Ref): + self.bindings[expected] = copy.deepcopy(actual) + + def _dictionary(self, args: tuple[str, ...]) -> dict: + result = {} + positional = [] + for arg in args: + key, separator, value = arg.partition("=") + if separator: + result[key] = self.value(value) + else: + positional.append(self.value(arg)) + for index in range(0, len(positional) - 1, 2): + result[str(positional[index])] = positional[index + 1] + return result + + def _expectation(self, request: dict) -> None: + if not self.comment: + return + if self.comment != request["request_comment"] or self.comment.lstrip().lower().startswith("agent"): + if self.comment not in request["expectation_comments"]: + request["expectation_comments"].append(self.comment) + + def _request_from(self, expression: str): + value = self.value(expression) + if isinstance(value, Response): + return self.by_id.get(value.request_id) + if isinstance(value, Ref): + return self.by_id.get(value.request_id) + return None + + def _assign(self, assign, value) -> None: + if not assign: + return + name = _variable_name(str(assign[0]).rstrip("=")) + if name: + self.env[name] = value + + def value(self, expression: str, seen: set[str] | None = None): + raw = str(expression) + match = VARIABLE.match(raw) + if match: + name, suffix = match.groups() + response_part = None + if name.endswith(".json()"): + response_part = "body" + name = name[:-7] + elif name.endswith(".status_code"): + response_part = "status" + name = name[:-12] + if name in self.env: + if seen is None: + seen = set() + if name in seen: + return Unknown(raw) + seen = {*seen, name} + value = self.env[name] + if isinstance(value, str) and VARIABLE.match(value): + value = self.value(value, seen) + if response_part == "body" and isinstance(value, Response): + value = Ref(value.request_id) + elif response_part == "status" and isinstance(value, Response): + return value + elif response_part is not None: + return Unknown(raw) + return _apply_suffix(value, suffix, raw) + if not suffix: + return Unknown(raw) + + unknown = False + + def replace(variable_match): + nonlocal unknown + name = variable_match.group(1) + if name not in self.env: + unknown = True + return variable_match.group(0) + value = self.env[name] + if isinstance(value, str) and VARIABLE.match(value): + value = self.value(value) + if isinstance(value, (dict, list, Ref, Response, Unknown)): + unknown = True + return variable_match.group(0) + return str(value) + + resolved = VARIABLE_ANYWHERE.sub(replace, raw) + return Unknown(raw) if unknown else resolved + + def finish(self) -> list[dict]: + result = [] + for request in self.requests: + status = request["status"] + body = None + resolution = "unresolved" + if status == 200: + if request["operation"].startswith("Query Entities"): + ids = self.resolve(request["expected_ids"]) + if isinstance(ids, list) and all(not isinstance(item, Unknown) for item in ids): + bodies = [self.resolve(Ref(request["id"], ("entity", str(entity_id)))) for entity_id in ids] + if all(_is_concrete(item) for item in bodies): + body = sorted(bodies, key=lambda item: str(item.get("id", ""))) + resolution = "exact" + else: + candidate = self.resolve(Ref(request["id"])) + if _is_concrete(candidate): + body = candidate + resolution = "exact" + elif status is not None: + resolution = "status-only" + + result.append( + { + "id": request["id"], + "order": request["order"], + "method": request["method"], + "endpoint": request["endpoint"], + "broker": request["broker"], + "status": status, + "request_comment": request["request_comment"], + "expectation_comments": request["expectation_comments"], + "expected_body": body, + "resolution": resolution, + "source_line": request["source_line"], + } + ) + return result + + def finish_timeline(self) -> list[dict]: + timeline = copy.deepcopy(self.timeline_steps) + for step in timeline: + step["group"] = step["order"] + return timeline + + def resolve(self, value, trail: set[Ref] | None = None): + if isinstance(value, Unknown): + return value + if isinstance(value, Ref): + if trail is None: + trail = set() + if value in trail: + return Unknown(str(value)) + trail = {*trail, value} + if value in self.bindings: + return self.resolve(self.bindings[value], trail) + for length in range(len(value.path) - 1, -1, -1): + parent = Ref(value.request_id, value.path[:length]) + if parent not in self.bindings: + continue + resolved = self.resolve(self.bindings[parent], trail) + return _apply_path(resolved, value.path[length:], str(value)) + return Unknown(str(value)) + if isinstance(value, dict): + return {key: self.resolve(item, trail) for key, item in value.items()} + if isinstance(value, list): + return [self.resolve(item, trail) for item in value] + return value + + +def _parsed_keyword(model, name: str): + for section in model.sections: + if type(section).__name__ != "KeywordSection": + continue + for keyword in section.body: + if str(keyword.name) == name: + return keyword + return None + + +def _template_variables(test, keyword) -> dict: + if not test.body: + return {} + call = test.body[0] + positional = [] + named = [] + names = set(keyword.args.argument_names) + for argument in (str(item) for item in call.args): + name, separator, value = argument.partition("=") + if separator and name in names: + named.append((name, value)) + else: + positional.append(argument) + try: + mapped_positional, mapped_named = keyword.args.map(positional, named) + except Exception as error: + raise ExtractionError(f"Unable to map template arguments for '{test.name}': {error}") from error + values = {name: str(value) for name, value in zip(keyword.args.positional, mapped_positional)} + values.update((name, str(value)) for name, value in mapped_named) + for name, value in keyword.args.defaults.items(): + values.setdefault(name, str(value)) + return values + + +def _initial_environment(env: dict) -> dict: + result = {} + for key, value in env.items(): + name = _variable_name(str(key)) + if name: + result[name] = copy.deepcopy(value) + return result + + +def _broker_environment(env: dict) -> dict: + return {f"${{{name}}}": value for name, value in env.items() if isinstance(value, str)} + + +def _variable_name(value: str) -> str | None: + match = re.match(r"^[$@&]\{([^}]+)\}$", value) + return match.group(1) if match else None + + +def _comment_text(comment) -> str: + token = next((token for token in comment.tokens if token.type == "COMMENT"), None) + return token.value.lstrip("#").strip() if token else "" + + +def _apply_suffix(value, suffix: str, expression: str): + if not suffix: + return value + parts = PATH_PART.findall(suffix) + if "".join(f"[{part}]" for part in parts) != suffix: + return Unknown(expression) + if isinstance(value, Ref): + return Ref(value.request_id, value.path + tuple(_path_key(part) for part in parts)) + return _apply_path(value, tuple(_path_key(part) for part in parts), expression) + + +def _apply_path(value, path: tuple, expression: str): + current = value + for index, key in enumerate(path): + if isinstance(current, Unknown): + return current + if isinstance(current, Ref): + return Ref(current.request_id, current.path + path[index:]) + try: + current = current[key] + except (KeyError, IndexError, TypeError): + return Unknown(expression) + return current + + +def _path_key(value: str): + return int(value) if value.isdigit() else value + + +def _is_concrete(value) -> bool: + if isinstance(value, (Unknown, Ref, Response)): + return False + if isinstance(value, dict): + return all(_is_concrete(item) for item in value.values()) + if isinstance(value, list): + return all(_is_concrete(item) for item in value) + return True + + +def _request_ids_in_value(value) -> list[str]: + if isinstance(value, (Response, Ref)): + return [value.request_id] + if isinstance(value, dict): + return [request_id for item in value.values() for request_id in _request_ids_in_value(item)] + if isinstance(value, (list, tuple)): + return [request_id for item in value for request_id in _request_ids_in_value(item)] + return [] diff --git a/tools/interoperability_visualizer/extractor.py b/tools/interoperability_visualizer/extractor.py new file mode 100644 index 0000000000000000000000000000000000000000..ccfb92a9c94774322476850c474748c400e60d97 --- /dev/null +++ b/tools/interoperability_visualizer/extractor.py @@ -0,0 +1,401 @@ +import ast +import copy +import json +import re +from collections import defaultdict +from pathlib import Path + +from robot.api import TestSuiteBuilder + + +MODES = ("inclusive", "auxiliary", "redirect", "exclusive") +MODE_COLORS = { + "inclusive": "#009E73", + "auxiliary": "#0072B2", + "redirect": "#E69F00", + "exclusive": "#CC79A7", +} +VARIABLE_PATTERN = re.compile(r"^\$\{[^}]+\}$") +BROKER_PATTERN = re.compile(r"^\$\{(b\d+)_url\}$") + + +class ExtractionError(ValueError): + pass + + +def extract_source(source: Path | str, repo_root: Path | str) -> dict: + root = Path(repo_root).resolve() + source_path = Path(source) + if not source_path.is_absolute(): + source_path = root / source_path + source_path = source_path.resolve() + + if not source_path.exists(): + raise ExtractionError(f"Source does not exist: {source_path}") + if source_path.is_file(): + if source_path.suffix.lower() != ".robot": + raise ExtractionError(f"Source is not a Robot file: {source_path}") + robot_files = [source_path] + else: + robot_files = sorted(source_path.rglob("*.robot")) + if not robot_files: + raise ExtractionError(f"No Robot files found in: {source_path}") + + config_variables = _load_config_variables(root / "resources" / "variables.py") + tests = [] + for robot_file in robot_files: + tests.extend(_extract_robot_file(robot_file, root, config_variables)) + + return { + "modes": [ + { + "id": mode, + "label": mode.capitalize(), + "color": MODE_COLORS[mode], + } + for mode in MODES + ], + "tests": tests, + } + + +def _extract_robot_file(robot_file: Path, root: Path, config_variables: dict[str, str]) -> list[dict]: + try: + suite = TestSuiteBuilder().build(str(robot_file)) + except Exception as error: + raise ExtractionError(f"Unable to parse {robot_file}: {error}") from error + + suite_variables = _suite_variables(suite) + variables = {"${EMPTY}": "", **config_variables, **suite_variables} + relative_robot_file = _relative_path(robot_file, root) + results = [] + + for test in suite.tests: + setup = next((keyword for keyword in suite.resource.keywords if keyword.name == test.setup.name), None) + if setup is None: + raise ExtractionError( + f"{relative_robot_file}: setup keyword '{test.setup.name}' for '{test.name}' was not found" + ) + + setup_variables = _setup_variables(test, setup) + env = {**variables, **setup_variables} + entities_by_broker: dict[str, list[dict]] = defaultdict(list) + registrations = [] + registration_context = "${ngsild_test_suite_context}" + + for keyword in setup.body: + if keyword.name == "Set Test Variable" and len(keyword.args) >= 2: + env[str(keyword.args[0])] = str(keyword.args[1]) + continue + + if keyword.name == "Create Entity": + entity = _extract_entity(keyword.args, env, root, relative_robot_file, test.name) + entities_by_broker[entity["broker"]].append( + { + "source_file": entity["source_file"], + "payload": entity["payload"], + } + ) + continue + + if keyword.name == "Create List" and len(keyword.args) == 5: + args = tuple(str(argument) for argument in keyword.args) + mode = args[2].lower() + if mode in MODES: + registrations.append( + _extract_registration( + args, + len(registrations) + 1, + env, + root, + relative_robot_file, + test.name, + ) + ) + elif _registration_like(args, env): + raise ExtractionError( + f"{relative_robot_file}: unsupported registration mode '{args[2]}' in '{test.name}'" + ) + continue + + if keyword.name == "Compose IOP Configuration": + registration_context = _compose_context(keyword.args) + + resolved_registration_context = _resolve( + registration_context, + env, + f"{relative_robot_file}: registration context in '{test.name}'", + ) + for registration in registrations: + if resolved_registration_context != "": + registration["payload"]["@context"] = resolved_registration_context + + broker_ids = set(entities_by_broker) + for registration in registrations: + broker_ids.add(registration["source"]) + broker_ids.add(registration["target"]) + brokers = [ + { + "id": broker_id, + "entities": entities_by_broker.get(broker_id, []), + } + for broker_id in sorted(broker_ids, key=_broker_sort_key) + ] + + from expected_outputs import extract_expected_steps + + steps, timeline_steps = extract_expected_steps( + robot_file, + suite, + test, + env, + root, + relative_robot_file, + ) + + results.append( + { + "name": test.name, + "suite": relative_robot_file, + "objective": str(suite.doc), + "tags": [str(tag) for tag in test.tags], + "context": resolved_registration_context, + "brokers": brokers, + "registrations": registrations, + "steps": steps, + "timeline_steps": timeline_steps, + } + ) + + return results + + +def _extract_entity(args, env: dict, root: Path, robot_file: str, test_name: str) -> dict: + values = _call_arguments(args, ("filename", "entity_id", "local", "broker_url", "context"), "Create Entity") + for required in ("filename", "entity_id", "broker_url"): + if required not in values: + raise ExtractionError(f"{robot_file}: Create Entity lacks {required} in '{test_name}'") + + filename = _resolve(values["filename"], env, f"{robot_file}: entity filename in '{test_name}'") + entity_id = _resolve(values["entity_id"], env, f"{robot_file}: entity ID in '{test_name}'") + context = _resolve( + values.get("context", "${ngsild_test_suite_context}"), + env, + f"{robot_file}: entity context in '{test_name}'", + ) + broker = _broker_name(values["broker_url"], env, f"{robot_file}: entity broker in '{test_name}'") + payload_path = _payload_path(root / "data" / "entities", filename, root, robot_file) + payload = _load_payload(payload_path, robot_file) + payload["id"] = entity_id + payload["@context"] = context + + return { + "broker": broker, + "source_file": _relative_path(payload_path, root), + "payload": payload, + } + + +def _extract_registration( + args: tuple[str, ...], + index: int, + env: dict, + root: Path, + robot_file: str, + test_name: str, +) -> dict: + entity_id = _resolve(args[0], env, f"{robot_file}: registration entity ID in '{test_name}'") + filename = _resolve(args[1], env, f"{robot_file}: registration filename in '{test_name}'") + mode = args[2].lower() + target = _broker_name(args[3], env, f"{robot_file}: registration endpoint in '{test_name}'") + source = _broker_name(args[4], env, f"{robot_file}: registering broker in '{test_name}'") + payload_path = _payload_path(root / "data", filename, root, robot_file) + payload = _load_payload(payload_path, robot_file) + payload["id"] = f"${{registration_id{index}}}" + payload["endpoint"] = f"${{{target}_url}}" + payload["mode"] = mode + if entity_id != "": + _inject_entity_id(payload, entity_id) + + return { + "id": f"registration-{index}", + "source": source, + "target": target, + "mode": mode, + "source_file": _relative_path(payload_path, root), + "payload": payload, + } + + +def _inject_entity_id(value, entity_id: str) -> None: + if isinstance(value, dict): + entities = value.get("entities") + if isinstance(entities, list): + for entity in entities: + if isinstance(entity, dict): + entity["id"] = entity_id + for nested in value.values(): + _inject_entity_id(nested, entity_id) + elif isinstance(value, list): + for nested in value: + _inject_entity_id(nested, entity_id) + + +def _call_arguments(args, signature: tuple[str, ...], keyword_name: str) -> dict[str, str]: + values = {} + position = 0 + for argument in (str(item) for item in args): + key, separator, value = argument.partition("=") + if separator and key in signature: + values[key] = value + continue + while position < len(signature) and signature[position] in values: + position += 1 + if position >= len(signature): + raise ExtractionError(f"{keyword_name} has an unexpected argument '{argument}'") + values[signature[position]] = argument + position += 1 + return values + + +def _setup_variables(test, setup) -> dict[str, str]: + argument_names = set(setup.args.argument_names) + positional = [] + named = [] + for argument in (str(item) for item in test.setup.args): + name, separator, value = argument.partition("=") + if separator and name in argument_names: + named.append((name, value)) + else: + positional.append(argument) + + try: + mapped_positional, mapped_named = setup.args.map(positional, named) + except Exception as error: + raise ExtractionError(f"Unable to map setup arguments for '{test.name}': {error}") from error + + variables = { + f"${{{name}}}": str(value) + for name, value in zip(setup.args.positional, mapped_positional) + } + variables.update((f"${{{name}}}", str(value)) for name, value in mapped_named) + for name, value in setup.args.defaults.items(): + variables.setdefault(f"${{{name}}}", str(value)) + return variables + + +def _suite_variables(suite) -> dict[str, str]: + variables = {} + for variable in suite.resource.variables: + values = tuple(str(value) for value in variable.value) + if len(values) != 1: + continue + variables[str(variable.name)] = values[0] + return variables + + +def _load_config_variables(path: Path) -> dict[str, str]: + if not path.is_file(): + raise ExtractionError(f"Variable file does not exist: {path}") + try: + tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) + except (OSError, SyntaxError) as error: + raise ExtractionError(f"Unable to parse variable file {path}: {error}") from error + + variables = {} + for node in tree.body: + if not isinstance(node, ast.Assign) or len(node.targets) != 1 or not isinstance(node.targets[0], ast.Name): + continue + try: + value = ast.literal_eval(node.value) + except (ValueError, TypeError): + continue + if isinstance(value, (str, int, float, bool)) or value is None: + variables[f"${{{node.targets[0].id}}}"] = str(value) if value is not None else "" + return variables + + +def _resolve(value: str, env: dict[str, str], field: str) -> str: + original = str(value) + result = original + visited = set() + while VARIABLE_PATTERN.match(result): + if result in visited: + raise ExtractionError(f"{field} contains a circular variable reference: {original}") + visited.add(result) + if result not in env: + raise ExtractionError(f"{field} contains an undefined variable: {result}") + result = str(env[result]) + return result + + +def _broker_name(value: str, env: dict[str, str], field: str) -> str: + original = str(value) + result = original + visited = set() + while result not in visited: + match = BROKER_PATTERN.match(result) + if match is not None: + return match.group(1) + visited.add(result) + if result not in env: + break + result = str(env[result]) + raise ExtractionError(f"{field} is not a broker variable: {original}") + + +def _registration_like(args: tuple[str, ...], env: dict[str, str]) -> bool: + try: + _broker_name(args[3], env, "Registration endpoint") + _broker_name(args[4], env, "Registering broker") + return True + except ExtractionError: + return False + + +def _compose_context(args) -> str: + string_args = tuple(str(argument) for argument in args) + for argument in string_args: + key, separator, value = argument.partition("=") + if separator and key == "ld_context": + return value + if len(string_args) > 1: + return string_args[1] + return "${ngsild_test_suite_context}" + + +def _payload_path(base: Path, relative: str, root: Path, robot_file: str) -> Path: + normalized = relative.replace("\\", "/") + candidate = (base / normalized).resolve() + resolved_base = base.resolve() + if candidate != resolved_base and resolved_base not in candidate.parents: + raise ExtractionError(f"{robot_file}: unsafe payload path '{relative}'") + if not candidate.is_file(): + raise ExtractionError(f"{robot_file}: payload does not exist: {_display_path(candidate, root)}") + return candidate + + +def _load_payload(path: Path, robot_file: str) -> dict: + try: + payload = json.loads(path.read_text(encoding="utf-8")) + except (OSError, UnicodeError, json.JSONDecodeError) as error: + raise ExtractionError(f"{robot_file}: invalid JSON payload {_display_path(path, path.parents[2])}: {error}") from error + if not isinstance(payload, dict): + raise ExtractionError(f"{robot_file}: payload must be a JSON object: {path}") + return copy.deepcopy(payload) + + +def _relative_path(path: Path, root: Path) -> str: + try: + return path.resolve().relative_to(root.resolve()).as_posix() + except ValueError: + return path.resolve().as_posix() + + +def _display_path(path: Path, root: Path) -> str: + return _relative_path(path, root) + + +def _broker_sort_key(broker_id: str) -> tuple[int, str]: + match = re.match(r"^b(\d+)$", broker_id) + return (int(match.group(1)), broker_id) if match else (10**9, broker_id) diff --git a/tools/interoperability_visualizer/generate.py b/tools/interoperability_visualizer/generate.py new file mode 100644 index 0000000000000000000000000000000000000000..652ae38a591f853008579501475e9dc940396233 --- /dev/null +++ b/tools/interoperability_visualizer/generate.py @@ -0,0 +1,85 @@ +import argparse +import json +import sys +from pathlib import Path + +from extractor import ExtractionError, extract_source + + +TOOL_DIR = Path(__file__).resolve().parent +REPO_ROOT = TOOL_DIR.parents[1] +DEFAULT_SOURCE = REPO_ROOT / "IOP_TP" / "NGSI-LD" / "Interoperability" +DEFAULT_OUTPUT_NAME = "interoperability-architecture.html" +DATA_TOKEN = "__IOP_VIEWER_DATA__" + + +def validate_output_name(value: str) -> str: + if ( + value in {"", ".", ".."} + or "/" in value + or "\\" in value + or Path(value).name != value + or not value.lower().endswith(".html") + ): + raise argparse.ArgumentTypeError("output name must be a single .html filename") + return value + + +def output_path(output_name: str, tool_dir: Path = TOOL_DIR) -> Path: + name = validate_output_name(output_name) + return tool_dir / "output" / name + + +def render_viewer(data: dict, template: str) -> str: + if template.count(DATA_TOKEN) != 1: + raise ValueError(f"Template must contain exactly one {DATA_TOKEN} token") + serialized = json.dumps(data, ensure_ascii=False, separators=(",", ":")) + serialized = serialized.replace("&", "\\u0026").replace("<", "\\u003c").replace(">", "\\u003e") + return template.replace(DATA_TOKEN, serialized) + + +def resolve_source(value: str | None) -> Path: + if value is None: + return DEFAULT_SOURCE + source = Path(value).expanduser() + if source.is_absolute(): + return source + cwd_source = Path.cwd() / source + return cwd_source if cwd_source.exists() else REPO_ROOT / source + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description="Generate the NGSI-LD interoperability architecture viewer.") + parser.add_argument( + "--source", + help="Robot file or directory. Defaults to IOP_TP/NGSI-LD/Interoperability.", + ) + parser.add_argument( + "--output-name", + default=DEFAULT_OUTPUT_NAME, + type=validate_output_name, + help=f"Filename inside the ignored output directory. Defaults to {DEFAULT_OUTPUT_NAME}.", + ) + return parser + + +def main(argv: list[str] | None = None) -> int: + parser = build_parser() + args = parser.parse_args(argv) + destination = output_path(args.output_name) + try: + data = extract_source(resolve_source(args.source), REPO_ROOT) + template = (TOOL_DIR / "viewer_template.html").read_text(encoding="utf-8") + document = render_viewer(data, template) + destination.parent.mkdir(parents=True, exist_ok=True) + destination.write_text(document, encoding="utf-8") + except (ExtractionError, OSError, UnicodeError, ValueError) as error: + print(f"Error: {error}", file=sys.stderr) + return 1 + + print(destination) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tools/interoperability_visualizer/tests/test_visualizer.py b/tools/interoperability_visualizer/tests/test_visualizer.py new file mode 100644 index 0000000000000000000000000000000000000000..4551946ccfc1b140c388b95b2a3420e6869219aa --- /dev/null +++ b/tools/interoperability_visualizer/tests/test_visualizer.py @@ -0,0 +1,429 @@ +import argparse +import json +import re +import sys +import tempfile +import unittest +from pathlib import Path + + +ROOT = Path(__file__).resolve().parents[3] +TOOL_DIR = ROOT / "tools" / "interoperability_visualizer" +sys.path.insert(0, str(TOOL_DIR)) + +from extractor import ExtractionError, extract_source +from generate import DATA_TOKEN, render_viewer, validate_output_name + + +class Fixture: + def __init__(self): + self.temporary_directory = tempfile.TemporaryDirectory() + self.root = Path(self.temporary_directory.name) + self.robot_file = self.root / "IOP_TP" / "NGSI-LD" / "Interoperability" / "fixture.robot" + self.entity_file = self.root / "data" / "entities" / "interoperability" / "entity.jsonld" + self.inclusive_file = ( + self.root + / "data" + / "csourceRegistrations" + / "interoperability" + / "registration-inclusive.jsonld" + ) + self.redirect_file = ( + self.root + / "data" + / "csourceRegistrations" + / "interoperability" + / "registration-redirect.jsonld" + ) + variables_file = self.root / "resources" / "variables.py" + variables_file.parent.mkdir(parents=True) + variables_file.write_text( + "core_context = 'https://example.test/core.jsonld'\n" + "ngsild_test_suite_context = 'https://example.test/user.jsonld'\n", + encoding="utf-8", + ) + self.entity_file.parent.mkdir(parents=True) + self.entity_file.write_text( + json.dumps({"id": "urn:old", "type": "Vehicle", "speed": {"type": "Property", "value": 10}}), + encoding="utf-8", + ) + registration = { + "id": "urn:old-registration", + "type": "ContextSourceRegistration", + "information": [{"entities": [{"type": "Vehicle"}]}], + "mode": "inclusive", + "endpoint": "xxx", + } + self.inclusive_file.parent.mkdir(parents=True) + self.inclusive_file.write_text(json.dumps(registration), encoding="utf-8") + registration["mode"] = "redirect" + self.redirect_file.write_text(json.dumps(registration), encoding="utf-8") + self.write_robot() + + def cleanup(self): + self.temporary_directory.cleanup() + + def write_robot( + self, + entity_payload: str = "interoperability/entity.jsonld", + first_mode: str = "inclusive", + first_endpoint: str = "${b2_url}", + first_registration: str = "${inclusive_registration}", + scenario_body: str = " No Operation", + ): + self.robot_file.parent.mkdir(parents=True, exist_ok=True) + self.robot_file.write_text( + f"""*** Settings *** +Documentation Fixture objective +Test Template Run Scenario + + +*** Variables *** +${{entity_payload}} {entity_payload} +${{inclusive_registration}} csourceRegistrations/interoperability/registration-inclusive.jsonld +${{redirect_registration}} csourceRegistrations/interoperability/registration-redirect.jsonld +${{b1_url}} ${{EMPTY}} +${{b2_url}} ${{EMPTY}} + + +*** Test Cases *** +IOP_999_01_01 Default Context + [Tags] since_v1.6.1 iop default-context 4_3_3 + [Setup] Prepare Configuration ${{core_context}} + ${{core_context}} + +IOP_999_01_02 User Context + [Tags] since_v1.6.1 iop user-context 4_3_3 + [Setup] Prepare Configuration ${{ngsild_test_suite_context}} + ${{ngsild_test_suite_context}} + + +*** Keywords *** +Run Scenario + [Arguments] ${{context}} +{scenario_body} + +Prepare Configuration + [Arguments] ${{context}} + Set Test Variable ${{entity_id}} urn:ngsi-ld:Vehicle:1 + Create Entity + ... ${{entity_payload}} + ... ${{entity_id}} + ... broker_url=${{b2_url}} + ... context=${{context}} + @{{first_set}}= Create List + ... ${{entity_id}} + ... {first_registration} + ... {first_mode} + ... {first_endpoint} + ... ${{b1_url}} + @{{second_set}}= Create List + ... ${{EMPTY}} + ... ${{redirect_registration}} + ... redirect + ... ${{b2_url}} + ... ${{b1_url}} + @{{configuration}}= Create List ${{first_set}} ${{second_set}} + Compose IOP Configuration ${{configuration}} ld_context=${{context}} +""", + encoding="utf-8", + ) + + +class ExtractorTest(unittest.TestCase): + def setUp(self): + self.fixture = Fixture() + self.addCleanup(self.fixture.cleanup) + + def test_extracts_direction_parallel_modes_and_effective_payloads(self): + data = extract_source(self.fixture.robot_file, self.fixture.root) + + self.assertEqual(len(data["tests"]), 2) + default_test, user_test = data["tests"] + self.assertEqual([broker["id"] for broker in default_test["brokers"]], ["b1", "b2"]) + self.assertEqual([registration["source"] for registration in default_test["registrations"]], ["b1", "b1"]) + self.assertEqual([registration["target"] for registration in default_test["registrations"]], ["b2", "b2"]) + self.assertEqual([registration["mode"] for registration in default_test["registrations"]], ["inclusive", "redirect"]) + + b2 = next(broker for broker in default_test["brokers"] if broker["id"] == "b2") + self.assertEqual(len(b2["entities"]), 1) + self.assertEqual(b2["entities"][0]["payload"]["id"], "urn:ngsi-ld:Vehicle:1") + self.assertEqual(b2["entities"][0]["payload"]["@context"], "https://example.test/core.jsonld") + + registration = default_test["registrations"][0] + self.assertEqual(registration["payload"]["id"], "${registration_id1}") + self.assertEqual(registration["payload"]["endpoint"], "${b2_url}") + self.assertEqual(registration["payload"]["mode"], "inclusive") + self.assertEqual(registration["payload"]["@context"], "https://example.test/core.jsonld") + self.assertEqual( + registration["payload"]["information"][0]["entities"][0]["id"], + "urn:ngsi-ld:Vehicle:1", + ) + self.assertEqual(user_test["context"], "https://example.test/user.jsonld") + self.assertEqual( + user_test["registrations"][0]["payload"]["@context"], + "https://example.test/user.jsonld", + ) + + def test_rejects_unsafe_payload_path(self): + self.fixture.write_robot(entity_payload="../../outside.json") + with self.assertRaisesRegex(ExtractionError, "unsafe payload path"): + extract_source(self.fixture.robot_file, self.fixture.root) + + def test_rejects_missing_payload(self): + self.fixture.write_robot(entity_payload="interoperability/missing.jsonld") + with self.assertRaisesRegex(ExtractionError, "payload does not exist"): + extract_source(self.fixture.robot_file, self.fixture.root) + + def test_rejects_malformed_json(self): + self.fixture.entity_file.write_text("{", encoding="utf-8") + with self.assertRaisesRegex(ExtractionError, "invalid JSON payload"): + extract_source(self.fixture.robot_file, self.fixture.root) + + def test_rejects_unsupported_mode(self): + self.fixture.write_robot(first_mode="mystery") + with self.assertRaisesRegex(ExtractionError, "unsupported registration mode 'mystery'"): + extract_source(self.fixture.robot_file, self.fixture.root) + + def test_rejects_invalid_broker_variable(self): + self.fixture.write_robot(first_endpoint="${peer_url}") + with self.assertRaisesRegex(ExtractionError, "is not a broker variable"): + extract_source(self.fixture.robot_file, self.fixture.root) + + +class ExpectedOutputsTest(unittest.TestCase): + def setUp(self): + self.fixture = Fixture() + self.addCleanup(self.fixture.cleanup) + + def extract_test(self): + return extract_source(self.fixture.robot_file, self.fixture.root)["tests"][0] + + def extract_steps(self): + return self.extract_test()["steps"] + + def test_reconstructs_retrieve_body_endpoint_and_nested_comments(self): + self.fixture.write_robot( + scenario_body=""" # Client retrieves the Vehicle + ${response}= Retrieve Entity + ... ${entity_id} + ... attrs=speed + ... local=true + ... broker_url=${b2_url} + ... context=${context} + ... type=Vehicle + # Agent checks the exact entity: + # - only the speed attribute is returned + Check Response Status Code 200 ${response.status_code} + ${expected}= Load Entity ${entity_payload} ${entity_id} + Keep In Dictionary ${expected} id type speed + Check Resource Set To ${expected} ${response.json()}""" + ) + + extracted = self.extract_test() + step = extracted["steps"][0] + self.assertEqual(step["method"], "GET") + self.assertEqual( + step["endpoint"], + "/ngsi-ld/v1/entities/urn:ngsi-ld:Vehicle:1?type=Vehicle&attrs=speed&local=true", + ) + self.assertEqual(step["broker"], "b2") + self.assertEqual(step["status"], 200) + self.assertEqual(step["resolution"], "exact") + self.assertEqual(set(step["expected_body"]), {"id", "type", "speed"}) + self.assertEqual(step["request_comment"], "Client retrieves the Vehicle") + self.assertIn("\n- only the speed attribute is returned", step["expectation_comments"][0]) + self.assertGreater(step["source_line"], 0) + timeline = extracted["timeline_steps"] + self.assertEqual(len(timeline), 2) + self.assertEqual(timeline[0]["text"], "Client retrieves the Vehicle") + self.assertEqual( + timeline[1]["text"], + "Agent checks the exact entity:\n- only the speed attribute is returned", + ) + self.assertEqual(timeline[0]["request_ids"], ["step-1"]) + self.assertEqual(timeline[1]["request_ids"], ["step-1"]) + self.assertEqual([timeline_step["group"] for timeline_step in timeline], [1, 2]) + self.assertEqual( + [operation["keyword"] for operation in timeline[1]["operations"]], + ["Check Response Status Code", "Load Entity", "Keep In Dictionary", "Check Resource Set To"], + ) + + def test_resolves_post_query_and_cross_broker_payload_reference(self): + self.fixture.write_robot( + scenario_body=""" # Client queries b1 via POST + &{selector}= Create Dictionary type=Vehicle + @{selectors}= Create List ${selector} + ${response_b1}= Query Entities Via POST + ... entities=${selectors} + ... broker_url=${b1_url} + ... context=${context} + # Agent checks b1 + Check Response Status Code 200 ${response_b1.status_code} + ${ids_b1}= Create List ${entity_id} + Check Response Body Containing Entities URIS set to ${ids_b1} ${response_b1.json()} + ${entity_b1}= Get Value From JSON ${response_b1.json()} $[?(@.id=='${entity_id}')] + # Client queries b2 + ${response_b2}= Query Entities + ... entity_types=Vehicle + ... broker_url=${b2_url} + ... context=${context} + Check Response Status Code 200 ${response_b2.status_code} + ${ids_b2}= Create List ${entity_id} + Check Response Body Containing Entities URIS set to ${ids_b2} ${response_b2.json()} + ${entity_b2}= Get Value From JSON ${response_b2.json()} $[?(@.id=='${entity_id}')] + ${expected_b2}= Load Entity ${entity_payload} ${entity_id} + Check Resource Set To ${expected_b2} ${entity_b2}[0] + # Agent checks b1 combines b2 + ${expected_b1}= Load Entity ${entity_payload} ${entity_id} + Keep In Dictionary ${expected_b1} id type + Set To Dictionary ${expected_b1} speed=${entity_b2}[0][speed] + Check Resource Set To ${expected_b1} ${entity_b1}[0]""" + ) + + first, second = self.extract_steps() + self.assertEqual(first["endpoint"], "/ngsi-ld/v1/entityOperations/query") + self.assertEqual(first["request_comment"], "Client queries b1 via POST") + self.assertEqual(first["resolution"], "exact") + self.assertEqual(first["expected_body"][0]["speed"], second["expected_body"][0]["speed"]) + self.assertEqual(second["endpoint"], "/ngsi-ld/v1/entities?type=Vehicle") + + def test_creation_and_not_found_are_status_only(self): + self.fixture.write_robot( + scenario_body=""" # Client creates an entity + ${response}= Create Entity + ... ${entity_payload} + ... ${entity_id} + ... broker_url=${b1_url} + ... context=${context} + Check Response Status Code 201 ${response.status_code} + # Client retrieves a missing entity + ${response}= Retrieve Entity + ... urn:ngsi-ld:Vehicle:missing + ... broker_url=${b1_url} + ... context=${context} + Check Response Status Code 404 ${response.status_code}""" + ) + + steps = self.extract_steps() + self.assertEqual([step["status"] for step in steps], [201, 404]) + self.assertTrue(all(step["expected_body"] is None for step in steps)) + self.assertTrue(all(step["resolution"] == "status-only" for step in steps)) + + def test_unsupported_body_expression_remains_unresolved(self): + self.fixture.write_robot( + scenario_body=""" # Client retrieves an entity + ${response}= Retrieve Entity + ... ${entity_id} + ... broker_url=${b1_url} + Check Response Status Code 200 ${response.status_code} + Check Resource Set To ${unsupported_expression} ${response.json()}""" + ) + + step = self.extract_steps()[0] + self.assertEqual(step["status"], 200) + self.assertEqual(step["resolution"], "unresolved") + self.assertIsNone(step["expected_body"]) + + +class GeneratorTest(unittest.TestCase): + def test_output_name_rejects_paths_and_non_html_files(self): + for value in ("../viewer.html", "folder/viewer.html", "folder\\viewer.html", "viewer.json", ""): + with self.subTest(value=value): + with self.assertRaises(argparse.ArgumentTypeError): + validate_output_name(value) + self.assertEqual(validate_output_name("viewer.html"), "viewer.html") + + def test_render_viewer_escapes_embedded_markup(self): + rendered = render_viewer({"value": "") + self.assertNotIn(" + + +