KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use _assert_all method in localstack
Best Python code snippet using localstack_python
replica_side_filtering_test.py
Source:replica_side_filtering_test.py 
...70 @param row Expected result row from query71 """72 decorated_query = self._decorate_query(query)73 assert_one(self.session, decorated_query, row)74 def _assert_all(self, query, rows):75 """76 Assert query returns all expected rows in the correct order.77 @param query Query to run78 @param rows Expected result rows from query79 """80 decorated_query = self._decorate_query(query)81 assert_all(self.session, decorated_query, rows)82 def _decorate_query(self, query):83 return query if self.create_index() else query + " ALLOW FILTERING"84 def _skip_if_index_on_static_is_not_supported(self):85 if self.create_index() and self.cluster.version() < '3.4':86 pytest.skip('Secondary indexes on static column are not supported before 3.4 (CASSANDRA-8103)')87 def _skip_if_filtering_partition_columns_is_not_supported(self):88 if not self.create_index() and self.cluster.version() < '3.11':89 pytest.skip('Filtering of partition key parts is not supported before 3.11 (CASSANDRA-13275)')90 @abstractmethod91 def create_index(self):92 """93 :return:``True`` if the tests should create an index, ``False`` if they should use ``ALLOW FILTERING`` instead94 """95 pass96 def test_update_on_skinny_table(self):97 self._prepare_cluster(98 create_table="CREATE TABLE t (k int PRIMARY KEY, v text)",99 create_index="CREATE INDEX ON t(v)",100 both_nodes=["INSERT INTO t(k, v) VALUES (0, 'old')"],101 only_node1=["UPDATE t SET v = 'new' WHERE k = 0"])102 self._assert_none("SELECT * FROM t WHERE v = 'old'")103 self._assert_one("SELECT * FROM t WHERE v = 'new'", row=[0, 'new'])104 def test_update_on_wide_table(self):105 self._prepare_cluster(106 create_table="CREATE TABLE t (k int, c int, v text, s int STATIC, PRIMARY KEY (k, c))",107 create_index="CREATE INDEX ON t(v)",108 both_nodes=["INSERT INTO t(k, s) VALUES (0, 9)",109 "INSERT INTO t(k, c, v) VALUES (0, -1, 'old')",110 "INSERT INTO t(k, c, v) VALUES (0, 0, 'old')",111 "INSERT INTO t(k, c, v) VALUES (0, 1, 'old')"],112 only_node1=["UPDATE t SET v = 'new' WHERE k = 0 AND c = 0"])113 self._assert_all("SELECT * FROM t WHERE v = 'old'", rows=[[0, -1, 9, 'old'], [0, 1, 9, 'old']])114 self._assert_all("SELECT * FROM t WHERE v = 'new'", rows=[[0, 0, 9, 'new']])115 def test_update_on_static_column_with_empty_partition(self):116 self._skip_if_index_on_static_is_not_supported()117 self._prepare_cluster(118 create_table="CREATE TABLE t (k int, c int, v int, s text STATIC, PRIMARY KEY (k, c))",119 create_index="CREATE INDEX ON t(s)",120 both_nodes=["INSERT INTO t(k, s) VALUES (0, 'old')",121 "INSERT INTO t(k, s) VALUES (1, 'old')"],122 only_node1=["UPDATE t SET s = 'new' WHERE k = 0"])123 self._assert_one("SELECT * FROM t WHERE s = 'old'", row=[1, None, 'old', None])124 self._assert_one("SELECT * FROM t WHERE s = 'new'", row=[0, None, 'new', None])125 def test_update_on_static_column_with_not_empty_partition(self):126 self._skip_if_index_on_static_is_not_supported()127 self._prepare_cluster(128 create_table="CREATE TABLE t (k int, c int, v int, s text STATIC, PRIMARY KEY (k, c))",129 create_index="CREATE INDEX ON t(s)",130 both_nodes=["INSERT INTO t(k, s) VALUES (0, 'old')",131 "INSERT INTO t(k, s) VALUES (1, 'old')",132 "INSERT INTO t(k, c, v) VALUES (0, 10, 100)",133 "INSERT INTO t(k, c, v) VALUES (0, 20, 200)",134 "INSERT INTO t(k, c, v) VALUES (1, 30, 300)",135 "INSERT INTO t(k, c, v) VALUES (1, 40, 400)"],136 only_node1=["UPDATE t SET s = 'new' WHERE k = 0"])137 self._assert_all("SELECT * FROM t WHERE s = 'old'", rows=[[1, 30, 'old', 300], [1, 40, 'old', 400]])138 self._assert_all("SELECT * FROM t WHERE s = 'new'", rows=[[0, 10, 'new', 100], [0, 20, 'new', 200]])139 def test_update_on_collection(self):140 self._prepare_cluster(141 create_table="CREATE TABLE t (k int PRIMARY KEY, v set<int>)",142 create_index="CREATE INDEX ON t(v)",143 both_nodes=["INSERT INTO t(k, v) VALUES (0, {-1, 0, 1})"],144 only_node1=["UPDATE t SET v = v - {0} WHERE k = 0"])145 self._assert_none("SELECT * FROM t WHERE v CONTAINS 0")146 self._assert_one("SELECT * FROM t WHERE v CONTAINS 1", row=[0, [-1, 1]])147 def test_complementary_deletion_with_limit_on_partition_key_column_with_empty_partitions(self):148 self._skip_if_filtering_partition_columns_is_not_supported()149 self._prepare_cluster(150 create_table="CREATE TABLE t (k1 int, k2 int, c int, s int STATIC, PRIMARY KEY ((k1, k2), c))",151 create_index="CREATE INDEX ON t(k1)",152 both_nodes=["INSERT INTO t (k1, k2, s) VALUES (0, 1, 10)",153 "INSERT INTO t (k1, k2, s) VALUES (0, 2, 20)"],154 only_node1=["DELETE FROM t WHERE k1 = 0 AND k2 = 1"],155 only_node2=["DELETE FROM t WHERE k1 = 0 AND k2 = 2"])156 self._assert_none("SELECT * FROM t WHERE k1 = 0 LIMIT 1")157 def test_complementary_deletion_with_limit_on_partition_key_column_with_not_empty_partitions(self):158 self._skip_if_filtering_partition_columns_is_not_supported()159 self._prepare_cluster(160 create_table="CREATE TABLE t (k1 int, k2 int, c int, s int STATIC, PRIMARY KEY ((k1, k2), c))",161 create_index="CREATE INDEX ON t(k1)",162 both_nodes=["INSERT INTO t (k1, k2, c, s) VALUES (0, 1, 10, 100)",163 "INSERT INTO t (k1, k2, c, s) VALUES (0, 2, 20, 200)"],164 only_node1=["DELETE FROM t WHERE k1 = 0 AND k2 = 1"],165 only_node2=["DELETE FROM t WHERE k1 = 0 AND k2 = 2"])166 self._assert_none("SELECT * FROM t WHERE k1 = 0 LIMIT 1")167 def test_complementary_deletion_with_limit_on_clustering_key_column(self):168 self._prepare_cluster(169 create_table="CREATE TABLE t (k int, c int, PRIMARY KEY (k, c))",170 create_index="CREATE INDEX ON t(c)",171 both_nodes=["INSERT INTO t (k, c) VALUES (1, 0)",172 "INSERT INTO t (k, c) VALUES (2, 0)"],173 only_node1=["DELETE FROM t WHERE k = 1"],174 only_node2=["DELETE FROM t WHERE k = 2"])175 self._assert_none("SELECT * FROM t WHERE c = 0 LIMIT 1")176 def test_complementary_deletion_with_limit_on_static_column_with_empty_partitions(self):177 self._skip_if_index_on_static_is_not_supported()178 self._prepare_cluster(179 create_table="CREATE TABLE t (k int, c int, s int STATIC, PRIMARY KEY (k, c))",180 create_index="CREATE INDEX ON t(s)",181 both_nodes=["INSERT INTO t (k, s) VALUES (1, 0)",182 "INSERT INTO t (k, s) VALUES (2, 0)"],183 only_node1=["DELETE FROM t WHERE k = 1"],184 only_node2=["DELETE FROM t WHERE k = 2"])185 self._assert_none("SELECT * FROM t WHERE s = 0 LIMIT 1")186 def test_complementary_deletion_with_limit_on_static_column_with_empty_partitions_and_rows_after(self):187 self._skip_if_index_on_static_is_not_supported()188 self._prepare_cluster(189 create_table="CREATE TABLE t (k int, c int, s int STATIC, PRIMARY KEY (k, c))",190 create_index="CREATE INDEX ON t(s)",191 both_nodes=["INSERT INTO t (k, s) VALUES (1, 0)",192 "INSERT INTO t (k, s) VALUES (2, 0)",193 "INSERT INTO t (k, s) VALUES (3, 0)",194 "INSERT INTO t (k, c) VALUES (3, 1)",195 "INSERT INTO t (k, c) VALUES (3, 2)"],196 only_node1=["DELETE FROM t WHERE k = 1"],197 only_node2=["DELETE FROM t WHERE k = 2"])198 self._assert_one("SELECT * FROM t WHERE s = 0 LIMIT 1", row=[3, 1, 0])199 self._assert_all("SELECT * FROM t WHERE s = 0 LIMIT 10", rows=[[3, 1, 0], [3, 2, 0]])200 self._assert_all("SELECT * FROM t WHERE s = 0", rows=[[3, 1, 0], [3, 2, 0]])201 def test_complementary_deletion_with_limit_on_static_column_with_not_empty_partitions(self):202 self._skip_if_index_on_static_is_not_supported()203 self._prepare_cluster(204 create_table="CREATE TABLE t (k int, c int, s int STATIC, v int, PRIMARY KEY (k, c))",205 create_index="CREATE INDEX ON t(s)",206 both_nodes=["INSERT INTO t (k, c, v, s) VALUES (1, 10, 100, 0)",207 "INSERT INTO t (k, c, v, s) VALUES (2, 20, 200, 0)"],208 only_node1=["DELETE FROM t WHERE k = 1"],209 only_node2=["DELETE FROM t WHERE k = 2"])210 self._assert_none("SELECT * FROM t WHERE s = 0 LIMIT 1")211 def test_complementary_deletion_with_limit_on_static_column_with_not_empty_partitions_and_rows_after(self):212 self._skip_if_index_on_static_is_not_supported()213 self._prepare_cluster(214 create_table="CREATE TABLE t (k int, c int, s int STATIC, v int, PRIMARY KEY(k, c))",215 create_index="CREATE INDEX ON t(s)",216 both_nodes=["INSERT INTO t (k, c, v, s) VALUES (1, 10, 100, 0)",217 "INSERT INTO t (k, c, v, s) VALUES (2, 20, 200, 0)",218 "INSERT INTO t (k, s) VALUES (3, 0)",219 "INSERT INTO t (k, c) VALUES (3, 1)",220 "INSERT INTO t (k, c) VALUES (3, 2)"],221 only_node1=["DELETE FROM t WHERE k = 1"],222 only_node2=["DELETE FROM t WHERE k = 2"])223 self._assert_one("SELECT * FROM t WHERE s = 0 LIMIT 1", row=[3, 1, 0, None])224 self._assert_all("SELECT * FROM t WHERE s = 0 LIMIT 10", rows=[[3, 1, 0, None], [3, 2, 0, None]])225 self._assert_all("SELECT * FROM t WHERE s = 0", rows=[[3, 1, 0, None], [3, 2, 0, None]])226 def test_complementary_deletion_with_limit_on_regular_column(self):227 self._prepare_cluster(228 create_table="CREATE TABLE t (k int, c int, v int, PRIMARY KEY (k, c))",229 create_index="CREATE INDEX ON t(v)",230 both_nodes=["INSERT INTO t (k, c, v) VALUES (0, 1, 0)",231 "INSERT INTO t (k, c, v) VALUES (0, 2, 0)"],232 only_node1=["DELETE FROM t WHERE k = 0 AND c = 1"],233 only_node2=["DELETE FROM t WHERE k = 0 AND c = 2"])234 self._assert_none("SELECT * FROM t WHERE v = 0 LIMIT 1")235 def test_complementary_deletion_with_limit_and_rows_after(self):236 self._prepare_cluster(237 create_table="CREATE TABLE t (k int, c int, v int, PRIMARY KEY (k, c))",238 create_index="CREATE INDEX ON t(v)",239 both_nodes=["INSERT INTO t (k, c, v) VALUES (0, 1, 0)",240 "INSERT INTO t (k, c, v) VALUES (0, 2, 0)",241 "INSERT INTO t (k, c, v) VALUES (0, 3, 0)"],242 only_node1=["DELETE FROM t WHERE k = 0 AND c = 1",243 "INSERT INTO t (k, c, v) VALUES (0, 4, 0)"],244 only_node2=["INSERT INTO t (k, c, v) VALUES (0, 5, 0)",245 "DELETE FROM t WHERE k = 0 AND c = 2"])246 self._assert_one("SELECT * FROM t WHERE v = 0 LIMIT 1", row=[0, 3, 0])247 self._assert_all("SELECT * FROM t WHERE v = 0 LIMIT 2", rows=[[0, 3, 0], [0, 4, 0]])248 self._assert_all("SELECT * FROM t WHERE v = 0 LIMIT 3", rows=[[0, 3, 0], [0, 4, 0], [0, 5, 0]])249 self._assert_all("SELECT * FROM t WHERE v = 0 LIMIT 4", rows=[[0, 3, 0], [0, 4, 0], [0, 5, 0]])250 def test_complementary_deletion_with_limit_and_rows_between(self):251 self._prepare_cluster(252 create_table="CREATE TABLE t (k int, c int, v int, PRIMARY KEY (k, c))",253 create_index="CREATE INDEX ON t(v)",254 both_nodes=["INSERT INTO t (k, c, v) VALUES (0, 1, 0)",255 "INSERT INTO t (k, c, v) VALUES (0, 4, 0)"],256 only_node1=["DELETE FROM t WHERE k = 0 AND c = 1"],257 only_node2=["INSERT INTO t (k, c, v) VALUES (0, 2, 0)",258 "INSERT INTO t (k, c, v) VALUES (0, 3, 0)",259 "DELETE FROM t WHERE k = 0 AND c = 4"])260 self._assert_one("SELECT * FROM t WHERE v = 0 LIMIT 1", row=[0, 2, 0])261 self._assert_all("SELECT * FROM t WHERE v = 0 LIMIT 2", rows=[[0, 2, 0], [0, 3, 0]])262 self._assert_all("SELECT * FROM t WHERE v = 0 LIMIT 3", rows=[[0, 2, 0], [0, 3, 0]])263 def test_complementary_update_with_limit_on_static_column_with_empty_partitions(self):264 self._skip_if_index_on_static_is_not_supported()265 self._prepare_cluster(266 create_table="CREATE TABLE t (k int, c int, s text STATIC, v int, PRIMARY KEY (k, c))",267 create_index="CREATE INDEX ON t(s)",268 both_nodes=["INSERT INTO t (k, s) VALUES (1, 'old')",269 "INSERT INTO t (k, s) VALUES (2, 'old')"],270 only_node1=["UPDATE t SET s = 'new' WHERE k = 1"],271 only_node2=["UPDATE t SET s = 'new' WHERE k = 2"])272 self._assert_none("SELECT * FROM t WHERE s = 'old' LIMIT 1")273 self._assert_one("SELECT k, c, v, s FROM t WHERE s = 'new' LIMIT 1", row=[1, None, None, 'new'])274 self._assert_all("SELECT k, c, v, s FROM t WHERE s = 'new'",275 rows=[[1, None, None, 'new'], [2, None, None, 'new']])276 def test_complementary_update_with_limit_on_static_column_with_not_empty_partitions(self):277 self._skip_if_index_on_static_is_not_supported()278 self._prepare_cluster(279 create_table="CREATE TABLE t (k int, c int, s text STATIC, v int, PRIMARY KEY (k, c))",280 create_index="CREATE INDEX ON t(s)",281 both_nodes=["INSERT INTO t (k, c, v, s) VALUES (1, 10, 100, 'old')",282 "INSERT INTO t (k, c, v, s) VALUES (2, 20, 200, 'old')"],283 only_node1=["UPDATE t SET s = 'new' WHERE k = 1"],284 only_node2=["UPDATE t SET s = 'new' WHERE k = 2"])285 self._assert_none("SELECT * FROM t WHERE s = 'old' LIMIT 1")286 self._assert_one("SELECT k, c, v, s FROM t WHERE s = 'new' LIMIT 1", row=[1, 10, 100, 'new'])287 self._assert_all("SELECT k, c, v, s FROM t WHERE s = 'new'", rows=[[1, 10, 100, 'new'], [2, 20, 200, 'new']])288 def test_complementary_update_with_limit_on_regular_column(self):289 self._prepare_cluster(290 create_table="CREATE TABLE t (k int, c int, v text, PRIMARY KEY (k, c))",291 create_index="CREATE INDEX ON t(v)",292 both_nodes=["INSERT INTO t (k, c, v) VALUES (0, 1, 'old')",293 "INSERT INTO t (k, c, v) VALUES (0, 2, 'old')"],294 only_node1=["UPDATE t SET v = 'new' WHERE k = 0 AND c = 1"],295 only_node2=["UPDATE t SET v = 'new' WHERE k = 0 AND c = 2"])296 self._assert_none("SELECT * FROM t WHERE v = 'old' LIMIT 1")297 self._assert_one("SELECT * FROM t WHERE v = 'new' LIMIT 1", row=[0, 1, 'new'])298 self._assert_all("SELECT * FROM t WHERE v = 'new'", rows=[[0, 1, 'new'], [0, 2, 'new']])299 def test_complementary_update_with_limit_and_rows_between(self):300 self._prepare_cluster(301 create_table="CREATE TABLE t (k int, c int, v text, PRIMARY KEY (k, c))",302 create_index="CREATE INDEX ON t(v)",303 both_nodes=["INSERT INTO t (k, c, v) VALUES (0, 1, 'old')",304 "INSERT INTO t (k, c, v) VALUES (0, 4, 'old')"],305 only_node1=["UPDATE t SET v = 'new' WHERE k = 0 AND c = 1"],306 only_node2=["INSERT INTO t (k, c, v) VALUES (0, 2, 'old')",307 "INSERT INTO t (k, c, v) VALUES (0, 3, 'old')",308 "UPDATE t SET v = 'new' WHERE k = 0 AND c = 4"])309 self._assert_one("SELECT * FROM t WHERE v = 'old' LIMIT 1", row=[0, 2, 'old'])310 self._assert_all("SELECT * FROM t WHERE v = 'old' LIMIT 2", rows=[[0, 2, 'old'], [0, 3, 'old']])311 self._assert_all("SELECT * FROM t WHERE v = 'old' LIMIT 3", rows=[[0, 2, 'old'], [0, 3, 'old']])312 self._assert_one("SELECT * FROM t WHERE v = 'new' LIMIT 1", row=[0, 1, 'new'])313 self._assert_all("SELECT * FROM t WHERE v = 'new'", rows=[[0, 1, 'new'], [0, 4, 'new']])314 def test_partition_deletion_on_skinny_table(self):315 self._prepare_cluster(316 create_table="CREATE TABLE t (k int PRIMARY KEY, v text)",317 create_index="CREATE INDEX ON t(v)",318 only_node1=["INSERT INTO t (k, v) VALUES (0, 'old') USING TIMESTAMP 1"],319 only_node2=["DELETE FROM t WHERE k = 0"])320 self._assert_none("SELECT * FROM t WHERE v = 'old' LIMIT 1")321 self._assert_none("SELECT * FROM t WHERE v = 'old'")322 def test_partition_deletion_on_wide_table(self):323 self._prepare_cluster(324 create_table="CREATE TABLE t (k int, c int, v text, PRIMARY KEY (k, c))",325 create_index="CREATE INDEX ON t(v)",326 only_node1=["INSERT INTO t (k, c, v) VALUES (0, 1, 'old') USING TIMESTAMP 1"],327 only_node2=["DELETE FROM t WHERE k = 0"])328 self._assert_none("SELECT * FROM t WHERE v = 'old' LIMIT 1")329 self._assert_none("SELECT * FROM t WHERE v = 'old'")330 def test_row_deletion_on_wide_table(self):331 self._prepare_cluster(332 create_table="CREATE TABLE t (k int, c int, v text, PRIMARY KEY (k, c))",333 create_index="CREATE INDEX ON t(v)",334 only_node1=["INSERT INTO t (k, c, v) VALUES (0, 1, 'old') USING TIMESTAMP 1"],335 only_node2=["DELETE FROM t WHERE k = 0 AND c = 1"])336 self._assert_none("SELECT * FROM t WHERE v = 'old' LIMIT 1")337 self._assert_none("SELECT * FROM t WHERE v = 'old'")338 def test_range_deletion_on_wide_table(self):339 self._prepare_cluster(340 create_table="CREATE TABLE t (k int, c int, v text, PRIMARY KEY (k, c))",341 create_index="CREATE INDEX ON t(v)",342 only_node1=["INSERT INTO t (k, c, v) VALUES (0, 1, 'old') USING TIMESTAMP 1",343 "INSERT INTO t (k, c, v) VALUES (0, 2, 'old') USING TIMESTAMP 1",344 "INSERT INTO t (k, c, v) VALUES (0, 3, 'old') USING TIMESTAMP 1",345 "INSERT INTO t (k, c, v) VALUES (0, 4, 'old') USING TIMESTAMP 1"],346 only_node2=["DELETE FROM t WHERE k = 0 AND c > 1 AND c < 4"])347 self._assert_one("SELECT * FROM t WHERE v = 'old' LIMIT 1", row=[0, 1, 'old'])348 self._assert_all("SELECT * FROM t WHERE v = 'old'", rows=[[0, 1, 'old'], [0, 4, 'old']])349 def test_mismatching_insertions_on_skinny_table(self):350 self._prepare_cluster(351 create_table="CREATE TABLE t (k int PRIMARY KEY, v text)",352 create_index="CREATE INDEX ON t(v)",353 only_node1=["INSERT INTO t (k, v) VALUES (0, 'old') USING TIMESTAMP 1"],354 only_node2=["INSERT INTO t (k, v) VALUES (0, 'new') USING TIMESTAMP 2"])355 self._assert_none("SELECT * FROM t WHERE v = 'old' LIMIT 1")356 self._assert_none("SELECT * FROM t WHERE v = 'old'")357 self._assert_one("SELECT * FROM t WHERE v = 'new'", row=[0, 'new'])358 def test_mismatching_insertions_on_wide_table(self):359 self._prepare_cluster(360 create_table="CREATE TABLE t (k int, c int, v text, PRIMARY KEY (k, c))",361 create_index="CREATE INDEX ON t(v)",362 only_node1=["INSERT INTO t (k, c, v) VALUES (0, 1, 'old') USING TIMESTAMP 1"],363 only_node2=["INSERT INTO t (k, c, v) VALUES (0, 1, 'new') USING TIMESTAMP 2"])364 self._assert_none("SELECT * FROM t WHERE v = 'old' LIMIT 1")365 self._assert_none("SELECT * FROM t WHERE v = 'old'")366 self._assert_one("SELECT * FROM t WHERE v = 'new'", row=[0, 1, 'new'])367 def test_consistent_skinny_table(self):368 self._prepare_cluster(369 create_table="CREATE TABLE t (k int PRIMARY KEY, v text)",370 create_index="CREATE INDEX ON t(v)",371 both_nodes=["INSERT INTO t(k, v) VALUES (1, 'old')", # updated to 'new'372 "INSERT INTO t(k, v) VALUES (2, 'old')",373 "INSERT INTO t(k, v) VALUES (3, 'old')", # updated to 'new'374 "INSERT INTO t(k, v) VALUES (4, 'old')",375 "INSERT INTO t(k, v) VALUES (5, 'old')", # deleted partition376 "UPDATE t SET v = 'new' WHERE k = 1",377 "UPDATE t SET v = 'new' WHERE k = 3",378 "DELETE FROM t WHERE k = 5"])379 self._assert_one("SELECT * FROM t WHERE v = 'old' LIMIT 1", row=[2, 'old'])380 self._assert_one("SELECT * FROM t WHERE v = 'new' LIMIT 1", row=[1, 'new'])381 self._assert_all("SELECT * FROM t WHERE v = 'old'", rows=[[2, 'old'], [4, 'old']])382 self._assert_all("SELECT * FROM t WHERE v = 'new'", rows=[[1, 'new'], [3, 'new']])383 def test_consistent_wide_table(self):384 self._prepare_cluster(385 create_table="CREATE TABLE t (k int, c int, v text, PRIMARY KEY (k, c))",386 create_index="CREATE INDEX ON t(v)",387 both_nodes=["INSERT INTO t(k, c, v) VALUES (0, 1, 'old')", # updated to 'new'388 "INSERT INTO t(k, c, v) VALUES (0, 2, 'old')",389 "INSERT INTO t(k, c, v) VALUES (0, 3, 'old')", # updated to 'new'390 "INSERT INTO t(k, c, v) VALUES (0, 4, 'old')",391 "INSERT INTO t(k, c, v) VALUES (0, 5, 'old')", # deleted row392 "INSERT INTO t(k, c, v) VALUES (1, 1, 'old')", # deleted partition393 "INSERT INTO t(k, c, v) VALUES (1, 2, 'old')", # deleted partition394 "UPDATE t SET v = 'new' WHERE k = 0 AND c = 1",395 "UPDATE t SET v = 'new' WHERE k = 0 AND c = 3",396 "DELETE FROM t WHERE k = 0 AND c = 5",397 "DELETE FROM t WHERE k = 1"])398 self._assert_one("SELECT * FROM t WHERE v = 'old' LIMIT 1", row=[0, 2, 'old'])399 self._assert_one("SELECT * FROM t WHERE v = 'new' LIMIT 1", row=[0, 1, 'new'])400 self._assert_all("SELECT * FROM t WHERE v = 'old'", rows=[[0, 2, 'old'], [0, 4, 'old']])401 self._assert_all("SELECT * FROM t WHERE v = 'new'", rows=[[0, 1, 'new'], [0, 3, 'new']])402 def test_count(self):403 self._prepare_cluster(404 create_table="CREATE TABLE t (k int PRIMARY KEY, v text)",405 create_index="CREATE INDEX ON t(v)",406 both_nodes=["INSERT INTO t(k, v) VALUES (1, 'old')",407 "INSERT INTO t(k, v) VALUES (2, 'old')",408 "INSERT INTO t(k, v) VALUES (3, 'old')",409 "INSERT INTO t(k, v) VALUES (4, 'old')",410 "INSERT INTO t(k, v) VALUES (5, 'old')"],411 only_node1=["UPDATE t SET v = 'new' WHERE k = 2",412 "UPDATE t SET v = 'new' WHERE k = 4"])413 self._assert_one("SELECT COUNT(*) FROM t WHERE v = 'old' LIMIT 1", row=[3])414 self._assert_one("SELECT COUNT(*) FROM t WHERE v = 'old'", row=[3])415 self._assert_one("SELECT COUNT(*) FROM t WHERE v = 'new'", row=[2])...
test_stereo.py
Source:test_stereo.py
...152 left_rectified_camera, right_rectified_camera = _SmokeTestData._create_stereo_camera(153 batch_size, device, dtype, tx_fx154 )155 stereo_camera = StereoCamera(left_rectified_camera, right_rectified_camera)156 def _assert_all(x, y):157 assert torch.all(torch.eq(x, y))158 _assert_all(stereo_camera.fx, _TestParams.fx)159 _assert_all(stereo_camera.fy, _TestParams.fy)160 _assert_all(stereo_camera.cx_left, _TestParams.cx)161 _assert_all(stereo_camera.cy, _TestParams.cy)162 _assert_all(stereo_camera.tx, -tx_fx / _TestParams.fx)163 assert stereo_camera.Q.shape == (batch_size, 4, 4)164 assert stereo_camera.Q.dtype in (torch.float16, torch.float32, torch.float64)165 @staticmethod166 def test_stereo_camera_attributes_real(batch_size, device, dtype):167 """Test proper setup of the class for real data."""168 left_rectified_camera, right_rectified_camera = _RealTestData._get_real_stereo_camera(batch_size, device, dtype)169 stereo_camera = StereoCamera(left_rectified_camera, right_rectified_camera)170 assert_allclose(stereo_camera.fx, left_rectified_camera[..., 0, 0])171 assert_allclose(stereo_camera.fy, left_rectified_camera[..., 1, 1])172 assert_allclose(stereo_camera.cx_left, left_rectified_camera[..., 0, 2])173 assert_allclose(stereo_camera.cy, left_rectified_camera[..., 1, 2])174 assert_allclose(stereo_camera.tx, -right_rectified_camera[..., 0, 3] / right_rectified_camera[..., 0, 0])175 assert stereo_camera.Q.shape == (batch_size, 4, 4)176 assert stereo_camera.Q.dtype in (torch.float16, torch.float32, torch.float64)...
Automation Testing Tutorials
Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run localstack automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
