fix: refactor initial_conditions handling and add IOP preconditions tests (d0b2b3a0) · Commits · CIM - Context Information Management / NGSI-LD Test Suite

doc/analysis/generaterobotdata.py

+132 −7

Original line number	Diff line number	Diff line
		@@ -5,6 +5,7 @@ from analysis.parseapiutilsfile import ParseApiUtilsFile
		from analysis.parsevariablesfile import ParseVariablesFile
		from analysis.initial_setup import InitialSetup
		from re import match, findall, finditer, sub, MULTILINE
		from urllib.parse import urlparse


		class GenerateRobotData:
		@@ -647,12 +648,13 @@ class GenerateRobotData:
		version = 'v1.3.1'
		tp_id = self.generate_name_iop()
		reference, clauses = self.generate_reference(version=version)
		test_doc = {}
		# Add test case documentation
		initial_conditions = {}
		if self.robot.test_case_names:
		test_name = self.robot.test_case_names[0]
		test_doc = self.robot.get_iop_documentation_data(test_name=test_name)
		test_doc['registrations_established'] = self.generate_iop_registrations(test_name=test_name)
		initial_conditions = {
		'pre_conditions': self.generate_iop_preconditions(test_name=test_name),
		'registrations_established': self.generate_iop_registrations(test_name=test_name)
		}

		self.test_suite = {
		'tp_id': tp_id,
		@@ -663,13 +665,12 @@ class GenerateRobotData:
		'clauses': clauses,
		'pics_selection': str(),
		'keywords': [x.to_dict()['name'] for x in list(self.suite.resource.keywords)],
		'initial_conditions': test_doc,
		'initial_conditions': initial_conditions,
		'teardown': str(self.suite.teardown),
		'test_cases': list()
		}

		def generate_iop_registrations(self, test_name: str) -> str:
		"""Generate the registration description from the test setup."""
		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:
		raise ValueError(f"IOP test '{test_name}' was not found")
		@@ -679,6 +680,130 @@ class GenerateRobotData:
		if setup is None:
		raise ValueError(f"IOP setup keyword '{setup_name}' was not found")

		return setup

		def _resolve_iop_value(self, value: str, field: str) -> str:
		original = str(value)
		result = original
		visited = set()

		while match(r'^\$\{[^}]+\}$', result):
		if result in visited:
		raise ValueError(f"{field} variable '{original}' contains a circular reference")
		visited.add(result)

		if result in self.robot.variables:
		result = str(self.robot.variables[result])
		continue

		try:
		result = str(self.config_variables.get_variable(result))
		except KeyError as error:
		raise ValueError(f"{field} variable '{result}' is not defined") from error

		return result

		def _iop_filename(self, value: str, field: str) -> str:
		resolved = self._resolve_iop_value(value=value, field=field)
		path = urlparse(resolved).path.replace('\\', '/').rstrip('/')
		filename = basename(path)
		if filename == '':
		raise ValueError(f"{field} value '{resolved}' does not contain a filename")
		return filename

		def _iop_broker_name(self, value: str) -> str:
		original = str(value)
		result = original
		visited = set()

		while result not in visited:
		broker_match = match(r'^\$\{(b\d+)_url\}$', result)
		if broker_match is not None:
		return broker_match.group(1)

		visited.add(result)
		if result not in self.robot.variables:
		break
		result = str(self.robot.variables[result])

		raise ValueError(f"Create Entity broker variable '{original}' is invalid")

		@staticmethod
		def _create_entity_arguments(args) -> dict:
		signature = ('filename', 'entity_id', 'local', 'broker_url', 'context')
		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 ValueError(f"Create Entity has an unexpected argument '{argument}'")
		values[signature[position]] = argument
		position += 1

		return values

		def generate_iop_preconditions(self, test_name: str) -> list:
		setup = self._get_iop_setup(test_name=test_name)
		contexts = []
		data = []

		for keyword in setup.body:
		if keyword.name != 'Create Entity':
		continue

		arguments = self._create_entity_arguments(keyword.args)
		if 'filename' not in arguments:
		raise ValueError('Create Entity does not define a payload filename')
		if 'broker_url' not in arguments:
		raise ValueError('Create Entity does not define broker_url')

		broker = self._iop_broker_name(arguments['broker_url'])
		payload_filename = self._iop_filename(
		value=arguments['filename'],
		field='Create Entity payload'
		)
		data.append(f"{broker} contains {payload_filename}.")

		if 'context' in arguments:
		context_filename = self._iop_filename(
		value=arguments['context'],
		field='Create Entity context'
		)
		contexts.append(
		f"The {context_filename} user context is used when creating on {broker}."
		)

		return [
		{
		'type': 'context',
		'top-level': (
		'Context used for creation in following brokers:'
		if contexts else
		'No user context used.'
		),
		'nested-object': contexts
		},
		{
		'type': 'data',
		'top-level': (
		'Data in following brokers:'
		if data else
		'No data in any broker.'
		),
		'nested-object': data
		}
		]

		def generate_iop_registrations(self, test_name: str) -> str:
		"""Generate the registration description from the test setup."""
		setup = self._get_iop_setup(test_name=test_name)
		modes = {'inclusive', 'exclusive', 'redirect', 'auxiliary'}
		registrations = []

