Add IOP Viewer scripts (!308) · Merge requests · CIM - Context Information Management / NGSI-LD Test Suite

doc/analysis/generaterobotdata.py

+11 −0

Changes for doc/analysis/generaterobotdata.py: 11 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -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:

doc/tests/test_iop_preconditions.py

+32 −4

Changes for doc/tests/test_iop_preconditions.py: 32 added lines, 4 removed lines.
Original line number	Diff line number	Diff line
		@@ -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],
		{

tools/interoperability_visualizer/tests/test_visualizer.py

0 → 100644

+429 −0

Changes for tools/interoperability_visualizer/tests/test_visualizer.py: 429 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		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": "</script><script src='x'>&"}, f"<script>{DATA_TOKEN}</script>")
		self.assertNotIn("</script><script src", rendered)
		self.assertIn("\\u003c/script\\u003e", rendered)
		self.assertIn("\\u0026", rendered)

		def test_template_is_offline_and_contains_the_legend(self):
		template = (TOOL_DIR / "viewer_template.html").read_text(encoding="utf-8")
		self.assertIsNone(re.search(r"(?:src\|href)\s=\s['\"]https?://", template, re.IGNORECASE))
		for label, color in (
		("Inclusive", "#009E73"),
		("Auxiliary", "#0072B2"),
		("Redirect", "#E69F00"),
		("Exclusive", "#CC79A7"),
		):
		self.assertIn(label, template)
		self.assertIn(color, template)
		for value in (
		"Expected broker responses",
		">Preview</button>",
		">Expected output</button>",
		"Broker view",
		"Timeline view",
		"view-expectations",
		'setSurfaceView("expectations")',
		"response-step",
		"response-expanded",
		"response-hierarchy",
		"status-code",
		"responseItem(step)",
		"timeline-step",
		"timelineGroup(steps, requestsById)",
		"Assertions and expected output after Step",
		"Robot operation",
		"correlated expected HTTP response",
		"level-guide",
		"initialPositions(test)",
		"levels.set(target, nextLevel)",
		"renderLevels()",
		):
		self.assertIn(value, template)

		def test_output_directory_is_git_ignored(self):
		gitignore = (ROOT / ".gitignore").read_text(encoding="utf-8").splitlines()
		self.assertIn("/tools/interoperability_visualizer/output/", gitignore)


		class CorpusIntegrationTest(unittest.TestCase):
		def test_extracts_complete_interoperability_corpus(self):
		data = extract_source(ROOT / "IOP_TP" / "NGSI-LD" / "Interoperability", ROOT)
		tests = data["tests"]
		suites = {test["suite"] for test in tests}
		first_per_suite = {test["suite"]: test for test in tests}

		self.assertEqual(len(suites), 24)
		self.assertEqual(len(tests), 48)
		self.assertEqual(sum(len(test["registrations"]) for test in first_per_suite.values()), 82)
		steps = [step for test in first_per_suite.values() for step in test["steps"]]
		all_steps = [step for test in tests for step in test["steps"]]
		timeline_steps = [step for test in first_per_suite.values() for step in test["timeline_steps"]]
		all_timeline_steps = [step for test in tests for step in test["timeline_steps"]]
		self.assertEqual(len(steps), 91)
		self.assertEqual(len(all_steps), 182)
		self.assertEqual(len(timeline_steps), 133)
		self.assertEqual(len(all_timeline_steps), 266)
		self.assertTrue(all(step["text"] and step["operations"] for step in timeline_steps))
		self.assertTrue(all(step["request_ids"] for step in timeline_steps))
		grouped_example = next(
		test
		for test in tests
		if test["suite"].endswith("RetrieveEntity/IOP_002_04_02.robot")
		)
		self.assertEqual(
		[step["group"] for step in grouped_example["timeline_steps"]],
		[1, 2, 3, 4, 5, 6],
		)
		self.assertEqual(
		{status: sum(step["status"] == status for step in steps) for status in (200, 201, 404)},
		{200: 74, 201: 8, 404: 9},
		)
		self.assertTrue(all(step["resolution"] == "exact" for step in steps if step["status"] == 200))
		for step in steps:
		if isinstance(step["expected_body"], list):
		ids = [entity["id"] for entity in step["expected_body"]]
		self.assertEqual(ids, sorted(ids))
		self.assertEqual(
		{registration["mode"] for test in tests for registration in test["registrations"]},
		{"inclusive", "auxiliary", "redirect", "exclusive"},
		)


		if __name__ == "__main__":
		unittest.main()

tools/interoperability_visualizer/README.md

0 → 100644

+70 −0

Changes for tools/interoperability_visualizer/README.md: 70 added lines, 0 removed lines.
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.
		- 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'
		```

tools/interoperability_visualizer/expected_outputs.py

0 → 100644

+554 −0

Changes for tools/interoperability_visualizer/expected_outputs.py: 554 added lines, 0 removed lines.
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 []

Add IOP Viewer scripts

Merge request reports