1 # SPDX-License-Identifier: GPL-2.0-only
3 # Copyright (C) 2019 EfficiOS Inc.
8 from utils
import run_in_component_init
9 from bt2
import stream
as bt2_stream
10 from bt2
import field
as bt2_field
13 class PacketTestCase(unittest
.TestCase
):
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
()
21 clock_class
, tc
= run_in_component_init(create_tc_cc
)
23 # stream event context
24 sec
= tc
.create_structure_field_class()
26 ("cpu_id", tc
.create_signed_integer_field_class(8)),
27 ("stuff", tc
.create_double_precision_real_field_class()),
33 pc
= tc
.create_structure_field_class()
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)),
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,
50 ec
= tc
.create_structure_field_class()
52 ("ant", tc
.create_signed_integer_field_class(16)),
53 ("msg", tc
.create_string_field_class()),
57 ep
= tc
.create_structure_field_class()
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)),
65 event_class
= sc
.create_event_class(name
="ec", payload_field_class
=ep
)
66 event_class
.common_context_field_class
= ec
72 stream
= trace
.create_stream(sc
)
75 return stream
.create_packet(), stream
, pc
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
)
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
)
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
)
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
)
95 def test_no_context_field(self
):
96 packet
, _
, _
= self
._create
_packet
(with_pc
=False)
97 self
.assertIsNone(packet
.context_field
)
100 if __name__
== "__main__":