Skip to content
Kirak Docs

Testing

How to test Kirak applications and how the runtime’s own test suite is organized.

Test Setup

Section titled “Test Setup”

Install test dependencies:

Terminal window
pip install pytest pytest-asyncio

Configure pytest for async tests in pyproject.toml or pytest.ini:

pyproject.toml
[tool.pytest.ini_options]
asyncio_mode = "auto"

Or pytest.ini:

[pytest]
asyncio_mode = auto

Environment Fixtures

Section titled “Environment Fixtures”

Kirak reads configuration from environment variables at import time. Use monkeypatch to inject test values without touching real .env files.

The pattern used in Kirak’s own test suite (tests/conftest.py):

import pytest


@pytest.fixture(autouse=True)
def set_test_env(monkeypatch):
    monkeypatch.setenv("KIRAK_AUTH_JWT_SECRET_KEY", "test-secret-key-must-be-at-least-32-bytes!!")
    monkeypatch.setenv("KIRAK_AUTH_VERIFICATION_KEY", "")

autouse=True applies this fixture to every test in the session automatically. Settings that are not env vars – access-token TTL, email-verification-required, and most other auth behavior toggles – live in kirak.json/the manifest, not KIRAK_AUTH_* env vars. Override those via a mock config object (see MockConfig in tests/conftest.py), not monkeypatch.setenv.

Unit Testing Operations

Section titled “Unit Testing Operations”

Kirak operations are plain async functions. Test them directly without spinning up a server.

Pattern: mock the DB and module instance

# tests/conftest.py -- reusable mocks (from kirak's own test suite)


class MockDialect:
    def placeholder(self, index):
        return "%s"


class MockCursor:
    def __init__(self, fetchone_result=None):
        self.fetchone_result = fetchone_result
        self.executed = []


    async def execute(self, query, params=None):
        self.executed.append((query, params))


    async def fetchone(self):
        return self.fetchone_result


    async def __aenter__(self):
        return self


    async def __aexit__(self, *args):
        pass


class MockConnection:
    def __init__(self, cursor):
        self._cursor = cursor


    def cursor(self):
        return self._cursor


    def transaction(self):
        return MockTransaction()


    async def __aenter__(self):
        return self


    async def __aexit__(self, *args):
        pass


class MockTransaction:
    async def __aenter__(self):
        return None


    async def __aexit__(self, *args):
        pass


class MockDB:
    dialect = MockDialect()


    def __init__(self, fetchone_result=None):
        self._cursor = MockCursor(fetchone_result)
        self._conn = MockConnection(self._cursor)


    def acquire(self):
        return self._conn


@pytest.fixture()
def mock_db():
    return MockDB()

Example: testing the login operation

import pytest
from kirak.auth.operations.login import _login
from kirak.core.exceptions import ValidationError, AuthenticationError




class TestLogin:
    async def test_missing_email_raises_validation_error(self, mock_auth):
        # _login raises rather than returning an error dict for missing input.
        with pytest.raises(ValidationError, match="Email and password are required"):
            await _login(mock_auth, "users", {"password": "password123"})


    async def test_nonexistent_user_raises_authentication_error(self, mock_auth):
        mock_auth.kirak.configure("execute_query", None)
        with pytest.raises(AuthenticationError):
            await _login(mock_auth, "users", {
                "email": "ghost@example.com",
                "password": "password123",
            })


    async def test_valid_login_returns_tokens(self, mock_auth, sample_user):
        # Queue two sequential DB results: user lookup, MFA check
        mock_auth.kirak.configure("execute_query", [sample_user, None])
        result = await _login(mock_auth, "users", {
            "email": "test@example.com",
            "password": "password123",
        })
        assert result["status"] == "success"
        assert "accessToken" in result["data"]
        assert "refreshToken" in result["data"]
        assert "password" not in result["data"]["user"]

Operations raise KirakException subclasses (ValidationError, AuthenticationError, PermissionDenied, …) for failure cases – they don’t return an error dict. The dict-with-statusCode/status/error shape only appears on the HTTP response, built by create_error_response() in kirak/core/error_response.py from whatever exception propagated. Test the exception directly at the operation level; test the envelope shape at the HTTP/integration level.

Testing bcrypt Without Slowdown

Section titled “Testing bcrypt Without Slowdown”

bcrypt at its default cost (13 rounds) is intentionally slow – it will make tests take minutes if used naively. Use cost 4 for tests (same algorithm, fast):

import bcrypt
import pytest


_TEST_PASSWORD = "password123"
_TEST_BCRYPT_HASH = bcrypt.hashpw(
    _TEST_PASSWORD.encode(), bcrypt.gensalt(rounds=4)
).decode()


@pytest.fixture()
def test_password():
    return _TEST_PASSWORD


@pytest.fixture()
def test_bcrypt_hash():
    return _TEST_BCRYPT_HASH


@pytest.fixture()
def sample_user(test_bcrypt_hash):
    return {
        "id": 1,
        "email": "test@example.com",
        "password": test_bcrypt_hash,
        "role": "user",
        "is_active": True,
        "is_verified": True,
    }

Testing Hooks

Section titled “Testing Hooks”

Test hooks as standalone async functions:

async def test_after_create_hook_enriches_result():
    from myapp.hooks import add_order_total


    mock_result = {
        "statusCode": 200,
        "status": "success",
        "message": "orders created successfully",
        "data": {"id": 42, "status": "pending"},
    }
    enriched = await add_order_total(mock_result)
    assert "total" in enriched["data"]

Test hook registration in isolation using a bare BaseModule:

