Files
beer_data/tests/test_analysis.py
T
hermes 76a4d0805c fix(analysis): the review findings - rising gravity, undated readings, the band
A second, deeper review of PR #12 found five things the first review did
not.  All are fixed here.

The band was not the band
-------------------------
The fermentation curve's shaded range was drawn as ONE dataset with
`fill: "+1"`, which fills to the NEXT dataset - and the next dataset was
the median line, not the maximum.  So the band shaded min-to-median, the
upper half of the range was drawn as a hairline but never shaded, and the
tooltip said "range min to max" while the picture showed something else.
The band is now two datasets - the min curve filled down to a max curve -
with the max hidden from the legend.

Rising gravity
--------------
`attenuation_percent` rejected a final gravity below 1.000 and one
implying more than 100% attenuation, but a run whose gravity went UP
produced a NEGATIVE attenuation - which was then averaged into the
headline figure and plotted as a negative bar.  Verified against the live
code: OG 1010 -> FG 1020 gave -100.0%.  Now rejected, and reported by
`implausible_runs` with the reason "the gravity went up, which is usually
a mis-ordered pair of readings".

Undated readings
----------------
An undated reading sorted to the front of the run, so it became the
original gravity - which put the OG after the FG and produced a negative
attenuation.  The OG now comes from the earliest *dated* reading, and
`days_to_target` skips the readings it cannot place on the timeline rather
than giving up on the whole run.  Verified: readings 1050 (dated), 1005
(undated), 1010 (dated) gave None instead of 24 days, and an OG of 1005
instead of 1050.

Active runs on the curve
------------------------
The curve chart filtered on `curve.length > 1` only, so it plotted the
active runs as well - a partial fermentation on a chart meant to describe
how a completed one behaves.  It now excludes them, along with the
suspect runs.

Four fetches, one payload
-------------------------
`/api/runs` carries every reading of every run, and four charts each
fetched it independently - so the analysis page made two identical
round-trips of the whole collection.  There is now one shared fetch.

Also
----
- The "Readings that need a second look" table links to the run, so a
  suspect reading can be corrected in one click.
- `days_to_1015` and the `curves` variable were computed and never used.
  Removed.
- A reading below 1.000 now says what it is - "beer cannot be less dense
  than water, so this is almost certainly a typo (a 1.012 reading entered
  as 012?)".  The reading is still SAVED: rejecting it would lose the
  brewer's actual observation, and seeing it recorded is how they spot the
  mistake.

142 tests (up from 134).
2026-11-09 15:55:56 +10:00

511 lines
17 KiB
Python

