# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
#
-from collections import OrderedDict
import unittest
+import utils
from utils import run_in_component_init
+from bt2 import stream as bt2_stream
+from bt2 import field as bt2_field
class PacketTestCase(unittest.TestCase):
# stream event context
sec = tc.create_structure_field_class()
- sec += OrderedDict((
+ sec += [
('cpu_id', tc.create_signed_integer_field_class(8)),
('stuff', tc.create_real_field_class()),
- ))
+ ]
# packet context
pc = None
if with_pc:
pc = tc.create_structure_field_class()
- pc += OrderedDict((
+ pc += [
('something', tc.create_signed_integer_field_class(8)),
('something_else', tc.create_real_field_class()),
('events_discarded', tc.create_unsigned_integer_field_class(64)),
('packet_seq_num', tc.create_unsigned_integer_field_class(64)),
- ))
+ ]
# stream class
- sc = tc.create_stream_class(default_clock_class=clock_class,
- event_common_context_field_class=sec,
- packet_context_field_class=pc,
- supports_packets=True)
+ sc = tc.create_stream_class(
+ default_clock_class=clock_class,
+ event_common_context_field_class=sec,
+ packet_context_field_class=pc,
+ supports_packets=True,
+ )
# event context
ec = tc.create_structure_field_class()
- ec += OrderedDict((
+ ec += [
('ant', tc.create_signed_integer_field_class(16)),
('msg', tc.create_string_field_class()),
- ))
+ ]
# event payload
ep = tc.create_structure_field_class()
- ep += OrderedDict((
+ ep += [
('giraffe', tc.create_signed_integer_field_class(32)),
('gnu', tc.create_signed_integer_field_class(8)),
('mosquito', tc.create_signed_integer_field_class(8)),
- ))
+ ]
# event class
event_class = sc.create_event_class(name='ec', payload_field_class=ep)
def test_attr_stream(self):
packet, stream, _ = self._create_packet(with_pc=True)
self.assertEqual(packet.stream.addr, stream.addr)
+ self.assertIs(type(packet.stream), bt2_stream._Stream)
+
+ def test_const_attr_stream(self):
+ packet = utils.get_const_packet_beginning_message().packet
+ self.assertIs(type(packet.stream), bt2_stream._StreamConst)
def test_context_field(self):
packet, stream, pc_fc = self._create_packet(with_pc=True)
- self.assertEqual(packet.context_field.field_class.addr, pc_fc.addr)
+ self.assertEqual(packet.context_field.cls.addr, pc_fc.addr)
+ self.assertIs(type(packet.context_field), bt2_field._StructureField)
+
+ def test_const_context_field(self):
+ packet = utils.get_const_packet_beginning_message().packet
+ self.assertIs(type(packet.context_field), bt2_field._StructureFieldConst)
def test_no_context_field(self):
packet, _, _ = self._create_packet(with_pc=False)