src.ctf.fs: add stream port name to trace-info query, use it for stream intersection

[babeltrace.git] / tests / plugins / ctf / test_query_trace_info.py

diff --git a/tests/plugins/ctf/test_query_trace_info.py b/tests/plugins/ctf/test_query_trace_info.py
index c8b5923cd9e8aec34cf7565b63c697f53a7700c7..053c0e1e43c20b4a7f8892d96100bb09c8903ccd 100644 (file)
--- a/tests/plugins/ctf/test_query_trace_info.py
+++ b/tests/plugins/ctf/test_query_trace_info.py
@@ -17,6 +17,7 @@
 import unittest
 import bt2
 import os
+import re
 
 
 test_ctf_traces_path = os.environ['TEST_CTF_TRACES_PATH']
@@ -121,5 +122,54 @@ class QueryTraceInfoClockOffsetTestCase(unittest.TestCase):
                 'clock-class-offset-ns': None,
             })
 
+
+class QueryTraceInfoPortNameTestCase(unittest.TestCase):
+    def setUp(self):
+        ctf = bt2.find_plugin("ctf")
+        self._fs = ctf.source_component_classes["fs"]
+
+        self._executor = bt2.QueryExecutor()
+
+    def test_trace_uuid_stream_class_id_no_stream_id(self):
+        res = self._executor.query(
+            self._fs,
+            "trace-info",
+            {
+                "paths": [
+                    os.path.join(
+                        test_ctf_traces_path, "intersection", "3eventsintersect"
+                    )
+                ]
+            },
+        )
+        self.assertEqual(len(res), 1)
+        trace = res[0]
+        streams = sorted(trace["streams"], key=sort_by_begin)
+        self.assertEqual(len(streams), 2)
+        self.assertRegexpMatches(
+            str(streams[0]["port-name"]),
+            r"^7afe8fbe-79b8-4f6a-bbc7-d0c782e7ddaf \| 0 \| .*/tests/ctf-traces/intersection/3eventsintersect/test_stream_0$",
+        )
+        self.assertRegexpMatches(
+            str(streams[1]["port-name"]),
+            r"^7afe8fbe-79b8-4f6a-bbc7-d0c782e7ddaf \| 0 \| .*/tests/ctf-traces/intersection/3eventsintersect/test_stream_1$",
+        )
+
+    def test_trace_uuid_no_stream_class_id_no_stream_id(self):
+        res = self._executor.query(
+            self._fs,
+            "trace-info",
+            {"paths": [os.path.join(test_ctf_traces_path, "succeed", "succeed1")]},
+        )
+        self.assertEqual(len(res), 1)
+        trace = res[0]
+        streams = sorted(trace["streams"], key=sort_by_begin)
+        self.assertEqual(len(streams), 1)
+        self.assertRegexpMatches(
+            str(streams[0]["port-name"]),
+            r"^2a6422d0-6cee-11e0-8c08-cb07d7b3a564 \| .*/tests/ctf-traces/succeed/succeed1/dummystream$",
+        )
+
+
 if __name__ == '__main__':
     unittest.main()