"""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) 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'
([^<]+)
', 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_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" )