tests/test_sequences_through_the_routes.py
Rama main · réplica de lectura del repositorio original.
"""Generated sequences of what people do, through the real routes, checked
after every step for what must always hold.
A route test makes one request against a state it set up. What test users ran
into came from sequences instead: a ticker removed while it was in the plan, a
currency corrected after trades were recorded under the old one, a holding
page left behind once its last trade moved away. Hypothesis drives the app
through generated sequences across two portfolios, and shrinks a failing one
to the shortest sequence that still fails.
After every step:
- nothing answered 500, and every page of the active portfolio renders;
- no instrument's units go below zero anywhere in its timeline, whichever route
wrote the rows — the rule every write route is meant to keep;
- a refused request changed nothing;
- the holdings export lists exactly the open positions the stored trades make;
- nothing done in one portfolio changed the other's rows, except a move into it.
"""
from __future__ import annotations
import csv
import datetime as dt
import io
import os
import shutil
from decimal import Decimal
from types import SimpleNamespace
import pytest
from fastapi.testclient import TestClient
from hypothesis import Phase, settings
from hypothesis import strategies as st
from hypothesis.stateful import RuleBasedStateMachine, invariant, rule, run_state_machine_as_test
from sqlalchemy import select, text
import factories as fac
from app import auth as auth_mod
from app import clock, pricefeed, queries, tenancy
from app.models import (
MARKET_OPEN,
Dividend,
HoldingPref,
Instrument,
InvestmentPlan,
Trade,
User,
trade_order,
)
from appcore import make_session_factory
from appcore.config import DatabaseSettings
from test_routes import PASSWORD, pre_auth_csrf
HTML = {"accept": "text/html"}
TICKERS = ["ACME", "NOVA", "WIDGET", "ALPHA"]
NEW_TICKERS = ["ZULU", "GAMMA"]
# What a lookup says for a new one, so adding never reaches the network.
YAHOO = {"ZULU.AX": {"longName": "Zulu Ltd", "currency": "AUD", "quoteType": "EQUITY"},
"GAMMA": {"longName": "Gamma Inc", "currency": "USD", "quoteType": "EQUITY"}}
class Walk(RuleBasedStateMachine):
def __init__(self, env: SimpleNamespace):
super().__init__()
self.env = env
self.factory = env.fresh()
app = env.app_module
self._restore = (app.SessionLocal, app.app.state.session_factory)
app.SessionLocal = self.factory
app.app.state.session_factory = self.factory
for cache in (queries._series_cache, queries._holdings_cache, queries._grouped_cache):
cache.clear()
self.client = TestClient(app.app)
with self.factory() as s:
user = fac.make_user(s, "walker@example.test",
password_hash=auth_mod.hash_password(PASSWORD))
first = fac.make_portfolio(s, "First", owner=user)
second = fac.make_portfolio(s, "Second", owner=user)
day = clock.today() - dt.timedelta(days=400)
for ticker, exchange, currency in (("ACME", "ASX", "AUD"),
("NOVA", "NASDAQ", "USD"),
("WIDGET", "ASX", "AUD"),
("ALPHA", "ASX", "AUD")):
inst = fac.make_instrument(s, ticker, exchange=exchange, currency=currency,
asset_class="share")
fac.add_prices(s, inst, [(day, "10.00"), (clock.today(), "11.00")])
fac.add_fx_series(s, "USDAUD", [(day, "1.50"), (clock.today(), "1.55")])
for ticker, portfolio in (("ACME", first), ("NOVA", first), ("WIDGET", second)):
inst = s.scalar(select(Instrument).where(Instrument.ticker == ticker))
fac.hold(s, inst, portfolio_id=portfolio.id)
s.commit()
self.user_id, self.portfolios = user.id, [first.id, second.id]
token = pre_auth_csrf(self.client)
self.client.post("/login", data={"email": "walker@example.test", "password": PASSWORD,
"_csrf": token}, headers=HTML, follow_redirects=False)
self.active = self.portfolios[0]
self._post("/portfolio/switch", {"portfolio_id": str(self.active)})
def teardown(self):
app = self.env.app_module
app.SessionLocal, app.app.state.session_factory = self._restore
self.factory.kw["bind"].dispose()
# ------------------------------------------------------------------ #
# Reading what is stored, across both portfolios
# ------------------------------------------------------------------ #
def _rows(self) -> dict:
with self.factory() as s:
tenancy.allow_unscoped(s)
trades = sorted(
(t.id, t.portfolio_id, t.instrument_id, t.date, t.time or MARKET_OPEN, t.type,
t.quantity, t.unit_price, t.brokerage, t.fx_rate)
for t in s.scalars(select(Trade)))
dividends = sorted((d.id, d.portfolio_id, d.instrument_id, d.date, d.cash_amount,
d.fx_rate) for d in s.scalars(select(Dividend)))
prefs = sorted((p.portfolio_id, p.instrument_id, p.drp, p.note)
for p in s.scalars(select(HoldingPref)))
plans = sorted((p.portfolio_id, tuple(e.instrument_id for e in p.entries),
p.interval_days) for p in s.scalars(select(InvestmentPlan)))
instruments = sorted((i.id, i.ticker, i.exchange, i.currency, i.asset_class, i.name,
i.yahoo_symbol, i.active) for i in s.scalars(select(Instrument)))
return {"trades": trades, "dividends": dividends, "prefs": prefs, "plans": plans,
"instruments": instruments}
def _mine(self, rows: dict, portfolio: int) -> set[int]:
"""Instrument ids a portfolio has (decisions.md #139)."""
return ({t[2] for t in rows["trades"] if t[1] == portfolio}
| {d[2] for d in rows["dividends"] if d[1] == portfolio}
| {p[1] for p in rows["prefs"] if p[0] == portfolio}
| {i for p in rows["plans"] if p[0] == portfolio for i in p[1]})
def _of(self, rows: dict, portfolio: int) -> dict:
"""One portfolio's own rows, rates left out (a currency change resets them
everywhere, on purpose)."""
return {"trades": [t[:9] for t in rows["trades"] if t[1] == portfolio],
"dividends": [d[:5] for d in rows["dividends"] if d[1] == portfolio],
"prefs": [p for p in rows["prefs"] if p[0] == portfolio],
"plans": [p for p in rows["plans"] if p[0] == portfolio]}
def _id(self, ticker: str) -> int | None:
return next((i[0] for i in self._rows()["instruments"] if i[1] == ticker), None)
# ------------------------------------------------------------------ #
# Requests
# ------------------------------------------------------------------ #
def _csrf(self) -> str:
from app.models import UserSession
with self.factory() as s:
return s.scalars(select(UserSession).order_by(UserSession.id.desc())).first().csrf_token
def _post(self, path: str, data: dict):
resp = self.client.post(path, data={"_csrf": self._csrf(), **data}, headers=HTML,
follow_redirects=False)
assert resp.status_code < 500, (path, data, resp.text[:300])
return resp
def _changing(self, path: str, data: dict, *, moves_to: int | None = None) -> bool:
"""Post, and check what a refusal and an acceptance may each change."""
before = self._rows()
resp = self._post(path, data)
after = self._rows()
accepted = resp.status_code == 303 and "error=" not in resp.headers.get("location", "")
if not accepted:
assert after == before, (path, data, resp.status_code, "a refusal changed rows")
for other in self.portfolios:
if other not in (self.active, moves_to):
assert self._of(after, other) == self._of(before, other), (path, data)
return accepted
@staticmethod
def _day(days_ago: int) -> str:
return (clock.today() - dt.timedelta(days=days_ago)).isoformat()
# ------------------------------------------------------------------ #
# What people do
# ------------------------------------------------------------------ #
@rule()
def switch(self):
self.active = next(p for p in self.portfolios if p != self.active)
self._post("/portfolio/switch", {"portfolio_id": str(self.active)})
@rule(ticker=st.sampled_from(TICKERS + NEW_TICKERS), kind=st.sampled_from(["buy", "sell"]),
days_ago=st.integers(0, 900), units=st.sampled_from(["1", "2.5", "10", "0.0001"]),
price=st.sampled_from(["0", "1.25", "40"]), brokerage=st.sampled_from(["0", "9.95"]))
def record(self, ticker, kind, days_ago, units, price, brokerage):
rows = self._rows()
inst_id = self._id(ticker)
listed = inst_id is not None and inst_id in self._mine(rows, self.active)
data = {"type": kind, "trade_date": self._day(days_ago), "quantity": units,
"unit_price": price, "brokerage": brokerage}
if listed:
data["instrument_id"] = str(inst_id)
else:
exchange = "NASDAQ" if ticker in ("NOVA", "GAMMA") else "ASX"
data.update(instrument_id="new", new_ticker=ticker, new_exchange=exchange,
new_asset_class="share",
new_currency="USD" if exchange == "NASDAQ" else "AUD")
held = sum((t[6] if t[5] != "sell" else -t[6] for t in rows["trades"]
if t[1] == self.active and t[2] == inst_id), Decimal(0))
accepted = self._changing("/trade/new", data)
if kind == "buy" and listed:
assert accepted, ("a buy of something listed is never refused", data)
if kind == "sell" and Decimal(units) > held:
assert not accepted, ("more than is held cannot be sold", data, held)
@rule(pick=st.integers(0, 10**6), days_ago=st.integers(0, 900),
units=st.sampled_from(["1", "5", "0.5"]))
def edit(self, pick, days_ago, units):
mine = [t for t in self._rows()["trades"] if t[1] == self.active]
if not mine:
return
t = mine[pick % len(mine)]
self._changing(f"/trade/{t[0]}/edit", {
"type": t[5], "trade_date": self._day(days_ago), "quantity": units,
"unit_price": str(t[7]), "brokerage": str(t[8])})
@rule(pick=st.integers(0, 10**6))
def delete(self, pick):
mine = [t for t in self._rows()["trades"] if t[1] == self.active]
if mine:
self._changing(f"/trade/{mine[pick % len(mine)][0]}/delete", {})
@rule(pick=st.integers(0, 10**6))
def move_everything(self, pick):
rows = self._rows()
mine = [t for t in rows["trades"] if t[1] == self.active]
if not mine:
return
inst = mine[pick % len(mine)][2]
target = next(p for p in self.portfolios if p != self.active)
picked = [f"trade:{t[0]}" for t in mine if t[2] == inst] + [
f"dividend:{d[0]}" for d in rows["dividends"] if d[1] == self.active and d[2] == inst]
self._changing(f"/trade/{mine[pick % len(mine)][0]}/move",
{"target": str(target), "row": picked}, moves_to=target)
@rule(ticker=st.sampled_from(TICKERS + NEW_TICKERS))
def add_holding(self, ticker):
exchange = "NASDAQ" if ticker in ("NOVA", "GAMMA") else "ASX"
self._changing("/holdings/add", {"ticker": ticker, "exchange": exchange,
"asset_class": "share"})
@rule(ticker=st.sampled_from(TICKERS + NEW_TICKERS), everything=st.booleans())
def remove_holding(self, ticker, everything):
inst = self._id(ticker)
if inst is None:
return
if everything:
self._changing(f"/holdings/{inst}/delete-trades", {})
self._changing(f"/holdings/{inst}/remove", {})
@rule(ticker=st.sampled_from(TICKERS + NEW_TICKERS),
currency=st.sampled_from(["AUD", "USD"]), cls=st.sampled_from(["etf", "share"]))
def correct_details(self, ticker, currency, cls):
inst = self._id(ticker)
if inst is None:
return
self._changing(f"/holdings/{inst}/pref", {
"catalogue": "1", "name": f"{ticker} corrected", "asset_class": cls,
"currency": currency, "yahoo_symbol": "", "note": "kept", "drp": "1"})
@rule(picks=st.lists(st.sampled_from(TICKERS + NEW_TICKERS), max_size=3, unique=True),
interval=st.sampled_from([7, 14, 28]))
def plan(self, picks, interval):
self._changing("/schedule/save", {
"name": "Plan", "interval_days": str(interval), "amount": "500",
"brokerage": "9.95", "start_date": "", "tickers": ",".join(picks)})
# ------------------------------------------------------------------ #
# What must hold after every step
# ------------------------------------------------------------------ #
@invariant()
def no_timeline_goes_below_zero(self):
rows = self._rows()
by_holding: dict = {}
for t in rows["trades"]:
by_holding.setdefault((t[1], t[2]), []).append(t)
for key, trades in by_holding.items():
held = Decimal(0)
for t in sorted(trades, key=lambda r: trade_order(SimpleNamespace(
date=r[3], time=r[4], id=r[0]))):
held += -t[6] if t[5] == "sell" else t[6]
assert held >= 0, (key, trades)
@invariant()
def every_page_renders(self):
rows = self._rows()
tickers = {i[1] for i in rows["instruments"] if i[0] in self._mine(rows, self.active)}
pages = ["/", "/holdings", "/charts", "/schedule", "/export", "/trade/new"]
pages += [f"/holding/{t}" for t in sorted(tickers)]
with self.factory() as s:
tenancy.bind(s, self.active, self.user_id)
from app import fyreport
pages += [f"/?fy={fy}" for fy in fyreport.available_fys(s)]
for page in pages:
resp = self.client.get(page, headers=HTML)
assert resp.status_code == 200, (page, resp.status_code, resp.text[:300])
@invariant()
def the_export_lists_the_open_positions(self):
rows = self._rows()
units: dict = {}
for t in rows["trades"]:
if t[1] == self.active:
units[t[2]] = units.get(t[2], Decimal(0)) + (-t[6] if t[5] == "sell" else t[6])
active = {i[0]: i for i in rows["instruments"]}
want = {active[i][1]: u.quantize(Decimal("0.00000001")) for i, u in units.items()
if u > 0 and active[i][7]}
resp = self.client.get("/export/download?report=holdings&fmt=csv")
assert resp.status_code == 200
got = {r["Ticker"]: Decimal(r["Units"]) for r in csv.DictReader(io.StringIO(
resp.content.decode("utf-8-sig")))}
assert got == want
def _yahoo(monkeypatch):
class Ticker:
def __init__(self, symbol):
self.symbol = symbol
def get_info(self):
return dict(YAHOO.get(self.symbol, {}))
monkeypatch.setattr(pricefeed, "_yf", SimpleNamespace(Ticker=Ticker))
@pytest.fixture
def env(app_module, template_db, tmp_path, monkeypatch):
_yahoo(monkeypatch)
made = []
def fresh():
if os.environ.get("STOCKTAKE_TEST_DB", "sqlite") == "postgres":
factory = make_session_factory(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"]))
with factory() as s:
names = [r[0] for r in s.execute(text(
"SELECT tablename FROM pg_tables WHERE schemaname = 'public' "
"AND tablename <> 'alembic_version'"))]
s.execute(text("TRUNCATE " + ", ".join(f'"{n}"' for n in names)
+ " RESTART IDENTITY CASCADE"))
s.commit()
else:
path = tmp_path / f"walk{len(made)}.db"
shutil.copy(template_db, path)
factory = make_session_factory(DatabaseSettings(type="sqlite", path=str(path)))
tenancy.install(factory)
made.append(factory)
return factory
return SimpleNamespace(app_module=app_module, fresh=fresh)
def test_generated_sequences_keep_what_must_always_hold(env):
# No shrinking but when exploring. A failing walk is at most 12 steps,
# readable as it stands, and each attempt to shrink one builds a fresh
# database: Hypothesis spent its whole 300 s cap on that before printing
# anything, which read as a hang, while memory doubled.
exploring = os.environ.get("HYPOTHESIS_PROFILE") == "explore"
run_state_machine_as_test(lambda: Walk(env), settings=settings(
max_examples=settings.default.max_examples // 4 or 1, stateful_step_count=12,
phases=list(Phase) if exploring else [Phase.explicit, Phase.reuse, Phase.generate]))
def test_the_walk_starts_where_its_rules_expect(env):
"""The checks prove nothing if the starting state is not what the rules
assume: two portfolios, each with a holding, and a session in the first."""
walk = Walk(env)
try:
rows = walk._rows()
assert [len(walk._mine(rows, p)) for p in walk.portfolios] == [2, 1]
assert walk.client.get("/holdings", headers=HTML).status_code == 200
with walk.factory() as s:
assert s.scalar(select(User.email)) == "walker@example.test"
walk.no_timeline_goes_below_zero()
walk.every_page_renders()
walk.the_export_lists_the_open_positions()
finally:
walk.teardown()