| 1 | # SPDX-License-Identifier: GPL-2.0-only |
| 2 | # |
| 3 | # Copyright (C) 2019 EfficiOS Inc. |
| 4 | # |
| 5 | |
| 6 | import unittest |
| 7 | import utils |
| 8 | from utils import run_in_component_init |
| 9 | from bt2 import stream as bt2_stream |
| 10 | from bt2 import field as bt2_field |
| 11 | |
| 12 | |
| 13 | class PacketTestCase(unittest.TestCase): |
| 14 | @staticmethod |
| 15 | def _create_packet(with_pc): |
| 16 | def create_tc_cc(comp_self): |
| 17 | cc = comp_self._create_clock_class(frequency=1000, name="my_cc") |
| 18 | tc = comp_self._create_trace_class() |
| 19 | return cc, tc |
| 20 | |
| 21 | clock_class, tc = run_in_component_init(create_tc_cc) |
| 22 | |
| 23 | # stream event context |
| 24 | sec = tc.create_structure_field_class() |
| 25 | sec += [ |
| 26 | ("cpu_id", tc.create_signed_integer_field_class(8)), |
| 27 | ("stuff", tc.create_double_precision_real_field_class()), |
| 28 | ] |
| 29 | |
| 30 | # packet context |
| 31 | pc = None |
| 32 | if with_pc: |
| 33 | pc = tc.create_structure_field_class() |
| 34 | pc += [ |
| 35 | ("something", tc.create_signed_integer_field_class(8)), |
| 36 | ("something_else", tc.create_double_precision_real_field_class()), |
| 37 | ("events_discarded", tc.create_unsigned_integer_field_class(64)), |
| 38 | ("packet_seq_num", tc.create_unsigned_integer_field_class(64)), |
| 39 | ] |
| 40 | |
| 41 | # stream class |
| 42 | sc = tc.create_stream_class( |
| 43 | default_clock_class=clock_class, |
| 44 | event_common_context_field_class=sec, |
| 45 | packet_context_field_class=pc, |
| 46 | supports_packets=True, |
| 47 | ) |
| 48 | |
| 49 | # event context |
| 50 | ec = tc.create_structure_field_class() |
| 51 | ec += [ |
| 52 | ("ant", tc.create_signed_integer_field_class(16)), |
| 53 | ("msg", tc.create_string_field_class()), |
| 54 | ] |
| 55 | |
| 56 | # event payload |
| 57 | ep = tc.create_structure_field_class() |
| 58 | ep += [ |
| 59 | ("giraffe", tc.create_signed_integer_field_class(32)), |
| 60 | ("gnu", tc.create_signed_integer_field_class(8)), |
| 61 | ("mosquito", tc.create_signed_integer_field_class(8)), |
| 62 | ] |
| 63 | |
| 64 | # event class |
| 65 | event_class = sc.create_event_class(name="ec", payload_field_class=ep) |
| 66 | event_class.common_context_field_class = ec |
| 67 | |
| 68 | # trace |
| 69 | trace = tc() |
| 70 | |
| 71 | # stream |
| 72 | stream = trace.create_stream(sc) |
| 73 | |
| 74 | # packet |
| 75 | return stream.create_packet(), stream, pc |
| 76 | |
| 77 | def test_attr_stream(self): |
| 78 | packet, stream, _ = self._create_packet(with_pc=True) |
| 79 | self.assertEqual(packet.stream.addr, stream.addr) |
| 80 | self.assertIs(type(packet.stream), bt2_stream._Stream) |
| 81 | |
| 82 | def test_const_attr_stream(self): |
| 83 | packet = utils.get_const_packet_beginning_message().packet |
| 84 | self.assertIs(type(packet.stream), bt2_stream._StreamConst) |
| 85 | |
| 86 | def test_context_field(self): |
| 87 | packet, stream, pc_fc = self._create_packet(with_pc=True) |
| 88 | self.assertEqual(packet.context_field.cls.addr, pc_fc.addr) |
| 89 | self.assertIs(type(packet.context_field), bt2_field._StructureField) |
| 90 | |
| 91 | def test_const_context_field(self): |
| 92 | packet = utils.get_const_packet_beginning_message().packet |
| 93 | self.assertIs(type(packet.context_field), bt2_field._StructureFieldConst) |
| 94 | |
| 95 | def test_no_context_field(self): |
| 96 | packet, _, _ = self._create_packet(with_pc=False) |
| 97 | self.assertIsNone(packet.context_field) |
| 98 | |
| 99 | |
| 100 | if __name__ == "__main__": |
| 101 | unittest.main() |