Skip to content

Testing

Vectis uses pytest with pytest-asyncio. Tests run against a real PostgreSQL database (not SQLite) for production parity with JSONB, BigInteger PKs, and async drivers.

Running Tests

cd vectis/backend
pytest                            # all tests
pytest tests/test_strategy.py -v  # single file
pytest -k "test_resolve"          # pattern match

Test Database Setup

The conftest.py fixture creates all tables at session start and drops them after. A dedicated vectis_test database is configured via DATABASE_URL env var. reset_engine() clears the app's singleton so it picks up the test URL.

Note

Set DATABASE_URL to a separate vectis_test database. The fixture runs Base.metadata.create_all at session start and drop_all at teardown.

Shared Fixtures

@pytest.fixture
def event_bus():
    return EventBus()

@pytest.fixture
def strategy_resolver():
    return StrategyResolver()

@pytest.fixture
def staff_context():
    return RequestContext(
        user_id=1, user_type="staff", is_authenticated=True,
        is_staff=True, permissions=["*"],
    )

Testing Services

@pytest.mark.asyncio
async def test_create_product():
    factory = get_session_factory()
    async with factory() as session:
        svc = ProductService(session)
        product = await svc.create(name="Widget", slug="widget", status="draft")
        assert product.id is not None
        assert product.name == "Widget"
        await session.commit()

Testing Strategies

class MockTax:
    pass

def test_register_and_resolve(strategy_resolver):
    impl = MockTax()
    strategy_resolver.register(MockTax, impl, name="default")
    assert strategy_resolver.resolve(MockTax) is impl

def test_resolve_all(strategy_resolver):
    a, b = MockTax(), MockTax()
    strategy_resolver.register(MockTax, a, name="state")
    strategy_resolver.register(MockTax, b, name="county")
    assert len(strategy_resolver.resolve_all(MockTax)) == 2

Testing the EventBus

@pytest.mark.asyncio
async def test_priority_ordering(event_bus):
    order = []
    async def high(e): order.append("high")
    async def low(e): order.append("low")
    event_bus.subscribe("test", low, priority=100)
    event_bus.subscribe("test", high, priority=10)
    await event_bus.emit(Event(type="test"))
    assert order == ["high", "low"]

Integration Tests

@pytest.mark.asyncio
async def test_health_check(client):
    resp = await client.post("/graphql", json={"query": "{ health }"})
    assert resp.status_code == 200
    assert resp.json()["data"]["health"] == "Vectis Commerce API is healthy"

Warning

Integration tests share the database. Clean up test data or use unique identifiers to avoid collisions.

Test Organization

tests/
├── conftest.py              # Shared fixtures
├── test_strategy.py         # Strategy resolver unit tests
├── test_events.py           # EventBus unit tests
├── test_pricing.py          # Pricing module tests
├── test_cart_order_flow.py  # Cart-to-order integration
├── test_graphql_api.py      # GraphQL resolver tests
└── test_rbac.py             # Permission and role tests