Testing the command "cyclonedds typeof" with a derived topic

tests/test_topic_inheritance.py

@@ -11,14 +11,15 @@
 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")
 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)