Schema Design
Testing
FraiseQL applications have three distinct testing layers, each validating a different concern:
| Layer | What it tests | Speed |
|---|---|---|
| Schema tests | Type definitions and schema compilation | Instant |
| Database tests | SQL views and PostgreSQL functions | Fast (real DB, transaction rollback) |
| Integration tests | End-to-end GraphQL API behaviour | Slower (real server) |
Start with database tests — they give the most coverage at the best speed.
Test Setup
Section titled “Test Setup”Dependencies
Section titled “Dependencies”# Install test dependenciesuv add --dev pytest pytest-asyncio httpx testcontainers factory-boy[tool.pytest.ini_options]asyncio_mode = "auto"testpaths = ["tests"]
[tool.uv]dev-dependencies = [ "pytest>=8.0", "pytest-asyncio>=0.23", "httpx>=0.27", "testcontainers[postgres]>=4.0", "factory-boy>=3.3",]# Install test dependenciesnpm install --save-dev vitest @vitest/ui fraiseql testcontainers{ "scripts": { "test": "vitest run", "test:watch": "vitest", "test:ui": "vitest --ui" }, "devDependencies": { "vitest": "^1.0", "fraiseql": "latest", "testcontainers": "^10.0" }}Test Database Fixture
Section titled “Test Database Fixture”import pytestimport psycopgimport subprocessfrom testcontainers.postgres import PostgresContainer
@pytest.fixture(scope="session")def postgres(): """Spin up a throwaway PostgreSQL container for the test session.""" with PostgresContainer("postgres:16-alpine") as pg: yield pg
@pytest.fixture(scope="session")def db_url(postgres): return postgres.get_connection_url().replace("postgresql+psycopg2", "postgresql")
@pytest.fixture(scope="session")def migrated_db(db_url): """Apply your schema (tables, views, functions) to the test database.""" subprocess.run( ["fraiseql", "db", "build", "--database-url", db_url], check=True, ) return db_url
@pytest.fixturedef db(migrated_db): """Per-test database connection that rolls back after each test.""" with psycopg.connect(migrated_db) as conn: conn.autocommit = False yield conn conn.rollback()import { PostgreSqlContainer, StartedPostgreSqlContainer } from 'testcontainers';import { execSync } from 'child_process';import { TestClient } from 'fraiseql/testing';
let container: StartedPostgreSqlContainer;let testDatabaseUrl: string;
beforeAll(async () => { container = await new PostgreSqlContainer('postgres:16-alpine').start(); testDatabaseUrl = container.getConnectionUri();
// Apply schema (tables, views, functions) to the test database execSync(`fraiseql db build --database-url ${testDatabaseUrl}`, { stdio: 'inherit', });});
afterAll(async () => { await container.stop();});
export function getTestDatabaseUrl(): string { return testDatabaseUrl;}
export function createTestClient(): TestClient { return new TestClient({ schemaPath: './schema.ts', databaseUrl: testDatabaseUrl, });}1. Schema Tests
Section titled “1. Schema Tests”Verify that your type definitions compile to the expected GraphQL schema:
import fraiseqlfrom schema import User, Post, Comment
def test_user_type_has_expected_fields(): schema = fraiseql.compile_schema() user_type = schema.types["User"]
assert "id" in user_type.fields assert "email" in user_type.fields assert "username" in user_type.fields assert user_type.fields["id"].scalar == "ID"
def test_post_has_author_relationship(): schema = fraiseql.compile_schema() post_type = schema.types["Post"]
assert "author" in post_type.fields assert post_type.fields["author"].type_name == "User"
def test_mutations_are_registered(): schema = fraiseql.compile_schema()
assert "createPost" in schema.mutations assert "publishPost" in schema.mutations assert "deletePost" in schema.mutations
def test_schema_compiles_without_error(): """Full compilation should produce valid schema JSON.""" schema_json = fraiseql.export_schema_json() assert "types" in schema_json assert "queries" in schema_json assert "mutations" in schema_jsonimport { compileSchema, exportSchemaJson } from 'fraiseql';import { describe, it, expect } from 'vitest';
describe('Schema compilation', () => { it('User type has expected fields', () => { const schema = compileSchema(); const userType = schema.types['User'];
expect(userType.fields).toHaveProperty('id'); expect(userType.fields).toHaveProperty('email'); expect(userType.fields).toHaveProperty('username'); expect(userType.fields['id'].scalar).toBe('ID'); });
it('Post has author relationship', () => { const schema = compileSchema(); const postType = schema.types['Post'];
expect(postType.fields).toHaveProperty('author'); expect(postType.fields['author'].typeName).toBe('User'); });
it('mutations are registered', () => { const schema = compileSchema();
expect(schema.mutations).toHaveProperty('createPost'); expect(schema.mutations).toHaveProperty('publishPost'); expect(schema.mutations).toHaveProperty('deletePost'); });
it('schema compiles without error', () => { const schemaJson = exportSchemaJson(); expect(schemaJson).toHaveProperty('types'); expect(schemaJson).toHaveProperty('queries'); expect(schemaJson).toHaveProperty('mutations'); });});2. Database Tests
Section titled “2. Database Tests”Test SQL views and PostgreSQL functions directly. These tests skip the HTTP layer and call the database layer that FraiseQL actually uses at runtime.
Testing Views
Section titled “Testing Views”import pytestimport psycopg
class TestUserView: def test_v_user_returns_data_column(self, db): """v_user must expose a JSONB data column.""" with db.cursor() as cur: cur.execute(""" INSERT INTO tb_user (username, email, identifier) VALUES ('alice', 'alice@example.com', 'alice@example.com') RETURNING id """) user_id = cur.fetchone()[0]
cur.execute("SELECT data FROM v_user WHERE id = %s", (user_id,)) data = cur.fetchone()[0]
assert data["email"] == "alice@example.com" assert data["username"] == "alice" assert "id" in data # Surrogate key must NOT be exposed assert "pk_user" not in data
def test_v_user_filters_by_id(self, db): """v_user supports filtering by id column.""" with db.cursor() as cur: cur.execute(""" INSERT INTO tb_user (username, email, identifier) VALUES ('bob', 'bob@example.com', 'bob@example.com') RETURNING id """) user_id = cur.fetchone()[0]
cur.execute("SELECT COUNT(*) FROM v_user WHERE id = %s", (user_id,)) count = cur.fetchone()[0]
assert count == 1
class TestPostView: def test_v_post_nests_author(self, db): """v_post must include nested author data from v_user.""" with db.cursor() as cur: # Create author cur.execute(""" INSERT INTO tb_user (username, email, identifier) VALUES ('carol', 'carol@example.com', 'carol@example.com') RETURNING pk_user, id """) pk_user, user_id = cur.fetchone()
# Create post cur.execute(""" INSERT INTO tb_post (fk_user, title, content, slug, identifier) VALUES (%s, 'Hello World', 'Content here', 'hello-world', 'hello-world') RETURNING id """, (pk_user,)) post_id = cur.fetchone()[0]
cur.execute("SELECT data FROM v_post WHERE id = %s", (post_id,)) data = cur.fetchone()[0]
assert data["title"] == "Hello World" assert data["author"]["username"] == "carol" assert data["author"]["id"] == str(user_id)
def test_v_post_filters_by_is_published(self, db): """is_published column available for automatic-where filtering.""" with db.cursor() as cur: cur.execute(""" INSERT INTO tb_user (username, email, identifier) VALUES ('dave', 'dave@example.com', 'dave@example.com') RETURNING pk_user """) pk_user = cur.fetchone()[0]
cur.execute(""" INSERT INTO tb_post (fk_user, title, content, slug, identifier, is_published) VALUES (%s, 'Draft', 'Draft content', 'draft', 'draft', false), (%s, 'Published', 'Published content', 'published', 'published', true) """, (pk_user, pk_user))
cur.execute("SELECT COUNT(*) FROM v_post WHERE is_published = true") published_count = cur.fetchone()[0]
assert published_count >= 1import { describe, it, expect, beforeAll, afterAll } from 'vitest';import { sql } from 'fraiseql/testing';import { getTestDatabaseUrl } from './setup';import postgres from 'postgres';
let db: ReturnType<typeof postgres>;
beforeAll(() => { db = postgres(getTestDatabaseUrl());});
afterAll(async () => { await db.end();});
describe('v_user view', () => { it('returns a JSONB data column', async () => { const [{ id: userId }] = await db` INSERT INTO tb_user (username, email, identifier) VALUES ('alice', 'alice@example.com', 'alice@example.com') RETURNING id `; const [{ data }] = await db` SELECT data FROM v_user WHERE id = ${userId} `;
expect(data.email).toBe('alice@example.com'); expect(data.username).toBe('alice'); expect(data).toHaveProperty('id'); // Surrogate key must NOT be exposed expect(data).not.toHaveProperty('pk_user');
await db`ROLLBACK`; });
it('filters by id column', async () => { const [{ id: userId }] = await db` INSERT INTO tb_user (username, email, identifier) VALUES ('bob', 'bob@example.com', 'bob@example.com') RETURNING id `; const [{ count }] = await db` SELECT COUNT(*)::int AS count FROM v_user WHERE id = ${userId} `;
expect(count).toBe(1); await db`ROLLBACK`; });});
describe('v_post view', () => { it('nests author data from v_user', async () => { const [{ pk_user, id: userId }] = await db` INSERT INTO tb_user (username, email, identifier) VALUES ('carol', 'carol@example.com', 'carol@example.com') RETURNING pk_user, id `; const [{ id: postId }] = await db` INSERT INTO tb_post (fk_user, title, content, slug, identifier) VALUES (${pk_user}, 'Hello World', 'Content here', 'hello-world', 'hello-world') RETURNING id `; const [{ data }] = await db` SELECT data FROM v_post WHERE id = ${postId} `;
expect(data.title).toBe('Hello World'); expect(data.author.username).toBe('carol'); expect(data.author.id).toBe(String(userId));
await db`ROLLBACK`; });
it('exposes is_published for filtering', async () => { const [{ pk_user }] = await db` INSERT INTO tb_user (username, email, identifier) VALUES ('dave', 'dave@example.com', 'dave@example.com') RETURNING pk_user `; await db` INSERT INTO tb_post (fk_user, title, content, slug, identifier, is_published) VALUES (${pk_user}, 'Draft', 'Draft content', 'draft', 'draft', false), (${pk_user}, 'Published', 'Published content', 'published', 'published', true) `; const [{ count }] = await db` SELECT COUNT(*)::int AS count FROM v_post WHERE is_published = true `;
expect(count).toBeGreaterThanOrEqual(1); await db`ROLLBACK`; });});Testing PostgreSQL Functions
Section titled “Testing PostgreSQL Functions”import pytestimport psycopg
class TestCreateUser: def test_creates_user_and_returns_data(self, db): """fn_create_user inserts a row and returns via v_user.""" with db.cursor() as cur: cur.execute(""" SELECT data FROM fn_create_user( '{"username": "eve", "email": "eve@example.com"}'::jsonb ) """) user = cur.fetchone()[0]
assert user["username"] == "eve" assert user["email"] == "eve@example.com" assert "id" in user
def test_rejects_duplicate_email(self, db): """fn_create_user raises an error on duplicate email.""" with db.cursor() as cur: cur.execute(""" INSERT INTO tb_user (username, email, identifier) VALUES ('frank', 'frank@example.com', 'frank@example.com') """)
with pytest.raises(psycopg.errors.RaiseException) as exc_info: cur.execute(""" SELECT data FROM fn_create_user( '{"username": "frank2", "email": "frank@example.com"}'::jsonb ) """)
assert "already exists" in str(exc_info.value).lower()
def test_rejects_missing_required_fields(self, db): """fn_create_user raises when required fields are absent.""" with db.cursor() as cur: with pytest.raises(psycopg.errors.RaiseException): cur.execute(""" SELECT data FROM fn_create_user('{"username": "nomail"}'::jsonb) """)
class TestPublishPost: def test_sets_is_published(self, db): """fn_publish_post sets is_published = true.""" with db.cursor() as cur: # Setup cur.execute(""" INSERT INTO tb_user (username, email, identifier) VALUES ('grace', 'grace@example.com', 'grace@example.com') RETURNING pk_user, id """) pk_user, user_id = cur.fetchone()
cur.execute(""" INSERT INTO tb_post (fk_user, title, content, slug, identifier) VALUES (%s, 'Draft Post', 'Content', 'draft-post', 'draft-post') RETURNING id """, (pk_user,)) post_id = cur.fetchone()[0]
# Publish cur.execute( "SELECT data FROM fn_publish_post(%s::uuid, %s::uuid)", (post_id, user_id), ) post = cur.fetchone()[0]
assert post["isPublished"] is True
def test_rejects_wrong_author(self, db): """fn_publish_post raises if caller is not the author.""" with db.cursor() as cur: cur.execute(""" INSERT INTO tb_user (username, email, identifier) VALUES ('hank', 'hank@example.com', 'hank@example.com'), ('ivy', 'ivy@example.com', 'ivy@example.com') RETURNING pk_user, id """) rows = cur.fetchall() hank_pk, hank_id = rows[0] _ivy_pk, ivy_id = rows[1]
cur.execute(""" INSERT INTO tb_post (fk_user, title, content, slug, identifier) VALUES (%s, 'Hank Post', 'Content', 'hank-post', 'hank-post') RETURNING id """, (hank_pk,)) post_id = cur.fetchone()[0]
with pytest.raises(psycopg.errors.RaiseException): cur.execute( "SELECT data FROM fn_publish_post(%s::uuid, %s::uuid)", (post_id, ivy_id), # ivy tries to publish hank's post )import { describe, it, expect, beforeAll, afterAll } from 'vitest';import { getTestDatabaseUrl } from './setup';import postgres from 'postgres';
let db: ReturnType<typeof postgres>;
beforeAll(() => { db = postgres(getTestDatabaseUrl());});
afterAll(async () => { await db.end();});
describe('fn_create_user', () => { it('inserts a user and returns data via v_user', async () => { const [user] = await db` SELECT data FROM fn_create_user( '{"username": "eve", "email": "eve@example.com"}'::jsonb ) `;
expect(user.data.username).toBe('eve'); expect(user.data.email).toBe('eve@example.com'); expect(user.data).toHaveProperty('id');
await db`ROLLBACK`; });
it('raises an error on duplicate email', async () => { await db` INSERT INTO tb_user (username, email, identifier) VALUES ('frank', 'frank@example.com', 'frank@example.com') `;
await expect( db`SELECT data FROM fn_create_user( '{"username": "frank2", "email": "frank@example.com"}'::jsonb )` ).rejects.toThrow(/already exists/i);
await db`ROLLBACK`; });
it('raises when required fields are absent', async () => { await expect( db`SELECT data FROM fn_create_user('{"username": "nomail"}'::jsonb)` ).rejects.toThrow();
await db`ROLLBACK`; });});
describe('fn_publish_post', () => { it('sets is_published to true', async () => { const [{ pk_user, id: userId }] = await db` INSERT INTO tb_user (username, email, identifier) VALUES ('grace', 'grace@example.com', 'grace@example.com') RETURNING pk_user, id `; const [{ id: postId }] = await db` INSERT INTO tb_post (fk_user, title, content, slug, identifier) VALUES (${pk_user}, 'Draft Post', 'Content', 'draft-post', 'draft-post') RETURNING id `;
const [{ data: post }] = await db` SELECT data FROM fn_publish_post(${postId}::uuid, ${userId}::uuid) `;
expect(post.isPublished).toBe(true); await db`ROLLBACK`; });
it('raises if caller is not the author', async () => { const rows = await db` INSERT INTO tb_user (username, email, identifier) VALUES ('hank', 'hank@example.com', 'hank@example.com'), ('ivy', 'ivy@example.com', 'ivy@example.com') RETURNING pk_user, id `; const hankPk = rows[0].pk_user; const ivyId = rows[1].id;
const [{ id: postId }] = await db` INSERT INTO tb_post (fk_user, title, content, slug, identifier) VALUES (${hankPk}, 'Hank Post', 'Content', 'hank-post', 'hank-post') RETURNING id `;
// ivy tries to publish hank's post await expect( db`SELECT data FROM fn_publish_post(${postId}::uuid, ${ivyId}::uuid)` ).rejects.toThrow();
await db`ROLLBACK`; });});3. Integration Tests
Section titled “3. Integration Tests”End-to-end tests that send GraphQL requests to a running FraiseQL server:
import pytestimport httpximport subprocessimport time
@pytest.fixture(scope="session")def fraiseql_server(migrated_db): """Start a FraiseQL server against the test database.""" proc = subprocess.Popen([ "fraiseql", "serve", "--database-url", migrated_db, "--port", "18080", "--no-playground", ]) # Wait for server to be ready for _ in range(20): try: httpx.get("http://localhost:18080/health", timeout=1) break except httpx.ConnectError: time.sleep(0.5) yield proc proc.terminate() proc.wait()
@pytest.fixturedef client(fraiseql_server): return httpx.Client(base_url="http://localhost:18080", timeout=10)import pytest
class TestUserQueries: def test_list_users(self, client): response = client.post("/graphql", json={ "query": "query { users(limit: 10) { id username email } }" }) assert response.status_code == 200 data = response.json() assert "errors" not in data assert isinstance(data["data"]["users"], list)
def test_single_user_by_id(self, client, created_user): response = client.post("/graphql", json={ "query": """ query GetUser($id: ID!) { user(id: $id) { id username email } } """, "variables": {"id": created_user["id"]}, }) data = response.json() assert data["data"]["user"]["id"] == created_user["id"]
class TestUserMutations: def test_create_user(self, client): response = client.post("/graphql", json={ "query": """ mutation { createUser(input: { username: "newuser" email: "newuser@example.com" }) { id username email } } """ }) data = response.json() assert "errors" not in data user = data["data"]["createUser"] assert user["username"] == "newuser" assert user["id"] is not None
def test_create_user_duplicate_email(self, client, created_user): response = client.post("/graphql", json={ "query": """ mutation CreateUser($input: CreateUserInput!) { createUser(input: $input) { id } } """, "variables": { "input": { "username": "other", "email": created_user["email"], } }, }) data = response.json() assert "errors" in data assert "already exists" in data["errors"][0]["message"].lower()
class TestPostQueries: def test_post_includes_nested_author(self, client, created_post): response = client.post("/graphql", json={ "query": """ query { posts(limit: 10) { id title author { id username } } } """ }) data = response.json() posts = data["data"]["posts"] assert len(posts) > 0 assert posts[0]["author"]["username"] is not Noneimport { describe, it, expect, beforeAll, afterAll } from 'vitest';import { TestClient } from 'fraiseql/testing';import { createTestClient } from './setup';
let client: TestClient;
beforeAll(() => { client = createTestClient();});
afterAll(async () => { await client.close();});
describe('User queries', () => { it('lists users', async () => { const result = await client.query(` query { users(limit: 10) { id username email } } `);
expect(result.errors).toBeUndefined(); expect(Array.isArray(result.data.users)).toBe(true); });
it('fetches a single user by id', async () => { const created = await client.mutate(` mutation { createUser(input: { username: "querytest", email: "querytest@example.com" }) { id username } } `); const userId = created.data.createUser.id;
const result = await client.query( `query GetUser($id: ID!) { user(id: $id) { id username email } }`, { id: userId } );
expect(result.data.user.id).toBe(userId); });});
describe('User mutations', () => { it('creates a user', async () => { const result = await client.mutate(` mutation { createUser(input: { username: "newuser" email: "newuser@example.com" }) { id username email } } `);
expect(result.errors).toBeUndefined(); expect(result.data.createUser.username).toBe('newuser'); expect(result.data.createUser.id).toBeTruthy(); });
it('returns an error on duplicate email', async () => { await client.mutate(` mutation { createUser(input: { username: "orig", email: "dup@example.com" }) { id } } `);
const result = await client.mutate(` mutation CreateUser($input: CreateUserInput!) { createUser(input: $input) { id } } `, { input: { username: 'other', email: 'dup@example.com' } });
expect(result.errors).toBeDefined(); expect(result.errors![0].message.toLowerCase()).toContain('already exists'); });});
describe('Post queries', () => { it('includes nested author data', async () => { const result = await client.query(` query { posts(limit: 10) { id title author { id username } } } `);
const posts = result.data.posts; expect(posts.length).toBeGreaterThan(0); expect(posts[0].author.username).toBeTruthy(); });});Test Factories
Section titled “Test Factories”import factoryimport uuid
class UserFactory(factory.DictFactory): id = factory.LazyFunction(lambda: str(uuid.uuid4())) username = factory.Sequence(lambda n: f"user{n}") email = factory.Sequence(lambda n: f"user{n}@example.com") bio = factory.Faker("sentence")
class PostFactory(factory.DictFactory): id = factory.LazyFunction(lambda: str(uuid.uuid4())) title = factory.Faker("sentence", nb_words=5) content = factory.Faker("paragraph") slug = factory.Sequence(lambda n: f"post-{n}")from tests.factories import UserFactory
@pytest.fixturedef created_user(client): """Create a user via the API and return the created record.""" data = UserFactory() response = client.post("/graphql", json={ "query": """ mutation CreateUser($input: CreateUserInput!) { createUser(input: $input) { id username email } } """, "variables": {"input": {"username": data["username"], "email": data["email"]}}, }) return response.json()["data"]["createUser"]let userSeq = 0;let postSeq = 0;
export function userFactory(overrides: Partial<{ username: string; email: string; bio: string }> = {}) { const n = ++userSeq; return { username: `user${n}`, email: `user${n}@example.com`, bio: 'A test user biography.', ...overrides, };}
export function postFactory(overrides: Partial<{ title: string; content: string; slug: string }> = {}) { const n = ++postSeq; return { title: `Test Post ${n}`, content: 'Lorem ipsum dolor sit amet.', slug: `post-${n}`, ...overrides, };}import { TestClient } from 'fraiseql/testing';import { userFactory } from './factories';
export async function createTestUser(client: TestClient, overrides = {}) { const data = userFactory(overrides); const result = await client.mutate(` mutation CreateUser($input: CreateUserInput!) { createUser(input: $input) { id username email } } `, { input: { username: data.username, email: data.email } }); return result.data.createUser;}Running Tests
Section titled “Running Tests”# All testsuv run pytest
# Database tests only (fast)uv run pytest tests/test_views.py tests/test_functions.py
# With coverageuv run pytest --cov=schema --cov-report=html
# In paralleluv run pytest -n auto# All testsnpm test
# Database tests only (fast)npx vitest run tests/views.test.ts tests/functions.test.ts
# With coveragenpx vitest run --coverage
# Watch mode during developmentnpx vitestCI Pipeline
Section titled “CI Pipeline”name: Tests
on: [push, pull_request]
jobs: test: runs-on: ubuntu-latest
services: postgres: image: postgres:16-alpine env: POSTGRES_PASSWORD: postgres POSTGRES_DB: test options: >- --health-cmd pg_isready --health-interval 10s --health-timeout 5s --health-retries 5 ports: - 5432:5432
steps: - uses: actions/checkout@v4
- name: Install uv uses: astral-sh/setup-uv@v3
- name: Install dependencies run: uv sync
- name: Build database schema env: DATABASE_URL: postgresql://postgres:postgres@localhost:5432/test run: uv run fraiseql db build
- name: Run tests env: DATABASE_URL: postgresql://postgres:postgres@localhost:5432/test run: uv run pytest --cov --cov-report=xml
- uses: codecov/codecov-action@v4Next Steps
Section titled “Next Steps”Custom Queries
Observers
Troubleshooting