22 22
from cassandra.cqlengine import management
23 23 24 24 25 -
class TestPolymorphicClassConstruction(BaseCassEngTestCase):
26 - 27 -
def test_multiple_polymorphic_key_failure(self):
28 -
""" Tests that defining a model with more than one polymorphic key fails """
29 -
with self.assertRaises(models.ModelDefinitionException):
30 -
class M(models.Model):
31 - 32 -
partition = columns.Integer(primary_key=True)
33 -
type1 = columns.Integer(polymorphic_key=True)
34 -
type2 = columns.Integer(polymorphic_key=True)
35 - 36 -
def test_no_polymorphic_key_column_failure(self):
37 -
with self.assertRaises(models.ModelDefinitionException):
38 -
class M(models.Model):
39 -
__polymorphic_key__ = 1
40 - 41 -
def test_polymorphic_key_inheritance(self):
42 -
""" Tests that polymorphic_key attribute is not inherited """
43 -
class Base(models.Model):
44 - 45 -
partition = columns.Integer(primary_key=True)
46 -
type1 = columns.Integer(polymorphic_key=True)
47 - 48 -
class M1(Base):
49 -
__polymorphic_key__ = 1
50 - 51 -
class M2(M1):
52 -
pass
53 - 54 -
assert M2.__polymorphic_key__ is None
55 - 56 -
def test_polymorphic_metaclass(self):
57 -
""" Tests that the model meta class configures polymorphic models properly """
58 -
class Base(models.Model):
59 - 60 -
partition = columns.Integer(primary_key=True)
61 -
type1 = columns.Integer(polymorphic_key=True)
62 - 63 -
class M1(Base):
64 -
__polymorphic_key__ = 1
65 - 66 -
assert Base._is_polymorphic
67 -
assert M1._is_polymorphic
68 - 69 -
assert Base._is_polymorphic_base
70 -
assert not M1._is_polymorphic_base
71 - 72 -
assert Base._discriminator_column is Base._columns['type1']
73 -
assert M1._discriminator_column is M1._columns['type1']
74 - 75 -
assert Base._discriminator_column_name == 'type1'
76 -
assert M1._discriminator_column_name == 'type1'
77 - 78 -
def test_table_names_are_inherited_from_poly_base(self):
79 -
class Base(models.Model):
80 - 81 -
partition = columns.Integer(primary_key=True)
82 -
type1 = columns.Integer(polymorphic_key=True)
83 - 84 -
class M1(Base):
85 -
__polymorphic_key__ = 1
86 - 87 -
assert Base.column_family_name() == M1.column_family_name()
88 - 89 -
def test_collection_columns_cant_be_polymorphic_keys(self):
90 -
with self.assertRaises(models.ModelDefinitionException):
91 -
class Base(models.Model):
92 - 93 -
partition = columns.Integer(primary_key=True)
94 -
type1 = columns.Set(columns.Integer, polymorphic_key=True)
95 - 96 - 97 -
class PolyBase(models.Model):
98 - 99 -
partition = columns.UUID(primary_key=True, default=uuid.uuid4)
100 -
row_type = columns.Integer(polymorphic_key=True)
101 - 102 - 103 -
class Poly1(PolyBase):
104 -
__polymorphic_key__ = 1
105 -
data1 = columns.Text()
106 - 107 - 108 -
class Poly2(PolyBase):
109 -
__polymorphic_key__ = 2
110 -
data2 = columns.Text()
111 - 112 - 113 -
class TestPolymorphicModel(BaseCassEngTestCase):
114 - 115 -
@classmethod
116 -
def setUpClass(cls):
117 -
super(TestPolymorphicModel, cls).setUpClass()
118 -
management.sync_table(Poly1)
119 -
management.sync_table(Poly2)
120 - 121 -
@classmethod
122 -
def tearDownClass(cls):
123 -
super(TestPolymorphicModel, cls).tearDownClass()
124 -
management.drop_table(Poly1)
125 -
management.drop_table(Poly2)
126 - 127 -
def test_saving_base_model_fails(self):
128 -
with self.assertRaises(models.PolymorphicModelException):
129 -
PolyBase.create()
130 - 131 -
def test_saving_subclass_saves_poly_key(self):
132 -
p1 = Poly1.create(data1='pickle')
133 -
p2 = Poly2.create(data2='bacon')
134 - 135 -
assert p1.row_type == Poly1.__polymorphic_key__
136 -
assert p2.row_type == Poly2.__polymorphic_key__
137 - 138 -
def test_query_deserialization(self):
139 -
p1 = Poly1.create(data1='pickle')
140 -
p2 = Poly2.create(data2='bacon')
141 - 142 -
p1r = PolyBase.get(partition=p1.partition)
143 -
p2r = PolyBase.get(partition=p2.partition)
144 - 145 -
assert isinstance(p1r, Poly1)
146 -
assert isinstance(p2r, Poly2)
147 - 148 -
def test_delete_on_polymorphic_subclass_does_not_include_polymorphic_key(self):
149 -
p1 = Poly1.create()
150 -
session = get_session()
151 -
with mock.patch.object(session, 'execute') as m:
152 -
Poly1.objects(partition=p1.partition).delete()
153 - 154 -
# make sure our polymorphic key isn't in the CQL
155 -
# not sure how we would even get here if it was in there
156 -
# since the CQL would fail.
157 - 158 -
self.assertNotIn("row_type", m.call_args[0][0].query_string)
159 - 160 - 161 -
class UnindexedPolyBase(models.Model):
162 - 163 -
partition = columns.UUID(primary_key=True, default=uuid.uuid4)
164 -
cluster = columns.UUID(primary_key=True, default=uuid.uuid4)
165 -
row_type = columns.Integer(polymorphic_key=True)
166 - 167 - 168 -
class UnindexedPoly1(UnindexedPolyBase):
169 -
__polymorphic_key__ = 1
170 -
data1 = columns.Text()
171 - 172 - 173 -
class UnindexedPoly2(UnindexedPolyBase):
174 -
__polymorphic_key__ = 2
175 -
data2 = columns.Text()
176 - 177 - 178 -
class UnindexedPoly3(UnindexedPoly2):
179 -
__polymorphic_key__ = 3
180 -
data3 = columns.Text()
181 - 182 - 183 -
class TestUnindexedPolymorphicQuery(BaseCassEngTestCase):
184 - 185 -
@classmethod
186 -
def setUpClass(cls):
187 -
super(TestUnindexedPolymorphicQuery, cls).setUpClass()
188 -
management.sync_table(UnindexedPoly1)
189 -
management.sync_table(UnindexedPoly2)
190 -
management.sync_table(UnindexedPoly3)
191 - 192 -
cls.p1 = UnindexedPoly1.create(data1='pickle')
193 -
cls.p2 = UnindexedPoly2.create(partition=cls.p1.partition, data2='bacon')
194 -
cls.p3 = UnindexedPoly3.create(partition=cls.p1.partition, data3='turkey')
195 - 196 -
@classmethod
197 -
def tearDownClass(cls):
198 -
super(TestUnindexedPolymorphicQuery, cls).tearDownClass()
199 -
management.drop_table(UnindexedPoly1)
200 -
management.drop_table(UnindexedPoly2)
201 -
management.drop_table(UnindexedPoly3)
202 - 203 -
def test_non_conflicting_type_results_work(self):
204 -
p1, p2, p3 = self.p1, self.p2, self.p3
205 -
assert len(list(UnindexedPoly1.objects(partition=p1.partition, cluster=p1.cluster))) == 1
206 -
assert len(list(UnindexedPoly2.objects(partition=p1.partition, cluster=p2.cluster))) == 1
207 -
assert len(list(UnindexedPoly3.objects(partition=p1.partition, cluster=p3.cluster))) == 1
208 - 209 -
def test_subclassed_model_results_work_properly(self):
210 -
p1, p2, p3 = self.p1, self.p2, self.p3
211 -
assert len(list(UnindexedPoly2.objects(partition=p1.partition, cluster__in=[p2.cluster, p3.cluster]))) == 2
212 - 213 -
def test_conflicting_type_results(self):
214 -
with self.assertRaises(models.PolymorphicModelException):
215 -
list(UnindexedPoly1.objects(partition=self.p1.partition))
216 -
with self.assertRaises(models.PolymorphicModelException):
217 -
list(UnindexedPoly2.objects(partition=self.p1.partition))
218 - 219 - 220 -
class IndexedPolyBase(models.Model):
221 - 222 -
partition = columns.UUID(primary_key=True, default=uuid.uuid4)
223 -
cluster = columns.UUID(primary_key=True, default=uuid.uuid4)
224 -
row_type = columns.Integer(polymorphic_key=True, index=True)
225 - 226 - 227 -
class IndexedPoly1(IndexedPolyBase):
228 -
__polymorphic_key__ = 1
229 -
data1 = columns.Text()
230 - 231 - 232 -
class IndexedPoly2(IndexedPolyBase):
233 -
__polymorphic_key__ = 2
234 -
data2 = columns.Text()
235 - 236 - 237 -
class TestIndexedPolymorphicQuery(BaseCassEngTestCase):
238 - 239 -
@classmethod
240 -
def setUpClass(cls):
241 -
super(TestIndexedPolymorphicQuery, cls).setUpClass()
242 -
management.sync_table(IndexedPoly1)
243 -
management.sync_table(IndexedPoly2)
244 - 245 -
cls.p1 = IndexedPoly1.create(data1='pickle')
246 -
cls.p2 = IndexedPoly2.create(partition=cls.p1.partition, data2='bacon')
247 - 248 -
@classmethod
249 -
def tearDownClass(cls):
250 -
super(TestIndexedPolymorphicQuery, cls).tearDownClass()
251 -
management.drop_table(IndexedPoly1)
252 -
management.drop_table(IndexedPoly2)
253 - 254 -
def test_success_case(self):
255 -
assert len(list(IndexedPoly1.objects(partition=self.p1.partition))) == 1
256 -
assert len(list(IndexedPoly2.objects(partition=self.p1.partition))) == 1
257 - 258 - 259 -
#########
260 -
# Repeated tests for 'discriminator' properties, following deprecation of polymorphic variants
261 -
#########
262 25
class TestInheritanceClassConstruction(BaseCassEngTestCase):
263 26 264 27
def test_multiple_discriminator_value_failure(self):
RetroSearch is an open source project built by @garambo | Open a GitHub Issue
Search and Browse the WWW like it's 1997 | Search results from DuckDuckGo
HTML:
3.2
| Encoding:
UTF-8
| Version:
0.7.4