3 # The MIT License (MIT)
5 # Copyright (C) 2016 - Jérémie Galarneau <jeremie.galarneau@efficios.com>
7 # Permission is hereby granted, free of charge, to any person obtaining a copy
8 # of this software and associated documentation files (the "Software"), to deal
9 # in the Software without restriction, including without limitation the rights
10 # to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
11 # copies of the Software, and to permit persons to whom the Software is
12 # furnished to do so, subject to the following conditions:
14 # The above copyright notice and this permission notice shall be included in
15 # all copies or substantial portions of the Software.
17 # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
18 # IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
19 # FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
20 # AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
21 # LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
22 # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
26 import babeltrace
.writer
as btw
27 import babeltrace
.reader
as btr
33 def __init__(self
, stream_id
, timestamp
=None, end_of_packet
=False):
34 self
.stream_id
= stream_id
35 self
.timestamp
= timestamp
36 self
.end_of_packet
= end_of_packet
40 def __init__(self
, timestamps
):
41 self
.timestamps
= timestamps
44 class TraceIntersectionTestCase(unittest
.TestCase
):
45 def _create_trace(self
, stream_descriptions
):
46 trace_path
= tempfile
.mkdtemp()
47 trace
= btw
.Writer(trace_path
)
48 clock
= btw
.Clock('test_clock')
49 clock
.uuid
= self
._clock
_uuid
50 trace
.add_clock(clock
)
52 integer_field_type
= btw
.IntegerFieldDeclaration(32)
54 event_class
= btw
.EventClass('simple_event')
55 event_class
.add_field(integer_field_type
, 'int_field')
57 stream_class
= btw
.StreamClass('test_stream')
58 stream_class
.add_event_class(event_class
)
59 stream_class
.clock
= clock
63 for stream_id
, stream_packets
in enumerate(stream_descriptions
):
64 stream
= trace
.create_stream(stream_class
)
65 streams
.append(stream
)
67 for packet
in stream_packets
:
68 for timestamp
in packet
.timestamps
:
69 stream_entries
.append(Entry(stream_id
, timestamp
))
70 # Mark the last inserted entry as the end of packet
71 stream_entries
[len(stream_entries
) - 1].end_of_packet
= True
73 # Sort stream entries which will provide us with a time-ordered list of
74 # events to insert in the streams.
75 for entry
in sorted(stream_entries
, key
=lambda entry
: entry
.timestamp
):
76 clock
.time
= entry
.timestamp
77 event
= btw
.Event(event_class
)
78 event
.payload('int_field').value
= entry
.stream_id
79 streams
[entry
.stream_id
].append_event(event
)
80 if entry
.end_of_packet
is True:
81 streams
[entry
.stream_id
].flush()
86 self
._clock
_uuid
= uuid
.uuid4()
87 self
._trace
_path
_early
= self
._create
_trace
(
89 [Packet(range(1, 7)), Packet(range(11, 18))],
90 [Packet(range(8, 15)), Packet(range(22, 24)), Packet(range(30, 60))],
91 [Packet(range(11, 14))]
94 self
._trace
_path
_late
= self
._create
_trace
(
96 [Packet(range(100, 105)), Packet(range(109, 120))],
97 [Packet(range(88, 95)), Packet(range(96, 110)), Packet(range(112, 140))],
98 [Packet(range(99, 105))]
102 self
._expected
_timestamps
_early
= []
103 for ts
in range(11, 14):
104 for stream
in range(3):
105 self
._expected
_timestamps
_early
.append(ts
)
107 self
._expected
_timestamps
_late
= []
108 for ts
in range(100, 105):
109 for stream
in range(3):
110 self
._expected
_timestamps
_late
.append(ts
)
112 self
._expected
_timestamps
_union
= (self
._expected
_timestamps
_early
+
113 self
._expected
_timestamps
_late
)
116 shutil
.rmtree(self
._trace
_path
_early
)
117 shutil
.rmtree(self
._trace
_path
_late
)
121 def _check_trace_expected_timestamps(trace_paths
, expected_timestamps
):
122 traces
= btr
.TraceCollection(intersect_mode
=True)
123 for trace_path
in trace_paths
:
124 trace_handle
= traces
.add_trace(trace_path
, 'ctf')
125 if trace_handle
is None:
126 print('# Failed to open trace at {}'.format(trace_path
))
128 for event
in traces
.events
:
129 expected_timestamp
= expected_timestamps
.pop(0)
130 if event
.timestamp
!= expected_timestamp
:
131 print('# Unexpected timestamp ({}), expected {}'.format(
132 event
.timestamp
, expected_timestamp
))
136 def test_trace_early(self
):
137 self
._check
_trace
_expected
_timestamps
([self
._trace
_path
_early
],
138 self
._expected
_timestamps
_early
)
140 def test_trace_late(self
):
141 self
._check
_trace
_expected
_timestamps
([self
._trace
_path
_late
],
142 self
._expected
_timestamps
_late
)
144 def test_trace_intersection(self
):
145 self
._check
_trace
_expected
_timestamps
([self
._trace
_path
_early
,
146 self
._trace
_path
_late
],
147 self
._expected
_timestamps
_union
)
This page took 0.040767 seconds and 4 git commands to generate.