Commit | Line | Data |
---|---|---|
ae0bfae8 | 1 | from bt2 import value |
1286dcbb PP |
2 | import unittest |
3 | import copy | |
4 | import bt2 | |
5 | ||
6 | ||
7 | class QueryExecutorTestCase(unittest.TestCase): | |
8 | def test_query(self): | |
9 | class MySink(bt2._UserSinkComponent): | |
10 | def _consume(self): | |
11 | pass | |
12 | ||
13 | @classmethod | |
83da519a | 14 | def _query(cls, query_exec, obj, params, log_level): |
1286dcbb PP |
15 | nonlocal query_params |
16 | query_params = params | |
17 | return { | |
18 | 'null': None, | |
19 | 'bt2': 'BT2', | |
20 | } | |
21 | ||
22 | query_params = None | |
23 | params = { | |
24 | 'array': ['coucou', 23, None], | |
25 | 'other_map': { | |
26 | 'yes': 'yeah', | |
27 | '19': 19, | |
28 | 'minus 1.5': -1.5, | |
29 | }, | |
30 | 'null': None, | |
31 | } | |
32 | ||
33 | res = bt2.QueryExecutor().query(MySink, 'obj', params) | |
34 | self.assertEqual(query_params, params) | |
35 | self.assertEqual(res, { | |
36 | 'null': None, | |
37 | 'bt2': 'BT2', | |
38 | }) | |
39 | del query_params | |
40 | ||
41 | def test_query_params_none(self): | |
42 | class MySink(bt2._UserSinkComponent): | |
43 | def _consume(self): | |
44 | pass | |
45 | ||
46 | @classmethod | |
83da519a | 47 | def _query(cls, query_exec, obj, params, log_level): |
1286dcbb PP |
48 | nonlocal query_params |
49 | query_params = params | |
50 | ||
51 | query_params = 23 | |
52 | res = bt2.QueryExecutor().query(MySink, 'obj', None) | |
53 | self.assertEqual(query_params, None) | |
54 | del query_params | |
55 | ||
999e7442 PP |
56 | def test_query_logging_level(self): |
57 | class MySink(bt2._UserSinkComponent): | |
58 | def _consume(self): | |
59 | pass | |
60 | ||
61 | @classmethod | |
62 | def _query(cls, query_exec, obj, params, log_level): | |
63 | nonlocal query_log_level | |
64 | query_log_level = log_level | |
65 | ||
66 | query_log_level = None | |
67 | res = bt2.QueryExecutor().query(MySink, 'obj', None, | |
68 | bt2.LoggingLevel.INFO) | |
69 | self.assertEqual(query_log_level, bt2.LoggingLevel.INFO) | |
70 | del query_log_level | |
71 | ||
1286dcbb PP |
72 | def test_query_gen_error(self): |
73 | class MySink(bt2._UserSinkComponent): | |
74 | def _consume(self): | |
75 | pass | |
76 | ||
77 | @classmethod | |
83da519a | 78 | def _query(cls, query_exec, obj, params, log_level): |
1286dcbb PP |
79 | raise ValueError |
80 | ||
81 | with self.assertRaises(bt2.Error): | |
82 | res = bt2.QueryExecutor().query(MySink, 'obj', [17, 23]) | |
83 | ||
84 | def test_query_invalid_object(self): | |
85 | class MySink(bt2._UserSinkComponent): | |
86 | def _consume(self): | |
87 | pass | |
88 | ||
89 | @classmethod | |
83da519a | 90 | def _query(cls, query_exec, obj, params, log_level): |
1286dcbb PP |
91 | raise bt2.InvalidQueryObject |
92 | ||
93 | with self.assertRaises(bt2.InvalidQueryObject): | |
94 | res = bt2.QueryExecutor().query(MySink, 'obj', [17, 23]) | |
95 | ||
999e7442 PP |
96 | def test_query_logging_level_invalid_type(self): |
97 | class MySink(bt2._UserSinkComponent): | |
98 | def _consume(self): | |
99 | pass | |
100 | ||
101 | @classmethod | |
102 | def _query(cls, query_exec, obj, params, log_level): | |
103 | pass | |
104 | ||
105 | with self.assertRaises(TypeError): | |
106 | res = bt2.QueryExecutor().query(MySink, 'obj', [17, 23], 'yeah') | |
107 | ||
108 | def test_query_logging_level_invalid_value(self): | |
109 | class MySink(bt2._UserSinkComponent): | |
110 | def _consume(self): | |
111 | pass | |
112 | ||
113 | @classmethod | |
114 | def _query(cls, query_exec, obj, params, log_level): | |
115 | pass | |
116 | ||
117 | with self.assertRaises(ValueError): | |
118 | res = bt2.QueryExecutor().query(MySink, 'obj', [17, 23], 12345) | |
119 | ||
1286dcbb PP |
120 | def test_query_invalid_params(self): |
121 | class MySink(bt2._UserSinkComponent): | |
122 | def _consume(self): | |
123 | pass | |
124 | ||
125 | @classmethod | |
83da519a | 126 | def _query(cls, query_exec, obj, params, log_level): |
1286dcbb PP |
127 | raise bt2.InvalidQueryParams |
128 | ||
129 | with self.assertRaises(bt2.InvalidQueryParams): | |
130 | res = bt2.QueryExecutor().query(MySink, 'obj', [17, 23]) | |
131 | ||
132 | def test_query_try_again(self): | |
133 | class MySink(bt2._UserSinkComponent): | |
134 | def _consume(self): | |
135 | pass | |
136 | ||
137 | @classmethod | |
83da519a | 138 | def _query(cls, query_exec, obj, params, log_level): |
1286dcbb PP |
139 | raise bt2.TryAgain |
140 | ||
141 | with self.assertRaises(bt2.TryAgain): | |
142 | res = bt2.QueryExecutor().query(MySink, 'obj', [17, 23]) | |
143 | ||
144 | def test_cancel_no_query(self): | |
145 | query_exec = bt2.QueryExecutor() | |
146 | self.assertFalse(query_exec.is_canceled) | |
147 | query_exec.cancel() | |
148 | self.assertTrue(query_exec.is_canceled) | |
149 | ||
150 | def test_query_canceled(self): | |
151 | class MySink(bt2._UserSinkComponent): | |
152 | def _consume(self): | |
153 | pass | |
154 | ||
155 | @classmethod | |
83da519a | 156 | def _query(cls, query_exec, obj, params, log_level): |
1286dcbb PP |
157 | raise bt2.TryAgain |
158 | ||
159 | query_exec = bt2.QueryExecutor() | |
160 | query_exec.cancel() | |
161 | ||
162 | with self.assertRaises(bt2.QueryExecutorCanceled): | |
163 | res = query_exec.query(MySink, 'obj', [17, 23]) |