1 # SPDX-License-Identifier: GPL-2.0-only
3 # Copyright (C) 2019 EfficiOS Inc.
9 from bt2
import field
as bt2_field
10 from bt2
import stream
as bt2_stream
11 from utils
import run_in_component_init
14 class PacketTestCase(unittest
.TestCase
):
16 def _create_packet(with_pc
):
17 def create_tc_cc(comp_self
):
18 cc
= comp_self
._create
_clock
_class
(frequency
=1000, name
="my_cc")
19 tc
= comp_self
._create
_trace
_class
()
22 clock_class
, tc
= run_in_component_init(create_tc_cc
)
24 # stream event context
25 sec
= tc
.create_structure_field_class()
27 ("cpu_id", tc
.create_signed_integer_field_class(8)),
28 ("stuff", tc
.create_double_precision_real_field_class()),
34 pc
= tc
.create_structure_field_class()
36 ("something", tc
.create_signed_integer_field_class(8)),
37 ("something_else", tc
.create_double_precision_real_field_class()),
38 ("events_discarded", tc
.create_unsigned_integer_field_class(64)),
39 ("packet_seq_num", tc
.create_unsigned_integer_field_class(64)),
43 sc
= tc
.create_stream_class(
44 default_clock_class
=clock_class
,
45 event_common_context_field_class
=sec
,
46 packet_context_field_class
=pc
,
47 supports_packets
=True,
51 ec
= tc
.create_structure_field_class()
53 ("ant", tc
.create_signed_integer_field_class(16)),
54 ("msg", tc
.create_string_field_class()),
58 ep
= tc
.create_structure_field_class()
60 ("giraffe", tc
.create_signed_integer_field_class(32)),
61 ("gnu", tc
.create_signed_integer_field_class(8)),
62 ("mosquito", tc
.create_signed_integer_field_class(8)),
66 event_class
= sc
.create_event_class(name
="ec", payload_field_class
=ep
)
67 event_class
.common_context_field_class
= ec
73 stream
= trace
.create_stream(sc
)
76 return stream
.create_packet(), stream
, pc
78 def test_attr_stream(self
):
79 packet
, stream
, _
= self
._create
_packet
(with_pc
=True)
80 self
.assertEqual(packet
.stream
.addr
, stream
.addr
)
81 self
.assertIs(type(packet
.stream
), bt2_stream
._Stream
)
83 def test_const_attr_stream(self
):
84 packet
= utils
.get_const_packet_beginning_message().packet
85 self
.assertIs(type(packet
.stream
), bt2_stream
._StreamConst
)
87 def test_context_field(self
):
88 packet
, stream
, pc_fc
= self
._create
_packet
(with_pc
=True)
89 self
.assertEqual(packet
.context_field
.cls
.addr
, pc_fc
.addr
)
90 self
.assertIs(type(packet
.context_field
), bt2_field
._StructureField
)
92 def test_const_context_field(self
):
93 packet
= utils
.get_const_packet_beginning_message().packet
94 self
.assertIs(type(packet
.context_field
), bt2_field
._StructureFieldConst
)
96 def test_no_context_field(self
):
97 packet
, _
, _
= self
._create
_packet
(with_pc
=False)
98 self
.assertIsNone(packet
.context_field
)
101 if __name__
== "__main__":