| 1 | from collections import OrderedDict |
| 2 | import unittest |
| 3 | from utils import run_in_component_init |
| 4 | |
| 5 | |
| 6 | class PacketTestCase(unittest.TestCase): |
| 7 | @staticmethod |
| 8 | def _create_packet(with_pc): |
| 9 | def create_tc_cc(comp_self): |
| 10 | cc = comp_self._create_clock_class(frequency=1000, name='my_cc') |
| 11 | tc = comp_self._create_trace_class() |
| 12 | return cc, tc |
| 13 | |
| 14 | clock_class, tc = run_in_component_init(create_tc_cc) |
| 15 | |
| 16 | # stream event context |
| 17 | sec = tc.create_structure_field_class() |
| 18 | sec += OrderedDict(( |
| 19 | ('cpu_id', tc.create_signed_integer_field_class(8)), |
| 20 | ('stuff', tc.create_real_field_class()), |
| 21 | )) |
| 22 | |
| 23 | # packet context |
| 24 | pc = None |
| 25 | if with_pc: |
| 26 | pc = tc.create_structure_field_class() |
| 27 | pc += OrderedDict(( |
| 28 | ('something', tc.create_signed_integer_field_class(8)), |
| 29 | ('something_else', tc.create_real_field_class()), |
| 30 | ('events_discarded', tc.create_unsigned_integer_field_class(64)), |
| 31 | ('packet_seq_num', tc.create_unsigned_integer_field_class(64)), |
| 32 | )) |
| 33 | |
| 34 | # stream class |
| 35 | sc = tc.create_stream_class(default_clock_class=clock_class, |
| 36 | event_common_context_field_class=sec, |
| 37 | packet_context_field_class=pc) |
| 38 | |
| 39 | # event context |
| 40 | ec = tc.create_structure_field_class() |
| 41 | ec += OrderedDict(( |
| 42 | ('ant', tc.create_signed_integer_field_class(16)), |
| 43 | ('msg', tc.create_string_field_class()), |
| 44 | )) |
| 45 | |
| 46 | # event payload |
| 47 | ep = tc.create_structure_field_class() |
| 48 | ep += OrderedDict(( |
| 49 | ('giraffe', tc.create_signed_integer_field_class(32)), |
| 50 | ('gnu', tc.create_signed_integer_field_class(8)), |
| 51 | ('mosquito', tc.create_signed_integer_field_class(8)), |
| 52 | )) |
| 53 | |
| 54 | # event class |
| 55 | event_class = sc.create_event_class(name='ec', payload_field_class=ep) |
| 56 | event_class.common_context_field_class = ec |
| 57 | |
| 58 | # trace |
| 59 | trace = tc() |
| 60 | |
| 61 | # stream |
| 62 | stream = trace.create_stream(sc) |
| 63 | |
| 64 | # packet |
| 65 | return stream.create_packet(), stream, pc |
| 66 | |
| 67 | def test_attr_stream(self): |
| 68 | packet, stream, _ = self._create_packet(with_pc=True) |
| 69 | self.assertEqual(packet.stream.addr, stream.addr) |
| 70 | |
| 71 | def test_context_field(self): |
| 72 | packet, stream, pc_fc = self._create_packet(with_pc=True) |
| 73 | self.assertEqual(packet.context_field.field_class.addr, pc_fc.addr) |
| 74 | |
| 75 | def test_no_context_field(self): |
| 76 | packet, _, _ = self._create_packet(with_pc=False) |
| 77 | self.assertIsNone(packet.context_field) |