Testing

Strategies for testing applications that use Ryx.

In-Memory SQLite

The fastest way to test — no files, no cleanup:

import asyncio
import pytest
import ryx
from ryx import Model, CharField, IntField

class Post(Model):
    title  = CharField(max_length=200)
    views  = IntField(default=0)

@pytest.fixture(scope="session")
def event_loop():
    loop = asyncio.new_event_loop()
    yield loop
    loop.close()

@pytest.fixture(scope="session")
async def db(event_loop):
    """Setup database once per session."""
    await ryx.setup("sqlite::memory:")
    from ryx.migrations import MigrationRunner
    await MigrationRunner([Post]).migrate()

@pytest.fixture(autouse=True)
async def clean_db(db):
    """Clean tables before each test."""
    await ryx.executor_helpers.raw_execute("DELETE FROM posts")

@pytest.mark.asyncio
async def test_create_post():
    post = await Post.objects.create(title="Test", views=10)
    assert post.pk is not None
    assert post.title == "Test"

@pytest.mark.asyncio
async def test_filter_posts():
    await Post.objects.create(title="A", views=100)
    await Post.objects.create(title="B", views=0)

    popular = await Post.objects.filter(views__gt=50)
    assert len(popular) == 1
    assert popular[0].title == "A"

Test Fixtures

Define test-specific models in conftest.py:

# conftest.py
import pytest
import ryx
from ryx import Model, CharField, ForeignKey

class Author(Model):
    name = CharField(max_length=100)

class Post(Model):
    title  = CharField(max_length=200)
    author = ForeignKey(Author, on_delete="CASCADE")

@pytest.fixture(scope="session")
async def db():
    await ryx.setup("sqlite::memory:")
    from ryx.migrations import MigrationRunner
    await MigrationRunner([Author, Post]).migrate()

@pytest.fixture
async def author(db):
    return await Author.objects.create(name="Test Author")

@pytest.fixture
async def post(author):
    return await Post.objects.create(title="Test Post", author=author)

Testing with Transactions

Wrap each test in a transaction that rolls back:

@pytest.fixture(autouse=True)
async def transactional_db(db):
    """Wrap each test in a transaction that rolls back."""
    async with ryx.transaction() as tx:
        yield
        await tx.rollback()  # Undo everything

Mocking the Rust Core

For unit tests that don't need a database, mock ryx_core:

from unittest.mock import AsyncMock, patch

@patch("ryx_core.QueryBuilder")
def test_queryset_builds_correct_query(mock_builder):
    mock_builder.fetch_all = AsyncMock(return_value=[
        {"id": 1, "title": "Mocked", "views": 100}
    ])

    # Your test code here

Testing Signals

@pytest.mark.asyncio
async def test_post_save_signal():
    call_count = 0

    async def handler(sender, instance, created, **kwargs):
        nonlocal call_count
        call_count += 1

    post_save.connect(handler, sender=Post)
    try:
        await Post.objects.create(title="Signal test")
        assert call_count == 1
    finally:
        post_save.disconnect(handler, sender=Post)

Testing Validation

@pytest.mark.asyncio
async def test_validation_errors():
    post = Post(title="x")  # Too short if min_length=5
    try:
        await post.full_clean()
        assert False, "Should have raised"
    except ValidationError as e:
        assert "title" in e.errors

Running Tests

# With pytest
pytest tests/ -v --asyncio-mode=auto

# With the test runner
python test.py --all

Next Steps

→ Deployment — Production setup