feat: add viewer logic (1e36e53f) · Commits · CIM - Context Information Management / NGSI-LD Test Suite

.gitignore

+2 −1

Original line number	Diff line number	Diff line
		@@ -5,3 +5,4 @@ results
		.vscode/
		.venv
		.DS_Store
		/tools/interoperability_visualizer/output/

tools/interoperability_visualizer/README.md

0 → 100644

+69 −0

Original line number	Diff line number	Diff line
		# 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.
		- 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 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'
		```

tools/interoperability_visualizer/expected_outputs.py

0 → 100644

+554 −0

Original line number	Diff line number	Diff line
		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 []

tools/interoperability_visualizer/extractor.py

0 → 100644

+401 −0

File added.

Preview size limit exceeded, changes collapsed.

tools/interoperability_visualizer/generate.py

0 → 100644

+85 −0

File added.

Preview size limit exceeded, changes collapsed.