diff --git a/tests/test_topic_inheritance.py b/tests/test_topic_inheritance.py index da82388d..52f30ce0 100755 --- a/tests/test_topic_inheritance.py +++ b/tests/test_topic_inheritance.py @@ -11,7 +11,7 @@ from cyclonedds.sub import DataReader from cyclonedds.util import duration - +BASE_IDL = 'module Hierarchy { @mutable struct Base { string fieldA; }; };' @dataclass @annotate.mutable @annotate.autoid("sequential") @@ -19,6 +19,7 @@ class Base(idl.IdlStruct, typename="Hierarchy.Base"): fieldA: str +DERIVED_IDL = 'module Hierarchy { struct Derived : Base { string fieldB; }; };' @dataclass @annotate.mutable @annotate.autoid("sequential") @@ -73,11 +74,8 @@ async def test_base_and_derived_topics(): fieldB='Dolor') tasks = [ - # This should not be necessary. I suspect there is a bug in CycloneDDS _subscriber(Base), _publisher(Base, base), - - # This is the part of the test that really matters _subscriber(Derived), _publisher(Derived, derived), ] @@ -87,6 +85,21 @@ async def test_base_and_derived_topics(): assert results[2] == results[3] +@pytest.mark.asyncio +async def test_cyclonedds_typeof_command(): + ''' + Executes the command "cyclonedds typeof" and compares the results with the + expected IDL. + ''' + tasks = [ + _subscriber(Base), + _subscriber(Derived), + _type_checker(Base, BASE_IDL), + _type_checker(Derived, DERIVED_IDL), + ] + await asyncio.gather(*tasks) + + async def _publisher(topicClass, value, timeout=2): ''' Publishes an update with a given value. @@ -109,3 +122,29 @@ async def _subscriber(topicClass, timeout=2): reader = DataReader(participant, topic) async for update in reader.read_aiter(timeout=duration(seconds=timeout)): return update + + +async def _type_checker(topicClass, expectedIdl): + ''' + Executes the command "cyclonedds typeof" and compares the result with the + expected IDL. + ''' + def _normalise(text): + text = text.replace('\n', '') + text = ' '.join(text.split()) + return text + + import subprocess + result = subprocess.run( + ['cyclonedds', 'typeof', topicClass.__name__, '--suppress-progress-bar'], + stdout = subprocess.PIPE, + check = True) + outputLines = result.stdout.decode().splitlines() + + # Skips the first lines of the output (As defined in participant...) + firstIdlLine = 0 + while firstIdlLine < len(outputLines) and not outputLines[firstIdlLine].startswith('module'): + firstIdlLine += 1 + actualIdl = '\n'.join(outputLines[firstIdlLine:]) + + assert _normalise(actualIdl) == _normalise(expectedIdl)