Testing Guide¶
FenLiu uses tryke as its test framework. Tests use a describe/test/expect/fixture style with dependency injection via Depends.
Running Tests¶
# All tests
uv run tryke test
# Single file
uv run tryke test tests/test_spam_scoring.py
# Via nox (also runs linting, type-checking, complexipy)
uv run nox -s fenliu-tryke
There is no watch mode or coverage report command — run the suite directly.
Test Structure¶
Tests are organised by area of concern:
test_spam_scoring.py— Spam scoring rulestest_curated.py— Curated queue API (ack/nack/error/requeue)test_api.py— General REST API endpointstest_reblog_controls.py— Blocklist and reblog filter endpointstest_topical.py— Topical-tag delivery logictest_liveviews.py— LiveView render/event handlerstest_main.py— App wiring, middleware, stats/post-details pagestest_schemas.py— Pydantic schema validationtest_display.py—process_post_for_display/process_stream_for_displaytest_middleware.py— API key and UI auth middlewaretest_ui_auth.py— UI authentication flowstest_api_keys.py— API key managementtest_feedback_stats.py— Feedback stats CLItest_pattern_blocking.py— Blocklist pattern matchingtest_training.py— ML training pipeline
Writing Tests¶
File header¶
Every test file must set the three env-var guards before any fenliu import, or parallel test runs will fail:
import os
os.environ["DATABASE_URL"] = "sqlite:///:memory:"
os.environ.setdefault("UI_AUTH_ENABLED", "false")
os.environ.setdefault("SECRET_KEY", "test-secret-key-not-for-production")
from tryke import describe, expect, fixture, test, Depends
from fenliu.models import Post
Basic test shape¶
from tryke import describe, expect, test
with describe("ContentLengthRule"):
@test("short content raises spam score")
def test_short_content() -> None:
rule = ContentLengthRule()
result = rule.apply({"content": "hi"})
expect(result, "score").to_be_greater_than(0)
@test("normal content scores zero")
def test_normal_content() -> None:
rule = ContentLengthRule()
result = rule.apply({"content": "A normal length post about Python."})
expect(result, "score").to_equal(0)
describe is a context manager that groups tests. test decorates test functions. expect(value, label) returns a chainable assertion object — .to_equal(), .to_be_truthy(), .to_contain(), .to_raise(), .to_be_greater_than(), etc.
Fixtures and dependency injection¶
Fixtures are plain functions decorated with @fixture. Tests receive them via Annotated[T, Depends(fixture_fn)]:
from collections.abc import Generator
from typing import Annotated
from tryke import Depends, fixture
from sqlalchemy import create_engine
from sqlalchemy.orm import Session, sessionmaker
from fenliu.database import Base
@fixture
def db_session() -> Generator[Session, None, None]:
eng = create_engine(
"sqlite:///file:test?mode=memory&cache=shared&uri=true",
connect_args={"check_same_thread": False},
)
Base.metadata.create_all(bind=eng)
session = sessionmaker(bind=eng)()
try:
yield session
finally:
session.close()
Base.metadata.drop_all(bind=eng)
eng.dispose()
@test("stream is persisted")
def test_stream_persisted(
db: Annotated[Session, Depends(db_session)],
) -> None:
stream = HashtagStream(hashtag="python", instance="mastodon.social", active=True)
db.add(stream)
db.commit()
expect(stream.id, "id assigned").to_be_truthy()
Fixtures can depend on other fixtures the same way.
Note
Tryke does not resolve cross-module Depends() references. Each test file defines its own fixtures locally.
HTTP endpoint tests¶
Use starlette.testclient.TestClient (not FastAPI's):
from starlette.testclient import TestClient
from fenliu.main import app
@fixture
def client(
app_with_test_db: Annotated[ASGIApp, Depends(app_with_test_db)],
db_session: Annotated[Session, Depends(db_session)],
) -> Generator[TestClient, None, None]:
with TestClient(app_with_test_db) as c:
yield c
with describe("GetNext"):
@test("empty queue returns 204")
def test_empty_queue(
client: Annotated[TestClient, Depends(client)],
) -> None:
response = client.get("/api/v1/curated/next")
expect(response.status_code, "status").to_equal(204)
The app_with_test_db fixture patches get_db across all modules so the test session is used throughout the request lifecycle. See tests/test_curated.py for the canonical implementation.
Mocking external calls¶
Use unittest.mock.patch as a context manager inside the test body:
from unittest.mock import AsyncMock, patch
@test("fetch stores posts in database")
def test_fetch_stores_posts(
client: Annotated[TestClient, Depends(client)],
stream: Annotated[HashtagStream, Depends(stream)],
) -> None:
mock_posts = [{"id": "1", "content": "Hello", ...}]
with patch("fenliu.services.fediverse.FediverseClient.fetch_hashtag_timeline") as m:
m.return_value = mock_posts
response = client.post(f"/api/v1/hashtags/{stream.id}/fetch")
expect(response.status_code, "status").to_equal(200)
Test env-var guards¶
The three guards at the top of each file are load-order protection — they ensure the settings singleton is configured before any module-level code in fenliu runs. Missing even one causes intermittent failures when tests run in parallel.
Continuous Integration¶
Tests run automatically on every PR push via .woodpecker/. The fenliu pipeline runs:
ruff checkruff format --checkty checkcomplexipytryke test
Before pushing, run the same checks locally:
uv run nox -t fenliu
Or individually:
uv run --directory packages/fenliu ruff check .
uv run --directory packages/fenliu ty check .
uv run --directory packages/fenliu tryke test
Next Steps¶
- Local Setup — Set up development environment
- Contributing — Contribution guidelines
- System Design — Architecture overview