How to use db_for_read method in pytest-django

Best Python code snippet using pytest-django_python

tests.py

Source:tests.py

...23        self.assertFalse('auth' in REDSHIFT_APPs)24    def testDefaultDbForRead(self):25        router = DefaultRouter()26        # DB127        self.assertEqual(router.db_for_read(Account), 'default')28        self.assertEqual(router.db_for_read(Org), 'default')29        self.assertEqual(router.db_for_read(OrgMembership), 'default')30        self.assertEqual(router.db_for_read(Project), 'default')31        self.assertEqual(router.db_for_read(Device), 'default')32        self.assertEqual(router.db_for_read(StreamId), 'default')33        self.assertEqual(router.db_for_read(StreamVariable), 'default')34        self.assertEqual(router.db_for_read(StreamSystemVariable), 'default')35        self.assertEqual(router.db_for_read(Streamer), 'default')36        self.assertEqual(router.db_for_read(StreamerReport), 'default')37        self.assertEqual(router.db_for_read(StreamFilter), 'default')38        self.assertEqual(router.db_for_read(StreamFilterAction), 'default')39        self.assertEqual(router.db_for_read(StreamFilterTrigger), 'default')40        self.assertEqual(router.db_for_read(StreamEventData), 'default')41        # DB242        self.assertEqual(router.db_for_read(StreamData), None)43    def testDefaultDbForWrite(self):44        router = DefaultRouter()45        # DB146        self.assertEqual(router.db_for_write(Account), 'default')47        self.assertEqual(router.db_for_write(Org), 'default')48        self.assertEqual(router.db_for_write(OrgMembership), 'default')49        self.assertEqual(router.db_for_write(Project), 'default')50        self.assertEqual(router.db_for_write(Device), 'default')51        self.assertEqual(router.db_for_write(StreamId), 'default')52        self.assertEqual(router.db_for_write(StreamVariable), 'default')53        self.assertEqual(router.db_for_write(StreamSystemVariable), 'default')54        self.assertEqual(router.db_for_write(Streamer), 'default')55        self.assertEqual(router.db_for_write(StreamerReport), 'default')56        self.assertEqual(router.db_for_write(StreamFilter), 'default')57        self.assertEqual(router.db_for_write(StreamFilterAction), 'default')58        self.assertEqual(router.db_for_write(StreamFilterTrigger), 'default')59        self.assertEqual(router.db_for_write(StreamEventData), 'default')60        # DB261        self.assertEqual(router.db_for_write(StreamData), None)62    def testDefaultDbAllowRelation(self):63        router = DefaultRouter()64        self.assertTrue(router.allow_relation(obj1=Org, obj2=Project))65        self.assertTrue(router.allow_relation(obj1=Device, obj2=Project))66        self.assertTrue(router.allow_relation(obj1=StreamId, obj2=Project))67        self.assertTrue(router.allow_relation(obj1=Streamer, obj2=Device))68        self.assertFalse(router.allow_relation(obj1=Org, obj2=StreamData))69        self.assertFalse(router.allow_relation(obj1=Org, obj2=StreamData))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)...

dbrouter.py

Source:dbrouter.py

...3    """4    A router to control all database operations on models in the website contenttypes application.5    """6    # route_app_labels = {'website'}7    def db_for_read(self, model, **hints):8        """9        Attempts to read auth and contenttypes models go to auth_db.10        """11        if model == User or model == IndustryData or model == Sentences or model == SentenceResults or model == SentenceRank:12            return 'website'13        return None14    def db_for_write(self, model, **hints):15        """16        Attempts to write auth and contenttypes models go to auth_db.17        """18        if model == User or model == IndustryData or model == Sentences or model == SentenceResults or model == SentenceRank:19            return 'website'20        return None21# class UserDBRouter(object):22#     def db_for_read(self, model, **hints):23#         """ reading SomeModel from otherdb """24#         if model == User:25#             return 'website'26#         return None27#     def db_for_write(self, model, **hints):28#         """ writing SomeModel to otherdb """29#         if model == User:30#             return 'website'31#         return None32# class IndustryDataDBRouter(object):33#     def db_for_read(self, model, **hints):34#         """ reading SomeModel from otherdb """35#         if model == IndustryData:36#             return 'website'37#         return None38#     def db_for_write(self, model, **hints):39#         """ writing SomeModel to otherdb """40#         if model == IndustryData:41#             return 'website'42#         return None43# class SentencesDBRouter(object):44#     def db_for_read(self, model, **hints):45#         """ reading SomeModel from otherdb """46#         if model == Sentences:47#             return 'website'48#         return None49#     def db_for_write(self, model, **hints):50#         """ writing SomeModel to otherdb """51#         if model == Sentences:52#             return 'website'53#         return None54# class SentenceResultsDBRouter(object):55#     def db_for_read(self, model, **hints):56#         """ reading SomeModel from otherdb """57#         if model == SentenceResults:58#             return 'website'59#         return None60#     def db_for_write(self, model, **hints):61#         """ writing SomeModel to otherdb """62#         if model == SentenceResults:63#             return 'website'64#         return None65# class UserDBRouter(object):66#     def db_for_read(self, model, **hints):67#         """ reading SomeModel from otherdb """68#         if model == User:69#             return 'website'70#         return None71#     def db_for_write(self, model, **hints):72#         """ writing SomeModel to otherdb """73#         if model == User:74#             return 'website'...

dbrouters.py

Source:dbrouters.py

...16    db_name = None17    db_tables = ()18    app_labels = ()1920    def db_for_read(self, model, **hints):21        if (model._meta.db_table in self.db_tables) or (model._meta.app_label in self.app_labels):22            return self.db_name23        return None242526class MetaDBRouter(_Router):27    db_name = 'metadb'2829    if settings.DATABASES.get(db_name):30        db_tables = _get_tables(db_name, db_label=db_name)3132    def db_for_write(self, model, **hints):33        return self.db_for_read(model, **hints)343536class DjangoRouter(_Router):37    db_name = 'django'3839    if settings.DATABASES.get(db_name):40        db_tables = _get_tables(db_name, db_label=db_name)4142    def db_for_write(self, model, **hints):43        return self.db_for_read(model, **hints)444546class CRSdbRouter(_Router):47    db_name = 'CRSdb'48    if settings.DATABASES.get(db_name):49        db_tables = _get_tables(db_name, db_label=db_name)5051    def db_for_write(self, model, **hints):
...

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.