Testing

FraiseQL applications have three distinct testing layers, each validating a different concern:

Layer	What it tests	Speed
Schema export tests	Type definitions produce valid schema JSON	Instant
Database tests	SQL views and PostgreSQL functions	Fast (real DB, transaction rollback)
Integration tests	End-to-end GraphQL API behaviour	Slower (real server)

Understanding the FraiseQL Architecture

FraiseQL’s Python SDK is compile-time only. The workflow is:

Define your schema with @fraiseql.type, @fraiseql.query, @fraiseql.mutation
Export the schema: fraiseql.export_schema("schema.json") (writes JSON to disk)
Compile: fraiseql compile produces schema.compiled.json
Run the Rust server pointing to your compiled schema and PostgreSQL

There is no runtime Python execution, no Python test client, and no Python server. Tests either talk directly to PostgreSQL (database tests) or send HTTP requests to a running Rust server (integration tests).

Test Setup

Dependencies

# Install test dependencies
uv add --dev pytest pytest-asyncio httpx testcontainers factory-boy psycopg

[tool.pytest.ini_options]
asyncio_mode = "auto"
testpaths = ["tests"]

[dependency-groups]
dev = [
    "pytest>=8.0",
    "pytest-asyncio>=0.23",
    "httpx>=0.27",
    "testcontainers[postgres]>=4.0",
    "factory-boy>=3.3",
    "psycopg[binary]>=3.1",
]

Test Database Fixture

import pytest
import psycopg
import subprocess
from 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.

    Replace the confiture command with however your project applies DDL.
    A common approach: run confiture (FraiseQL's migration tool) or
    execute your SQL migration scripts directly.
    """
    subprocess.run(
        ["confiture", "build", "--env", "test"],
        env={"DATABASE_URL": db_url, **__import__("os").environ},
        check=True,
    )
    return db_url

@pytest.fixture
def 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()

1. Schema Export Tests

Verify that your type definitions produce valid schema JSON. Because FraiseQL is compile-time only, schema tests work by calling fraiseql.export_schema() and inspecting the resulting JSON file.

import json
import fraiseql
import pytest

# Import your schema module so decorators register types/queries
import schema  # noqa: F401  — side-effects register types with SchemaRegistry


def test_schema_exports_without_error(tmp_path):
    """export_schema() should write valid JSON without raising."""
    output = tmp_path / "schema.json"
    fraiseql.export_schema(str(output))
    assert output.exists()

    with output.open() as f:
        data = json.load(f)

    # Top-level keys must be present
    assert "types" in data
    assert "queries" in data
    assert "mutations" in data


def test_user_type_is_registered(tmp_path):
    """User type must appear in the exported schema."""
    output = tmp_path / "schema.json"
    fraiseql.export_schema(str(output))

    with output.open() as f:
        data = json.load(f)

    type_names = {t["name"] for t in data["types"]}
    assert "User" in type_names


def test_user_type_has_expected_fields(tmp_path):
    """User type must declare id, email, and username fields."""
    output = tmp_path / "schema.json"
    fraiseql.export_schema(str(output))

    with output.open() as f:
        data = json.load(f)

    user_type = next(t for t in data["types"] if t["name"] == "User")
    field_names = {f["name"] for f in user_type["fields"]}

    assert "id" in field_names
    assert "email" in field_names
    assert "username" in field_names


def test_mutations_are_registered(tmp_path):
    """Core mutations must appear in the exported schema."""
    output = tmp_path / "schema.json"
    fraiseql.export_schema(str(output))

    with output.open() as f:
        data = json.load(f)

    mutation_names = {m["name"] for m in data["mutations"]}
    assert "createPost" in mutation_names
    assert "publishPost" in mutation_names


def test_queries_have_sql_source(tmp_path):
    """Every query must declare a sql_source (view name)."""
    output = tmp_path / "schema.json"
    fraiseql.export_schema(str(output))

    with output.open() as f:
        data = json.load(f)

    for query in data["queries"]:
        assert "sql_source" in query, (
            f"Query {query['name']!r} is missing sql_source — "
            "add sql_source= to @fraiseql.query"
        )

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

import pytest
import 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."""
        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 >= 1

Testing PostgreSQL Functions

Mutation functions return a mutation_response composite type. Test the status and entity fields:

import pytest
import psycopg


class TestCreateUser:
    def test_creates_user_and_returns_success(self, db):
        """fn_create_user inserts a row and returns status='success'."""
        with db.cursor() as cur:
            cur.execute("""
                SELECT status, entity_id, entity
                FROM fn_create_user('{"username": "eve", "email": "eve@example.com"}'::jsonb)
            """)
            row = cur.fetchone()
            status, entity_id, entity = row

        assert status == "success"
        assert entity_id is not None
        assert entity["username"] == "eve"
        assert entity["email"] == "eve@example.com"

    def test_rejects_duplicate_email(self, db):
        """fn_create_user returns conflict status 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')
            """)

            cur.execute("""
                SELECT status FROM fn_create_user(
                    '{"username": "frank2", "email": "frank@example.com"}'::jsonb
                )
            """)
            status = cur.fetchone()[0]

        assert status.startswith("conflict:")

    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 status 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 status, entity FROM fn_publish_post(%s::uuid, %s::uuid)",
                (post_id, user_id),
            )
            status, entity = cur.fetchone()

        assert status == "success"
        assert entity["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 status FROM fn_publish_post(%s::uuid, %s::uuid)",
                    (post_id, ivy_id),  # ivy tries to publish hank's post
                )

