Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More
How to use db_for_write method in pytest-django
Best Python code snippet using pytest-django_python
test_routers.py
Source:test_routers.py 
...148 self.user_c = user_model.objects.create_user('musubi')149 HorizontalMetadata.objects.create(group='a', key=self.user_a.id, index=1)150 HorizontalMetadata.objects.create(group='a', key=self.user_b.id, index=2)151 HorizontalMetadata.objects.create(group='a', key=self.user_c.id, index=3)152 def test_db_for_write(self):153 with patch.object(154 HorizontalRouter, 'db_for_write', wraps=self.router.db_for_write,155 ) as mock_db_for_write:156 OneModel.objects.create(user=self.user_a, spam='1st')157 mock_db_for_write.assert_any_call(OneModel, horizontal_key=self.user_a.id)158 self.assertEqual(159 'a1-primary',160 self.router.db_for_write(OneModel, horizontal_key=self.user_a.id),161 )162 with patch.object(163 HorizontalRouter, 'db_for_write', wraps=self.router.db_for_write,164 ) as mock_db_for_write:165 OneModel.objects.create(user=self.user_b, spam='2nd')166 mock_db_for_write.assert_any_call(OneModel, horizontal_key=self.user_b.id)167 self.assertEqual(168 'a2-primary',169 self.router.db_for_write(OneModel, horizontal_key=self.user_b.id),170 )171 with patch.object(172 HorizontalRouter, 'db_for_write', wraps=self.router.db_for_write,173 ) as mock_db_for_write:174 OneModel.objects.create(user=self.user_c, spam='1st')175 mock_db_for_write.assert_any_call(OneModel, horizontal_key=self.user_c.id)176 self.assertEqual(177 'a3',178 self.router.db_for_write(OneModel, horizontal_key=self.user_c.id),179 )180 def test_db_for_write_by_id(self):181 with patch.object(182 HorizontalRouter, 'db_for_write', wraps=self.router.db_for_write,183 ) as mock_db_for_write:184 OneModel.objects.create(user_id=self.user_a.id, spam='1st')185 mock_db_for_write.assert_any_call(OneModel, horizontal_key=self.user_a.id)186 self.assertEqual(187 'a1-primary',188 self.router.db_for_write(OneModel, horizontal_key=self.user_a.id),189 )190 def test_db_for_write_other_databases(self):191 with patch.object(192 HorizontalRouter, 'db_for_write', wraps=self.router.db_for_write,193 ) as mock_db_for_write:194 new_user = user_model.objects.create_user('pizza')195 mock_db_for_write.assert_any_call(user_model, instance=new_user)196 self.assertIsNone(self.router.db_for_write(user_model, instance=new_user))197 def test_db_for_read(self):198 with patch.object(199 HorizontalRouter, 'db_for_read', wraps=self.router.db_for_read,200 ) as mock_db_for_read:201 list(OneModel.objects.filter(user=self.user_a))202 mock_db_for_read.assert_any_call(OneModel, horizontal_key=self.user_a.id)203 self.assertIn(204 self.router.db_for_read(OneModel, horizontal_key=self.user_a.id),205 ['a1-replica-1', 'a1-replica-2'],206 )207 with patch.object(208 HorizontalRouter, 'db_for_read', wraps=self.router.db_for_read,209 ) as mock_db_for_read:210 list(OneModel.objects.filter(user=self.user_b))...
tests.py
Source:tests.py
...42 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)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'))...
routers.py
Source:routers.py
1MY_APPS = [2 'budget_app',3 'calend_app',4 'car_app',5 'computers_app',6 'hrefs_app',7 'humans_app',8 'menu_app',9 'metro_app',10]11MY_DB = [12 'budget_db',13 'calend_db',14 'car_db',15 'computers_db',16 'hrefs_db',17 'humans_db',18 'menu_db',19 'metro_db',20]21class BudgetRouter (object):22 '''ÐеÑенапÑавление в ÐÐ budget'''23 def db_for_read (self, model, **hints):24 if model._meta.app_label == 'budget_app': return 'budget_db'25 return None26 def db_for_write (self, model, **hints):27 if model._meta.app_label == 'budget_app': return 'budget_db'28 return None29 def allow_relation (self, obj1, obj2, **hints):30 if obj1._meta.app_label == 'budget_app' or \31 obj2._meta.app_label == 'budget_app':32 return True33 return None34 def allow_migrate (self, db, app_label, model = None, **hints):35 if app_label == 'budget_app': return db == 'budget_db'36 return None37class CalendRouter (object):38 '''ÐеÑенапÑавление в ÐÐ calend'''39 def db_for_read (self, model, **hints):40 if model._meta.app_label == 'calend_app': return 'calend_db'41 return None42 def db_for_write (self, model, **hints):43 if model._meta.app_label == 'calend_app': return 'calend_db'44 return None45 def allow_relation (self, obj1, obj2, **hints):46 if obj1._meta.app_label == 'calend_app' or \47 obj2._meta.app_label == 'calend_app':48 return True49 return None50 def allow_migrate (self, db, app_label, model = None, **hints):51 if app_label == 'calend_app': return db == 'calend_db'52 return None53class CarRouter (object):54 '''ÐеÑенапÑавление в ÐÐ car'''55 def db_for_read (self, model, **hints):56 if model._meta.app_label == 'car_app': return 'car_db'57 return None58 def db_for_write (self, model, **hints):59 if model._meta.app_label == 'car_app': return 'car_db'60 return None61 def allow_relation (self, obj1, obj2, **hints):62 if obj1._meta.app_label == 'car_app' or \63 obj2._meta.app_label == 'car_app':64 return True65 return None66 def allow_migrate (self, db, app_label, model = None, **hints):67 if app_label == 'car_app': return db == 'car_db'68 return None69class ComputersRouter (object):70 '''ÐеÑенапÑавление в ÐÐ computers'''71 def db_for_read (self, model, **hints):72 if model._meta.app_label == 'computers_app': return 'computers_db'73 return None74 def db_for_write (self, model, **hints):75 if model._meta.app_label == 'computers_app': return 'computers_db'76 return None77 def allow_relation (self, obj1, obj2, **hints):78 if obj1._meta.app_label == 'computers_app' or \79 obj2._meta.app_label == 'computers_app':80 return True81 return None82 def allow_migrate (self, db, app_label, model = None, **hints):83 if app_label == 'computers_app': return db == 'computers_db'84 return None85class HrefsRouter (object):86 '''ÐеÑенапÑавление в ÐÐ hrefs'''87 def db_for_read (self, model, **hints):88 if model._meta.app_label == 'hrefs_app': return 'hrefs_db'89 return None90 def db_for_write (self, model, **hints):91 if model._meta.app_label == 'hrefs_app': return 'hrefs_db'92 return None93 def allow_relation (self, obj1, obj2, **hints):94 if obj1._meta.app_label == 'hrefs_app' or \95 obj2._meta.app_label == 'hrefs_app':96 return True97 return None98 def allow_migrate (self, db, app_label, model = None, **hints):99 if app_label == 'hrefs_app': return db == 'hrefs_db'100 return None101class HumansRouter (object):102 '''ÐеÑенапÑавление в ÐÐ humans'''103 def db_for_read (self, model, **hints):104 if model._meta.app_label == 'humans_app': return 'humans_db'105 return None106 def db_for_write (self, model, **hints):107 if model._meta.app_label == 'humans_app': return 'humans_db'108 return None109 def allow_relation (self, obj1, obj2, **hints):110 if obj1._meta.app_label == 'humans_app' or \111 obj2._meta.app_label == 'humans_app':112 return True113 return None114 def allow_migrate (self, db, app_label, model = None, **hints):115 if app_label == 'humans_app': return db == 'humans_db'116 return None117class MenuRouter (object):118 '''ÐеÑенапÑавление в ÐÐ menu'''119 def db_for_read (self, model, **hints):120 if model._meta.app_label == 'menu_app': return 'menu_db'121 return None122 def db_for_write (self, model, **hints):123 if model._meta.app_label == 'menu_app': return 'menu_db'124 return None125 def allow_relation (self, obj1, obj2, **hints):126 if obj1._meta.app_label == 'menu_app' or \127 obj2._meta.app_label == 'menu_app':128 return True129 return None130 def allow_migrate (self, db, app_label, model = None, **hints):131 if app_label == 'menu_app': return db == 'menu_db'132 return None133class MetroRouter (object):134 '''ÐеÑенапÑавление в ÐÐ metro'''135 def db_for_read (self, model, **hints):136 if model._meta.app_label == 'metro_app': return 'metro_db'137 return None138 def db_for_write (self, model, **hints):139 if model._meta.app_label == 'metro_app': return 'metro_db'140 return None141 def allow_relation (self, obj1, obj2, **hints):142 if obj1._meta.app_label == 'metro_app' or \143 obj2._meta.app_label == 'metro_app':144 return True145 return None146 def allow_migrate (self, db, app_label, model = None, **hints):147 if app_label == 'metro_app': return db == 'metro_db'148 return None149class DefaultRouter (object):150 '''ÐеÑенапÑавление в ÐÐ default'''151 def db_for_read (self, model, **hints):152 return 'default'153 def db_for_write (self, model, **hints):154 return 'default'155 def allow_relation (self, obj1, obj2, **hints):156 return True157 def allow_migrate (self, db, app_label, model = None, **hints):158 if app_label not in MY_APPS and \159 db not in MY_DB: 160 return True...
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 pytest-django 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?
