tests/conftest.py
Rama main · réplica de lectura del repositorio original.
"""Shared fixtures. Three things here are load-bearing.
1. **Environment before imports.** `appcore.config` reads `APP_CONFIG_FILE` at
module import time, so the env vars are set at the top of this file.
2. **Schema from the real migration chain**, built once per session and copied
per test. `Base.metadata.create_all` would test a schema production never
has.
3. **Cache clearing.** `queries` caches by portfolio id against a fingerprint,
every test gets id 1, and two tests can share a fingerprint — so
`_clear_caches` is autouse. Do not remove it. `_discard_wizard_draft` is
autouse for the same reason: the draft is process-global, and the app
refuses pages while one is in progress.
SQLite by default; Postgres is an untested claim otherwise:
STOCKTAKE_TEST_DB=postgres PGHOST=localhost PGPORT=5432 \
PGDATABASE=stocktake_test PGUSER=postgres PGPASSWORD=postgres \
pytest tests/
Isolation differs by backend, so assert on answers, never on a backend's
representation — `appcore.ensure_utc` is what papers over the difference.
"""
from __future__ import annotations
import os
import shutil
import sys
import tempfile
from pathlib import Path
# --- environment, before any appcore/app import (see note 1 above) ----------- #
TESTS_DIR = Path(__file__).parent
APP_ROOT = TESTS_DIR.parent
_TMP_ROOT = Path(tempfile.mkdtemp(prefix="portfolio-tests-"))
BACKEND = os.environ.get("STOCKTAKE_TEST_DB", "sqlite").lower()
if BACKEND not in ("sqlite", "postgres"):
raise RuntimeError(f"STOCKTAKE_TEST_DB must be sqlite or postgres, got {BACKEND!r}")
os.environ["APP_CONFIG_FILE"] = str(TESTS_DIR / "config.test.yaml")
# A broker CSV waiting for its commit goes under /scratch, which is the
# container's: a laptop has no such folder to write to.
os.environ["APP_IMPORTS__STAGING_DIR"] = str(_TMP_ROOT / "staged")
if BACKEND == "sqlite":
os.environ["APP_DATABASE__PATH"] = str(_TMP_ROOT / "app.db")
os.environ.pop("APP_DATABASE__TYPE", None)
else:
# app.main reads these at import time for its own session factory.
os.environ["APP_DATABASE__TYPE"] = "postgres"
os.environ["APP_DATABASE__HOST"] = os.environ.get("PGHOST", "localhost")
os.environ["APP_DATABASE__PORT"] = os.environ.get("PGPORT", "5432")
os.environ["APP_DATABASE__NAME"] = os.environ.get("PGDATABASE", "stocktake_test")
os.environ["APP_DATABASE__USER"] = os.environ.get("PGUSER", "postgres")
os.environ["APP_DATABASE__PASSWORD"] = os.environ.get("PGPASSWORD", "postgres")
# `app` is a package under apps/portfolio; tests may be invoked from anywhere.
sys.path.insert(0, str(APP_ROOT))
sys.path.insert(0, str(TESTS_DIR))
import pytest # noqa: E402
from hypothesis import HealthCheck # noqa: E402
from hypothesis import settings as generated # noqa: E402
from sqlalchemy import text # noqa: E402
from app import pricefeed, providers, queries, tenancy # noqa: E402
from app.models import Portfolio, PortfolioMember, User # noqa: E402
from appcore import make_session_factory, upgrade_to_head # noqa: E402
from appcore.config import DatabaseSettings # noqa: E402
from appcore.db import make_engine # noqa: E402
# Generated cases (the tests using Hypothesis) are the same every run, so a red
# build can be reproduced from its log. `HYPOTHESIS_PROFILE=explore` searches
# afresh, with far more of them. Function-scoped fixtures are fine here: each
# case rolls back what it wrote rather than relying on a fresh fixture.
_RELAXED = [HealthCheck.function_scoped_fixture, HealthCheck.too_slow,
HealthCheck.data_too_large]
generated.register_profile("default", derandomize=True, database=None, deadline=None,
max_examples=40, suppress_health_check=_RELAXED)
generated.register_profile("explore", deadline=None, max_examples=1500,
suppress_health_check=_RELAXED)
generated.load_profile(os.environ.get("HYPOTHESIS_PROFILE", "default"))
# A frozen "today" for every test that depends on the current date. Chosen so
# the reference portfolio (which starts 2019) fills the 1d/1m/6m/1y/3y/5y
# performance windows but NOT 10y/20y — the "record doesn't reach back that
# far" path gets exercised too.
FROZEN_TODAY = "2026-08-02"
# --------------------------------------------------------------------------- #
# Database
# --------------------------------------------------------------------------- #
def _pg_settings() -> DatabaseSettings:
return DatabaseSettings(
type="postgres",
host=os.environ["APP_DATABASE__HOST"],
port=int(os.environ["APP_DATABASE__PORT"]),
name=os.environ["APP_DATABASE__NAME"],
user=os.environ["APP_DATABASE__USER"],
password=os.environ["APP_DATABASE__PASSWORD"],
)
@pytest.fixture(scope="session")
def template_db():
"""The migrated schema, built once per session.
SQLite gets a template file to copy; Postgres gets its one database
migrated in place (there is no file to copy, so isolation is by truncation
instead — see `db_path`).
"""
if BACKEND == "postgres":
upgrade_to_head(_pg_settings(), str(APP_ROOT / "app" / "migrations"))
return None
path = _TMP_ROOT / "template.db"
upgrade_to_head(
DatabaseSettings(type="sqlite", path=str(path)), str(APP_ROOT / "app" / "migrations")
)
return path
@pytest.fixture
def db_path(template_db, tmp_path: Path):
"""A clean database for one test."""
if BACKEND == "postgres":
# Empty every table but keep the schema (and alembic's own bookkeeping).
engine = make_engine(_pg_settings())
with engine.begin() as conn:
names = [r[0] for r in conn.execute(text(
"SELECT tablename FROM pg_tables WHERE schemaname = 'public' "
"AND tablename <> 'alembic_version'"))]
if names:
conn.execute(text(
"TRUNCATE " + ", ".join(f'"{n}"' for n in names) + " RESTART IDENTITY CASCADE"))
engine.dispose()
return None
path = tmp_path / "portfolio.db"
shutil.copy(template_db, path)
return path
@pytest.fixture
def session_factory(db_path):
settings = _pg_settings() if BACKEND == "postgres" else DatabaseSettings(
type="sqlite", path=str(db_path))
factory = make_session_factory(settings)
tenancy.install(factory)
yield factory
# Every test builds its own engine, and an engine holds a pool of up to ten
# connections. Undisposed, they come back only when CPython collects the
# engine — so the suite's connection use depended on garbage-collection
# timing rather than on anything it does, and a machine slow enough to hold
# more engines at once exhausted Postgres's 100 and errored. Reported from
# a machine with tesseract installed, where the OCR tests run.
factory.kw["bind"].dispose()
@pytest.fixture(autouse=True)
def _clear_caches():
"""Stop one test's cached series being served to the next (see note 3)."""
for cache in (queries._series_cache, queries._holdings_cache, queries._grouped_cache):
cache.clear()
yield
for cache in (queries._series_cache, queries._holdings_cache, queries._grouped_cache):
cache.clear()
@pytest.fixture(autouse=True)
def _fresh_provider(monkeypatch):
"""The app keeps one identity provider per configuration, with what its
discovery said, and every test that configures one uses the same issuer
with a discovery document of its own."""
from app import federation
monkeypatch.setattr(federation, "_provider", None)
@pytest.fixture(autouse=True)
def _no_stop(monkeypatch):
"""No test may stop the test run. The real `request_stop` sends SIGTERM to
this very process, and Finish asks for it wherever a restart would bring
the app back: in a container, the image's test stage included. A test
about stopping replaces it with its own."""
from app import lifecycle
monkeypatch.setattr(lifecycle, "request_stop", lambda reason: None)
@pytest.fixture(autouse=True)
def _discard_wizard_draft():
"""A wizard draft is process-global, so one test's leaks into the next.
It never showed until the app started REFUSING pages while a wizard is in
progress: before that nothing outside /setup consulted it, and a stale
draft was invisible. Now a test that walks the wizard makes every later
test's request redirect.
"""
from app import setupwizard
setupwizard.discard()
yield
setupwizard.discard()
@pytest.fixture(autouse=True)
def _restore_settings():
"""Put the live settings object back the way the test found it.
`/admin/settings` calls `configfile.apply_live`, which mutates the ONE
global `settings` the whole app reads. A test that saves settings therefore
changes the app for every test after it, in file order.
That was harmless while every editable setting only affected background
behaviour. It stopped being harmless when `features.dca_schedule` landed:
the settings tests post partial forms, an absent checkbox means False, and
28 unrelated plan tests started getting 410 from a feature they never
touched. The fix belongs here rather than in those tests — the next global
setting will do the same thing.
"""
from app.main import settings as live
before = live.model_copy(deep=True)
yield
for name in type(live).model_fields:
setattr(live, name, getattr(before, name))
class _UnreachableYahoo:
"""Where yfinance would be, until a test stubs `pricefeed._yf` itself.
yfinance talks through curl_cffi, which is C and never touches Python's
socket module, so this door is the only one to shut. It fails the test
rather than raising: `lookup` catches Exception, and a test would otherwise
make the real call and pass as if offline.
"""
def __getattr__(self, name: str):
if name.startswith("__"):
raise AttributeError(name)
pytest.fail(
f"a test tried to reach Yahoo (yfinance.{name}). Stub `pricefeed._yf` "
"(see test_pricefeed.stub_yf) or `pricefeed.lookup` instead.",
pytrace=False)
@pytest.fixture(autouse=True)
def _no_network(monkeypatch):
"""Nothing in this suite may talk to the internet.
Not a style rule — it caught a real bug. When the price feed gained fallback
providers, a test that stubbed Yahoo into returning nothing fell straight
through to the live Frankfurter API and passed against real market data. It
would have failed on a plane, in CI without egress, and differently every
day.
So the shared HTTP helper, and appcore's for an identity provider, fail the
test here by default. A test that wants a provider's parsing exercised
stubs `providers._get_json` itself with a canned payload, which is both
hermetic and far clearer about what shape it is testing against.
They FAIL the test rather than raise: the callers catch Exception
(`providers.fetch` turns any failure into the next provider), so an
AssertionError was swallowed and the test passed as if offline. And a call
from a worker thread, a quote refresh after sign-in say, never reaches the
test at all, so every attempt is kept as well, and one left at the end
fails it there. Yields the list, for the guard's own tests.
"""
from appcore import oidc
attempts: list[str] = []
def _blocked(url, *_args, **_kwargs):
attempts.append(str(url))
pytest.fail(f"a test tried to reach the network: {url}. Stub "
"`providers._get_json` with a canned payload instead.", pytrace=False)
monkeypatch.setattr(providers, "_get_json", _blocked)
monkeypatch.setattr(oidc, "_fetch", _blocked)
monkeypatch.setattr(pricefeed, "_yf", _UnreachableYahoo())
# The other way out, and a nastier one: adding an instrument kicks a real
# feed run onto a worker thread that outlives the test — decisions.md #53.
import app.main as main_module
monkeypatch.setattr(main_module, "_kick_feed", lambda: None)
# Every sign-in asks for live prices the same way, on a worker thread that
# outlives the test, and it reaches Yahoo. A test of it captures the real
# function at import, as test_feed_kick does with `_kick_feed`.
monkeypatch.setattr(main_module, "refresh_quotes_soon", lambda: None)
yield attempts
if attempts:
pytest.fail(f"a test tried to reach the network: {attempts}", pytrace=False)
@pytest.fixture
def db(session_factory):
"""An *unbound* session: no portfolio on it.
Personal-data queries raise `TenancyError` here by design — use this for
market-data work (instruments, prices, FX) and for testing that the
fail-closed guarantee actually fails closed.
"""
session = session_factory()
try:
yield session
session.commit()
finally:
session.close()
@pytest.fixture
def owner(db):
"""A user + their portfolio + the owner membership. Returns (user, portfolio)."""
user = User(
email="owner@example.test",
name="Test Owner",
password_hash="x", # tests that care about login set a real hash
is_admin=True,
)
db.add(user)
db.flush()
portfolio = Portfolio(name="Test portfolio")
db.add(portfolio)
db.flush()
db.add(PortfolioMember(portfolio_id=portfolio.id, user_id=user.id, role="owner"))
db.flush()
return user, portfolio
@pytest.fixture
def portfolio(session_factory):
"""An owner and a portfolio, committed, as (portfolio_id, user_id): for the
generated cases, each of which rolls back only the rows it wrote."""
from factories import make_portfolio, make_user
with session_factory() as s:
user = make_user(s, "owner@example.test")
p = make_portfolio(s, "Generated", owner=user)
s.commit()
return p.id, user.id
@pytest.fixture
def pf(db, owner):
"""A session bound to one portfolio — what a logged-in request gets.
Yields the session; `pf.info` carries the portfolio/user ids. This is the
fixture most tests want.
"""
user, portfolio = owner
tenancy.bind(db, portfolio.id, user.id)
return db
# --------------------------------------------------------------------------- #
# Route-level testing
# --------------------------------------------------------------------------- #
@pytest.fixture(scope="session")
def app_module(template_db):
"""The imported FastAPI app.
`app.main` does config load + migrations + session factory at *import*
time, so it can only be imported once per process — hence session scope.
Per-test isolation comes from `client`, which repoints both session
factories at a fresh database.
The test config sets `price_feed.enabled: false`, so importing this does
not start the feed's background task or touch the network.
"""
from app import main
return main
@pytest.fixture
def feed_on(app_module, monkeypatch):
"""The price feed on, for a test that stubs what the feed would be asked.
The test config turns it off, and the feed governs every outbound request
(#9): a stub behind it is never reached, and a test of what the feed says
passes having checked nothing.
"""
monkeypatch.setattr(app_module.settings.price_feed, "enabled", True)
@pytest.fixture
def client(app_module, session_factory, db_path):
"""A TestClient whose app talks to this test's database.
Two factories must be repointed, because the app reaches the database two
ways: module-global `SessionLocal` (the pre-login/`anon()` paths) and
`app.state.session_factory` (everything behind `auth.scoped_session`).
Patching only one leaves half the app on the session-scoped database.
Constructed WITHOUT a `with` block on purpose: Starlette only runs the
lifespan for a context-managed client, and the lifespan starts the price
feed. Belt and braces with `price_feed.enabled: false`.
"""
from fastapi.testclient import TestClient
original_local = app_module.SessionLocal
original_state = app_module.app.state.session_factory
app_module.SessionLocal = session_factory
app_module.app.state.session_factory = session_factory
try:
yield TestClient(app_module.app)
finally:
app_module.SessionLocal = original_local
app_module.app.state.session_factory = original_state