|
| 1 | +""" |
| 2 | +Pets example demonstrating per-test rollback with before_each/after_each. |
| 3 | +
|
| 4 | +This example shows how pgsql-test provides complete test isolation through |
| 5 | +automatic transaction rollback. Each test starts with a clean slate, |
| 6 | +regardless of what previous tests inserted. |
| 7 | +
|
| 8 | +Key concept: before_each() creates a savepoint, after_each() rolls back to it. |
| 9 | +""" |
| 10 | + |
| 11 | +import pytest |
| 12 | + |
| 13 | +from pysql_test import get_connections, seed |
| 14 | + |
| 15 | + |
| 16 | +@pytest.fixture |
| 17 | +def pets_db(): |
| 18 | + """ |
| 19 | + Create an isolated test database with a simple pets schema. |
| 20 | +
|
| 21 | + The before_each()/after_each() pattern ensures each test: |
| 22 | + 1. Starts with only the seeded data (empty pets table) |
| 23 | + 2. Can insert/modify data freely during the test |
| 24 | + 3. Has all changes rolled back automatically after the test |
| 25 | + """ |
| 26 | + conn = get_connections( |
| 27 | + seed_adapters=[ |
| 28 | + seed.fn(lambda ctx: ctx["pg"].query(""" |
| 29 | + CREATE TABLE pets ( |
| 30 | + id SERIAL PRIMARY KEY, |
| 31 | + name TEXT NOT NULL, |
| 32 | + species TEXT NOT NULL, |
| 33 | + age INTEGER |
| 34 | + ) |
| 35 | + """)) |
| 36 | + ] |
| 37 | + ) |
| 38 | + db = conn.db |
| 39 | + db.before_each() # Begin transaction + create savepoint |
| 40 | + yield db |
| 41 | + db.after_each() # Rollback to savepoint (undo all changes) |
| 42 | + conn.teardown() |
| 43 | + |
| 44 | + |
| 45 | +# ============================================================================= |
| 46 | +# Test 1: Insert a pet and verify it exists |
| 47 | +# ============================================================================= |
| 48 | +def test_insert_pet(pets_db): |
| 49 | + """Insert a pet and verify it exists in the database.""" |
| 50 | + pets_db.execute( |
| 51 | + "INSERT INTO pets (name, species, age) VALUES (%s, %s, %s)", |
| 52 | + ("Buddy", "dog", 3), |
| 53 | + ) |
| 54 | + |
| 55 | + pet = pets_db.one("SELECT * FROM pets WHERE name = %s", ("Buddy",)) |
| 56 | + |
| 57 | + assert pet["name"] == "Buddy" |
| 58 | + assert pet["species"] == "dog" |
| 59 | + assert pet["age"] == 3 |
| 60 | + |
| 61 | + |
| 62 | +# ============================================================================= |
| 63 | +# Test 2: Verify the table is empty (previous insert was rolled back!) |
| 64 | +# ============================================================================= |
| 65 | +def test_table_empty_after_rollback(pets_db): |
| 66 | + """ |
| 67 | + Verify that the previous test's insert was rolled back. |
| 68 | +
|
| 69 | + Even though test_insert_pet inserted 'Buddy', that change was |
| 70 | + automatically rolled back by after_each(). This test starts fresh. |
| 71 | + """ |
| 72 | + count = pets_db.one("SELECT COUNT(*) as count FROM pets") |
| 73 | + |
| 74 | + # Table should be empty - Buddy was rolled back! |
| 75 | + assert count["count"] == 0 |
| 76 | + |
| 77 | + |
| 78 | +# ============================================================================= |
| 79 | +# Test 3: Insert multiple pets |
| 80 | +# ============================================================================= |
| 81 | +def test_insert_multiple_pets(pets_db): |
| 82 | + """Insert multiple pets and verify the count.""" |
| 83 | + pets_db.execute( |
| 84 | + "INSERT INTO pets (name, species, age) VALUES (%s, %s, %s)", |
| 85 | + ("Whiskers", "cat", 5), |
| 86 | + ) |
| 87 | + pets_db.execute( |
| 88 | + "INSERT INTO pets (name, species, age) VALUES (%s, %s, %s)", |
| 89 | + ("Goldie", "fish", 1), |
| 90 | + ) |
| 91 | + pets_db.execute( |
| 92 | + "INSERT INTO pets (name, species, age) VALUES (%s, %s, %s)", |
| 93 | + ("Rex", "dog", 7), |
| 94 | + ) |
| 95 | + |
| 96 | + count = pets_db.one("SELECT COUNT(*) as count FROM pets") |
| 97 | + assert count["count"] == 3 |
| 98 | + |
| 99 | + # Verify we can query specific pets |
| 100 | + cats = pets_db.many("SELECT * FROM pets WHERE species = %s", ("cat",)) |
| 101 | + assert len(cats) == 1 |
| 102 | + assert cats[0]["name"] == "Whiskers" |
| 103 | + |
| 104 | + |
| 105 | +# ============================================================================= |
| 106 | +# Test 4: Verify table is empty again (all 3 pets were rolled back!) |
| 107 | +# ============================================================================= |
| 108 | +def test_table_empty_again(pets_db): |
| 109 | + """ |
| 110 | + Verify that ALL previous inserts were rolled back. |
| 111 | +
|
| 112 | + The 3 pets from test_insert_multiple_pets are gone. |
| 113 | + Each test truly starts with a clean slate. |
| 114 | + """ |
| 115 | + count = pets_db.one("SELECT COUNT(*) as count FROM pets") |
| 116 | + |
| 117 | + # Table should be empty - all pets were rolled back! |
| 118 | + assert count["count"] == 0 |
| 119 | + |
| 120 | + |
| 121 | +# ============================================================================= |
| 122 | +# Test 5: Demonstrate update rollback |
| 123 | +# ============================================================================= |
| 124 | +def test_update_rollback(pets_db): |
| 125 | + """ |
| 126 | + Demonstrate that updates are also rolled back. |
| 127 | +
|
| 128 | + Insert a pet, update it, verify the update - all rolled back after. |
| 129 | + """ |
| 130 | + # Insert |
| 131 | + pets_db.execute( |
| 132 | + "INSERT INTO pets (name, species, age) VALUES (%s, %s, %s)", |
| 133 | + ("Max", "dog", 2), |
| 134 | + ) |
| 135 | + |
| 136 | + # Update |
| 137 | + pets_db.execute( |
| 138 | + "UPDATE pets SET age = %s WHERE name = %s", |
| 139 | + (3, "Max"), |
| 140 | + ) |
| 141 | + |
| 142 | + # Verify update worked within this test |
| 143 | + pet = pets_db.one("SELECT * FROM pets WHERE name = %s", ("Max",)) |
| 144 | + assert pet["age"] == 3 # Updated age |
| 145 | + |
| 146 | + |
| 147 | +# ============================================================================= |
| 148 | +# Test 6: Final verification - still empty! |
| 149 | +# ============================================================================= |
| 150 | +def test_final_empty_check(pets_db): |
| 151 | + """ |
| 152 | + Final check: table is still empty after all previous tests. |
| 153 | +
|
| 154 | + This proves that before_each()/after_each() provides complete |
| 155 | + isolation for every single test, no matter what operations were performed. |
| 156 | + """ |
| 157 | + count = pets_db.one("SELECT COUNT(*) as count FROM pets") |
| 158 | + assert count["count"] == 0 |
| 159 | + |
| 160 | + # We can safely insert knowing it won't affect other tests |
| 161 | + pets_db.execute( |
| 162 | + "INSERT INTO pets (name, species, age) VALUES (%s, %s, %s)", |
| 163 | + ("Luna", "cat", 4), |
| 164 | + ) |
| 165 | + |
| 166 | + # Verify our insert worked |
| 167 | + pet = pets_db.one("SELECT * FROM pets WHERE name = %s", ("Luna",)) |
| 168 | + assert pet["name"] == "Luna" |
| 169 | + |
| 170 | + # Luna will be rolled back after this test completes |
0 commit comments