from kirak.core.base_module import BaseModule


class TestModule(BaseModule):
    def __init__(self):
        super().__init__()


    def _setup_module(self):
        self.slug = "test"
        self._default_model = "test"


async def test_hook_fires():
    mod = TestModule()
    received = []


    @mod.hook("after_create")
    async def capture(result):
        received.append(result)
        return result


    result = {"status": "success", "data": {"id": 1}}
    # run_hooks requires current_user as the 4th positional arg, even if unused by the hook.
    await mod.run_hooks("test", "after_create", result, {"user_id": 1, "role": "user"})
    assert received == [result]

Testing Webhook Handlers

Section titled “Testing Webhook Handlers”

Mock the provider’s signature verification:

from unittest.mock import patch
import pytest
from kirak.core.exceptions import KirakException


stripe = pytest.importorskip("stripe")


async def test_valid_stripe_webhook(stripe_provider):
    fake_event = {
        "type": "checkout.session.completed",
        "data": {"object": {"id": "cs_test_123"}},
    }
    with patch("stripe.Webhook.construct_event", return_value=fake_event):
        result = await stripe_provider._verify_webhook(b'{"type":"test"}', "valid_sig")


    assert result["event"] == fake_event


async def test_invalid_stripe_signature(stripe_provider):
    with patch(
        "stripe.Webhook.construct_event",
        side_effect=stripe.error.SignatureVerificationError("bad sig", "sig_header"),
    ):
        with pytest.raises(KirakException) as exc_info:
            await stripe_provider._verify_webhook(b'{"type":"test"}', "invalid_sig")


    assert exc_info.value.status_code == 400
    assert exc_info.value.code == "INVALID_WEBHOOK_SIGNATURE"

Use pytest.importorskip("stripe") to skip the test automatically if the stripe package isn’t installed.

Integration Tests

Section titled “Integration Tests”

Integration tests hit a real database. Kirak’s own integration tests require KIRAK_INTEGRATION=1 and live MySQL/PostgreSQL instances.

Start the test databases:

Terminal window
docker compose -f tests/docker-compose.yml up -d

The databases run on non-standard ports to avoid conflicts:

  • MySQL: 127.0.0.1:3307
  • PostgreSQL: 127.0.0.1:5433

Run integration tests:

Terminal window
KIRAK_INTEGRATION=1 pytest tests/test_core/ -v -m integration

Unit tests (no DB required):

Terminal window
pytest tests/ -v -m "not integration"

Environment variables for integration tests:

Variable Default
TEST_MYSQL_HOST 127.0.0.1
TEST_MYSQL_PORT 3307
TEST_MYSQL_USER kirak
TEST_MYSQL_PASSWORD kirak_test_pw
TEST_MYSQL_DB kirak_test
TEST_POSTGRES_HOST 127.0.0.1
TEST_POSTGRES_PORT 5433
TEST_POSTGRES_USER kirak
TEST_POSTGRES_PASSWORD kirak_test_pw
TEST_POSTGRES_DB kirak_test

Testing with CrudEngine Directly

Section titled “Testing with CrudEngine Directly”

For integration tests, create a CrudEngine with a real DB pool:

from kirak.core.engine import CrudEngine
from kirak.core.database.factory import create_database_driver
from kirak.core.context import set_user_context, reset_user_context


@pytest_asyncio.fixture(scope="module")
async def mysql_engine():
    # create_database_driver takes db_type plus a single config dict -- not individual kwargs.
    db = create_database_driver("mysql", {
        "host": "127.0.0.1",
        "port": 3307,
        "user": "kirak",
        "password": "kirak_test_pw",
        "database": "kirak_test",
        "pool_min": 1,
        "pool_max": 2,
    })
    await db.connect()
    yield CrudEngine(models=MY_TEST_MODELS, db_pool=db, auth=None)
    await db.disconnect()




async def test_create_and_fetch(mysql_engine):
    # dispatch() resolves current_user from context, not from the params dict --
    # set it before calling, the same way script-mode/hook calls do.
    token = set_user_context({"user_id": 999, "role": "admin"})
    try:
        result = await mysql_engine.create("items", {
            "data": {"name": "Widget", "slug": "widget"},
        })
        assert result["status"] == "success"
        item_id = result["data"]["id"]


        # fetch/search/count/exists read filters from "query", not "filter".
        fetched = await mysql_engine.fetch("items", {
            "query": {"id": item_id},
        })
        assert fetched["data"][0]["name"] == "Widget"
    finally:
        reset_user_context(token)

Test Structure Conventions

Section titled “Test Structure Conventions”

Follow Kirak’s own test layout:

tests/
  conftest.py              # shared fixtures (mock_db, mock_auth, sample_user, env)
  test_auth/
    test_login.py          # unit tests per operation
    test_register.py
    test_jwt.py
    test_token_blacklist.py
    test_rate_limit.py
  test_core/
    conftest.py            # integration fixtures (mysql_db, postgres_db, engines)
    test_crud_integration.py
    test_graphql.py
  test_payments/
    test_stripe_webhook.py
    test_razorpay_webhook.py
  test_scheduler/
    test_database_backend.py
  test_storage/
    test_local.py
  test_admin/
  test_ai/
  test_mcp/
  • One test file per module/operation.
  • Unit tests that mock the DB in tests/conftest.py.
  • Integration tests gated behind KIRAK_INTEGRATION=1 and marked with @pytest.mark.integration.
  • Use class grouping (class TestLoginValidation, class TestLoginSuccess) for related test cases.