Test user creation and retrieval using database fixture
Preconditions (3)
✅ Expected Result: The inserted user is found with the correct email, and the database remains unchanged after the test
Jump into concepts and practice - no test required
import pytest import psycopg2 from psycopg2.extras import RealDictCursor @pytest.fixture(scope='function') def db_connection(): conn = psycopg2.connect( dbname='testdb', user='testuser', password='testpass', host='localhost' ) conn.autocommit = False cursor = conn.cursor(cursor_factory=RealDictCursor) yield cursor conn.rollback() cursor.close() conn.close() def test_insert_and_retrieve_user(db_connection): cursor = db_connection insert_query = """ INSERT INTO users (username, email) VALUES (%s, %s) """ cursor.execute(insert_query, ('testuser', 'testuser@example.com')) select_query = """ SELECT username, email FROM users WHERE username = %s """ cursor.execute(select_query, ('testuser',)) user = cursor.fetchone() assert user is not None, "User should be found in database" assert user['email'] == 'testuser@example.com', f"Expected email 'testuser@example.com', got {user['email']}"
The db_connection fixture creates a database connection and cursor with a transaction started. It yields the cursor to the test, then rolls back any changes to keep the database clean.
The test test_insert_and_retrieve_user uses this fixture to insert a user with parameterized queries to avoid SQL injection. Then it queries the user back and asserts the email matches the expected value.
Using a fixture for connection management and rollback ensures tests do not affect each other or the real data, following best practices for database testing.
Now add data-driven testing with 3 different user inputs for username and email
yield?import pytest
@pytest.fixture
def sample_db():
data = {'count': 0}
yield data
data['count'] += 1
def test_increment(sample_db):
print(sample_db['count'])
sample_db['count'] += 5
print(sample_db['count'])@pytest.fixture
def test_db():
conn = connect_db()
conn.execute('CREATE TABLE users')
return conn
conn.execute('DROP TABLE users')
conn.close()