doc/analysis/parserobotfile.py

+0 −62

Original line number	Diff line number	Diff line
		@@ -657,68 +657,6 @@ class ParseRobotFile:
		if len(indexes) > 0:
		self.test_cases[self.test_case_names[-1]] = string_test_cases[indexes[-1]:]

		def _get_documentation_content(self, test_name: str) -> str:
		"""Helper: Extract raw documentation content after [Documentation] until next keyword block."""
		if test_name not in self.test_cases:
		return ""

		test_content = self.test_cases[test_name]
		match = re.search(r'\[Documentation\]\s*', test_content)
		if not match:
		return ""

		start_pos = match.end()
		doc_section = re.search(r'(.?)(?=\n\s\[\|\Z)', test_content[start_pos:], re.MULTILINE \| re.DOTALL)
		return doc_section.group(1) if doc_section else ""

		def get_iop_documentation_data(self, test_name: str) -> dict:
		"""Extract documentation in the Test Cases."""
		if test_name not in self.test_cases:
		return " "

		doc_content = self._get_documentation_content(test_name)
		if not doc_content:
		return " "

		label_pattern = re.compile(
		r'(?:\A[ \t]\|\n[ \t]\.{3}[ \t]+)([A-Za-z][A-Za-z \-]):[ \t]',
		re.MULTILINE
		)
		matches = list(label_pattern.finditer(doc_content))
		sections = {}
		for index, match in enumerate(matches):
		end = matches[index + 1].start() if index + 1 < len(matches) else len(doc_content)
		key = match.group(1).strip().lower().replace(' ', '_').replace('-', '_')
		value = re.sub(r'\n\s\.{3}\s', ' ', doc_content[match.end():end])
		sections[key] = value.replace('\n', ' ').strip().rstrip('.')

		if 'pre_conditions' in sections:
		sections['pre_conditions'] = self._format_iop_preconditions(sections['pre_conditions'])

		return sections

		@staticmethod
		def _format_iop_preconditions(preconditions: str):
		match = re.match(
		r'^(.*?)(Data (?:only on leaves\|on every broker))\.\s+((?:b\d+\s+contains\s+).+)$',
		preconditions,
		)
		if match is None:
		return preconditions

		prefix, data_location, broker_data = match.groups()
		broker_entries = re.split(r'\.\s+(?=b\d+\s+contains\s+)', broker_data)
		prefix_entries = [
		entry.strip()
		for entry in prefix.rstrip('. ').split('.')
		if entry.strip()
		]
		return [
		*prefix_entries,
		f"{data_location}:",
		[entry.rstrip('.') for entry in broker_entries],
		]

		def get_iop_test_tags(self, test_name: str) -> list:
		"""Extract [Tags] from a specific IOP test case."""
		if test_name not in self.test_cases:

doc/tests/test_iop_preconditions.py

0 → 100644

+195 −0

