Commit | Line | Data |
---|---|---|
fd878519 PP |
1 | import unittest |
2 | import bt2 | |
3 | ||
4 | ||
5 | class StreamClassSingleClockClassTestCase(unittest.TestCase): | |
6 | def setUp(self): | |
7 | self._sc = bt2.StreamClass() | |
8 | ||
9 | def tearDown(self): | |
10 | del self._sc | |
11 | ||
12 | def _test_add_sc_ft_vs_event_class(self, set_ft_func): | |
13 | cc = bt2.ClockClass('first_cc', 1000) | |
14 | scft = bt2.StructureFieldType() | |
15 | ft = bt2.IntegerFieldType(32, mapped_clock_class=cc) | |
16 | scft.append_field('salut', ft) | |
17 | set_ft_func(scft) | |
18 | cc = bt2.ClockClass('second_cc', 1000) | |
19 | payloadft = bt2.StructureFieldType() | |
20 | ft = bt2.IntegerFieldType(32, mapped_clock_class=cc) | |
21 | payloadft.append_field('zorg', ft) | |
22 | ec = bt2.EventClass('ec', payload_field_type=payloadft) | |
23 | self._sc.add_event_class(ec) | |
24 | trace = bt2.Trace() | |
25 | ||
26 | with self.assertRaises(bt2.Error): | |
27 | trace.add_stream_class(self._sc) | |
28 | ||
29 | def test_add_sc_packet_context_vs_event_class(self): | |
30 | def func(ft): | |
31 | self._sc.packet_context_field_type = ft | |
32 | ||
33 | self._test_add_sc_ft_vs_event_class(func) | |
34 | ||
35 | def test_add_sc_event_context_vs_event_class(self): | |
36 | def func(ft): | |
37 | self._sc.event_context_field_type = ft | |
38 | ||
39 | self._test_add_sc_ft_vs_event_class(func) | |
40 | ||
41 | def test_add_sc_event_header_vs_event_class(self): | |
42 | def func(ft): | |
43 | self._sc.event_header_field_type = ft | |
44 | ||
45 | self._test_add_sc_ft_vs_event_class(func) | |
46 | ||
47 | def test_add_sc_event_class_vs_event_class(self): | |
48 | ehft = bt2.StructureFieldType() | |
49 | ft = bt2.IntegerFieldType(32) | |
50 | ehft.append_field('id', ft) | |
51 | self._sc.event_header_field_type = ehft | |
52 | cc = bt2.ClockClass('first_cc', 1000) | |
53 | payloadft = bt2.StructureFieldType() | |
54 | ft = bt2.IntegerFieldType(32, mapped_clock_class=cc) | |
55 | payloadft.append_field('zorg', ft) | |
56 | ec1 = bt2.EventClass('ec', payload_field_type=payloadft) | |
57 | cc = bt2.ClockClass('second_cc', 1000) | |
58 | payloadft = bt2.StructureFieldType() | |
59 | ft = bt2.IntegerFieldType(32, mapped_clock_class=cc) | |
60 | payloadft.append_field('logi', ft) | |
61 | ec2 = bt2.EventClass('ec', payload_field_type=payloadft) | |
62 | self._sc.add_event_class(ec1) | |
63 | self._sc.add_event_class(ec2) | |
64 | trace = bt2.Trace() | |
65 | ||
66 | with self.assertRaises(bt2.Error): | |
67 | trace.add_stream_class(self._sc) | |
68 | ||
69 | def _test_create_event_ft_vs_event_class(self, set_ft_func): | |
70 | cc = bt2.ClockClass('first_cc', 1000) | |
71 | scft = bt2.StructureFieldType() | |
72 | ft = bt2.IntegerFieldType(32, mapped_clock_class=cc) | |
73 | scft.append_field('salut', ft) | |
74 | set_ft_func(scft) | |
75 | cc = bt2.ClockClass('second_cc', 1000) | |
76 | payloadft = bt2.StructureFieldType() | |
77 | ft = bt2.IntegerFieldType(32, mapped_clock_class=cc) | |
78 | payloadft.append_field('zorg', ft) | |
79 | ec = bt2.EventClass('ec', payload_field_type=payloadft) | |
80 | self._sc.add_event_class(ec) | |
81 | ||
82 | with self.assertRaises(bt2.Error): | |
83 | ev = ec() | |
84 | ||
85 | def test_create_event_packet_context_vs_event_class(self): | |
86 | def func(ft): | |
87 | self._sc.packet_context_field_type = ft | |
88 | ||
89 | self._test_create_event_ft_vs_event_class(func) | |
90 | ||
91 | def test_create_event_event_context_vs_event_class(self): | |
92 | def func(ft): | |
93 | self._sc.event_context_field_type = ft | |
94 | ||
95 | self._test_create_event_ft_vs_event_class(func) | |
96 | ||
97 | def test_create_event_event_header_vs_event_class(self): | |
98 | def func(ft): | |
99 | self._sc.event_header_field_type = ft | |
100 | ||
101 | self._test_create_event_ft_vs_event_class(func) | |
102 | ||
103 | def test_create_event_event_class_vs_event_class(self): | |
104 | ehft = bt2.StructureFieldType() | |
105 | ft = bt2.IntegerFieldType(32) | |
106 | ehft.append_field('id', ft) | |
107 | self._sc.event_header_field_type = ehft | |
108 | cc = bt2.ClockClass('first_cc', 1000) | |
109 | payloadft = bt2.StructureFieldType() | |
110 | ft = bt2.IntegerFieldType(32, mapped_clock_class=cc) | |
111 | payloadft.append_field('zorg', ft) | |
112 | ec1 = bt2.EventClass('ec', payload_field_type=payloadft) | |
113 | cc = bt2.ClockClass('second_cc', 1000) | |
114 | payloadft = bt2.StructureFieldType() | |
115 | ft = bt2.IntegerFieldType(32, mapped_clock_class=cc) | |
116 | payloadft.append_field('logi', ft) | |
117 | ec2 = bt2.EventClass('ec', payload_field_type=payloadft) | |
118 | self._sc.add_event_class(ec1) | |
119 | self._sc.add_event_class(ec2) | |
120 | ||
121 | with self.assertRaises(bt2.Error): | |
122 | ev = ec1() | |
123 | ||
124 | def test_add_ec_after_add_sc(self): | |
125 | cc = bt2.ClockClass('first_cc', 1000) | |
126 | scft = bt2.StructureFieldType() | |
127 | ft = bt2.IntegerFieldType(32, mapped_clock_class=cc) | |
128 | scft.append_field('salut', ft) | |
129 | self._sc.packet_context_field_type = scft | |
130 | trace = bt2.Trace() | |
131 | trace.add_stream_class(self._sc) | |
132 | cc = bt2.ClockClass('second_cc', 1000) | |
133 | payloadft = bt2.StructureFieldType() | |
134 | ft = bt2.IntegerFieldType(32, mapped_clock_class=cc) | |
135 | payloadft.append_field('zorg', ft) | |
136 | ec = bt2.EventClass('ec', payload_field_type=payloadft) | |
137 | ||
138 | with self.assertRaises(bt2.Error): | |
139 | self._sc.add_event_class(ec) | |
140 | ||
141 | def test_add_ec_after_add_sc_single_cc(self): | |
142 | ehft = bt2.StructureFieldType() | |
143 | ft = bt2.IntegerFieldType(32) | |
144 | ehft.append_field('id', ft) | |
145 | self._sc.event_header_field_type = ehft | |
146 | trace = bt2.Trace() | |
147 | trace.add_stream_class(self._sc) | |
148 | cc = bt2.ClockClass('first_cc', 1000) | |
149 | payloadft = bt2.StructureFieldType() | |
150 | ft = bt2.IntegerFieldType(32, mapped_clock_class=cc) | |
151 | payloadft.append_field('zorg', ft) | |
152 | ec = bt2.EventClass('ec', payload_field_type=payloadft) | |
153 | self._sc.add_event_class(ec) | |
154 | cc = bt2.ClockClass('second_cc', 1000) | |
155 | payloadft = bt2.StructureFieldType() | |
156 | ft = bt2.IntegerFieldType(32, mapped_clock_class=cc) | |
157 | payloadft.append_field('lel', ft) | |
158 | ec = bt2.EventClass('ec', payload_field_type=payloadft) | |
159 | ||
160 | with self.assertRaises(bt2.Error): | |
161 | self._sc.add_event_class(ec) | |
162 | ||
163 | def test_add_ec_after_create_event(self): | |
164 | ehft = bt2.StructureFieldType() | |
165 | ft = bt2.IntegerFieldType(32) | |
166 | ehft.append_field('id', ft) | |
167 | self._sc.event_header_field_type = ehft | |
168 | cc = bt2.ClockClass('first_cc', 1000) | |
169 | payloadft = bt2.StructureFieldType() | |
170 | ft = bt2.IntegerFieldType(32, mapped_clock_class=cc) | |
171 | payloadft.append_field('zorg', ft) | |
172 | ec1 = bt2.EventClass('ec', payload_field_type=payloadft) | |
173 | self._sc.add_event_class(ec1) | |
174 | ev = ec1() | |
175 | cc = bt2.ClockClass('second_cc', 1000) | |
176 | payloadft = bt2.StructureFieldType() | |
177 | ft = bt2.IntegerFieldType(32, mapped_clock_class=cc) | |
178 | payloadft.append_field('logi', ft) | |
179 | ec2 = bt2.EventClass('ec', payload_field_type=payloadft) | |
180 | ||
181 | with self.assertRaises(bt2.Error): | |
182 | self._sc.add_event_class(ec2) | |
183 | ||
184 | def test_sc_clock_matches_expected_clock_class(self): | |
185 | clock = bt2.CtfWriterClock('sc_clock') | |
186 | self._sc.clock = clock | |
187 | writer = bt2.CtfWriter('/tmp') | |
188 | writer.add_clock(clock) | |
189 | writer.trace.add_stream_class(self._sc) | |
190 | cc = bt2.ClockClass('other_cc', 1000) | |
191 | payloadft = bt2.StructureFieldType() | |
192 | ft = bt2.IntegerFieldType(32, mapped_clock_class=cc) | |
193 | payloadft.append_field('zorg', ft) | |
194 | ec = bt2.EventClass('ec', payload_field_type=payloadft) | |
195 | ||
196 | with self.assertRaises(bt2.Error): | |
197 | self._sc.add_event_class(ec) |