"""Tests for the deeper analysis: attenuation, fermentation curves, timings."""
from __future__ import annotations
from datetime import date
from pathlib import Path
import pytest
from brewing import (Run, attenuation_percent, days_to_target,
fermentation_curve, implausible_runs)
from conftest import seed_readings
def _doc(run_id="1", beer_type="Ale", sg="1045", date_="2026-01-01",
final="False", comment=""):
return {
"_id": f"id-{run_id}-{date_}-{sg}",
"beer_run_id": run_id,
"beer_type": beer_type,
"sg": sg,
"date": date_,
"final_reading": final,
"comment": comment,
}
# ------------------------------------------------------------ attenuation #
def test_attenuation_percent():
"""Attenuation is the fraction of the original gravity fermented away."""
# OG 1050, FG 1010: 40 of the 50 points above 1.000 went away.
run = Run.from_docs("1", [
_doc(sg="1050", date_="2026-01-01"),
_doc(sg="1010", date_="2026-01-20", final="True"),
])
assert attenuation_percent(run) == 80.0
def test_attenuation_percent_of_a_full_fermentation():
"""A wort that finishes at 1.000 has 100% attenuation."""
run = Run.from_docs("2", [
_doc(sg="1050", date_="2026-01-01"),
_doc(sg="1000", date_="2026-01-20", final="True"),
])
assert attenuation_percent(run) == 100.0
def test_attenuation_percent_needs_a_gravity_above_1000():
"""A run that started at or below 1.000 has no meaningful attenuation."""
run = Run.from_docs("3", [
_doc(sg="1000", date_="2026-01-01"),
_doc(sg="0995", date_="2026-01-20", final="True"),
])
assert attenuation_percent(run) is None
def test_attenuation_percent_of_an_unfinished_run_uses_its_current_gravity():
"""The current gravity stands in for the final one until the run ends."""
run = Run.from_docs("4", [
_doc(sg="1050", date_="2026-01-01"),
_doc(sg="1020", date_="2026-01-10"),
])
assert attenuation_percent(run) == 60.0
# ------------------------------------------------------- fermentation curve #
def test_fermentation_curve_is_empty_for_a_single_reading():
"""One reading is no curve."""
run = Run.from_docs("5", [_doc(sg="1045", date_="2026-01-01")])
assert fermentation_curve(run) == []
def test_fermentation_curve_progresses_from_zero_to_one():
"""The curve is the fraction of the gravity drop completed."""
run = Run.from_docs("6", [
_doc(sg="1050", date_="2026-01-01"),
_doc(sg="1030", date_="2026-01-10"),
_doc(sg="1010", date_="2026-01-20", final="True"),
])
curve = fermentation_curve(run)
assert len(curve) == 3
assert curve[0]["progress"] == 0.0
assert curve[1]["progress"] == 0.5
assert curve[2]["progress"] == 1.0
def test_fermentation_curve_carries_the_reading_details():
run = Run.from_docs("7", [
_doc(sg="1050", date_="2026-01-01", comment="brew day"),
_doc(sg="1010", date_="2026-01-20", final="True", comment="kegging"),
])
curve = fermentation_curve(run)
assert curve[0]["date"] == "2026-01-01"
assert curve[0]["sg_decimal"] == "1.050"
assert curve[0]["point_scale"] == "50"
assert curve[0]["comment"] == "brew day"
assert curve[1]["final"] is True
def test_fermentation_curve_is_empty_when_the_gravity_never_falls():
"""A run that only goes up has no drop to measure progress against."""
run = Run.from_docs("8", [
_doc(sg="1010", date_="2026-01-01"),
_doc(sg="1020", date_="2026-01-10", final="True"),
])
assert fermentation_curve(run) == []
# ------------------------------------------------------------- days to target #
def test_days_to_target():
"""How long the brew took to reach a given gravity."""
run = Run.from_docs("9", [
_doc(sg="1050", date_="2026-01-01"),
_doc(sg="1020", date_="2026-01-08"),
_doc(sg="1005", date_="2026-01-20", final="True"),
])
# 1020 is still above 1015, so the run does not reach 1015 until the
# third reading, 19 days in.
assert days_to_target(run, 1015) == 19
assert days_to_target(run, 1010) == 19
# 1020 is below 1025, so that target is reached on the second reading.
assert days_to_target(run, 1025) == 7
assert days_to_target(run, 1000) is None
def test_days_to_target_returns_none_when_never_reached():
run = Run.from_docs("10", [
_doc(sg="1050", date_="2026-01-01"),
_doc(sg="1040", date_="2026-01-20", final="True"),
])
assert days_to_target(run, 1010) is None
# ------------------------------------------------------------------- summary #
def test_run_summary_includes_the_new_analysis_fields():
from brewing import run_summary
run = Run.from_docs("11", [
_doc(sg="1050", date_="2026-01-01"),
_doc(sg="1010", date_="2026-01-20", final="True"),
])
summary = run_summary(run)
assert summary["attenuation"] == 80.0
assert len(summary["curve"]) == 2
assert summary["curve"][0]["progress"] == 0.0
def test_attenuation_is_none_for_a_final_gravity_below_water():
"""Beer cannot be less dense than water. Run 12 in the live data has a
final gravity of 900, which would otherwise show as 338% attenuation."""
run = Run.from_docs("12", [
_doc(sg="1042", date_="2025-10-05"),
_doc(sg="900", date_="2025-11-30", final="True"),
])
assert attenuation_percent(run) is None
def test_attenuation_is_none_when_it_would_exceed_100_percent():
run = Run.from_docs("13", [
_doc(sg="1050", date_="2026-01-01"),
_doc(sg="1000", date_="2026-01-20", final="True"),
])
# 1050 -> 1000 is exactly 100%, which is the limit but still valid.
assert attenuation_percent(run) == 100.0
run2 = Run.from_docs("14", [
_doc(sg="1050", date_="2026-01-01"),
_doc(sg="1010", date_="2026-01-20", final="True"),
])
assert attenuation_percent(run2) == 80.0
def test_implausible_runs_are_identified():
from brewing import implausible_runs
good = Run.from_docs("1", [
_doc(sg="1050", date_="2026-01-01"),
_doc(sg="1010", date_="2026-01-20", final="True"),
])
bad = Run.from_docs("12", [
_doc(sg="1042", date_="2025-10-05"),
_doc(sg="900", date_="2025-11-30", final="True"),
])
problems = implausible_runs([good, bad])
assert len(problems) == 1
assert problems[0]["beer_run_id"] == "12"
assert problems[0]["fg_points"] == 900
assert "below 1.000" in problems[0]["problem"]
def test_implausible_runs_ignores_good_runs():
from brewing import implausible_runs
good = Run.from_docs("1", [
_doc(sg="1050", date_="2026-01-01"),
_doc(sg="1010", date_="2026-01-20", final="True"),
])
assert implausible_runs([good]) == []
# ------------------------------------------------------- chart rendering #
def test_charts_do_not_reverse_the_gravity_axis():
"""Gravity must FALL on screen as the beer ferments.
The gravity charts originally set `reverse: true` on the y axis, which
puts the HIGH reading at the bottom of the plot - so the fermentation
curve climbed left to right and read as gravity *rising*. A brewer
reads a fermentation as a descending line, so the axis must not be
reversed. The point-scale run chart had the same bug.
"""
charts_js = (
Path(__file__).resolve().parents[1]
/ "src" / "flask" / "static" / "js" / "charts.js"
).read_text()
assert "reverse: true" not in charts_js, (
"a gravity chart is reversing the y axis; the fermentation curve "
"would climb instead of descend"
)
def test_attenuation_chart_truncates_long_names():
"""Long beer names were clipped at the left edge of the plot.
The rotated x-axis labels were cut off ("...x crying out loud"), so the
names are trimmed to a readable length before they are drawn. The
tooltip and the table below carry the full name.
"""
charts_js = (
Path(__file__).resolve().parents[1]
/ "src" / "flask" / "static" / "js" / "charts.js"
).read_text()
assert "name.length > 18" in charts_js, (
"the attenuation chart is not trimming long beer names"
)
# ------------------------------------------- suspect runs are not counted #
def _two_runs(repo):
"""One good brew and one with an impossible final gravity.
The bad run mimics the live data: run 12 has a final gravity of 900,
which is below water and makes the run look like a 56-day brew that
reached 18.64% alcohol.
"""
seed_readings(repo, [
("1", "Good ale", "1050", "2026-01-01", False, ""),
("1", "Good ale", "1010", "2026-01-10", True, ""),
# 56 days, and a final gravity below 1.000.
("12", "Bad brew", "1042", "2026-01-01", False, ""),
("12", "Bad brew", "0900", "2026-02-26", True, "typo"),
])
def _stat_values(body):
"""The figures in the analysis stat tiles, in order.
The tiles are rendered as label/value pairs, so the numbers can be
pulled out without depending on the exact markup.
"""
import re
values = re.findall(r'<div class="stat__value">([^<]+)</div>', body)
return [v.strip() for v in values]
def test_a_suspect_run_is_excluded_from_the_brew_time_average(logged_in, repo):
"""The 56-day bad run must not drag the average brew time up.
This was the review finding: the run was excluded from the attenuation
average but not from the brew-time average, so one bad reading was
poisoning two different statistics.
"""
_two_runs(repo)
response = logged_in.get("/analysis")
assert response.status_code == 200
body = response.get_data(as_text=True)
values = _stat_values(body)
assert values, "the analysis page rendered no stat tiles"
# The stat tiles are: average attenuation, spread, average brew time,
# time to 1.010. The good run took 9 days; with the bad run folded
# in the average would be (9 + 56) / 2 = 32.5 days.
assert "9d" in values, f"expected a 9 day brew time, got {values}"
def test_a_suspect_run_is_excluded_from_the_time_to_1010_average(logged_in, repo):
"""Same for the "time to 1.010" figure."""
_two_runs(repo)
response = logged_in.get("/analysis")
body = response.get_data(as_text=True)
values = _stat_values(body)
assert values
# The good run reached 1.010 in 9 days; the bad one "reached" it in 56.
assert "9d" in values, f"expected 9 days to 1.010, got {values}"
def test_the_suspect_run_is_still_listed_for_review(logged_in, repo):
"""Excluding it from the averages must not hide it.
The run has to stay visible so it can be corrected.
"""
_two_runs(repo)
response = logged_in.get("/analysis")
body = response.get_data(as_text=True)
assert "Readings that need a second look" in body
assert "0.900" in body
def test_a_suspect_run_is_flagged_in_the_api(logged_in, repo):
"""The charts fetch /api/runs, so the flag has to be in the payload."""
_two_runs(repo)
response = logged_in.get("/api/runs")
assert response.status_code == 200
payload = response.get_json()
by_run = {str(r["beer_run_id"]): r for r in payload}
assert by_run["12"]["suspect"] is True
assert by_run["1"]["suspect"] is False
def test_a_suspect_run_is_not_the_strongest_brew_in_the_history(logged_in, repo):
"""The history page showed the bad run as the strongest beer.
An impossible final gravity produced an 18.64% ABV, which made run 12
the "strongest" brew in the history and pulled the average up.
The run's own row still shows its ABV - it is real data, and hiding it
would be worse - but it must not be counted as the strongest brew, and
it must not appear in the "Strongest" figure.
"""
_two_runs(repo)
response = logged_in.get("/history")
assert response.status_code == 200
body = response.get_data(as_text=True)
# The "Strongest" tile shows the good run's 4.76%, not the bad
# run's 18.64%. Find the Strongest tile and read its value.
import re
strongest = re.search(
r"Strongest.*?stat__value[^>]*>([^<]+)<", body, re.DOTALL
)
assert strongest, "the history page has no Strongest figure"
assert "18.64" not in strongest.group(1), (
f"the suspect run was counted as the strongest brew: {strongest.group(1)}"
)
def test_the_run_detail_api_also_flags_a_suspect_run(logged_in, repo):
"""The detail endpoint carries the same flag as the list endpoint.
A consumer of /api/runs/<run_id> must be able to tell an impossible
run from a good one.
"""
_two_runs(repo)
response = logged_in.get("/api/runs/12")
assert response.status_code == 200
assert response.get_json()["suspect"] is True
response = logged_in.get("/api/runs/1")
assert response.status_code == 200
assert response.get_json()["suspect"] is False
def test_the_charts_skip_suspect_runs():
"""The chart script must leave the suspect runs out.
A suspect run's fermentation curve plunges to a gravity below water,
which would drag the median line down and stretch the envelope.
"""
charts_js = (
Path(__file__).resolve().parents[1]
/ "src" / "flask" / "static" / "js" / "charts.js"
).read_text()
assert charts_js.count("!r.suspect") >= 2, (
"the charts must filter out the runs flagged as suspect"
)
# ------------------------------------------------- rising gravity #
def test_attenuation_is_none_when_the_gravity_went_up():
"""A negative attenuation is not a number a brewer can act on.
The guard rejected a final gravity below 1.000 and one implying more
than 100% attenuation, but a run whose gravity ROSE produced a
negative figure that was averaged into the headline number.
"""
run = Run.from_docs("30", [
_doc(sg="1010", date_="2026-01-01"),
_doc(sg="1020", date_="2026-01-20", final="True"),
])
assert attenuation_percent(run) is None
def test_a_rising_gravity_run_is_reported_as_implausible():
"""It must be listed for review, not just quietly dropped."""
run = Run.from_docs("31", [
_doc(sg="1010", date_="2026-01-01"),
_doc(sg="1020", date_="2026-01-20", final="True"),
])
problems = implausible_runs([run])
assert len(problems) == 1
assert problems[0]["beer_run_id"] == "31"
assert "went up" in problems[0]["problem"]
# ------------------------------------------------- undated readings #
def test_the_original_gravity_comes_from_the_first_dated_reading():
"""An undated reading cannot be placed on the timeline.
It used to sort to the front, so it became the OG - which put the OG
after the FG and produced a negative attenuation.
"""
run = Run.from_docs("32", [
_doc(sg="1050", date_="2026-01-01"),
{"_id": "x", "beer_run_id": "32", "beer_type": "Ale", "sg": "1005",
"date": None, "final_reading": "False", "comment": ""},
_doc(sg="1010", date_="2026-01-25", final="True"),
])
assert run.original_gravity() == 1050
assert run.final_gravity() == 1010
assert attenuation_percent(run) is not None
assert attenuation_percent(run) > 0
def test_days_to_target_skips_an_undated_reading():
"""One undated reading must not lose the whole figure.
The function gave up and returned None when the first reading at or
below the target had no date, even when a later reading did.
"""
run = Run.from_docs("33", [
_doc(sg="1050", date_="2026-01-01"),
{"_id": "y", "beer_run_id": "33", "beer_type": "Ale", "sg": "1005",
"date": None, "final_reading": "False", "comment": ""},
_doc(sg="1010", date_="2026-01-25", final="True"),
])
assert days_to_target(run, 1010) == 24
# ------------------------------------------------------- the charts #
def test_the_curve_chart_excludes_active_and_suspect_runs():
"""Only finished, plausible brews belong on the curve.
An active run's curve is a partial fermentation, and the curve is
meant to describe how a completed fermentation behaves.
"""
charts_js = (
Path(__file__).resolve().parents[1]
/ "src" / "flask" / "static" / "js" / "charts.js"
).read_text()
assert "!r.active" in charts_js, (
"the curve chart must leave the active runs out"
)
def test_the_curve_band_is_filled_to_the_max_not_the_median():
"""The shaded band has to span min to max.
With one dataset and `fill: "+1"` the band was filled to the NEXT
dataset - which was the median line, so the upper half of the range
was drawn as a hairline but never shaded, and the tooltip disagreed
with the shading.
"""
charts_js = (
Path(__file__).resolve().parents[1]
/ "src" / "flask" / "static" / "js" / "charts.js"
).read_text()
assert "envelopeMax" in charts_js, (
"the band needs a separate max dataset to fill down to"
)
assert 'label: "Range max"' in charts_js
def test_the_charts_fetch_the_run_payload_once():
"""/api/runs carries every reading of every run.
Four charts each fetched it independently, so the analysis page made
two identical round-trips of the whole collection.
"""
charts_js = (
Path(__file__).resolve().parents[1]
/ "src" / "flask" / "static" / "js" / "charts.js"
).read_text()
assert charts_js.count('fetch("/api/runs"') == 1, (
"the run payload should be fetched once and shared"
)
assert "function fetchRuns()" in charts_js