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.