KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use and_return method in autotest
Best Python code snippet using autotest_python
test_belongs_to_many.py
Source:test_belongs_to_many.py 
...28 )29 )30 )31 relation = self._get_relation()32 relation.get_parent().should_receive("get_connection_name").and_return(33 "foo.connection"34 )35 relation.get_query().get_query().should_receive("add_select").once().with_args(36 *[37 "roles.*",38 "user_role.user_id AS pivot_user_id",39 "user_role.role_id AS pivot_role_id",40 ]41 ).and_return(relation.get_query())42 relation.get_query().should_receive("get_models").once().and_return(models)43 relation.get_query().should_receive("eager_load_relations").once().with_args(44 models45 ).and_return(models)46 relation.get_related().should_receive("new_collection").replace_with(47 lambda l: Collection(l)48 )49 relation.get_query().should_receive("get_query").once().and_return(base_builder)50 results = relation.get()51 self.assertIsInstance(results, Collection)52 # Make sure the foreign keys were set on the pivot models53 self.assertEqual("user_id", results[0].pivot.get_foreign_key())54 self.assertEqual("role_id", results[0].pivot.get_other_key())55 self.assertEqual("john", results[0].name)56 self.assertEqual(1, results[0].pivot.user_id)57 self.assertEqual(2, results[0].pivot.role_id)58 self.assertEqual("foo.connection", results[0].pivot.get_connection_name())59 self.assertEqual("jane", results[1].name)60 self.assertEqual(3, results[1].pivot.user_id)61 self.assertEqual(4, results[1].pivot.role_id)62 self.assertEqual("foo.connection", results[1].pivot.get_connection_name())63 self.assertEqual("user_role", results[0].pivot.get_table())64 self.assertTrue(results[0].pivot.exists)65 def test_timestamps_can_be_retrieved_properly(self):66 model1 = OrmBelongsToManyModelStub()67 model1.fill(name="john", pivot_user_id=1, pivot_role_id=2)68 model2 = OrmBelongsToManyModelStub()69 model2.fill(name="jane", pivot_user_id=3, pivot_role_id=4)70 models = [model1, model2]71 base_builder = flexmock(72 Builder(73 QueryBuilder(74 MockConnection().prepare_mock(), QueryGrammar(), QueryProcessor()75 )76 )77 )78 relation = self._get_relation().with_timestamps()79 relation.get_parent().should_receive("get_connection_name").and_return(80 "foo.connection"81 )82 relation.get_query().get_query().should_receive("add_select").once().with_args(83 "roles.*",84 "user_role.user_id AS pivot_user_id",85 "user_role.role_id AS pivot_role_id",86 "user_role.created_at AS pivot_created_at",87 "user_role.updated_at AS pivot_updated_at",88 ).and_return(relation.get_query())89 relation.get_query().should_receive("get_models").once().and_return(models)90 relation.get_query().should_receive("eager_load_relations").once().with_args(91 models92 ).and_return(models)93 relation.get_related().should_receive("new_collection").replace_with(94 lambda l: Collection(l)95 )96 relation.get_query().should_receive("get_query").once().and_return(base_builder)97 results = relation.get()98 def test_models_are_properly_matched_to_parents(self):99 relation = self._get_relation()100 result1 = OrmBelongsToManyModelPivotStub()101 result1.pivot.user_id = 1102 result2 = OrmBelongsToManyModelPivotStub()103 result2.pivot.user_id = 2104 result3 = OrmBelongsToManyModelPivotStub()105 result3.pivot.user_id = 2106 model1 = OrmBelongsToManyModelStub()107 model1.id = 1108 model2 = OrmBelongsToManyModelStub()109 model2.id = 2110 model3 = OrmBelongsToManyModelStub()111 model3.id = 3112 relation.get_related().should_receive("new_collection").replace_with(113 lambda l=None: Collection(l)114 )115 models = relation.match(116 [model1, model2, model3], Collection([result1, result2, result3]), "foo"117 )118 self.assertEqual(1, models[0].foo[0].pivot.user_id)119 self.assertEqual(1, len(models[0].foo))120 self.assertEqual(2, models[1].foo[0].pivot.user_id)121 self.assertEqual(2, models[1].foo[1].pivot.user_id)122 self.assertEqual(2, len(models[1].foo))123 self.assertTrue(models[2].foo.is_empty())124 def test_relation_is_properly_initialized(self):125 relation = self._get_relation()126 relation.get_related().should_receive("new_collection").replace_with(127 lambda l=None: Collection(l or [])128 )129 model = flexmock(Model())130 model.should_receive("set_relation").once().with_args("foo", Collection)131 models = relation.init_relation([model], "foo")132 self.assertEqual([model], models)133 def test_eager_constraints_are_properly_added(self):134 relation = self._get_relation()135 relation.get_query().get_query().should_receive("where_in").once().with_args(136 "user_role.user_id", [1, 2]137 )138 model1 = OrmBelongsToManyModelStub()139 model1.id = 1140 model2 = OrmBelongsToManyModelStub()141 model2.id = 2142 relation.add_eager_constraints([model1, model2])143 def test_attach_inserts_pivot_table_record(self):144 flexmock(BelongsToMany, touch_if_touching=lambda: True)145 relation = self._get_relation()146 query = flexmock()147 query.should_receive("from_").once().with_args("user_role").and_return(query)148 query.should_receive("insert").once().with_args(149 [{"user_id": 1, "role_id": 2, "foo": "bar"}]150 ).and_return(True)151 mock_query_builder = flexmock()152 relation.get_query().should_receive("get_query").and_return(mock_query_builder)153 mock_query_builder.should_receive("new_query").once().and_return(query)154 relation.should_receive("touch_if_touching").once()155 relation.attach(2, {"foo": "bar"})156 def test_attach_multiple_inserts_pivot_table_record(self):157 flexmock(BelongsToMany, touch_if_touching=lambda: True)158 relation = self._get_relation()159 query = flexmock()160 query.should_receive("from_").once().with_args("user_role").and_return(query)161 query.should_receive("insert").once().with_args(162 [163 {"user_id": 1, "role_id": 2, "foo": "bar"},164 {"user_id": 1, "role_id": 3, "bar": "baz", "foo": "bar"},165 ]166 ).and_return(True)167 mock_query_builder = flexmock()168 relation.get_query().should_receive("get_query").and_return(mock_query_builder)169 mock_query_builder.should_receive("new_query").once().and_return(query)170 relation.should_receive("touch_if_touching").once()171 relation.attach([2, {3: {"bar": "baz"}}], {"foo": "bar"})172 def test_attach_inserts_pivot_table_records_with_timestamps_when_ncessary(self):173 flexmock(BelongsToMany, touch_if_touching=lambda: True)174 relation = self._get_relation().with_timestamps()175 query = flexmock()176 query.should_receive("from_").once().with_args("user_role").and_return(query)177 now = pendulum.now()178 query.should_receive("insert").once().with_args(179 [180 {181 "user_id": 1,182 "role_id": 2,183 "foo": "bar",184 "created_at": now,185 "updated_at": now,186 }187 ]188 ).and_return(True)189 mock_query_builder = flexmock()190 relation.get_query().should_receive("get_query").and_return(mock_query_builder)191 mock_query_builder.should_receive("new_query").once().and_return(query)192 relation.get_parent().should_receive("fresh_timestamp").once().and_return(now)193 relation.should_receive("touch_if_touching").once()194 relation.attach(2, {"foo": "bar"})195 def test_attach_inserts_pivot_table_records_with_a_created_at_timestamp(self):196 flexmock(BelongsToMany, touch_if_touching=lambda: True)197 relation = self._get_relation().with_pivot("created_at")198 query = flexmock()199 query.should_receive("from_").once().with_args("user_role").and_return(query)200 now = pendulum.now()201 query.should_receive("insert").once().with_args(202 [{"user_id": 1, "role_id": 2, "foo": "bar", "created_at": now}]203 ).and_return(True)204 mock_query_builder = flexmock()205 relation.get_query().should_receive("get_query").and_return(mock_query_builder)206 mock_query_builder.should_receive("new_query").once().and_return(query)207 relation.get_parent().should_receive("fresh_timestamp").once().and_return(now)208 relation.should_receive("touch_if_touching").once()209 relation.attach(2, {"foo": "bar"})210 def test_attach_inserts_pivot_table_records_with_an_updated_at_timestamp(self):211 flexmock(BelongsToMany, touch_if_touching=lambda: True)212 relation = self._get_relation().with_pivot("updated_at")213 query = flexmock()214 query.should_receive("from_").once().with_args("user_role").and_return(query)215 now = pendulum.now()216 query.should_receive("insert").once().with_args(217 [{"user_id": 1, "role_id": 2, "foo": "bar", "updated_at": now}]218 ).and_return(True)219 mock_query_builder = flexmock()220 relation.get_query().should_receive("get_query").and_return(mock_query_builder)221 mock_query_builder.should_receive("new_query").once().and_return(query)222 relation.get_parent().should_receive("fresh_timestamp").once().and_return(now)223 relation.should_receive("touch_if_touching").once()224 relation.attach(2, {"foo": "bar"})225 def test_detach_remove_pivot_table_record(self):226 flexmock(BelongsToMany, touch_if_touching=lambda: True)227 relation = self._get_relation()228 query = flexmock()229 query.should_receive("from_").once().with_args("user_role").and_return(query)230 query.should_receive("where").once().with_args("user_id", 1).and_return(query)231 query.should_receive("where_in").once().with_args("role_id", [1, 2, 3])232 query.should_receive("delete").once().and_return(True)233 mock_query_builder = flexmock()234 relation.get_query().should_receive("get_query").and_return(mock_query_builder)235 mock_query_builder.should_receive("new_query").once().and_return(query)236 relation.should_receive("touch_if_touching").once()237 self.assertTrue(relation.detach([1, 2, 3]))238 def test_detach_with_single_id_remove_pivot_table_record(self):239 flexmock(BelongsToMany, touch_if_touching=lambda: True)240 relation = self._get_relation()241 query = flexmock()242 query.should_receive("from_").once().with_args("user_role").and_return(query)243 query.should_receive("where").once().with_args("user_id", 1).and_return(query)244 query.should_receive("where_in").once().with_args("role_id", [1])245 query.should_receive("delete").once().and_return(True)246 mock_query_builder = flexmock()247 relation.get_query().should_receive("get_query").and_return(mock_query_builder)248 mock_query_builder.should_receive("new_query").once().and_return(query)249 relation.should_receive("touch_if_touching").once()250 self.assertTrue(relation.detach(1))251 def test_detach_clears_all_records_when_no_ids(self):252 flexmock(BelongsToMany, touch_if_touching=lambda: True)253 relation = self._get_relation()254 query = flexmock()255 query.should_receive("from_").once().with_args("user_role").and_return(query)256 query.should_receive("where").once().with_args("user_id", 1).and_return(query)257 query.should_receive("where_in").never()258 query.should_receive("delete").once().and_return(True)259 mock_query_builder = flexmock()260 relation.get_query().should_receive("get_query").and_return(mock_query_builder)261 mock_query_builder.should_receive("new_query").once().and_return(query)262 relation.should_receive("touch_if_touching").once()263 self.assertTrue(relation.detach())264 def test_create_creates_new_model_and_insert_attachment_record(self):265 flexmock(BelongsToMany, attach=lambda: True)266 relation = self._get_relation()267 model = flexmock()268 relation.get_related().should_receive("new_instance").once().and_return(269 model270 ).with_args({"foo": "bar"})271 model.should_receive("save").once()272 model.should_receive("get_key").and_return("foo")273 relation.should_receive("attach").once().with_args("foo", {"bar": "baz"}, True)274 self.assertEqual(model, relation.create({"foo": "bar"}, {"bar": "baz"}))275 def test_find_or_new_finds_model(self):276 flexmock(BelongsToMany)277 relation = self._get_relation()278 model = flexmock()279 model.foo = "bar"280 relation.get_query().should_receive("find").with_args("foo", None).and_return(281 model282 )283 relation.get_related().should_receive("new_instance").never()284 self.assertEqual("bar", relation.find_or_new("foo").foo)285 def test_find_or_new_returns_new_model(self):286 flexmock(BelongsToMany)287 relation = self._get_relation()288 model = flexmock()289 model.foo = "bar"290 relation.get_query().should_receive("find").with_args("foo", None).and_return(291 None292 )293 relation.get_related().should_receive("new_instance").once().and_return(model)294 self.assertEqual("bar", relation.find_or_new("foo").foo)295 def test_first_or_new_finds_first_model(self):296 flexmock(BelongsToMany)297 relation = self._get_relation()298 model = flexmock()299 model.foo = "bar"300 relation.get_query().should_receive("where").with_args(301 {"foo": "bar"}302 ).and_return(relation.get_query())303 relation.get_query().should_receive("first").once().and_return(model)304 relation.get_related().should_receive("new_instance").never()305 self.assertEqual("bar", relation.first_or_new({"foo": "bar"}).foo)306 def test_first_or_new_returns_new_model(self):307 flexmock(BelongsToMany)308 relation = self._get_relation()309 model = flexmock()310 model.foo = "bar"311 relation.get_query().should_receive("where").with_args(312 {"foo": "bar"}313 ).and_return(relation.get_query())314 relation.get_query().should_receive("first").once().and_return(None)315 relation.get_related().should_receive("new_instance").once().and_return(model)316 self.assertEqual("bar", relation.first_or_new({"foo": "bar"}).foo)317 def test_first_or_create_finds_first_model(self):318 flexmock(BelongsToMany)319 relation = self._get_relation()320 model = flexmock()321 model.foo = "bar"322 relation.get_query().should_receive("where").with_args(323 {"foo": "bar"}324 ).and_return(relation.get_query())325 relation.get_query().should_receive("first").once().and_return(model)326 relation.should_receive("create").never()327 self.assertEqual("bar", relation.first_or_create({"foo": "bar"}).foo)328 def test_first_or_create_returns_new_model(self):329 flexmock(BelongsToMany)330 relation = self._get_relation()331 model = flexmock()332 model.foo = "bar"333 relation.get_query().should_receive("where").with_args(334 {"foo": "bar"}335 ).and_return(relation.get_query())336 relation.get_query().should_receive("first").once().and_return(None)337 relation.should_receive("create").once().with_args(338 {"foo": "bar"}, {}, True339 ).and_return(model)340 self.assertEqual("bar", relation.first_or_create({"foo": "bar"}).foo)341 def test_update_or_create_finds_first_mode_and_updates(self):342 flexmock(BelongsToMany)343 relation = self._get_relation()344 model = flexmock()345 model.foo = "bar"346 relation.get_query().should_receive("where").with_args(347 {"foo": "bar"}348 ).and_return(relation.get_query())349 relation.get_query().should_receive("first").once().and_return(model)350 model.should_receive("fill").once()351 model.should_receive("save").once()352 relation.should_receive("create").never()353 self.assertEqual("bar", relation.update_or_create({"foo": "bar"}).foo)354 def test_update_or_create_returns_new_model(self):355 flexmock(BelongsToMany)356 relation = self._get_relation()357 model = flexmock()358 model.foo = "bar"359 relation.get_query().should_receive("where").with_args(360 {"foo": "bar"}361 ).and_return(relation.get_query())362 relation.get_query().should_receive("first").once().and_return(None)363 relation.should_receive("create").once().with_args(364 {"bar": "baz"}, None, True365 ).and_return(model)366 self.assertEqual(367 "bar", relation.update_or_create({"foo": "bar"}, {"bar": "baz"}).foo368 )369 def test_sync_syncs_intermediate_table_with_given_list(self):370 for list_ in [[2, 3, 4], ["2", "3", "4"]]:371 flexmock(BelongsToMany)372 relation = self._get_relation()373 query = flexmock()374 query.should_receive("from_").once().with_args("user_role").and_return(375 query376 )377 query.should_receive("where").once().with_args("user_id", 1).and_return(378 query379 )380 mock_query_builder = flexmock()381 relation.get_query().should_receive("get_query").and_return(382 mock_query_builder383 )384 mock_query_builder.should_receive("new_query").once().and_return(query)385 query.should_receive("lists").once().with_args("role_id").and_return(386 Collection([1, list_[0], list_[1]])387 )388 relation.should_receive("attach").once().with_args(list_[2], {}, False)389 relation.should_receive("detach").once().with_args([1])390 relation.get_related().should_receive("touches").and_return(False)391 relation.get_parent().should_receive("touches").and_return(False)392 self.assertEqual(393 {"attached": [list_[2]], "detached": [1], "updated": []},394 relation.sync(list_),395 )396 def test_sync_syncs_intermediate_table_with_given_list_and_attributes(self):397 flexmock(BelongsToMany)398 relation = self._get_relation()399 query = flexmock()400 query.should_receive("from_").once().with_args("user_role").and_return(query)401 query.should_receive("where").once().with_args("user_id", 1).and_return(query)402 mock_query_builder = flexmock()403 relation.get_query().should_receive("get_query").and_return(mock_query_builder)404 mock_query_builder.should_receive("new_query").once().and_return(query)405 query.should_receive("lists").once().with_args("role_id").and_return(406 Collection([1, 2, 3])407 )408 relation.should_receive("attach").once().with_args(4, {"foo": "bar"}, False)409 relation.should_receive("update_existing_pivot").once().with_args(410 3, {"bar": "baz"}, False411 ).and_return(True)412 relation.should_receive("detach").once().with_args([1])413 relation.should_receive("touch_if_touching").once()414 relation.get_related().should_receive("touches").and_return(False)415 relation.get_parent().should_receive("touches").and_return(False)416 self.assertEqual(417 {"attached": [4], "detached": [1], "updated": [3]},418 relation.sync([2, {3: {"bar": "baz"}}, {4: {"foo": "bar"}}]),419 )420 def test_sync_does_not_return_values_that_were_not_updated(self):421 flexmock(BelongsToMany)422 relation = self._get_relation()423 query = flexmock()424 query.should_receive("from_").once().with_args("user_role").and_return(query)425 query.should_receive("where").once().with_args("user_id", 1).and_return(query)426 mock_query_builder = flexmock()427 relation.get_query().should_receive("get_query").and_return(mock_query_builder)428 mock_query_builder.should_receive("new_query").once().and_return(query)429 query.should_receive("lists").once().with_args("role_id").and_return(430 Collection([1, 2, 3])431 )432 relation.should_receive("attach").once().with_args(4, {"foo": "bar"}, False)433 relation.should_receive("update_existing_pivot").once().with_args(434 3, {"bar": "baz"}, False435 ).and_return(False)436 relation.should_receive("detach").once().with_args([1])437 relation.should_receive("touch_if_touching").once()438 relation.get_related().should_receive("touches").and_return(False)439 relation.get_parent().should_receive("touches").and_return(False)440 self.assertEqual(441 {"attached": [4], "detached": [1], "updated": []},442 relation.sync([2, {3: {"bar": "baz"}}, {4: {"foo": "bar"}}]),443 )444 def test_touch_method_syncs_timestamps(self):445 relation = self._get_relation()446 relation.get_related().should_receive("get_updated_at_column").and_return(447 "updated_at"448 )449 now = pendulum.now()450 relation.get_related().should_receive("fresh_timestamp").and_return(now)451 relation.get_related().should_receive("get_qualified_key_name").and_return(452 "table.id"453 )454 relation.get_query().get_query().should_receive("select").once().with_args(455 "table.id"456 ).and_return(relation.get_query().get_query())457 relation.get_query().should_receive("lists").once().and_return(458 Collection([1, 2, 3])459 )460 query = flexmock()461 relation.get_related().should_receive("new_query").once().and_return(query)462 query.should_receive("where_in").once().with_args("id", [1, 2, 3]).and_return(463 query464 )465 query.should_receive("update").once().with_args({"updated_at": now})466 relation.touch()467 def test_touch_if_touching(self):468 flexmock(BelongsToMany)469 relation = self._get_relation()470 relation.should_receive("_touching_parent").once().and_return(True)471 relation.get_parent().should_receive("touch").once()472 relation.get_parent().should_receive("touches").once().with_args(473 "relation_name"474 ).and_return(True)475 relation.should_receive("touch").once()476 relation.touch_if_touching()477 def test_sync_method_converts_collection_to_list_of_keys(self):478 flexmock(BelongsToMany)479 relation = self._get_relation()480 query = flexmock()481 query.should_receive("from_").once().with_args("user_role").and_return(query)482 query.should_receive("where").once().with_args("user_id", 1).and_return(query)483 mock_query_builder = flexmock()484 relation.get_query().should_receive("get_query").and_return(mock_query_builder)485 mock_query_builder.should_receive("new_query").once().and_return(query)486 query.should_receive("lists").once().with_args("role_id").and_return(487 Collection([1, 2, 3])488 )489 collection = flexmock(Collection())490 collection.should_receive("model_keys").once().and_return([1, 2, 3])491 relation.should_receive("_format_sync_list").with_args([1, 2, 3]).and_return(492 {1: {}, 2: {}, 3: {}}493 )494 relation.sync(collection)495 def test_where_pivot_params_used_for_new_queries(self):496 flexmock(BelongsToMany)497 relation = self._get_relation()498 relation.get_query().should_receive("where").once().and_return(relation)499 query = flexmock()500 mock_query_builder = flexmock()501 relation.get_query().should_receive("get_query").and_return(mock_query_builder)502 mock_query_builder.should_receive("new_query").once().and_return(query)503 query.should_receive("from_").once().with_args("user_role").and_return(query)504 query.should_receive("where").once().with_args("user_id", 1).and_return(query)505 query.should_receive("where").once().with_args(506 "foo", "=", "bar", "and"507 ).and_return(query)508 query.should_receive("lists").once().with_args("role_id").and_return(509 Collection([1, 2, 3])510 )511 relation.should_receive("_format_sync_list").with_args([1, 2, 3]).and_return(512 {1: {}, 2: {}, 3: {}}513 )514 relation = relation.where_pivot("foo", "=", "bar")515 relation.sync([1, 2, 3])516 def _get_relation(self):517 builder, parent = self._get_relation_arguments()[:2]518 return BelongsToMany(519 builder, parent, "user_role", "user_id", "role_id", "relation_name"520 )521 def _get_relation_arguments(self):522 flexmock(Model).should_receive("_boot_columns").and_return(["name"])523 parent = flexmock(Model())524 parent.should_receive("get_key").and_return(1)525 parent.should_receive("get_created_at_column").and_return("created_at")526 parent.should_receive("get_updated_at_column").and_return("updated_at")527 query = flexmock(528 QueryBuilder(529 MockConnection().prepare_mock(), QueryGrammar(), QueryProcessor()530 )531 )532 flexmock(Builder)533 builder = Builder(query)534 builder.should_receive("get_query").and_return(query)535 related = flexmock(Model())536 builder.set_model(related)537 builder.should_receive("get_model").and_return(related)538 related.should_receive("new_query").and_return(builder)539 related.should_receive("get_key_name").and_return("id")540 related.should_receive("get_table").and_return("roles")541 related.should_receive("new_pivot").replace_with(lambda *args: Pivot(*args))542 builder.get_query().should_receive("join").at_least().once().with_args(543 "user_role", "roles.id", "=", "user_role.role_id"544 )545 builder.should_receive("where").at_least().once().with_args(546 "user_role.user_id", "=", 1547 )548 return builder, parent, "user_role", "user_id", "role_id", "relation_id"549class OrmBelongsToManyModelStub(Model):550 __guarded__ = []551class OrmBelongsToManyModelPivotStub(Model):552 __guarded__ = []553 def __init__(self):554 super(OrmBelongsToManyModelPivotStub, self).__init__()...
test_migrator.py
Source:test_migrator.py
...18 if PY3K:19 inspect.getargspec = self.orig20 def test_migrations_are_run_up_when_outstanding_migrations_exist(self):21 resolver_mock = flexmock(DatabaseManager)22 resolver_mock.should_receive("connection").and_return({})23 resolver = flexmock(DatabaseManager({}))24 connection = flexmock()25 connection.should_receive("transaction").twice().and_return(connection)26 resolver.should_receive("connection").and_return(connection)27 migrator = flexmock(28 Migrator(29 flexmock(DatabaseMigrationRepository(resolver, "migrations")), resolver30 )31 )32 g = flexmock(glob)33 g.should_receive("glob").with_args(34 os.path.join(os.getcwd(), "[0-9]*_*.py")35 ).and_return(36 [37 os.path.join(os.getcwd(), "2_bar.py"),38 os.path.join(os.getcwd(), "1_foo.py"),39 os.path.join(os.getcwd(), "3_baz.py"),40 ]41 )42 migrator.get_repository().should_receive("get_ran").once().and_return(["1_foo"])43 migrator.get_repository().should_receive(44 "get_next_batch_number"45 ).once().and_return(1)46 migrator.get_repository().should_receive("log").once().with_args("2_bar", 1)47 migrator.get_repository().should_receive("log").once().with_args("3_baz", 1)48 bar_mock = flexmock(MigrationStub())49 bar_mock.set_connection(connection)50 bar_mock.should_receive("up").once()51 baz_mock = flexmock(MigrationStub())52 baz_mock.set_connection(connection)53 baz_mock.should_receive("up").once()54 migrator.should_receive("_resolve").with_args(55 os.getcwd(), "2_bar"56 ).once().and_return(bar_mock)57 migrator.should_receive("_resolve").with_args(58 os.getcwd(), "3_baz"59 ).once().and_return(baz_mock)60 migrator.run(os.getcwd())61 def test_migrations_are_run_up_directly_if_transactional_is_false(self):62 resolver_mock = flexmock(DatabaseManager)63 resolver_mock.should_receive("connection").and_return({})64 resolver = flexmock(DatabaseManager({}))65 connection = flexmock()66 connection.should_receive("transaction").never()67 resolver.should_receive("connection").and_return(connection)68 migrator = flexmock(69 Migrator(70 flexmock(DatabaseMigrationRepository(resolver, "migrations")), resolver71 )72 )73 g = flexmock(glob)74 g.should_receive("glob").with_args(75 os.path.join(os.getcwd(), "[0-9]*_*.py")76 ).and_return(77 [78 os.path.join(os.getcwd(), "2_bar.py"),79 os.path.join(os.getcwd(), "1_foo.py"),80 os.path.join(os.getcwd(), "3_baz.py"),81 ]82 )83 migrator.get_repository().should_receive("get_ran").once().and_return(["1_foo"])84 migrator.get_repository().should_receive(85 "get_next_batch_number"86 ).once().and_return(1)87 migrator.get_repository().should_receive("log").once().with_args("2_bar", 1)88 migrator.get_repository().should_receive("log").once().with_args("3_baz", 1)89 bar_mock = flexmock(MigrationStub())90 bar_mock.transactional = False91 bar_mock.set_connection(connection)92 bar_mock.should_receive("up").once()93 baz_mock = flexmock(MigrationStub())94 baz_mock.transactional = False95 baz_mock.set_connection(connection)96 baz_mock.should_receive("up").once()97 migrator.should_receive("_resolve").with_args(98 os.getcwd(), "2_bar"99 ).once().and_return(bar_mock)100 migrator.should_receive("_resolve").with_args(101 os.getcwd(), "3_baz"102 ).once().and_return(baz_mock)103 migrator.run(os.getcwd())104 def test_up_migration_can_be_pretended(self):105 resolver_mock = flexmock(DatabaseManager)106 resolver_mock.should_receive("connection").and_return({})107 resolver = flexmock(DatabaseManager({}))108 connection = flexmock(Connection(None))109 connection.should_receive("get_logged_queries").twice().and_return([])110 resolver.should_receive("connection").with_args(None).and_return(connection)111 migrator = flexmock(112 Migrator(113 flexmock(DatabaseMigrationRepository(resolver, "migrations")), resolver114 )115 )116 g = flexmock(glob)117 g.should_receive("glob").with_args(118 os.path.join(os.getcwd(), "[0-9]*_*.py")119 ).and_return(120 [121 os.path.join(os.getcwd(), "2_bar.py"),122 os.path.join(os.getcwd(), "1_foo.py"),123 os.path.join(os.getcwd(), "3_baz.py"),124 ]125 )126 migrator.get_repository().should_receive("get_ran").once().and_return(["1_foo"])127 migrator.get_repository().should_receive(128 "get_next_batch_number"129 ).once().and_return(1)130 bar_mock = flexmock(MigrationStub())131 bar_mock.should_receive("get_connection").once().and_return(connection)132 bar_mock.should_receive("up").once()133 baz_mock = flexmock(MigrationStub())134 baz_mock.should_receive("get_connection").once().and_return(connection)135 baz_mock.should_receive("up").once()136 migrator.should_receive("_resolve").with_args(137 os.getcwd(), "2_bar"138 ).once().and_return(bar_mock)139 migrator.should_receive("_resolve").with_args(140 os.getcwd(), "3_baz"141 ).once().and_return(baz_mock)142 migrator.run(os.getcwd(), True)143 def test_nothing_is_done_when_no_migrations_outstanding(self):144 resolver_mock = flexmock(DatabaseManager)145 resolver_mock.should_receive("connection").and_return(None)146 resolver = flexmock(DatabaseManager({}))147 migrator = flexmock(148 Migrator(149 flexmock(DatabaseMigrationRepository(resolver, "migrations")), resolver150 )151 )152 g = flexmock(glob)153 g.should_receive("glob").with_args(154 os.path.join(os.getcwd(), "[0-9]*_*.py")155 ).and_return([os.path.join(os.getcwd(), "1_foo.py")])156 migrator.get_repository().should_receive("get_ran").once().and_return(["1_foo"])157 migrator.run(os.getcwd())158 def test_last_batch_of_migrations_can_be_rolled_back(self):159 resolver_mock = flexmock(DatabaseManager)160 resolver_mock.should_receive("connection").and_return({})161 resolver = flexmock(DatabaseManager({}))162 connection = flexmock()163 connection.should_receive("transaction").twice().and_return(connection)164 resolver.should_receive("connection").and_return(connection)165 migrator = flexmock(166 Migrator(167 flexmock(DatabaseMigrationRepository(resolver, "migrations")), resolver168 )169 )170 foo_migration = MigrationStub("foo")171 bar_migration = MigrationStub("bar")172 migrator.get_repository().should_receive("get_last").once().and_return(173 [foo_migration, bar_migration]174 )175 bar_mock = flexmock(MigrationStub())176 bar_mock.set_connection(connection)177 bar_mock.should_receive("down").once()178 foo_mock = flexmock(MigrationStub())179 foo_mock.set_connection(connection)180 foo_mock.should_receive("down").once()181 migrator.should_receive("_resolve").with_args(182 os.getcwd(), "bar"183 ).once().and_return(bar_mock)184 migrator.should_receive("_resolve").with_args(185 os.getcwd(), "foo"186 ).once().and_return(foo_mock)187 migrator.get_repository().should_receive("delete").once().with_args(188 bar_migration189 )190 migrator.get_repository().should_receive("delete").once().with_args(191 foo_migration192 )193 migrator.rollback(os.getcwd())194 def test_last_batch_of_migrations_can_be_rolled_back_directly_if_transactional_is_false(195 self196 ):197 resolver_mock = flexmock(DatabaseManager)198 resolver_mock.should_receive("connection").and_return({})199 resolver = flexmock(DatabaseManager({}))200 connection = flexmock()201 connection.should_receive("transaction").never()202 resolver.should_receive("connection").and_return(connection)203 migrator = flexmock(204 Migrator(205 flexmock(DatabaseMigrationRepository(resolver, "migrations")), resolver206 )207 )208 foo_migration = MigrationStub("foo")209 bar_migration = MigrationStub("bar")210 migrator.get_repository().should_receive("get_last").once().and_return(211 [foo_migration, bar_migration]212 )213 bar_mock = flexmock(MigrationStub())214 bar_mock.transactional = False215 bar_mock.set_connection(connection)216 bar_mock.should_receive("down").once()217 foo_mock = flexmock(MigrationStub())218 foo_mock.transactional = False219 foo_mock.set_connection(connection)220 foo_mock.should_receive("down").once()221 migrator.should_receive("_resolve").with_args(222 os.getcwd(), "bar"223 ).once().and_return(bar_mock)224 migrator.should_receive("_resolve").with_args(225 os.getcwd(), "foo"226 ).once().and_return(foo_mock)227 migrator.get_repository().should_receive("delete").once().with_args(228 bar_migration229 )230 migrator.get_repository().should_receive("delete").once().with_args(231 foo_migration232 )233 migrator.rollback(os.getcwd())234 def test_rollback_migration_can_be_pretended(self):235 resolver_mock = flexmock(DatabaseManager)236 resolver_mock.should_receive("connection").and_return({})237 resolver = flexmock(DatabaseManager({}))238 connection = flexmock(Connection(None))239 connection.should_receive("get_logged_queries").twice().and_return([])240 resolver.should_receive("connection").with_args(None).and_return(connection)241 migrator = flexmock(242 Migrator(243 flexmock(DatabaseMigrationRepository(resolver, "migrations")), resolver244 )245 )246 foo_migration = flexmock(MigrationStub("foo"))247 foo_migration.should_receive("get_connection").and_return(connection)248 bar_migration = flexmock(MigrationStub("bar"))249 bar_migration.should_receive("get_connection").and_return(connection)250 migrator.get_repository().should_receive("get_last").once().and_return(251 [foo_migration, bar_migration]252 )253 migrator.should_receive("_resolve").with_args(254 os.getcwd(), "bar"255 ).once().and_return(bar_migration)256 migrator.should_receive("_resolve").with_args(257 os.getcwd(), "foo"258 ).once().and_return(foo_migration)259 migrator.rollback(os.getcwd(), True)260 self.assertTrue(foo_migration.downed)261 self.assertFalse(foo_migration.upped)262 self.assertTrue(foo_migration.downed)263 self.assertFalse(foo_migration.upped)264 def test_nothing_is_rolled_back_when_nothing_in_repository(self):265 resolver = flexmock(DatabaseManager)266 resolver.should_receive("connection").and_return(None)267 migrator = flexmock(268 Migrator(269 flexmock(DatabaseMigrationRepository(resolver, "migrations")), resolver270 )271 )272 migrator.get_repository().should_receive("get_last").once().and_return([])273 migrator.rollback(os.getcwd())274class MigrationStub(Migration):275 def __init__(self, migration=None):276 self.migration = migration277 self.upped = False278 self.downed = False279 def up(self):280 self.upped = True281 def down(self):282 self.downed = True283 def __getitem__(self, item):...
test_migrate_command.py
Source:test_migrate_command.py
...7from .. import OratorCommandTestCase8class MigrateCommandTestCase(OratorCommandTestCase):9 def test_basic_migrations_call_migrator_with_proper_arguments(self):10 resolver = flexmock(DatabaseManager)11 resolver.should_receive("connection").and_return(None)12 migrator_mock = flexmock(Migrator)13 migrator_mock.should_receive("set_connection").once().with_args(None)14 migrator_mock.should_receive("run").once().with_args(15 os.path.join(os.getcwd(), "migrations"), False16 )17 migrator_mock.should_receive("get_notes").and_return([])18 migrator_mock.should_receive("repository_exists").once().and_return(True)19 command = flexmock(MigrateCommand())20 command.should_receive("_get_config").and_return({})21 command.should_receive("confirm").and_return(True)22 self.run_command(command)23 def test_migration_repository_create_when_necessary(self):24 resolver = flexmock(DatabaseManager)25 resolver.should_receive("connection").and_return(None)26 migrator_mock = flexmock(Migrator)27 migrator_mock.should_receive("set_connection").once().with_args(None)28 migrator_mock.should_receive("run").once().with_args(29 os.path.join(os.getcwd(), "migrations"), False30 )31 migrator_mock.should_receive("get_notes").and_return([])32 migrator_mock.should_receive("repository_exists").once().and_return(False)33 command = flexmock(MigrateCommand())34 command.should_receive("_get_config").and_return({})35 command.should_receive("confirm").and_return(True)36 command.should_receive("call").once().with_args(37 "migrate:install", [("--config", None)]38 )39 self.run_command(command)40 def test_migration_can_be_pretended(self):41 resolver = flexmock(DatabaseManager)42 resolver.should_receive("connection").and_return(None)43 migrator_mock = flexmock(Migrator)44 migrator_mock.should_receive("set_connection").once().with_args(None)45 migrator_mock.should_receive("run").once().with_args(46 os.path.join(os.getcwd(), "migrations"), True47 )48 migrator_mock.should_receive("get_notes").and_return([])49 migrator_mock.should_receive("repository_exists").once().and_return(True)50 command = flexmock(MigrateCommand())51 command.should_receive("_get_config").and_return({})52 command.should_receive("confirm").and_return(True)53 self.run_command(command, [("--pretend", True)])54 def test_migration_database_can_be_set(self):55 resolver = flexmock(DatabaseManager)56 resolver.should_receive("connection").and_return(None)57 migrator_mock = flexmock(Migrator)58 migrator_mock.should_receive("set_connection").once().with_args("foo")59 migrator_mock.should_receive("run").once().with_args(60 os.path.join(os.getcwd(), "migrations"), False61 )62 migrator_mock.should_receive("get_notes").and_return([])63 migrator_mock.should_receive("repository_exists").once().and_return(False)64 command = flexmock(MigrateCommand())65 command.should_receive("_get_config").and_return({})66 command.should_receive("confirm").and_return(True)67 command.should_receive("call").once().with_args(68 "migrate:install", [("--database", "foo"), ("--config", None)]69 )70 self.run_command(command, [("--database", "foo")])71 def test_migration_can_be_forced(self):72 resolver = flexmock(DatabaseManager)73 resolver.should_receive("connection").and_return(None)74 migrator_mock = flexmock(Migrator)75 migrator_mock.should_receive("set_connection").once().with_args(None)76 migrator_mock.should_receive("run").once().with_args(77 os.path.join(os.getcwd(), "migrations"), False78 )79 migrator_mock.should_receive("get_notes").and_return([])80 migrator_mock.should_receive("repository_exists").once().and_return(True)81 command = flexmock(MigrateCommand())82 command.should_receive("_get_config").and_return({})...
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 autotest 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?