3. Integration Tests

End-to-end tests send GraphQL requests to a running FraiseQL Rust server. The server is started against a test database with your compiled schema.

Export schema JSON: fraiseql.export_schema("schema.json")
Compile: fraiseql compile → schema.compiled.json
Start the Rust server: fraiseql-server --schema schema.compiled.json
Run tests via HTTP with httpx

import pytest
import httpx
import subprocess
import time
import os

@pytest.fixture(scope="session")
def fraiseql_server(migrated_db):
    """Start a FraiseQL server against the test database.

    Assumes schema.compiled.json has already been built
    (e.g., via fraiseql compile in a CI step or conftest setup).
    """
    proc = subprocess.Popen(
        ["fraiseql-server", "--schema", "schema.compiled.json", "--port", "18080"],
        env={**os.environ, "DATABASE_URL": migrated_db},
    )
    # Wait for server to be ready
    for _ in range(30):
        try:
            httpx.get("http://localhost:18080/health", timeout=1)
            break
        except httpx.ConnectError:
            time.sleep(0.5)
    yield proc
    proc.terminate()
    proc.wait()

@pytest.fixture
def 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 "errors" not in data
        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


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()
        assert "errors" not in data
        posts = data["data"]["posts"]
        assert len(posts) > 0
        assert posts[0]["author"]["username"] is not None

Test Factories

import factory
import uuid

class UserFactory(factory.DictFactory):
    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):
    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.fixture
def 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"]

Running Tests

# All tests
uv run pytest

# Database tests only (fast, no server required)
uv run pytest tests/test_views.py tests/test_functions.py

# Schema export tests only
uv run pytest tests/test_schema.py

# With coverage
uv run pytest --cov=schema --cov-report=html

# In parallel
uv run pytest -n auto

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: Apply database schema
        env:
          DATABASE_URL: postgresql://postgres:postgres@localhost:5432/test
        # Replace with your project's DDL migration command
        run: confiture build --env ci

      - name: Export and compile schema
        run: |
          uv run python schema.py   # runs export_schema("schema.json")
          fraiseql compile          # produces schema.compiled.json

      - name: Run tests
        env:
          DATABASE_URL: postgresql://postgres:postgres@localhost:5432/test
        run: uv run pytest --cov --cov-report=xml

      - uses: codecov/codecov-action@v4

Testing REST and gRPC Endpoints

Integration tests should cover all active transports.

REST testing with curl (the default path is /rest/v1; configure with [rest] path):

# Test list endpoint
curl -s http://localhost:8080/rest/v1/posts | jq .

# Test with auth
curl -s http://localhost:8080/rest/v1/posts \
  -H "Authorization: Bearer $TEST_TOKEN"

# Test POST
curl -s -X POST http://localhost:8080/rest/v1/posts \
  -H "Content-Type: application/json" \
  -d '{"title": "Test Post", "authorId": "..."}' | jq .

REST testing with OpenAPI:

# Retrieve OpenAPI spec
curl http://localhost:8080/rest/openapi.json > openapi.json
# Use with any OpenAPI test generator (Schemathesis, Dredd, etc.)

gRPC testing with grpcurl:

# List available services
grpcurl -plaintext localhost:50052 list

# Test a method
grpcurl -plaintext \
  -d '{"id": "test-id"}' \
  localhost:50052 \
  BlogService/GetPost

Proto generation for client testing:

fraiseql compile --output-proto schema.proto
# Use schema.proto to generate typed client code for testing

Integration tests should assert behavior on all transports — the same auth rules, rate limits, and error codes should apply regardless of transport.

Next Steps

Schema Design

Design schemas that are easy to test

Custom Queries

Test filtering and pagination

Observers

Test observer side effects

Troubleshooting

Debug failing tests