| 1 |
import datetime |
|---|
| 2 |
import pickle |
|---|
| 3 |
import sys |
|---|
| 4 |
import unittest |
|---|
| 5 |
|
|---|
| 6 |
import dejavu |
|---|
| 7 |
from dejavu import logic |
|---|
| 8 |
|
|---|
| 9 |
nums = logic.codewalk.numeric_opcodes |
|---|
| 10 |
|
|---|
| 11 |
lx = "logic.Expression(lambda x: " |
|---|
| 12 |
|
|---|
| 13 |
|
|---|
| 14 |
class ExpressionTests(unittest.TestCase): |
|---|
| 15 |
|
|---|
| 16 |
def test_Expression_creation(self): |
|---|
| 17 |
e = logic.Expression(lambda x: dejavu.icontains(x.Status, 'c')) |
|---|
| 18 |
self.assertEqual(repr(e), lx + "dejavu.icontains(x.Status, 'c'))") |
|---|
| 19 |
self.assertEqual(e.func.func_code.co_code, |
|---|
| 20 |
'd\x03\x00|\x00\x00i\x03\x00d\x01\x00\x83\x02\x00S') |
|---|
| 21 |
|
|---|
| 22 |
|
|---|
| 23 |
|
|---|
| 24 |
|
|---|
| 25 |
def build_expr(field, criteria): |
|---|
| 26 |
k = lambda x: dejavu.icontains(getattr(x, field), criteria) |
|---|
| 27 |
return logic.Expression(k) |
|---|
| 28 |
e = build_expr(u'GroupName', u'Westmin') |
|---|
| 29 |
self.assertEqual(repr(e), lx + "dejavu.icontains(x.GroupName, u'Westmin'))") |
|---|
| 30 |
|
|---|
| 31 |
|
|---|
| 32 |
e = logic.Expression(lambda x: x.FirstDate > dejavu.today()) |
|---|
| 33 |
self.assertEqual(repr(e), lx + "x.FirstDate > dejavu.today())") |
|---|
| 34 |
|
|---|
| 35 |
|
|---|
| 36 |
e = logic.Expression(lambda x: x.has('Job') and x.Field == 'BC') |
|---|
| 37 |
self.assertEqual(repr(e), lx + "(x.has('Job')) and (x.Field == 'BC'))") |
|---|
| 38 |
e = logic.Expression(lambda x: not x.has('Job') and x.Field == 'BC') |
|---|
| 39 |
self.assertEqual(repr(e), lx + "(not (x.has('Job'))) and (x.Field == 'BC'))") |
|---|
| 40 |
|
|---|
| 41 |
|
|---|
| 42 |
e = logic.Expression(lambda x, y, z: x.Field == 'BC' and |
|---|
| 43 |
y.Qty > 3 and z.Qty < 20) |
|---|
| 44 |
self.assertEqual(repr(e), |
|---|
| 45 |
"logic.Expression(lambda x, y, z: (x.Field == 'BC')" |
|---|
| 46 |
" and ((y.Qty > 3) and (z.Qty < 20)))") |
|---|
| 47 |
|
|---|
| 48 |
|
|---|
| 49 |
e = logic.Expression(lambda x: x.Name in ['George', 'John']) |
|---|
| 50 |
self.assertEqual(repr(e), lx + "x.Name in ['George', 'John'])") |
|---|
| 51 |
|
|---|
| 52 |
def test_pickling(self): |
|---|
| 53 |
|
|---|
| 54 |
e = logic.Expression(lambda x: True) |
|---|
| 55 |
e.__setstate__(("lambda x: x.TripStatus != 'Inquiry' and x.Field " |
|---|
| 56 |
"== 'BC' and x.StartDate >= 3", {})) |
|---|
| 57 |
self.assertEqual(e.code(), |
|---|
| 58 |
"lambda x: (x.TripStatus != 'Inquiry') and ((x.Field " |
|---|
| 59 |
"== 'BC') and (x.StartDate >= 3))") |
|---|
| 60 |
|
|---|
| 61 |
|
|---|
| 62 |
e = logic.Expression(lambda x: x.LastDate > datetime.date(2004, 3, 1)) |
|---|
| 63 |
p = pickle.dumps(e) |
|---|
| 64 |
f = pickle.loads(p) |
|---|
| 65 |
self.assertEqual(repr(e), repr(f)) |
|---|
| 66 |
|
|---|
| 67 |
e = logic.Expression(lambda x: dejavu.icontains(x.Status, 'c')) |
|---|
| 68 |
p = pickle.dumps(e) |
|---|
| 69 |
f = pickle.loads(p) |
|---|
| 70 |
self.assertEqual(repr(e), repr(f)) |
|---|
| 71 |
|
|---|
| 72 |
def test_Expression_addition(self): |
|---|
| 73 |
a = lambda x: x.Date == 3 |
|---|
| 74 |
b = lambda x: x.Qty == 5 |
|---|
| 75 |
e = logic.Expression(a) |
|---|
| 76 |
e += logic.Expression(b) |
|---|
| 77 |
self.assertEqual(e.code(), "lambda x: (x.Date == 3) and (x.Qty == 5)") |
|---|
| 78 |
|
|---|
| 79 |
|
|---|
| 80 |
|
|---|
| 81 |
f = logic.Expression(lambda x: ((x.DateFrom == None or x.DateFrom <= datetime.date(2005, 11, 17)) |
|---|
| 82 |
and (x.DateTo == None or x.DateTo >= datetime.date(2005, 11, 17)))) |
|---|
| 83 |
f += logic.Expression(lambda x: x.DirectoryID == None or x.DirectoryID == 0) |
|---|
| 84 |
self.assertEqual(f.code(), |
|---|
| 85 |
'lambda x: (((x.DateFrom == None) or (x.DateFrom <= datetime.date(2005, 11, 17))) ' |
|---|
| 86 |
'and ((x.DateTo == None) or (x.DateTo >= datetime.date(2005, 11, 17)))) ' |
|---|
| 87 |
'and ((x.DirectoryID == None) or (x.DirectoryID == 0))') |
|---|
| 88 |
|
|---|
| 89 |
def test_Aggregator(self): |
|---|
| 90 |
a = lambda x: x.Date == 3 |
|---|
| 91 |
b = lambda x: x.Qty == 5 |
|---|
| 92 |
merged_code = nums(['LOAD_FAST', 0, 0, |
|---|
| 93 |
'LOAD_ATTR', 1, 0, |
|---|
| 94 |
'LOAD_CONST', 1, 0, |
|---|
| 95 |
'COMPARE_OP', 2, 0, |
|---|
| 96 |
'JUMP_IF_FALSE', 13, 0, |
|---|
| 97 |
'POP_TOP', |
|---|
| 98 |
'LOAD_FAST', 0, 0, |
|---|
| 99 |
'LOAD_ATTR', 2, 0, |
|---|
| 100 |
'LOAD_CONST', 2, 0, |
|---|
| 101 |
'COMPARE_OP', 2, 0, |
|---|
| 102 |
'RETURN_VALUE']) |
|---|
| 103 |
|
|---|
| 104 |
ag = logic.Aggregator(a) |
|---|
| 105 |
ag.and_combine(b) |
|---|
| 106 |
self.assertEqual(ag.bytecode(), merged_code) |
|---|
| 107 |
|
|---|
| 108 |
|
|---|
| 109 |
ag.and_combine(lambda y, **kw: y.Size < kw['Size']) |
|---|
| 110 |
self.assertEqual(ag.bytecode(), nums(['LOAD_FAST', 0, 0, |
|---|
| 111 |
'LOAD_ATTR', 1, 0, |
|---|
| 112 |
'LOAD_CONST', 1, 0, |
|---|
| 113 |
'COMPARE_OP', 2, 0, |
|---|
| 114 |
'JUMP_IF_FALSE', 13, 0, |
|---|
| 115 |
'POP_TOP', |
|---|
| 116 |
'LOAD_FAST', 0, 0, |
|---|
| 117 |
'LOAD_ATTR', 2, 0, |
|---|
| 118 |
'LOAD_CONST', 2, 0, |
|---|
| 119 |
'COMPARE_OP', 2, 0, |
|---|
| 120 |
'JUMP_IF_FALSE', 17, 0, |
|---|
| 121 |
'POP_TOP', |
|---|
| 122 |
'LOAD_FAST', 0, 0, |
|---|
| 123 |
'LOAD_ATTR', 3, 0, |
|---|
| 124 |
'LOAD_FAST', 1, 0, |
|---|
| 125 |
'LOAD_CONST', 3, 0, |
|---|
| 126 |
'BINARY_SUBSCR', |
|---|
| 127 |
'COMPARE_OP', 0, 0, |
|---|
| 128 |
'RETURN_VALUE'])) |
|---|
| 129 |
f = ag.function() |
|---|
| 130 |
if sys.hexversion >= (2 << 24 | 4 << 16): |
|---|
| 131 |
|
|---|
| 132 |
self.assertEqual(logic.Expression(f).code(), |
|---|
| 133 |
"lambda x, **kw: ((x.Date == 3) and (x.Qty == 5)) " |
|---|
| 134 |
"and (x.Size < kw['Size'])") |
|---|
| 135 |
else: |
|---|
| 136 |
|
|---|
| 137 |
f2 = lambda x, **kw: ((x.Date == 3) and x.Qty == 5) and x.Size < kw['Size'] |
|---|
| 138 |
self.assertEqual(logic.Expression(f).code(), logic.Expression(f2).code()) |
|---|
| 139 |
self.assertEqual(f.func_code.co_code, f2.func_code.co_code) |
|---|
| 140 |
|
|---|
| 141 |
|
|---|
| 142 |
|
|---|
| 143 |
ag = logic.Aggregator(lambda x, **kw: x.GroupID == u'4') |
|---|
| 144 |
newfunc = lambda x, **kw: x.UserID == u'rbre' |
|---|
| 145 |
ag.and_combine(newfunc) |
|---|
| 146 |
self.assertEqual(ag.instr_index, ([None] * 12 + |
|---|
| 147 |
[ag.instr_index[12]] * 17)) |
|---|
| 148 |
|
|---|
| 149 |
self.assertEqual(ag._bytecode, nums(['LOAD_FAST', 0, 0, |
|---|
| 150 |
'LOAD_ATTR', 1, 0, |
|---|
| 151 |
'LOAD_CONST', 1, 0, |
|---|
| 152 |
'COMPARE_OP', 2, 0, |
|---|
| 153 |
'JUMP_IF_FALSE', 13, 0, |
|---|
| 154 |
'POP_TOP', |
|---|
| 155 |
'LOAD_FAST', 0, 0, |
|---|
| 156 |
'LOAD_ATTR', 1, 0, |
|---|
| 157 |
'LOAD_CONST', 1, 0, |
|---|
| 158 |
'COMPARE_OP', 2, 0, |
|---|
| 159 |
'RETURN_VALUE'])) |
|---|
| 160 |
|
|---|
| 161 |
self.assertEqual(ag.bytecode(), nums(['LOAD_FAST', 0, 0, |
|---|
| 162 |
'LOAD_ATTR', 1, 0, |
|---|
| 163 |
'LOAD_CONST', 1, 0, |
|---|
| 164 |
'COMPARE_OP', 2, 0, |
|---|
| 165 |
'JUMP_IF_FALSE', 13, 0, |
|---|
| 166 |
'POP_TOP', |
|---|
| 167 |
'LOAD_FAST', 0, 0, |
|---|
| 168 |
'LOAD_ATTR', 2, 0, |
|---|
| 169 |
'LOAD_CONST', 2, 0, |
|---|
| 170 |
'COMPARE_OP', 2, 0, |
|---|
| 171 |
'RETURN_VALUE'])) |
|---|
| 172 |
|
|---|
| 173 |
def test_filter(self): |
|---|
| 174 |
f = logic.filter(Date=3) |
|---|
| 175 |
self.assertEqual(map(ord, f.func.func_code.co_code), |
|---|
| 176 |
nums(['LOAD_FAST', 0, 0, |
|---|
| 177 |
'LOAD_ATTR', 1, 0, |
|---|
| 178 |
'LOAD_CONST', 1, 0, |
|---|
| 179 |
'COMPARE_OP', 2, 0, |
|---|
| 180 |
'RETURN_VALUE'])) |
|---|
| 181 |
|
|---|
| 182 |
f = logic.filter(Name='Harry', Weight=300) |
|---|
| 183 |
self.assertEqual(map(ord, f.func.func_code.co_code), |
|---|
| 184 |
nums(['LOAD_FAST', 0, 0, |
|---|
| 185 |
'LOAD_ATTR', 1, 0, |
|---|
| 186 |
'LOAD_CONST', 1, 0, |
|---|
| 187 |
'COMPARE_OP', 2, 0, |
|---|
| 188 |
'JUMP_IF_FALSE', 13, 0, |
|---|
| 189 |
'POP_TOP', |
|---|
| 190 |
'LOAD_FAST', 0, 0, |
|---|
| 191 |
'LOAD_ATTR', 2, 0, |
|---|
| 192 |
'LOAD_CONST', 2, 0, |
|---|
| 193 |
'COMPARE_OP', 2, 0, |
|---|
| 194 |
'RETURN_VALUE'])) |
|---|
| 195 |
|
|---|
| 196 |
def test_comparison(self): |
|---|
| 197 |
f = logic.comparison('Name', 2, 'Harry') |
|---|
| 198 |
g = logic.Expression(lambda x: x.Name == 'Harry') |
|---|
| 199 |
self.assertEqual(f.func.func_code, g.func.func_code) |
|---|
| 200 |
|
|---|
| 201 |
f = logic.comparison('Size', 4, 300) |
|---|
| 202 |
g = logic.Expression(lambda x: x.Size > 300) |
|---|
| 203 |
self.assertEqual(f.func.func_code, g.func.func_code) |
|---|
| 204 |
|
|---|
| 205 |
f = logic.comparison(u'ID', 2, u'30003') |
|---|
| 206 |
g = logic.Expression(lambda x: x.ID == u'30003') |
|---|
| 207 |
self.assertEqual(f.func.func_code, g.func.func_code) |
|---|
| 208 |
|
|---|
| 209 |
|
|---|
| 210 |
if __name__ == "__main__": |
|---|
| 211 |
unittest.main(__name__) |
|---|
| 212 |
|
|---|