Commit | Line | Data |
---|---|---|
c4239792 | 1 | from bt2 import value |
c7eee084 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 | |
14 | def _query(cls, query_exec, obj, params): | |
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 | |
47 | def _query(cls, query_exec, obj, params): | |
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 | ||
56 | def test_query_gen_error(self): | |
57 | class MySink(bt2._UserSinkComponent): | |
58 | def _consume(self): | |
59 | pass | |
60 | ||
61 | @classmethod | |
62 | def _query(cls, query_exec, obj, params): | |
63 | raise ValueError | |
64 | ||
65 | with self.assertRaises(bt2.Error): | |
66 | res = bt2.QueryExecutor().query(MySink, 'obj', [17, 23]) | |
67 | ||
68 | def test_query_invalid_object(self): | |
69 | class MySink(bt2._UserSinkComponent): | |
70 | def _consume(self): | |
71 | pass | |
72 | ||
73 | @classmethod | |
74 | def _query(cls, query_exec, obj, params): | |
75 | raise bt2.InvalidQueryObject | |
76 | ||
77 | with self.assertRaises(bt2.InvalidQueryObject): | |
78 | res = bt2.QueryExecutor().query(MySink, 'obj', [17, 23]) | |
79 | ||
80 | def test_query_invalid_params(self): | |
81 | class MySink(bt2._UserSinkComponent): | |
82 | def _consume(self): | |
83 | pass | |
84 | ||
85 | @classmethod | |
86 | def _query(cls, query_exec, obj, params): | |
87 | raise bt2.InvalidQueryParams | |
88 | ||
89 | with self.assertRaises(bt2.InvalidQueryParams): | |
90 | res = bt2.QueryExecutor().query(MySink, 'obj', [17, 23]) | |
91 | ||
92 | def test_query_try_again(self): | |
93 | class MySink(bt2._UserSinkComponent): | |
94 | def _consume(self): | |
95 | pass | |
96 | ||
97 | @classmethod | |
98 | def _query(cls, query_exec, obj, params): | |
99 | raise bt2.TryAgain | |
100 | ||
101 | with self.assertRaises(bt2.TryAgain): | |
102 | res = bt2.QueryExecutor().query(MySink, 'obj', [17, 23]) | |
103 | ||
104 | def test_cancel_no_query(self): | |
105 | query_exec = bt2.QueryExecutor() | |
106 | self.assertFalse(query_exec.is_canceled) | |
107 | query_exec.cancel() | |
108 | self.assertTrue(query_exec.is_canceled) | |
109 | ||
110 | def test_query_canceled(self): | |
111 | class MySink(bt2._UserSinkComponent): | |
112 | def _consume(self): | |
113 | pass | |
114 | ||
115 | @classmethod | |
116 | def _query(cls, query_exec, obj, params): | |
117 | raise bt2.TryAgain | |
118 | ||
119 | query_exec = bt2.QueryExecutor() | |
120 | query_exec.cancel() | |
121 | ||
122 | with self.assertRaises(bt2.QueryExecutorCanceled): | |
123 | res = query_exec.query(MySink, 'obj', [17, 23]) |