Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
68 changes: 68 additions & 0 deletions .github/workflows/test.yml
Original file line number Diff line number Diff line change
@@ -0,0 +1,68 @@
name: pysql-test CI

on:
push:
branches:
- main
- develop
pull_request:
branches:
- main
- develop
workflow_dispatch:

concurrency:
group: ${{ github.workflow }}-${{ github.ref }}-run-tests
cancel-in-progress: true

jobs:
test:
runs-on: ubuntu-latest

env:
PGHOST: localhost
PGPORT: 5432
PGUSER: postgres
PGPASSWORD: password

services:
pg_db:
image: ghcr.io/constructive-io/docker/postgres-plus:17
env:
POSTGRES_USER: postgres
POSTGRES_PASSWORD: password
options: >-
--health-cmd pg_isready
--health-interval 10s
--health-timeout 5s
--health-retries 5
ports:
- 5432:5432

steps:
- name: Checkout
uses: actions/checkout@v4

- name: Set up Python
uses: actions/setup-python@v5
with:
python-version: "3.12"

- name: Install Poetry
uses: snok/install-poetry@v1
with:
version: latest
virtualenvs-create: true
virtualenvs-in-project: true

- name: Install dependencies
run: poetry install

- name: Run linting
run: poetry run ruff check .

- name: Run type checking
run: poetry run mypy src --ignore-missing-imports

- name: Run tests
run: poetry run pytest -v
71 changes: 71 additions & 0 deletions tests/test_example.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,71 @@
"""
Simple example test demonstrating pysql-test usage.

This is a minimal example showing how to use the testing framework
for PostgreSQL integration tests with automatic database isolation.
"""

import pytest

from pysql_test import get_connections, seed


@pytest.fixture
def db():
"""
Create an isolated test database with sample schema.

Each test gets a fresh database that is automatically
cleaned up after the test completes.
"""
conn = get_connections(
seed_adapters=[
seed.fn(lambda ctx: ctx["pg"].query("""
CREATE TABLE users (
id SERIAL PRIMARY KEY,
name TEXT NOT NULL,
email TEXT UNIQUE
)
"""))
]
)
db = conn.db
db.before_each()
yield db
db.after_each()
conn.teardown()


def test_insert_and_query_user(db):
"""Test inserting and querying a user."""
# Insert a user
db.execute(
"INSERT INTO users (name, email) VALUES (%s, %s)",
("Alice", "alice@example.com"),
)

# Query the user
user = db.one("SELECT * FROM users WHERE name = %s", ("Alice",))

assert user["name"] == "Alice"
assert user["email"] == "alice@example.com"


def test_transaction_isolation(db):
"""Test that changes are rolled back between tests."""
# This insert will be rolled back after the test
db.execute(
"INSERT INTO users (name, email) VALUES (%s, %s)",
("Bob", "bob@example.com"),
)

# Verify the user exists within this test
count = db.one("SELECT COUNT(*) as count FROM users")
assert count["count"] == 1


def test_empty_table_after_rollback(db):
"""Verify previous test's data was rolled back."""
# Table should be empty because previous test's insert was rolled back
count = db.one("SELECT COUNT(*) as count FROM users")
assert count["count"] == 0