hermes
bbd9fe17e5
Review finding, and it was right. The PR claimed that runs with impossible readings were "excluded from the averages". That was only true of the attenuation average. Run 12's bad final gravity was still poisoning three other statistics: - **Brew time** - its 56 days were folded into the average, making a typical brew look like 13.7 days when it is 10.9. - **Time to 1.010** - its 56 days did the same to that figure (8.4d). - **The history page's ABV** - an 18.64% ABV, which made run 12 the "strongest" beer in the history and pulled the average up. - **The fermentation curve** - its curve plunges from 1.042 to 0.900, which dragged the median line down and stretched the min-max envelope to cover a gravity below water. One bad reading was reaching four different statistics. What changed ------------ - The suspect runs are computed once and excluded from every aggregate: the attenuation, brew-time and time-to-1.010 averages, the fastest and slowest brews, the history page's ABV average, and its strongest and lightest. - `/api/runs` now carries a `suspect` flag, and the charts skip the flagged runs - the attenuation chart and the fermentation curve. - The suspect run stays in the table, with a "not counted" chip and a tinted row, so it can be corrected. It is excluded from the figures, not hidden. The card header says "16 brews (15 counted)" and the prose says how many brews the figures are based on. Tests ----- 133 (up from 127). New coverage: the suspect run is excluded from the brew time and time-to-1.010 averages, it is still listed for review, it is flagged in the API, it is not the strongest brew in the history, and the charts filter it out. The `logged_in` fixture moved to conftest.py, where both test modules can use it.
693 lines
24 KiB
Python
693 lines
24 KiB
Python
"""Tests for the web layer: login, the brew-process page and the API."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from conftest import seed_readings
|
|
|
|
|
|
# ------------------------------------------------------------------ public #
|
|
|
|
def test_healthz_is_public(client):
|
|
response = client.get("/healthz")
|
|
assert response.status_code == 200
|
|
assert response.json == {"status": "ok"}
|
|
|
|
|
|
def test_readyz_checks_the_database(client):
|
|
response = client.get("/readyz")
|
|
assert response.status_code == 200
|
|
assert response.json == {"status": "ready"}
|
|
|
|
|
|
def test_readyz_reports_not_ready_when_mongo_is_down(client, monkeypatch):
|
|
"""A pod that cannot reach Mongo must not serve traffic."""
|
|
from pymongo.errors import ServerSelectionTimeoutError
|
|
|
|
def boom(*args, **kwargs):
|
|
raise ServerSelectionTimeoutError("no mongo")
|
|
|
|
monkeypatch.setattr("mongo.repository.BeerRepository.__init__",
|
|
lambda self, **kw: (_ for _ in ()).throw(ServerSelectionTimeoutError("no mongo")))
|
|
response = client.get("/readyz")
|
|
assert response.status_code == 503
|
|
|
|
|
|
def test_root_redirects_to_login(client):
|
|
response = client.get("/")
|
|
assert response.status_code == 302
|
|
assert "/login" in response.headers["Location"]
|
|
|
|
|
|
def test_login_page_renders(client):
|
|
response = client.get("/login")
|
|
assert response.status_code == 200
|
|
assert b"Brew Log" in response.data
|
|
|
|
|
|
def test_login_with_bad_credentials_is_rejected(client, repo):
|
|
repo.users.insert_one({"username": "beer", "password": "beer_pass"})
|
|
|
|
response = client.post("/login", data={"username": "beer", "password": "wrong"},
|
|
follow_redirects=True)
|
|
assert response.status_code == 200
|
|
assert b"not recognised" in response.data
|
|
|
|
|
|
def test_protected_pages_redirect_when_logged_out(client):
|
|
for path in ("/brews", "/runs/1"):
|
|
response = client.get(path)
|
|
assert response.status_code == 302, path
|
|
assert "/login" in response.headers["Location"], path
|
|
|
|
|
|
# ------------------------------------------------------------- brew pages #
|
|
|
|
def test_brews_page_lists_runs(logged_in, repo):
|
|
seed_readings(repo, [
|
|
("1", "Pale Ale", "1050", "2026-01-01", False, "start"),
|
|
("1", "Pale Ale", "1010", "2026-01-20", True, "Keghing"),
|
|
("2", "Ginger Beer", "1045", "2026-02-01", False, ""),
|
|
])
|
|
|
|
response = logged_in.get("/brews")
|
|
assert response.status_code == 200
|
|
body = response.data.decode()
|
|
assert "Pale Ale" in body
|
|
assert "Ginger Beer" in body
|
|
# The comment on the reading should be surfaced.
|
|
assert "Keghing" in body
|
|
|
|
|
|
def test_run_detail_page_renders(logged_in, repo):
|
|
seed_readings(repo, [
|
|
("1", "Pale Ale", "1050", "2026-01-01", False, "start"),
|
|
("1", "Pale Ale", "1010", "2026-01-20", True, "Keghing"),
|
|
])
|
|
|
|
response = logged_in.get("/runs/1")
|
|
assert response.status_code == 200
|
|
body = response.data.decode()
|
|
assert "Pale Ale" in body
|
|
assert "Keghing" in body
|
|
assert "5.25%" in body
|
|
|
|
|
|
# ------------------------------------------------------------- write paths #
|
|
|
|
def test_recording_a_reading_stores_it(logged_in, repo):
|
|
response = logged_in.post("/brews", data={
|
|
"beer_run_id": "5",
|
|
"beer_run_type": "Test Porter",
|
|
"Date": "2026-03-04",
|
|
"sg": "1045",
|
|
"comment": "Hops in",
|
|
}, follow_redirects=True)
|
|
assert response.status_code == 200
|
|
|
|
doc = repo.readings.find_one({"beer_run_id": "5"})
|
|
assert doc is not None
|
|
assert doc["sg"] == "1045"
|
|
assert doc["beer_type"] == "Test Porter"
|
|
assert doc["comment"] == "Hops in"
|
|
assert doc["final_reading"] == "False"
|
|
|
|
|
|
def test_recording_a_reading_twice_updates_in_place(logged_in, repo):
|
|
"""The unique key is (date, beer_run_id) — no duplicate rows."""
|
|
for sg in ("1045", "1035"):
|
|
logged_in.post("/brews", data={
|
|
"beer_run_id": "6",
|
|
"beer_run_type": "Test",
|
|
"Date": "2026-03-04",
|
|
"sg": sg,
|
|
})
|
|
|
|
count = repo.readings.count_documents({"beer_run_id": "6"})
|
|
doc = repo.readings.find_one({"beer_run_id": "6"})
|
|
assert count == 1
|
|
assert doc["sg"] == "1035"
|
|
|
|
|
|
def test_setting_a_final_reading_closes_the_run(logged_in, repo):
|
|
seed_readings(repo, [("7", "Ale", "1045", "2026-03-01", False, "")])
|
|
|
|
response = logged_in.post("/runs/7/final", data={"date": "2026-03-01"},
|
|
follow_redirects=True)
|
|
assert response.status_code == 200
|
|
|
|
doc = repo.readings.find_one({"beer_run_id": "7"})
|
|
assert doc["final_reading"] == "True"
|
|
|
|
|
|
def test_unsetting_a_final_reading_reactivates_the_run(logged_in, repo):
|
|
seed_readings(repo, [
|
|
("7", "Ale", "1045", "2026-03-01", False, ""),
|
|
("7", "Ale", "1010", "2026-03-20", True, ""),
|
|
])
|
|
|
|
logged_in.post("/runs/7/unfinal", data={"date": "2026-03-20"}, follow_redirects=True)
|
|
|
|
doc = repo.readings.find_one({"beer_run_id": "7", "date": "2026-03-20"})
|
|
assert doc["final_reading"] == "False"
|
|
|
|
|
|
# -------------------------------------------------------------------- API #
|
|
|
|
|
|
# ----------------------------------------------------------------- misc #
|
|
|
|
|
|
def test_the_chart_library_is_loaded_exactly_once(logged_in):
|
|
"""Regression: the chart script was included both in the base template
|
|
and again by each page, so every chart was drawn twice."""
|
|
for path in ("/brews", "/active", "/history"):
|
|
body = logged_in.get(path).data.decode()
|
|
assert body.count("js/charts.js") == 1, path
|
|
assert body.count("chart.umd.js") == 1, path
|
|
|
|
|
|
def test_static_assets_are_served(client):
|
|
for path in ("/static/js/app.js", "/static/css/app.css",
|
|
"/static/img/logo.svg"):
|
|
response = client.get(path)
|
|
assert response.status_code == 200, path
|
|
assert len(response.data) > 100, path
|
|
|
|
|
|
def test_logout_clears_the_session(logged_in):
|
|
response = logged_in.post("/logout", follow_redirects=True)
|
|
assert response.status_code == 200
|
|
# After logging out the protected page bounces back to the login page.
|
|
response = logged_in.get("/brews")
|
|
assert response.status_code == 302
|
|
assert "/login" in response.headers["Location"]
|
|
|
|
|
|
def test_out_of_range_reading_is_saved_but_warned(logged_in, repo):
|
|
response = logged_in.post("/brews", data={
|
|
"beer_run_id": "12",
|
|
"beer_run_type": "Odd",
|
|
"Date": "2026-05-01",
|
|
"sg": "950",
|
|
}, follow_redirects=True)
|
|
assert response.status_code == 200
|
|
assert b"outside the usual hydrometer range" in response.data
|
|
# It is still stored — the warning is advisory, not a rejection.
|
|
assert repo.readings.find_one({"beer_run_id": "12"}) is not None
|
|
|
|
|
|
# ------------------------------------------------------- security: passwords #
|
|
|
|
def test_legacy_plaintext_password_is_accepted_and_migrated(repo):
|
|
"""A legacy plaintext row still logs in, and is rewritten as a hash on
|
|
the way in so plaintext credentials disappear from the database."""
|
|
repo.users.insert_one({"username": "beer", "password": "beer_pass"})
|
|
|
|
assert repo.user_check("beer", "beer_pass") is True
|
|
|
|
doc = repo.users.find_one({"username": "beer"})
|
|
assert doc["password"] != "beer_pass"
|
|
assert "$" in doc["password"] # a werkzeug hash
|
|
assert "password_migrated_at" in doc
|
|
|
|
|
|
def test_a_failed_password_migration_still_lets_the_login_succeed(repo, caplog):
|
|
"""If the upgrade write fails, the login must still work — but the
|
|
failure is logged so the operator knows the password is still plaintext."""
|
|
import logging
|
|
|
|
from pymongo.errors import PyMongoError
|
|
|
|
repo.users.insert_one({"username": "beer", "password": "beer_pass"})
|
|
|
|
def boom(*args, **kwargs):
|
|
raise PyMongoError("write failed")
|
|
|
|
original = repo.users.update_one
|
|
repo.users.update_one = boom
|
|
|
|
with caplog.at_level(logging.WARNING):
|
|
assert repo.user_check("beer", "beer_pass") is True
|
|
|
|
repo.users.update_one = original
|
|
|
|
assert any("plain text" in rec.message for rec in caplog.records)
|
|
|
|
|
|
def test_hashed_password_verifies(repo):
|
|
from mongo.repository import hash_password
|
|
|
|
repo.users.insert_one({"username": "beer", "password": hash_password("beer_pass")})
|
|
|
|
assert repo.user_check("beer", "beer_pass") is True
|
|
assert repo.user_check("beer", "wrong") is False
|
|
|
|
|
|
def test_user_check_rejects_empty_credentials(repo):
|
|
repo.users.insert_one({"username": "beer", "password": "beer_pass"})
|
|
|
|
assert repo.user_check("", "") is False
|
|
assert repo.user_check("beer", "") is False
|
|
assert repo.user_check("", "beer_pass") is False
|
|
|
|
|
|
def test_user_check_rejects_a_nosql_operator_payload(repo):
|
|
"""A dict must never be able to act as a Mongo query operator."""
|
|
repo.users.insert_one({"username": "beer", "password": "beer_pass"})
|
|
|
|
assert repo.user_check({"$ne": ""}, "anything") is False
|
|
assert repo.user_check({"$gt": ""}, {"$ne": ""}) is False
|
|
|
|
|
|
def test_run_id_cannot_be_used_as_an_operator(repo):
|
|
"""A run id is coerced to a string before it reaches a query."""
|
|
from mongo.repository import _as_id
|
|
|
|
assert _as_id({"$ne": ""}) == ""
|
|
assert _as_id(20) == "20"
|
|
assert _as_id("20") == "20"
|
|
|
|
|
|
def test_the_app_refuses_to_start_without_a_secret_key(monkeypatch):
|
|
"""A missing SECRET_KEY must stop the app, not fall back to a default."""
|
|
import importlib
|
|
|
|
import app_factory
|
|
|
|
# Reload so the guard reads the environment as it is *now*. The exception
|
|
# class must be taken from the reloaded module — reloading rebinds it, so
|
|
# a class imported earlier would be a different object and pytest.raises
|
|
# would not match it.
|
|
monkeypatch.delenv("SECRET_KEY", raising=False)
|
|
importlib.reload(app_factory)
|
|
with pytest.raises(app_factory.ImproperlyConfigured):
|
|
app_factory.create_app()
|
|
|
|
monkeypatch.setenv("SECRET_KEY", "test-secret-key-not-a-real-secret")
|
|
importlib.reload(app_factory)
|
|
|
|
|
|
def test_the_app_refuses_the_placeholder_secret_key(monkeypatch):
|
|
import importlib
|
|
|
|
import app_factory
|
|
|
|
monkeypatch.setenv("SECRET_KEY", "change-me-in-the-deployment")
|
|
importlib.reload(app_factory)
|
|
with pytest.raises(app_factory.ImproperlyConfigured):
|
|
app_factory.create_app()
|
|
|
|
monkeypatch.setenv("SECRET_KEY", "test-secret-key-not-a-real-secret")
|
|
importlib.reload(app_factory)
|
|
|
|
|
|
# ------------------------------------------------- active beers and metadata #
|
|
|
|
def test_active_page_shows_only_unfinished_runs(logged_in, repo):
|
|
seed_readings(repo, [
|
|
("1", "Finished Beer", "1050", "2026-01-01", False, ""),
|
|
("1", "Finished Beer", "1010", "2026-01-20", True, ""),
|
|
("2", "Still Brewing", "1045", "2026-02-01", False, ""),
|
|
])
|
|
|
|
response = logged_in.get("/active")
|
|
assert response.status_code == 200
|
|
body = response.data.decode()
|
|
assert "Still Brewing" in body
|
|
assert "Finished Beer" not in body
|
|
|
|
|
|
def test_active_page_shows_the_tap_number(logged_in, repo):
|
|
seed_readings(repo, [("2", "Still Brewing", "1045", "2026-02-01", False, "")])
|
|
repo.upsert_run_metadata("2", tap_number=4)
|
|
|
|
body = logged_in.get("/active").data.decode()
|
|
assert "Tap 4" in body
|
|
|
|
|
|
def test_active_page_shows_the_projected_alcohol(logged_in, repo):
|
|
"""With two readings the first is the OG, so the estimate is meaningful."""
|
|
seed_readings(repo, [
|
|
("2", "Still Brewing", "1050", "2026-02-01", False, ""),
|
|
("2", "Still Brewing", "1020", "2026-02-10", False, ""),
|
|
])
|
|
|
|
body = logged_in.get("/active").data.decode()
|
|
# 30 point drop = 3.94%
|
|
assert "3.94%" in body
|
|
|
|
|
|
def test_active_page_shows_a_dash_when_no_og_is_known(logged_in, repo):
|
|
"""A single reading with no recorded OG has no reliable estimate."""
|
|
seed_readings(repo, [("2", "Still Brewing", "1045", "2026-02-01", False, "")])
|
|
|
|
body = logged_in.get("/active").data.decode()
|
|
assert "needs 2nd reading" in body
|
|
|
|
|
|
def test_active_page_shows_the_latest_comment(logged_in, repo):
|
|
seed_readings(repo, [
|
|
("2", "Still Brewing", "1050", "2026-02-01", False, "Hops in"),
|
|
("2", "Still Brewing", "1030", "2026-02-10", False, "Almost there"),
|
|
])
|
|
|
|
body = logged_in.get("/active").data.decode()
|
|
assert "Almost there" in body
|
|
|
|
|
|
def test_run_metadata_is_saved_and_surfaced(logged_in, repo):
|
|
seed_readings(repo, [("8", "Pale Ale", "1045", "2026-03-01", False, "")])
|
|
|
|
logged_in.post("/runs/8", data={
|
|
"name": "Canonical Name",
|
|
"tap_number": "3",
|
|
"og_points": "1050",
|
|
"notes": "Starter was a 50F mess.",
|
|
}, follow_redirects=True)
|
|
|
|
meta = repo.run_metadata("8")
|
|
assert meta["name"] == "Canonical Name"
|
|
assert meta["tap_number"] == 3
|
|
assert meta["og_points"] == 1050
|
|
assert meta["notes"] == "Starter was a 50F mess."
|
|
|
|
# The canonical name must appear on the brew-process page and the active page.
|
|
assert "Canonical Name" in logged_in.get("/brews").data.decode()
|
|
assert "Canonical Name" in logged_in.get("/active").data.decode()
|
|
|
|
|
|
def test_run_metadata_can_be_cleared(logged_in, repo):
|
|
seed_readings(repo, [("8", "Pale Ale", "1045", "2026-03-01", False, "")])
|
|
repo.upsert_run_metadata("8", name="Set Name", tap_number=2)
|
|
|
|
logged_in.post("/runs/8", data={"name": "", "tap_number": "", "og_points": "", "notes": ""},
|
|
follow_redirects=True)
|
|
|
|
meta = repo.run_metadata("8")
|
|
# The name was cleared, so it falls back to the reading-derived name.
|
|
assert not meta.get("name")
|
|
assert meta.get("tap_number") is None
|
|
|
|
|
|
def test_the_stored_name_overrides_the_drifting_reading_name(logged_in, repo):
|
|
seed_readings(repo, [
|
|
("9", "Old Name", "1050", "2026-03-01", False, ""),
|
|
("9", "Newer Name", "1040", "2026-03-10", False, ""),
|
|
])
|
|
repo.upsert_run_metadata("9", name="The Name I Want")
|
|
|
|
body = logged_in.get("/brews").data.decode()
|
|
assert "The Name I Want" in body
|
|
|
|
|
|
def test_next_free_tap_number_skips_the_ones_in_use(repo):
|
|
assert repo.next_free_tap_number() == 1
|
|
|
|
repo.upsert_run_metadata("1", tap_number=1)
|
|
assert repo.next_free_tap_number() == 2
|
|
|
|
repo.upsert_run_metadata("2", tap_number=2)
|
|
assert repo.next_free_tap_number() == 3
|
|
|
|
# A gap is filled before extending the range.
|
|
repo.upsert_run_metadata("3", tap_number=4)
|
|
assert repo.next_free_tap_number() == 3
|
|
|
|
|
|
def test_history_page_shows_completed_brews(logged_in, repo):
|
|
seed_readings(repo, [
|
|
("1", "Pale Ale", "1050", "2026-01-01", False, ""),
|
|
("1", "Pale Ale", "1010", "2026-01-20", True, "done"),
|
|
])
|
|
|
|
response = logged_in.get("/history")
|
|
assert response.status_code == 200
|
|
body = response.data.decode()
|
|
assert "Pale Ale" in body
|
|
assert "5.25%" in body # standard ABV for a 40 point drop
|
|
assert "Legacy" in body # the comparison table is present
|
|
|
|
|
|
def test_history_page_flags_an_implausible_reading(logged_in, repo):
|
|
seed_readings(repo, [
|
|
("1", "Pale Ale", "1050", "2026-01-01", False, ""),
|
|
("1", "Pale Ale", "900", "2026-01-20", True, "typo?"),
|
|
])
|
|
|
|
body = logged_in.get("/history").data.decode()
|
|
assert "worth a second look" in body
|
|
assert "typo?" in body
|
|
|
|
|
|
def test_partial_metadata_update_leaves_other_fields_alone(repo):
|
|
"""The backfill uses this: it must not wipe a tap number the operator
|
|
has already set."""
|
|
repo.upsert_run_metadata("1", name="A Name", tap_number=3, notes="Some notes")
|
|
|
|
repo.upsert_run_metadata_partial("1", name="A Better Name")
|
|
|
|
meta = repo.run_metadata("1")
|
|
assert meta["name"] == "A Better Name"
|
|
assert meta["tap_number"] == 3
|
|
assert meta["notes"] == "Some notes"
|
|
|
|
|
|
def test_partial_metadata_update_ignores_none(repo):
|
|
repo.upsert_run_metadata("1", name="A Name", tap_number=3)
|
|
|
|
repo.upsert_run_metadata_partial("1", name=None, tap_number=None)
|
|
|
|
meta = repo.run_metadata("1")
|
|
assert meta["name"] == "A Name"
|
|
assert meta["tap_number"] == 3
|
|
|
|
|
|
# ---------------------------------------------------------- security headers #
|
|
|
|
def test_security_headers_are_set(logged_in):
|
|
"""The app serves only its own vendored assets, so a strict CSP of
|
|
'self' is enforceable."""
|
|
response = logged_in.get("/brews")
|
|
headers = response.headers
|
|
|
|
assert headers.get("X-Content-Type-Options") == "nosniff"
|
|
assert headers.get("X-Frame-Options") == "DENY"
|
|
assert headers.get("Referrer-Policy") == "same-origin"
|
|
|
|
csp = headers.get("Content-Security-Policy")
|
|
assert csp is not None
|
|
assert "default-src 'self'" in csp
|
|
assert "script-src 'self'" in csp
|
|
assert "object-src 'none'" in csp
|
|
assert "frame-ancestors 'none'" in csp
|
|
# No CDN is used, so nothing external should be allowed.
|
|
assert "https://" not in csp
|
|
# 'unsafe-inline' would defeat the point of the policy.
|
|
assert "unsafe-inline" not in csp
|
|
assert "unsafe-eval" not in csp
|
|
|
|
|
|
def test_security_headers_are_set_on_the_login_page(client):
|
|
response = client.get("/login")
|
|
assert response.headers.get("X-Content-Type-Options") == "nosniff"
|
|
assert response.headers.get("Content-Security-Policy") is not None
|
|
|
|
|
|
# ------------------------------------------------------------------ indexes #
|
|
|
|
def test_the_repository_creates_its_indexes(repo):
|
|
"""The run metadata is always looked up by beer_run_id, and the readings
|
|
by (beer_run_id, date) — both need an index."""
|
|
index_names = repo.run_meta.index_information()
|
|
assert "beer_run_id_1" in index_names
|
|
|
|
reading_indexes = repo.readings.index_information()
|
|
assert "run_date_1" in reading_indexes
|
|
|
|
|
|
def test_no_template_has_an_inline_script_or_style(client):
|
|
"""The strict CSP (script-src 'self'; style-src 'self') blocks inline
|
|
scripts and style attributes, so the templates must have neither."""
|
|
import glob
|
|
import os
|
|
import re
|
|
|
|
template_dir = os.path.join(
|
|
os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
|
|
"src", "flask", "templates",
|
|
)
|
|
|
|
for path in glob.glob(os.path.join(template_dir, "*.html")):
|
|
with open(path) as fh:
|
|
body = fh.read()
|
|
|
|
# An inline <script> is one with content rather than a src attribute.
|
|
for match in re.finditer(r"<script(?![^>]*\bsrc=)[^>]*>(.*?)</script>",
|
|
body, re.S):
|
|
assert match.group(1).strip() == "", (
|
|
f"{os.path.basename(path)} has an inline <script>, which the "
|
|
"CSP blocks"
|
|
)
|
|
|
|
assert not re.search(r"\sstyle\s*=", body), (
|
|
f"{os.path.basename(path)} has an inline style attribute, which "
|
|
"the CSP blocks"
|
|
)
|
|
|
|
|
|
# -------------------------------------------------------------------- API #
|
|
|
|
def test_api_runs_returns_json(logged_in, repo):
|
|
seed_readings(repo, [
|
|
("1", "Pale Ale", "1050", "2026-01-01", False, ""),
|
|
("1", "Pale Ale", "1010", "2026-01-20", True, ""),
|
|
])
|
|
|
|
response = logged_in.get("/api/runs")
|
|
assert response.status_code == 200
|
|
payload = response.get_json()
|
|
assert len(payload) == 1
|
|
assert payload[0]["abv"] == 5.25
|
|
assert payload[0]["active"] is False
|
|
|
|
|
|
def test_api_run_detail_returns_json(logged_in, repo):
|
|
seed_readings(repo, [("11", "Cider", "1045", "2026-04-01", False, "first")])
|
|
|
|
response = logged_in.get("/api/runs/11")
|
|
assert response.status_code == 200
|
|
payload = response.get_json()
|
|
assert payload["beer_run_id"] == "11"
|
|
assert payload["name"] == "Cider"
|
|
assert len(payload["readings"]) == 1
|
|
|
|
|
|
def test_api_run_detail_404s_for_an_unknown_run(logged_in):
|
|
response = logged_in.get("/api/runs/does-not-exist")
|
|
assert response.status_code == 404
|
|
|
|
|
|
def test_the_api_is_protected(client):
|
|
"""The charts fetch from the API, so it must not be public."""
|
|
for path in ("/api/runs", "/api/runs/1"):
|
|
response = client.get(path)
|
|
assert response.status_code == 302, path
|
|
assert "/login" in response.headers["Location"], path
|
|
|
|
|
|
def test_api_runs_includes_the_canonical_name_and_tap(logged_in, repo):
|
|
seed_readings(repo, [("12", "Drifting Name", "1045", "2026-05-01", False, "")])
|
|
repo.upsert_run_metadata("12", name="Canonical", tap_number=7)
|
|
|
|
payload = logged_in.get("/api/runs").get_json()
|
|
entry = next(r for r in payload if r["beer_run_id"] == "12")
|
|
assert entry["canonical_name"] == "Canonical"
|
|
assert entry["tap_number"] == 7
|
|
|
|
|
|
def test_the_months_chart_includes_the_gaps(logged_in, repo):
|
|
"""Months with no brews must appear with a zero, or the chart spaces the
|
|
busy months evenly and reads as continuous activity."""
|
|
seed_readings(repo, [
|
|
("1", "Spring", "1050", "2026-01-05", False, ""),
|
|
("1", "Spring", "1010", "2026-01-20", True, ""),
|
|
("2", "Autumn", "1050", "2026-06-05", False, ""),
|
|
("2", "Autumn", "1010", "2026-06-20", True, ""),
|
|
])
|
|
|
|
from views.analysis import _month_range
|
|
|
|
months = _month_range("2026-01", "2026-06")
|
|
assert months == ["2026-01", "2026-02", "2026-03", "2026-04",
|
|
"2026-05", "2026-06"]
|
|
|
|
|
|
def test_the_months_chart_counts_zero_for_a_quiet_month(logged_in, repo):
|
|
"""The chart is fed by the API, so the quiet months are visible there."""
|
|
seed_readings(repo, [
|
|
("1", "Spring", "1050", "2026-01-05", False, ""),
|
|
("1", "Spring", "1010", "2026-01-20", True, ""),
|
|
("2", "Autumn", "1050", "2026-06-05", False, ""),
|
|
("2", "Autumn", "1010", "2026-06-20", True, ""),
|
|
])
|
|
|
|
import json
|
|
import re
|
|
|
|
body = logged_in.get("/history").data.decode()
|
|
# The page embeds the month labels for the chart.
|
|
assert "2026-06" in body
|
|
|
|
|
|
def test_month_range_covers_a_year_boundary():
|
|
from views.analysis import _month_range
|
|
|
|
assert _month_range("2025-11", "2026-02") == [
|
|
"2025-11", "2025-12", "2026-01", "2026-02",
|
|
]
|
|
assert _month_range("2026-05", "2026-05") == ["2026-05"]
|
|
|
|
|
|
# ------------------------------------------------------------- deep analysis #
|
|
|
|
def test_the_analysis_page_renders(logged_in, repo):
|
|
seed_readings(repo, [
|
|
("1", "Pale Ale", "1050", "2026-01-01", False, ""),
|
|
("1", "Pale Ale", "1010", "2026-01-20", True, "done"),
|
|
("2", "Ginger Beer", "1045", "2026-02-01", False, ""),
|
|
("2", "Ginger Beer", "1015", "2026-02-15", True, "done"),
|
|
])
|
|
|
|
response = logged_in.get("/analysis")
|
|
assert response.status_code == 200
|
|
body = response.data.decode()
|
|
assert "Pale Ale" in body
|
|
assert "Ginger Beer" in body
|
|
# The attenuation figures and the summary sentence.
|
|
assert "80.0%" in body
|
|
assert "attenuation" in body.lower()
|
|
|
|
|
|
def test_the_analysis_page_excludes_active_runs(logged_in, repo):
|
|
"""An unfinished brew has no final reading, so it is not comparable."""
|
|
seed_readings(repo, [
|
|
("1", "Finished", "1050", "2026-01-01", False, ""),
|
|
("1", "Finished", "1010", "2026-01-20", True, ""),
|
|
("2", "Still Going", "1045", "2026-02-01", False, ""),
|
|
])
|
|
|
|
body = logged_in.get("/analysis").data.decode()
|
|
assert "Finished" in body
|
|
assert "Still Going" not in body
|
|
|
|
|
|
def test_the_analysis_page_needs_at_least_one_finished_brew(logged_in, repo):
|
|
seed_readings(repo, [("1", "Just Started", "1045", "2026-02-01", False, "")])
|
|
|
|
response = logged_in.get("/analysis")
|
|
assert response.status_code == 200
|
|
assert b"No finished brews" in response.data
|
|
|
|
|
|
def test_the_analysis_page_is_protected(client):
|
|
response = client.get("/analysis")
|
|
assert response.status_code == 302
|
|
assert "/login" in response.headers["Location"]
|
|
|
|
|
|
def test_the_api_exposes_the_attenuation_and_curve(logged_in, repo):
|
|
"""The charts fetch the curve data from the API."""
|
|
seed_readings(repo, [
|
|
("1", "Pale Ale", "1050", "2026-01-01", False, ""),
|
|
("1", "Pale Ale", "1010", "2026-01-20", True, ""),
|
|
])
|
|
|
|
payload = logged_in.get("/api/runs").get_json()
|
|
entry = payload[0]
|
|
assert entry["attenuation"] == 80.0
|
|
assert len(entry["curve"]) == 2
|
|
assert entry["curve"][0]["progress"] == 0.0
|
|
assert entry["curve"][1]["progress"] == 1.0
|