import bt2
import os
import re
-from pathlib import PureWindowsPath, PurePosixPath
test_ctf_traces_path = os.environ['BT_CTF_TRACES_PATH']
},
).query()
- os_stream_path = PurePosixPath(
- '/tests/data/ctf-traces/intersection/3eventsintersect/'
- )
if os.environ['BT_OS_TYPE'] == 'mingw':
- os_stream_path = PureWindowsPath(os_stream_path)
+ os_stream_path = (
+ '\\tests\\data\\ctf-traces\\intersection\\3eventsintersect\\'
+ )
+ else:
+ os_stream_path = '/tests/data/ctf-traces/intersection/3eventsintersect/'
self.assertEqual(len(res), 1)
trace = res[0]
self.assertRegex(
str(streams[0]["port-name"]),
r"^7afe8fbe-79b8-4f6a-bbc7-d0c782e7ddaf \| 0 \| .*"
- + re.escape(str(os_stream_path / "test_stream_0"))
+ + re.escape(os_stream_path + "test_stream_0")
+ r"$",
)
self.assertRegex(
str(streams[1]["port-name"]),
r"^7afe8fbe-79b8-4f6a-bbc7-d0c782e7ddaf \| 0 \| .*"
- + re.escape(str(os_stream_path / "test_stream_1"))
+ + re.escape(os_stream_path + "test_stream_1")
+ r"$",
)
{"inputs": [os.path.join(test_ctf_traces_path, "succeed", "succeed1")]},
).query()
- os_stream_path = PurePosixPath(
- '/tests/data/ctf-traces/succeed/succeed1/dummystream'
- )
if os.environ['BT_OS_TYPE'] == 'mingw':
- os_stream_path = PureWindowsPath(os_stream_path)
+ os_stream_path = '\\tests\\data\\ctf-traces\\succeed\\succeed1\\dummystream'
+ else:
+ os_stream_path = '/tests/data/ctf-traces/succeed/succeed1/dummystream'
self.assertEqual(len(res), 1)
trace = res[0]
self.assertRegex(
str(streams[0]["port-name"]),
r"^2a6422d0-6cee-11e0-8c08-cb07d7b3a564 \| .*"
- + re.escape(str(os_stream_path))
+ + re.escape(os_stream_path)
+ r"$",
)
self.assertRaises(KeyError, lambda: trace['range-ns'])
self.assertRaises(KeyError, lambda: streams[0]['range-ns'])
+ def test_trace_with_tracefile_rotation(self):
+ res = bt2.QueryExecutor(
+ self._fs,
+ "babeltrace.trace-infos",
+ {
+ "inputs": [
+ os.path.join(
+ test_ctf_traces_path,
+ "succeed",
+ "lttng-tracefile-rotation",
+ "kernel",
+ )
+ ]
+ },
+ ).query()
+
+ self.assertEqual(len(res), 1)
+ trace = res[0]
+ streams = trace["stream-infos"]
+ self.assertEqual(len(streams), 4)
+
+ # Note: the end timestamps are not the end timestamps found in the
+ # index files, because fix_index_lttng_event_after_packet_bug changes
+ # them based on the time of the last event in the stream.
+
+ self.assertEqual(streams[0]['range-ns']['begin'], 1571261795455986789)
+ self.assertEqual(streams[0]['range-ns']['end'], 1571261797582611840)
+
+ self.assertEqual(streams[1]['range-ns']['begin'], 1571261795456368232)
+ self.assertEqual(streams[1]['range-ns']['end'], 1571261797577754111)
+
+ self.assertEqual(streams[2]['range-ns']['begin'], 1571261795456748255)
+ self.assertEqual(streams[2]['range-ns']['end'], 1571261797577727795)
+
+ self.assertEqual(streams[3]['range-ns']['begin'], 1571261795457285142)
+ self.assertEqual(streams[3]['range-ns']['end'], 1571261797582522088)
+
class QueryTraceInfoPacketTimestampQuirksTestCase(unittest.TestCase):
def setUp(self):