Original line number	Diff line number	Diff line
		import sys
		import tempfile
		import unittest
		from pathlib import Path


		ROOT = Path(__file__).resolve().parents[2]
		sys.path.insert(0, str(ROOT / 'doc'))

		from analysis.generaterobotdata import GenerateRobotData


		class IopPreconditionsTest(unittest.TestCase):
		def generate(self, variables: str, setup: str, test_body: str = ' No Operation'):
		robot = f'''* Settings *
		Documentation Objective
		Test Setup Prepare Preconditions


		* Variables *
		{variables}


		* Test Cases *
		IOP_999_01 Generated Preconditions
		[Documentation] Pre-conditions: documentation must be ignored.
		[Tags] since_v1.6.1 iop 4_3_3
		{test_body}


		* Keywords *
		Prepare Preconditions
		{setup}
		'''
		temporary_directory = tempfile.TemporaryDirectory()
		self.addCleanup(temporary_directory.cleanup)
		path = (
		Path(temporary_directory.name)
		/ 'IOP_TP'
		/ 'NGSI-LD'
		/ 'Interoperability'
		/ 'Consumption'
		/ 'Entity'
		/ 'QueryEntities'
		/ 'IOP_999_01.robot'
		)
		path.parent.mkdir(parents=True)
		path.write_text(robot, encoding='utf-8')

		generator = GenerateRobotData(robot_file=str(path), execdir=str(ROOT))
		generator.parse_robot()
		return generator.get_info()['initial_conditions']

		def test_generates_context_and_data_objects(self):
		result = self.generate(
		variables='''${first_payload} fixtures/first-entity.jsonld
		${context_path} https://example.test/contexts/example-context.jsonld?version=1
		${broker_alias} ${b2_url}
		${b2_url} ${EMPTY}
		${b3_url} ${EMPTY}''',
		setup=''' ${response}= Create Entity
		... ${first_payload}
		... urn:ngsi-ld:Example:1
		... broker_url=${broker_alias}
		... context=${context_path}
		Create Entity
		... second-entity.json
		... urn:ngsi-ld:Example:2
		... ${EMPTY}
		... ${b3_url}
		... ${ngsild_test_suite_context}''',
		test_body=''' Create Entity
		... ignored-body-entity.jsonld
		... urn:ngsi-ld:Example:3
		... broker_url=${b3_url}'''
		)

		self.assertEqual(
		result['pre_conditions'],
		[
		{
		'type': 'context',
		'top-level': 'Context used for creation in following brokers:',
		'nested-object': [
		'The example-context.jsonld user context is used when creating on b2.',
		(
		'The ngsi-ld-test-suite-compound.jsonld user context '
		'is used when creating on b3.'
		)
		]
		},
		{
		'type': 'data',
		'top-level': 'Data in following brokers:',
		'nested-object': [
		'b2 contains first-entity.jsonld.',
		'b3 contains second-entity.json.'
		]
		}
		]
		)

		def test_uses_empty_context_object_when_context_is_omitted(self):
		result = self.generate(
		variables='${b2_url} ${EMPTY}',
		setup=(
		' Create Entity repeated.jsonld urn:ngsi-ld:Example:1'
		' broker_url=${b2_url}\n'
		' Create Entity repeated.jsonld urn:ngsi-ld:Example:2'
		' broker_url=${b2_url}'
		)
		)

		self.assertEqual(
		result['pre_conditions'],
		[
		{
		'type': 'context',
		'top-level': 'No user context used.',
		'nested-object': []
		},
		{
		'type': 'data',
		'top-level': 'Data in following brokers:',
		'nested-object': [
		'b2 contains repeated.jsonld.',
		'b2 contains repeated.jsonld.'
		]
		}
		]
		)

		def test_uses_empty_objects_without_create_entity(self):
		result = self.generate(
		variables='${b1_url} ${EMPTY}',
		setup=' No Operation'
		)

		self.assertEqual(
		result['pre_conditions'],
		[
		{
		'type': 'context',
		'top-level': 'No user context used.',
		'nested-object': []
		},
		{
		'type': 'data',
		'top-level': 'No data in any broker.',
		'nested-object': []
		}
		]
		)

		def test_rejects_invalid_broker(self):
		with self.assertRaisesRegex(
		ValueError,
		r"Create Entity broker variable '\$\{broker_url\}' is invalid"
		):
		self.generate(
		variables='${broker_url} https://example.test',
		setup=(
		' Create Entity entity.jsonld urn:ngsi-ld:Example:1'
		' broker_url=${broker_url}'
		)
		)

		def test_rejects_undefined_payload_and_context_variables(self):
		with self.assertRaisesRegex(
		ValueError,
		r"Create Entity payload variable '\$\{missing_payload\}' is not defined"
		):
		self.generate(
		variables='${b1_url} ${EMPTY}',
		setup=(
		' Create Entity ${missing_payload} urn:ngsi-ld:Example:1'
		' broker_url=${b1_url}'
		)
		)

		with self.assertRaisesRegex(
		ValueError,
		r"Create Entity context variable '\$\{missing_context\}' is not defined"
		):
		self.generate(
		variables='${b1_url} ${EMPTY}',
		setup=(
		' Create Entity entity.jsonld urn:ngsi-ld:Example:1'
		' broker_url=${b1_url} context=${missing_context}'
		)
		)


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