KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use allow_migrate method in pytest-django

Best Python code snippet using pytest-django_python

test_routers.py

Source:test_routers.py

...14class HorizontalRouterMigrationTestCase(HorizontalBaseTestCase):15    def setUp(self):16        super(HorizontalRouterMigrationTestCase, self).setUp()17        self.router = HorizontalRouter()18    def test_allow_migrate(self):19        for expected_database in ('a1-primary', 'a1-replica-1', 'a1-replica-2', 'a2-primary',20                                  'a2-replica', 'a3'):21            self.assertTrue(22                self.router.allow_migrate(expected_database, 'tests', 'OneModel',23                                          model=OneModel))24            self.assertTrue(25                self.router.allow_migrate(expected_database, 'tests', 'ManyModel',26                                          model=OneModel))27        for expected_database in ('b1-primary', 'b1-replica-1', 'b1-replica-2', 'b2-primary',28                                  'b2-replica', 'b3'):29            self.assertTrue(30                self.router.allow_migrate(expected_database, 'tests', 'ProxyBaseModel',31                                          model=ProxyBaseModel))32            self.assertTrue(33                self.router.allow_migrate(expected_database, 'tests', 'ProxiedModel',34                                          model=ProxiedModel))35            self.assertTrue(36                self.router.allow_migrate(expected_database, 'tests', 'ConcreteModel',37                                          model=ConcreteModel))38    def test_allow_migrate_without_hints(self):39        for expected_database in ('a1-primary', 'a1-replica-1', 'a1-replica-2', 'a2-primary',40                                  'a2-replica', 'a3'):41            self.assertTrue(42                self.router.allow_migrate(expected_database, 'tests', 'OneModel'))43            self.assertTrue(44                self.router.allow_migrate(expected_database, 'tests', 'ManyModel'))45        for expected_database in ('b1-primary', 'b1-replica-1', 'b1-replica-2', 'b2-primary',46                                  'b2-replica', 'b3'):47            self.assertTrue(48                self.router.allow_migrate(expected_database, 'tests', 'ProxyBaseModel'))49            self.assertTrue(50                self.router.allow_migrate(expected_database, 'tests', 'ProxiedModel'))51            self.assertTrue(52                self.router.allow_migrate(expected_database, 'tests', 'ConcreteModel'))53    def test_not_allow_migrate(self):54        for horizontal_model in (OneModel, ManyModel, ProxyBaseModel, ProxiedModel, ConcreteModel):55            self.assertIsNone(56                self.router.allow_migrate('default', 'tests', horizontal_model.__name__,57                                          model=horizontal_model),58                "Other database",59            )60        self.assertIsNone(61            self.router.allow_migrate('a1-primary', user_model._meta.app_label, user_model.__name__,62                                      model=user_model),63            "Not configured for horizontal database groups",64        )65        self.assertIsNone(66            self.router.allow_migrate('a1-primary', 'tests', ProxyBaseModel.__name__,67                                      model=ProxyBaseModel),68            "Another horizontal group",69        )70        self.assertIsNone(71            self.router.allow_migrate('a1-primary', 'tests', ProxiedModel.__name__,72                                      model=ProxiedModel),73            "Another horizontal group",74        )75        self.assertIsNone(76            self.router.allow_migrate('a1-primary', 'tests', ConcreteModel.__name__,77                                      model=ConcreteModel),78            "Another horizontal group",79        )80    def test_not_allow_migrate_without_hints(self):81        for horizontal_model in (82            'OneModel', 'ManyModel', 'ProxyBaseModel', 'ProxiedModel', 'ConcreteModel'83        ):84            self.assertIsNone(85                self.router.allow_migrate('default', 'tests', horizontal_model),86                "Other database",87            )88        self.assertIsNone(89            self.router.allow_migrate(90                'a1-primary', user_model._meta.app_label, 'User', model=user_model),91            "Not configured for horizontal database groups",92        )93        self.assertIsNone(94            self.router.allow_migrate(95                'a1-primary', 'tests', 'ProxyBaseModel', model=ProxyBaseModel),96            "Another horizontal group",97        )98        self.assertIsNone(99            self.router.allow_migrate(100                'a1-primary', 'tests', 'ProxiedModel', model=ProxiedModel),101            "Another horizontal group",102        )103        self.assertIsNone(104            self.router.allow_migrate(105                'a1-primary', 'tests', 'ConcreteModel', model=ConcreteModel),106            "Another horizontal group",107        )108class HorizontalRouterRelationTestCase(HorizontalBaseTestCase):109    def setUp(self):110        super(HorizontalRouterRelationTestCase, self).setUp()111        self.router = HorizontalRouter()112        self.user_a = user_model.objects.create_user('spam')113        self.user_b = user_model.objects.create_user('egg')114    def test_allow_relation_in_same_database(self):115        HorizontalMetadata.objects.create(group='a', key=self.user_a.id, index=1)116        HorizontalMetadata.objects.create(group='a', key=self.user_b.id, index=1)117        one1 = OneModel.objects.create(user=self.user_a, spam='1st')118        one2 = OneModel.objects.create(user=self.user_b, spam='2nd')...

utils.py

Source:utils.py

