Make sure nested transactions which reference different databases work as expected.
(self)
| 31 | ENGINE_2.remove_db_file() |
| 32 | |
| 33 | async def run_nested(self): |
| 34 | """ |
| 35 | Make sure nested transactions which reference different databases work |
| 36 | as expected. |
| 37 | """ |
| 38 | async with Musician._meta.db.transaction(): |
| 39 | await Musician.create_table().run() |
| 40 | await Musician(name="Bob").save().run() |
| 41 | |
| 42 | async with Roadie._meta.db.transaction(): |
| 43 | await Roadie.create_table().run() |
| 44 | await Roadie(name="Dave").save().run() |
| 45 | |
| 46 | self.assertTrue(await Musician.table_exists().run()) |
| 47 | musician = await Musician.select("name").first().run() |
| 48 | assert musician is not None |
| 49 | self.assertEqual(musician["name"], "Bob") |
| 50 | |
| 51 | self.assertTrue(await Roadie.table_exists().run()) |
| 52 | roadie = await Roadie.select("name").first().run() |
| 53 | assert roadie is not None |
| 54 | self.assertEqual(roadie["name"], "Dave") |
| 55 | |
| 56 | def test_nested(self): |
| 57 | asyncio.run(self.run_nested()) |
no test coverage detected