Commit | Line | Data |
---|---|---|
811644b8 PP |
1 | from bt2 import values |
2 | import unittest | |
3 | import copy | |
4 | import bt2 | |
5 | ||
6 | ||
7 | class UserNotificationIteratorTestCase(unittest.TestCase): | |
8 | @staticmethod | |
9 | def _create_graph(src_comp_cls): | |
10 | class MySink(bt2._UserSinkComponent): | |
11 | def __init__(self, params): | |
12 | self._add_input_port('in') | |
13 | ||
14 | def _consume(self): | |
15 | next(self._notif_iter) | |
16 | ||
17 | def _port_connected(self, port, other_port): | |
18 | self._notif_iter = port.connection.create_notification_iterator() | |
19 | ||
20 | graph = bt2.Graph() | |
21 | src_comp = graph.add_component(src_comp_cls, 'src') | |
22 | sink_comp = graph.add_component(MySink, 'sink') | |
23 | graph.connect_ports(src_comp.output_ports['out'], | |
24 | sink_comp.input_ports['in']) | |
25 | return graph | |
26 | ||
27 | def test_init(self): | |
28 | class MyIter(bt2._UserNotificationIterator): | |
29 | def __init__(self): | |
30 | nonlocal initialized | |
31 | initialized = True | |
32 | ||
33 | class MySource(bt2._UserSourceComponent, | |
34 | notification_iterator_class=MyIter): | |
35 | def __init__(self, params): | |
36 | self._add_output_port('out') | |
37 | ||
38 | initialized = False | |
39 | graph = self._create_graph(MySource) | |
40 | self.assertTrue(initialized) | |
41 | ||
42 | def test_finalize(self): | |
43 | class MyIter(bt2._UserNotificationIterator): | |
44 | def _finalize(self): | |
45 | nonlocal finalized | |
46 | finalized = True | |
47 | ||
48 | class MySource(bt2._UserSourceComponent, | |
49 | notification_iterator_class=MyIter): | |
50 | def __init__(self, params): | |
51 | self._add_output_port('out') | |
52 | ||
53 | finalized = False | |
54 | graph = self._create_graph(MySource) | |
55 | del graph | |
56 | self.assertTrue(finalized) | |
57 | ||
58 | def test_component(self): | |
59 | class MyIter(bt2._UserNotificationIterator): | |
60 | def __init__(self): | |
61 | nonlocal salut | |
62 | salut = self._component._salut | |
63 | ||
64 | class MySource(bt2._UserSourceComponent, | |
65 | notification_iterator_class=MyIter): | |
66 | def __init__(self, params): | |
67 | self._add_output_port('out') | |
68 | self._salut = 23 | |
69 | ||
70 | salut = None | |
71 | graph = self._create_graph(MySource) | |
72 | self.assertEqual(salut, 23) | |
73 | ||
74 | def test_addr(self): | |
75 | class MyIter(bt2._UserNotificationIterator): | |
76 | def __init__(self): | |
77 | nonlocal addr | |
78 | addr = self.addr | |
79 | ||
80 | class MySource(bt2._UserSourceComponent, | |
81 | notification_iterator_class=MyIter): | |
82 | def __init__(self, params): | |
83 | self._add_output_port('out') | |
84 | ||
85 | addr = None | |
86 | graph = self._create_graph(MySource) | |
87 | self.assertIsNotNone(addr) | |
88 | self.assertNotEqual(addr, 0) | |
89 | ||
90 | ||
91 | class GenericNotificationIteratorTestCase(unittest.TestCase): | |
92 | def test_component(self): | |
93 | class MyIter(bt2._UserNotificationIterator): | |
94 | pass | |
95 | ||
96 | class MySource(bt2._UserSourceComponent, | |
97 | notification_iterator_class=MyIter): | |
98 | def __init__(self, params): | |
99 | self._add_output_port('out') | |
100 | ||
101 | class MySink(bt2._UserSinkComponent): | |
102 | def __init__(self, params): | |
103 | self._add_input_port('in') | |
104 | ||
105 | def _consume(self): | |
106 | next(self._notif_iter) | |
107 | ||
108 | def _port_connected(self, port, other_port): | |
109 | nonlocal upstream_comp | |
110 | self._notif_iter = port.connection.create_notification_iterator() | |
111 | upstream_comp = self._notif_iter.component | |
112 | ||
113 | upstream_comp = None | |
114 | graph = bt2.Graph() | |
115 | src_comp = graph.add_component(MySource, 'src') | |
116 | sink_comp = graph.add_component(MySink, 'sink') | |
117 | graph.connect_ports(src_comp.output_ports['out'], | |
118 | sink_comp.input_ports['in']) | |
119 | self.assertEqual(src_comp, upstream_comp) | |
120 | del upstream_comp |