...235                allow = method(obj1, obj2, **hints)236                if allow is not None:237                    return allow238        return obj1._state.db == obj2._state.db239    def allow_migrate(self, db, app_label, **hints):240        for router in self.routers:241            try:242                method = router.allow_migrate243            except AttributeError:244                # If the router doesn't have a method, skip to the next one.245                continue246            if six.PY3:247                sig = inspect.signature(router.allow_migrate)248                has_deprecated_signature = not any(249                    p.kind == inspect.Parameter.VAR_KEYWORD for p in sig.parameters.values()250                )251            else:252                argspec = inspect.getargspec(router.allow_migrate)253                has_deprecated_signature = len(argspec.args) == 3 and not argspec.keywords254            if has_deprecated_signature:255                warnings.warn(256                    "The signature of allow_migrate has changed from "257                    "allow_migrate(self, db, model) to "258                    "allow_migrate(self, db, app_label, model_name=None, **hints). "259                    "Support for the old signature will be removed in Django 1.10.",260                    RemovedInDjango110Warning)261                model = hints.get('model')262                allow = None if model is None else method(db, model)263            else:264                allow = method(db, app_label, **hints)265            if allow is not None:266                return allow267        return True268    def allow_migrate_model(self, db, model):269        return self.allow_migrate(270            db,271            model._meta.app_label,272            model_name=model._meta.model_name,273            model=model,274        )275    def get_migratable_models(self, app_config, db, include_auto_created=False):276        """277        Return app models allowed to be synchronized on provided db.278        """279        models = app_config.get_models(include_auto_created=include_auto_created)...

tests.py

Source:tests.py

...70        self.assertFalse(router.allow_relation(obj1=StreamEventData, obj2=StreamData))71    def testDefaultDbAllowMigrate(self):72        router = DefaultRouter()73        # DB174        self.assertTrue(router.allow_migrate(db='default', app_label='authentication'))75        self.assertTrue(router.allow_migrate(db='default', app_label='org'))76        self.assertTrue(router.allow_migrate(db='default', app_label='project'))77        self.assertTrue(router.allow_migrate(db='default', app_label='stream'))78        self.assertTrue(router.allow_migrate(db='default', app_label='physicaldevice'))79        self.assertTrue(router.allow_migrate(db='default', app_label='streamer'))80        self.assertTrue(router.allow_migrate(db='default', app_label='streamevent'))81        # DB282        self.assertIsNone(router.allow_migrate(db='streamdata', app_label='streamdata'))83    def testStreamDataDbForRead(self):84        router = StreamDataRouter()85        # DB186        self.assertEqual(router.db_for_read(Account), None)87        self.assertEqual(router.db_for_read(Org), None)88        self.assertEqual(router.db_for_read(OrgMembership), None)89        self.assertEqual(router.db_for_read(Project), None)90        self.assertEqual(router.db_for_read(Device), None)91        self.assertEqual(router.db_for_read(StreamId), None)92        self.assertEqual(router.db_for_read(StreamVariable), None)93        self.assertEqual(router.db_for_read(StreamSystemVariable), None)94        self.assertEqual(router.db_for_read(Streamer), None)95        self.assertEqual(router.db_for_read(StreamerReport), None)96        self.assertEqual(router.db_for_read(StreamFilter), None)97        self.assertEqual(router.db_for_read(StreamFilterAction), None)98        self.assertEqual(router.db_for_read(StreamFilterTrigger), None)99        self.assertEqual(router.db_for_read(StreamEventData), None)100        # DB2101        self.assertEqual(router.db_for_read(StreamData), 'streamdata')102    def testStreamDataDbForWrite(self):103        router = StreamDataRouter()104        # DB1105        self.assertEqual(router.db_for_write(Account), None)106        self.assertEqual(router.db_for_write(Org), None)107        self.assertEqual(router.db_for_write(OrgMembership), None)108        self.assertEqual(router.db_for_write(Project), None)109        self.assertEqual(router.db_for_write(Device), None)110        self.assertEqual(router.db_for_write(StreamId), None)111        self.assertEqual(router.db_for_write(StreamVariable), None)112        self.assertEqual(router.db_for_write(StreamSystemVariable), None)113        self.assertEqual(router.db_for_write(Streamer), None)114        self.assertEqual(router.db_for_write(StreamerReport), None)115        self.assertEqual(router.db_for_write(StreamFilter), None)116        self.assertEqual(router.db_for_write(StreamFilterAction), None)117        self.assertEqual(router.db_for_write(StreamFilterTrigger), None)118        self.assertEqual(router.db_for_write(StreamEventData), None)119        # DB2120        self.assertEqual(router.db_for_write(StreamData), 'streamdata')121    def testStreamDataDbAllowRelation(self):122        router = StreamDataRouter()123        self.assertFalse(router.allow_relation(obj1=Org, obj2=Project))124        self.assertFalse(router.allow_relation(obj1=Device, obj2=Project))125        self.assertFalse(router.allow_relation(obj1=StreamId, obj2=Project))126        self.assertFalse(router.allow_relation(obj1=Streamer, obj2=Device))127        self.assertFalse(router.allow_relation(obj1=Org, obj2=StreamData))128        self.assertFalse(router.allow_relation(obj1=Org, obj2=StreamData))129        self.assertFalse(router.allow_relation(obj1=StreamEventData, obj2=StreamData))130    def testStreamDataDbAllowMigrate(self):131        router = StreamDataRouter()132        # DB1133        self.assertIsNone(router.allow_migrate(db='default', app_label='authentication'))134        self.assertIsNone(router.allow_migrate(db='default', app_label='org'))135        self.assertIsNone(router.allow_migrate(db='default', app_label='project'))136        self.assertIsNone(router.allow_migrate(db='default', app_label='stream'))137        self.assertIsNone(router.allow_migrate(db='default', app_label='physicaldevice'))138        self.assertIsNone(router.allow_migrate(db='default', app_label='streamer'))139        self.assertIsNone(router.allow_migrate(db='default', app_label='streamevent'))140        # DB2...

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.