armistace/beer_data
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
8e4770f8707fde0ad823f25f9af0f3ce2f4ea727
beer_data/tests/test_brew_advice.py
T
hermes fb1088ceec fix(ui): make the advice pop-out fail-safe against a stale stylesheet 
The screenshot from Firefox showed the advice expanded inline in the
table cell, running down the column and off the screen, with the long
and short labels both showing ("BREWING ADVICEADVICE").  The page had
the new markup and Firefox was serving a stale stylesheet.

Two fixes.

The static files now send Cache-Control: no-store.  They sent
no-cache, which means "revalidate" - and Firefox was reusing them
without revalidating.  no-store is the only one that means "do not
keep this".  The HTML pages already sent no-store.

The pop-out carries the hidden attribute, and the script removes it
when the panel opens and puts it back when it closes.  That makes the
MARKUP fail-safe rather than the stylesheet: a stylesheet that does
not know about .advice-pop at all still leaves the panel hidden, so
the advice can never appear expanded inside the cell.

Verified by removing all eight .advice-pop rules from the live
stylesheet - the panel stays hidden, 0x0, and the advice is not
rendered.  Before this it would have rendered inline.

The button still opens it on tap and on hover, and the close button
still closes it.  The desktop is unchanged: the same full headings and
the same hover pop-out.

195 tests pass.
2026-11-09 22:41:20 +10:00

1212 lines
45 KiB
Python
Raw Blame History

"""Tests for the LLM brewing advice: the prompt, the storage and the display."""
from __future__ import annotations
import json
from contextlib import contextmanager
from datetime import date
import pytest
import brew_advice
from conftest import seed_readings
# ------------------------------------------------------------ the prompt #
def test_the_prompt_gives_the_model_the_brew_and_its_readings():
"""The model needs the name, the OG and every reading."""
run = {
"canonical_name": "Nocturna Christmas beer",
"og_decimal": 1.042,
"tap_number": 3,
"notes": "dark malt, 19C",
}
reading = {"date": "2026-01-10", "sg_decimal": 1.022,
"point_scale": "1022 points", "comment": "no change since day 7",
"final": False}
history = [
{"date": "2026-01-01", "sg_decimal": 1.042, "point_scale": "1042 points",
"comment": "", "final": False},
reading,
]
prompt = brew_advice.build_prompt(run, reading, history)
assert "Nocturna Christmas beer" in prompt
assert "1.042" in prompt
assert "Fermenter tap: 3" in prompt
assert "dark malt" in prompt
# Both readings, oldest first.
assert "2026-01-01" in prompt
assert "2026-01-10" in prompt
assert "no change since day 7" in prompt
assert "1.022" in prompt
def test_the_prompt_tells_the_model_not_to_call_a_reading_impossible():
"""A reading below 1.000 is a mis-entry, but the run may still be real.
This matters: the brewer's own run 12 was a genuine long fermentation
that the app flagged as impossible. The model must not repeat that
mistake - it should point at the odd reading and keep talking about
the brew.
"""
assert "Never say a reading is impossible" in brew_advice.SYSTEM_PROMPT
assert "still be real" in brew_advice.SYSTEM_PROMPT
def test_the_prompt_names_the_reading_to_advise_about():
"""The reading being advised about must be NAMED.
The brewer may be correcting an OLDER reading, so "the most recent
reading" is the wrong one - and the advice would be about the wrong
number. The reading is marked in the list and named in the
instruction.
"""
run = {"canonical_name": "Ale"}
older = {"_id": "old", "date": "2026-01-01", "sg_decimal": 1.042,
"point_scale": "1042 points", "comment": "", "final": False}
newer = {"_id": "new", "date": "2026-01-10", "sg_decimal": 1.030,
"point_scale": "1030 points", "comment": "", "final": False}
prompt = brew_advice.build_prompt(run, older, [older, newer])
# The older one is the subject.
assert "THIS IS THE READING TO ADVISE ON" in prompt
# And it is named, with its date and gravity.
assert "2026-01-01" in prompt
assert "1.042" in prompt
# And NOT the newer one.
assert "1.03" in prompt # it is in the history, which is right
assert prompt.count("THIS IS THE READING TO ADVISE ON") == 1
def test_the_prompt_says_when_a_reading_has_been_corrected():
"""A corrected reading may differ from what it said before."""
run = {"canonical_name": "Ale"}
reading = {"_id": "r1", "date": "2026-01-01", "sg_decimal": 1.012,
"point_scale": "1012 points", "comment": "", "final": False}
plain = brew_advice.build_prompt(run, reading, [reading])
assert "just corrected" not in plain
corrected = brew_advice.build_prompt(run, reading, [reading], corrected=True)
assert "just corrected" in corrected
assert "may differ" in corrected
def test_the_prompt_marks_the_final_reading():
run = {"canonical_name": "Ale"}
reading = {"date": "2026-01-20", "sg_decimal": 1.010, "point_scale": "1010 points",
"comment": "", "final": True}
prompt = brew_advice.build_prompt(run, reading, [reading])
assert "FINAL reading" in prompt
# ---------------------------------------------------------- the LLM call #
def test_a_successful_call_returns_the_text(monkeypatch):
"""A normal reply comes back as the advice."""
captured = {}
def fake_urlopen(request, timeout=None):
captured["url"] = request.full_url
captured["body"] = json.loads(request.data.decode())
class Response:
def read(self):
return json.dumps({
"choices": [{"message": {"content": "ADVICE: Leave it alone, it is still fermenting."}}]
}).encode()
def __enter__(self):
return self
def __exit__(self, *args):
return False
return Response()
monkeypatch.setattr(brew_advice.urllib.request, "urlopen", fake_urlopen)
ok, text = brew_advice.generate_advice(
{"canonical_name": "Ale"},
{"date": "2026-01-01", "sg_decimal": 1.042, "point_scale": "1042 points",
"comment": "", "final": False},
[{"date": "2026-01-01", "sg_decimal": 1.042, "point_scale": "1042 points",
"comment": "", "final": False}],
model="glm-5.3:cloud",
)
assert ok is True
assert text == "Leave it alone, it is still fermenting."
assert captured["url"].endswith("/v1/chat/completions")
assert captured["body"]["model"] == "glm-5.3:cloud"
def test_an_unreachable_model_does_not_raise(monkeypatch):
"""A failure is recorded, not raised.
The reading has to be saved either way, and a silent failure would be
indistinguishable from "the advice is still being generated".
"""
def boom(request, timeout=None):
raise brew_advice.urllib.error.URLError("connection refused")
monkeypatch.setattr(brew_advice.urllib.request, "urlopen", boom)
ok, text = brew_advice.generate_advice(
{"canonical_name": "Ale"},
{"date": "2026-01-01", "sg_decimal": 1.042, "point_scale": "1042 points",
"comment": "", "final": False},
[],
)
assert ok is False
assert "could not be generated" in text
def test_an_empty_reply_is_a_failure(monkeypatch):
"""A model that returns nothing has not given advice."""
class Response:
def read(self):
return json.dumps({"choices": [{"message": {"content": " "}}]}).encode()
def __enter__(self):
return self
def __exit__(self, *args):
return False
monkeypatch.setattr(brew_advice.urllib.request, "urlopen",
lambda request, timeout=None: Response())
ok, text = brew_advice.generate_advice({"canonical_name": "Ale"},
{"date": "2026-01-01"}, [])
assert ok is False
assert "could not be generated" in text
def test_the_token_budget_leaves_room_for_the_reasoning(monkeypatch):
"""The reasoning models emit their thinking first, out of the same budget.
At 500 tokens the reasoning used the whole allowance and the reply
came back EMPTY - which looks like a broken model rather than a
budget problem. The budget has to cover the reasoning AND the advice.
"""
captured = {}
def fake_urlopen(request, timeout=None):
captured["body"] = json.loads(request.data.decode())
class Response:
def read(self):
return json.dumps({
"choices": [{"message": {"content": "Rouse the yeast."}}]
}).encode()
def __enter__(self):
return self
def __exit__(self, *args):
return False
return Response()
monkeypatch.setattr(brew_advice.urllib.request, "urlopen", fake_urlopen)
brew_advice.generate_advice({"canonical_name": "Ale"},
{"date": "2026-01-01"}, [])
# The reasoning comes out of the same budget as the answer, and
# `think: false` does not stop it - the models just move the working
# into the content. The budget has to cover both.
assert captured["body"]["max_tokens"] >= 8000
def test_the_model_is_asked_for_an_advice_marker():
"""The marker is what makes the split deterministic.
Guessing where the advice begins by classifying prose failed in both
directions - it emptied legitimate advice, and it let the model's
working through in full. A marker the model writes itself is exact.
"""
assert "ADVICE:" in brew_advice.SYSTEM_PROMPT
def test_the_working_before_the_marker_is_dropped():
"""Everything before the marker is the model's working."""
reply = (
"Let me think about what this reading means. The original gravity "
"is 1.055 and the current reading is 1.030, which is a drop of 25 "
"points over nine days, and I need to work out what that means for "
"the expected attenuation for this style at this temperature.\n\n"
"ADVICE:\n"
"Warm the fermenter to 20C and give the yeast a gentle swirl to "
"rouse it, then take another reading in three days."
)
stripped = brew_advice._strip_reasoning(reply)
assert stripped.startswith("Warm the fermenter")
assert "Let me think" not in stripped
assert "1.055" not in stripped
def test_a_reply_with_the_marker_is_kept_whole():
"""A legitimate advice reply is never emptied.
Whatever length it is, and whatever hedging words it happens to
contain, the advice after the marker is the advice.
"""
advice = (
"Actually the drop is fine. Perhaps it will finish this week. Wait "
"a couple of days and check again, and if it has not moved, give "
"the yeast a gentle swirl and take another reading before you "
"consider repitching anything."
)
reply = "ADVICE:\n" + advice
assert brew_advice._strip_reasoning(reply) == advice
def test_the_marker_is_found_however_the_model_formats_it():
"""The model formats the marker its own way often enough to matter.
The instruction asks for "ADVICE:", but the model writes
"**ADVICE:**", "## ADVICE:" or puts it in a code block. Each of
those has to extract the advice cleanly, with no leftover markdown.
"""
advice = "Warm it to 20C and wait."
for reply in (
f"ADVICE:\n{advice}",
f"**ADVICE:**\n{advice}",
f"## ADVICE:\n{advice}",
f"```\nADVICE:\n{advice}\n```",
f"**ADVICE:**\n```\n{advice}\n```",
f"advice:\n{advice}",
):
assert brew_advice._strip_reasoning(reply) == advice, (
f"the marker was not extracted cleanly from {reply!r}"
)
def test_a_reply_without_the_marker_is_used_as_it_stands():
"""The safe direction when the model does not use the marker.
The brewer sees a slightly rambling reply rather than no advice at
all.
"""
reply = "The gravity looks fine for day nine. Give it three more days."
assert brew_advice._strip_reasoning(reply) == reply
def test_a_reply_with_only_working_after_the_marker_is_empty():
"""Nothing after the marker means there is no advice."""
assert brew_advice._strip_reasoning("ADVICE:") == ""
def test_an_over_long_reply_keeps_the_advice_at_the_end():
"""The advice is at the END of the reply, so a plain truncation from
the front would cut it off and leave only the model's working.
That is worse than no advice at all, because it looks like advice.
"""
working = (
"I need to work through what this reading means. The original gravity "
"is 1.055 and the current reading is 1.030. " + ("padding " * 260) +
"Leave the fermenter alone, take another reading in three days and "
"only then consider repitching."
)
final = brew_advice._trim_advice(working)
assert len(final) <= brew_advice.MAX_ADVICE_CHARS
assert "Leave the fermenter alone" in final, (
"the advice was cut off and only the model's working was left"
)
def test_a_short_reply_is_not_trimmed():
"""The trim must not touch a reply that already fits."""
reply = "The gravity looks fine for day nine. Give it three more days."
assert brew_advice._trim_advice(reply) == reply
def test_the_advice_is_truncated():
"""A runaway reply must not fill the reading document."""
long_text = "ADVICE: " + ("x" * 5000)
class Response:
def read(self):
return json.dumps({"choices": [{"message": {"content": long_text}}]}).encode()
def __enter__(self):
return self
def __exit__(self, *args):
return False
import brew_advice as ba
original = ba.urllib.request.urlopen
ba.urllib.request.urlopen = lambda request, timeout=None: Response()
try:
ok, text = ba.generate_advice({"canonical_name": "Ale"},
{"date": "2026-01-01"}, [])
finally:
ba.urllib.request.urlopen = original
assert ok is True
assert len(text) <= ba.MAX_ADVICE_CHARS
# ------------------------------------------------------------- storage #
def test_the_advice_is_stored_in_its_own_field(repo):
"""The brewer's comment is never overwritten."""
seed_readings(repo, [("1", "Ale", "1042", "2026-01-01", False, "my own comment")])
doc = repo.readings.find_one({"beer_run_id": "1"})
repo.set_reading_advice(str(doc["_id"]), "The advice", ok=True)
updated = repo.readings.find_one({"beer_run_id": "1"})
assert updated["advice"] == "The advice"
assert updated["advice_ok"] is True
# The brewer's own comment is untouched.
assert updated["comment"] == "my own comment"
def test_editing_a_reading_clears_its_stale_advice(repo):
"""The advice was written for the gravity that was there before.
If the reading is edited, that advice no longer describes it, so it
is cleared and a fresh one is generated. The brewer's own comment is
never cleared.
"""
seed_readings(repo, [("1", "Ale", "1042", "2026-01-01", False, "my comment")])
doc = repo.readings.find_one({"beer_run_id": "1"})
repo.set_reading_advice(str(doc["_id"]), "THE ADVICE", True)
# Edit the reading.
repo.upsert_reading(beer_run_id="1", beer_type="Ale", sg=1030,
reading_date="2026-01-01", final_reading=False,
comment="my comment")
after = repo.readings.find_one({"beer_run_id": "1"})
assert after.get("advice") is None, "the stale advice was kept"
assert after["comment"] == "my comment", "the brewer's comment was touched"
assert after["sg"] == "1030"
def test_the_advice_is_stored_on_a_reading_with_a_string_id(repo):
"""The original app's readings have 24-character STRING ids.
This is the shape every existing reading in the live database has -
the ids were generated by the old app as random characters, not by
Mongo. A helper that only accepted an ObjectId would have matched
nothing, and the advice would silently never be stored.
The mongomock fixture generates ObjectIds, so this has to be
constructed explicitly or the bug is invisible in the tests.
"""
repo.readings.insert_one({
"_id": "l78ZidBJ3QaYn98tvB88x4uM",
"beer_run_id": "1",
"beer_type": "Ale",
"sg": "1042",
"date": "2026-01-01",
"final_reading": "False",
"comment": "First ever try",
})
modified = repo.set_reading_advice("l78ZidBJ3QaYn98tvB88x4uM",
"THE ADVICE", True)
assert modified == 1, "the advice was not stored on a string-id reading"
after = repo.readings.find_one({"_id": "l78ZidBJ3QaYn98tvB88x4uM"})
assert after["advice"] == "THE ADVICE"
assert after["comment"] == "First ever try"
def test_the_advice_is_stored_on_a_reading_with_an_object_id(repo):
"""The other shape: a reading written by Mongo itself."""
seed_readings(repo, [("1", "Ale", "1042", "2026-01-01", False, "a comment")])
doc = repo.readings.find_one({"beer_run_id": "1"})
modified = repo.set_reading_advice(str(doc["_id"]), "THE ADVICE", True)
assert modified == 1
after = repo.readings.find_one({"beer_run_id": "1"})
assert after["advice"] == "THE ADVICE"
def test_writing_the_advice_does_not_leak_a_client_per_reading():
"""A MongoClient spawns three monitor threads and pymongo does not
reap them on close() - they go only when the client is collected.
So building a client per advice write accumulated three threads and a
connection pool for every reading recorded. The advice writer shares
one client for the life of the process instead.
"""
import inspect
from views import brews as brews_module
# The writer must NOT build a repository of its own - it has to use
# the shared one. A MongoClient spawns three monitor threads and
# pymongo does not reap them on close(), so a client per advice write
# accumulated three threads and a connection pool for every reading
# recorded.
source = inspect.getsource(brews_module._start_advice)
assert "BeerRepository()" not in source, (
"the advice writer builds its own repository - that leaks a Mongo "
"client (three monitor threads) per reading"
)
assert "get_shared_repository" in source, (
"the advice writer must use the shared repository"
)
@contextmanager
def _app_with_repo(repo):
"""An application context whose repository is *repo*.
The view reads the repository from ``app.extensions``, so pointing
that at a fixture is enough to drive the view's code directly.
"""
from app_factory import create_app
app = create_app({
"TESTING": True,
"WTF_CSRF_ENABLED": False,
"SECRET_KEY": "test-secret",
"MONGO_CLIENT": repo.client,
})
# The advice must be generated even though this is a test app.
app.config["TESTING"] = False
original = app.extensions["beer_repo"]
app.extensions["beer_repo"] = lambda: repo
with app.app_context():
try:
yield app
finally:
app.extensions["beer_repo"] = original
def test_the_advice_is_about_the_reading_that_was_saved():
"""Editing an OLDER reading must not generate advice about the newest.
The advice is written against a specific reading, so the reading being
saved has to be the one the advice is about.
"""
import time
import mongomock
from mongo.repository import BeerRepository
from views import brews as brews_module
from views.brews import _start_advice
captured = {}
def fake_generate(run, reading, history, writer=None, corrected=False):
captured["reading"] = reading
captured["history"] = history
captured["corrected"] = corrected
return True, "THE ADVICE"
original = brews_module.brew_advice.generate_advice_in_background
brews_module.brew_advice.generate_advice_in_background = fake_generate
try:
# Two readings; the OLDER one is the one being saved.
repo = BeerRepository(client=mongomock.MongoClient())
old_id = repo.upsert_reading(beer_run_id="1", beer_type="Ale", sg=1042,
reading_date="2026-01-01", final_reading=False,
comment="first")
repo.upsert_reading(beer_run_id="1", beer_type="Ale", sg=1030,
reading_date="2026-01-10", final_reading=False,
comment="second")
# The advice is only generated when it is enabled, and it is off
# in testing.
with _app_with_repo(repo):
_start_advice("1", old_id)
for _ in range(40):
time.sleep(0.05)
if captured:
break
finally:
brews_module.brew_advice.generate_advice_in_background = original
assert captured, "the advice was never generated"
assert captured["reading"]["_id"] == old_id
assert captured["reading"]["sg"] == 1042, (
"the advice was generated about the wrong reading"
)
# And the PROMPT must ask about that reading - not just pass it
# through. Asserting only which dict was passed would not have
# caught the prompt still saying "give advice for the most recent
# reading", which is the bug that made the model advise about the
# wrong number.
prompt = brew_advice.build_prompt(
{"canonical_name": "Ale"},
captured["reading"],
captured["history"],
)
assert "THIS IS THE READING TO ADVISE ON" in prompt
assert "2026-01-01" in prompt
assert "most recent reading" not in prompt
def test_the_shared_repository_is_built_only_once():
"""Concurrent advice threads must not each build their own client.
An unsynchronised check-then-set on a module global let eight threads
on a cold process each build their own repository - so the fix
reintroduced the very leak it was meant to remove, just bounded by a
race instead of by the number of readings.
"""
import threading
import mongomock
import mongo.repository as R
original_shared = R._shared_repository
built = []
original_init = R.BeerRepository.__init__
def counting_init(self, *args, **kwargs):
built.append(1)
original_init(self, *args, **kwargs)
R._shared_repository = None
R.BeerRepository.__init__ = counting_init
original_build = R._build_client
R._build_client = lambda: mongomock.MongoClient()
try:
instances = []
barrier = threading.Barrier(8)
def go():
barrier.wait() # release all eight at once
instances.append(id(R.get_shared_repository()))
threads = [threading.Thread(target=go) for _ in range(8)]
for t in threads:
t.start()
for t in threads:
t.join()
finally:
R._shared_repository = original_shared
R.BeerRepository.__init__ = original_init
R._build_client = original_build
assert len(built) == 1, f"built {len(built)} repositories, expected 1"
assert len(set(instances)) == 1, "the threads got different repositories"
def test_stale_advice_cannot_overwrite_a_correction(repo):
"""A slow advice thread must not overwrite a correction.
The advice is generated in the background. If the brewer corrects a
reading while the advice for the old gravity is still being
generated, the slow thread lands afterwards - and without a guard the
reading ends up with the final gravity from the correction and the
advice from before it.
The write is conditional on the reading still holding the gravity the
advice was generated for.
"""
seed_readings(repo, [("1", "Ale", "1050", "2026-01-01", False, "a comment")])
doc = repo.readings.find_one({"beer_run_id": "1"})
reading_id = str(doc["_id"])
# The advice for the ORIGINAL 1050 arrives late...
modified = repo.set_reading_advice(reading_id, "ADVICE FOR THE 1050",
True, expected_sg=1050)
assert modified == 1
# ...but the brewer has already corrected it to 1020, and the fresh
# advice has landed.
repo.upsert_reading(beer_run_id="1", beer_type="Ale", sg=1020,
reading_date="2026-01-01", final_reading=False,
comment="a comment")
repo.set_reading_advice(reading_id, "ADVICE FOR THE 1020", True,
expected_sg=1020)
# Now the late thread for the old 1050 finally lands.
repo.set_reading_advice(reading_id, "STALE ADVICE FOR THE 1050", True,
expected_sg=1050)
after = repo.readings.find_one({"beer_run_id": "1"})
assert after["sg"] == "1020"
assert after["advice"] == "ADVICE FOR THE 1020", (
f"the stale advice overwrote the correction: {after['advice']!r}"
)
def test_a_comment_only_correction_cannot_be_overwritten(repo):
"""A comment-only correction leaves the gravity unchanged.
So the gravity alone cannot tell the two generations apart - a late
thread for the old advice would still match and overwrite the fresh
one. The editing token is what tells them apart.
"""
# Written through upsert_reading, so it has an editing token.
reading_id = repo.upsert_reading(beer_run_id="1", beer_type="Ale", sg=1050,
reading_date="2026-01-01",
final_reading=False, comment="first comment")
doc = repo.readings.find_one({"beer_run_id": "1"})
token = doc.get("advice_token")
assert token, "the reading has no editing token"
# The advice thread starts on this generation.
# The brewer corrects the COMMENT only - the gravity is unchanged.
repo.upsert_reading(beer_run_id="1", beer_type="Ale", sg=1050,
reading_date="2026-01-01", final_reading=False,
comment="corrected comment")
assert repo.readings.find_one({"beer_run_id": "1"}).get("advice_token") != token, (
"the correction did not write a fresh token"
)
# The fresh advice lands.
fresh_token = repo.readings.find_one({"beer_run_id": "1"}).get("advice_token")
assert repo.set_reading_advice(reading_id, "FRESH ADVICE", True,
expected_sg=1050,
expected_token=fresh_token) == 1
# The late thread for the OLD generation finally lands.
stale = repo.set_reading_advice(reading_id, "STALE ADVICE", True,
expected_sg=1050, expected_token=token)
assert stale == 0, "the stale advice overwrote the correction"
after = repo.readings.find_one({"beer_run_id": "1"})
assert after["advice"] == "FRESH ADVICE"
assert after["comment"] == "corrected comment"
def test_the_token_is_the_primary_guard(repo):
"""The token changes on EVERY edit, so it catches a stale generation
whatever else changed.
The gravity alone is not enough - a comment-only correction leaves it
unchanged - and it can be unparseable for some readings, so
`expected_sg` can be None. The token is what the guard relies on.
"""
reading_id = repo.upsert_reading(beer_run_id="1", beer_type="Ale", sg=1050,
reading_date="2026-01-01",
final_reading=False, comment="first")
token = repo.readings.find_one({"beer_run_id": "1"})["advice_token"]
# A direct call with no generation context is not conditional - it is
# a deliberate write, so it goes through.
assert repo.set_reading_advice(reading_id, "DIRECT ADVICE", True) == 1
# The brewer edits the reading, so the generation changes.
repo.upsert_reading(beer_run_id="1", beer_type="Ale", sg=1050,
reading_date="2026-01-01", final_reading=False,
comment="edited")
# A write for the generation that no longer exists is refused, even
# with no gravity to check against.
assert repo.set_reading_advice(reading_id, "STALE ADVICE", True,
expected_sg=None,
expected_token=token) == 0
after = repo.readings.find_one({"beer_run_id": "1"})
# The edit cleared the advice, and the stale write did not put it back.
assert after.get("advice") is None
def test_the_advice_is_written_for_a_reading_that_predates_the_token(repo):
"""None of the existing readings have a token.
They all predate it, so a reading read from the database has an EMPTY
token. An empty token is not a usable guard - there is no token to
match - so the write must not be conditioned on it, or the advice
would never be written for any of them.
"""
# Inserted directly, the way the readings already in the database
# are - no token.
repo.readings.insert_one({
"_id": "l78ZidBJ3QaYn98tvB88x4uM",
"beer_run_id": "1",
"beer_type": "Ale",
"sg": "1042",
"date": "2024-09-01",
"final_reading": "False",
"comment": "a comment",
})
run = repo.run("1")
assert run.readings[0].advice_token == "", "the reading should have no token"
# The advice flow passes the token it read, which is empty.
written = repo.set_reading_advice("l78ZidBJ3QaYn98tvB88x4uM",
"THE ADVICE", True,
expected_sg=None,
expected_token=run.readings[0].advice_token)
assert written == 1, "the advice was never written for a pre-token reading"
after = repo.readings.find_one({"beer_run_id": "1"})
assert after["advice"] == "THE ADVICE"
assert after["comment"] == "a comment"
def test_a_pre_token_reading_is_still_guarded_against_a_stale_write(repo):
"""A reading that has never been edited has no token at all.
The guard matches on the ABSENCE of a token, so an edit still tells
the generations apart. Falling back to the gravity here would
re-open the hole, because a comment-only correction leaves the
gravity unchanged.
"""
repo.readings.insert_one({
"_id": "preToken0000000000000010",
"beer_run_id": "1",
"beer_type": "Ale",
"sg": "1050",
"date": "2026-11-01",
"final_reading": "False",
"comment": "first comment",
})
# The advice thread starts on this generation - no token.
assert repo.set_reading_advice("preToken0000000000000010", "FRESH ADVICE",
True, expected_sg=None,
expected_token="") == 1
# The brewer corrects the comment. The edit writes a token.
repo.upsert_reading(beer_run_id="1", beer_type="Ale", sg=1050,
reading_date="2026-11-01", final_reading=False,
comment="corrected comment")
# The late thread for the OLD generation lands. It carries no token,
# and the reading now HAS one - so it must be refused.
stale = repo.set_reading_advice("preToken0000000000000010", "STALE ADVICE",
True, expected_sg=None,
expected_token="")
assert stale == 0, "the stale advice overwrote the correction"
after = repo.readings.find_one({"beer_run_id": "1"})
assert after.get("advice") is None
assert after["comment"] == "corrected comment"
def test_the_advice_is_written_for_a_reading_with_no_token_field(repo):
"""All three shapes of "no token" must work.
A reading read from the database has no token at all; one written by
an older version could have the field present but empty, or null.
The guard matches on the absence of a token, so it has to cover all
three - otherwise the advice is silently never written.
"""
for index, extra in enumerate(({}, {"advice_token": ""}, {"advice_token": None})):
reading_id = f"noToken{index:04d}"
repo.readings.insert_one({
"_id": reading_id,
"beer_run_id": str(index),
"beer_type": "Ale",
"sg": "1050",
"date": "2026-01-01",
"final_reading": "False",
"comment": "c",
**extra,
})
written = repo.set_reading_advice(reading_id, "THE ADVICE", True,
expected_sg=None,
expected_token="")
assert written == 1, f"the advice was not written for {extra!r}"
def test_a_reply_cut_off_at_the_token_ceiling_is_marked(repo):
"""A reply that stopped at the token limit is incomplete.
It may be advice that stops mid-sentence, so it must not be stored
as if it were the whole answer.
"""
captured = {}
def fake_urlopen(request, timeout=None):
captured["body"] = json.loads(request.data.decode())
class Response:
def read(self):
return json.dumps({
"choices": [{
"finish_reason": "length",
"message": {"content": "ADVICE: The gravity looks fine so far, but"},
}],
}).encode()
def __enter__(self):
return self
def __exit__(self, *args):
return False
return Response()
import brew_advice as ba
original = ba.urllib.request.urlopen
ba.urllib.request.urlopen = fake_urlopen
try:
ok, text = ba.generate_advice({"canonical_name": "Ale"},
{"date": "2026-01-01"}, [])
finally:
ba.urllib.request.urlopen = original
assert ok is True
assert "cut off" in text
def test_the_reading_id_cannot_be_used_to_inject_a_query(repo):
"""The reading id is coerced before it reaches Mongo.
Every value that reaches a Mongo query is coerced first, because a
caller passing a dict would have it interpreted as a query operator
rather than a value - the classic NoSQL injection route. A reading
id that is not a plain ObjectId or string must match nothing.
"""
seed_readings(repo, [
("1", "Ale", "1042", "2026-01-01", False, ""),
("2", "Ale", "1030", "2026-01-01", False, ""),
])
for hostile in [{"$ne": None}, {"$gt": ""}, ["a"], 123, None,
"not-an-objectid", ""]:
modified = repo.set_reading_advice(hostile, "INJECTED", True)
assert modified == 0, f"{hostile!r} matched something"
# The helper itself must return something that is safe to put in a
# query - an empty list matches nothing. Returning None would raise
# "$in needs an array" if it were ever used without a guard.
from mongo.repository import _as_doc_id
for hostile in [{"$ne": None}, {"$gt": ""}, ["a"], 123, None, ""]:
assert _as_doc_id(hostile) == [], f"{hostile!r} produced a usable id"
# And no document was touched.
for doc in repo.readings.find({}):
assert doc.get("advice") is None
def test_a_failed_generation_is_stored_as_a_failure(repo):
"""The note says it could not be generated, and is flagged as such."""
seed_readings(repo, [("1", "Ale", "1042", "2026-01-01", False, "")])
doc = repo.readings.find_one({"beer_run_id": "1"})
repo.set_reading_advice(str(doc["_id"]), "Advice could not be generated.", ok=False)
updated = repo.readings.find_one({"beer_run_id": "1"})
assert updated["advice_ok"] is False
def test_the_latest_advice_comes_from_the_newest_reading_with_one():
"""The advice for the newest reading wins.
The advice is generated in the background, so the newest reading may
not have one yet - in which case the newest that does is used.
"""
from brewing import Run
run = Run.from_docs("1", [
{"_id": "a", "beer_run_id": "1", "sg": "1042", "date": "2026-01-01",
"final_reading": "False", "comment": "", "advice": "first advice"},
{"_id": "b", "beer_run_id": "1", "sg": "1030", "date": "2026-01-05",
"final_reading": "False", "comment": "", "advice": "second advice"},
{"_id": "c", "beer_run_id": "1", "sg": "1020", "date": "2026-01-10",
"final_reading": "False", "comment": "", "advice": ""},
])
assert run.latest_advice() == "second advice"
assert run.latest_advice_ok() is True
# ------------------------------------------------------------- the pages #
def test_recording_a_reading_saves_it_even_when_the_model_is_down(client, repo):
"""The model being down must never stop a reading being saved."""
repo.users.insert_one({"username": "beer", "password": "beer_pass"})
client.post("/login", data={"username": "beer", "password": "beer_pass"},
follow_redirects=True)
response = client.post("/brews", data={
"beer_run_id": "99",
"beer_run_type": "Test ale",
"Date": "2026-01-01",
"sg": "1042",
"comment": "a comment",
}, follow_redirects=True)
assert response.status_code == 200
assert repo.readings.count_documents({"beer_run_id": "99"}) == 1
def test_the_advice_is_shown_on_the_brewing_page(client, repo):
"""The advice appears next to the comment, clearly labelled."""
repo.users.insert_one({"username": "beer", "password": "beer_pass"})
client.post("/login", data={"username": "beer", "password": "beer_pass"},
follow_redirects=True)
seed_readings(repo, [("1", "Ale", "1042", "2026-01-01", False, "my comment")])
doc = repo.readings.find_one({"beer_run_id": "1"})
repo.set_reading_advice(str(doc["_id"]), "Rouse the yeast and wait.", ok=True)
response = client.get("/brews")
body = response.get_data(as_text=True)
assert "Rouse the yeast and wait." in body
# The brewer's own comment is still there.
assert "my comment" in body
# And it is labelled as advice, so the two are not confused.
assert "Brewing advice" in body
def test_the_advice_is_shown_on_the_active_beers_page(client, repo):
"""The active page is where the brewer looks during a fermentation."""
repo.users.insert_one({"username": "beer", "password": "beer_pass"})
client.post("/login", data={"username": "beer", "password": "beer_pass"},
follow_redirects=True)
seed_readings(repo, [("1", "Ale", "1042", "2026-01-01", False, "my comment")])
doc = repo.readings.find_one({"beer_run_id": "1"})
repo.set_reading_advice(str(doc["_id"]), "Warm it to 21C and wait.", ok=True)
response = client.get("/active")
body = response.get_data(as_text=True)
assert "Warm it to 21C and wait." in body
assert "Brewing advice" in body
def test_the_advice_in_the_table_is_a_popout_not_the_text(client, repo):
"""The advice in the table is a trigger, and the text is in a panel.
The advice is several hundred characters. In a table cell it made
the table wider than a phone, so the column was clipped off the
right edge and the advice could not be read.
"""
repo.users.insert_one({"username": "beer", "password": "beer_pass"})
client.post("/login", data={"username": "beer", "password": "beer_pass"},
follow_redirects=True)
seed_readings(repo, [("1", "Ale", "1042", "2026-01-01", False, "")])
doc = repo.readings.find_one({"beer_run_id": "1"})
repo.set_reading_advice(str(doc["_id"]), "Warm it to 21C and wait.", ok=True)
body = client.get("/brews").get_data(as_text=True)
# The trigger is a real button, so it works with a keyboard.
assert "advice-trigger" in body
assert 'class="advice-trigger' in body
# The text lives in the panel, not in the cell.
assert "advice-pop" in body
assert "advice-pop__text" in body
# The full advice is in the page - it is just not in the cell.
assert "Warm it to 21C and wait." in body
def test_the_tables_keep_their_columns_on_a_phone(client, repo):
"""The tables are readable on a phone, not squeezed into nothing.
The advice is a button and the table scrolls, so nothing is
truncated to an unreadable stub.
"""
repo.users.insert_one({"username": "beer", "password": "beer_pass"})
client.post("/login", data={"username": "beer", "password": "beer_pass"},
follow_redirects=True)
seed_readings(repo, [("1", "Cider raspberry elderberry", "1045", "2026-01-01", False, "")])
doc = repo.readings.find_one({"beer_run_id": "1"})
repo.set_reading_advice(str(doc["_id"]), "Warm it to 21C and wait.", ok=True)
body = client.get("/brews").get_data(as_text=True)
# The long heading has a short form for a phone, and the long form
# is still in the page for a desktop.
assert "Brewing advice" in body
assert ">Advice<" in body.replace("\n", "")
# The beer name is not truncated in the markup.
assert "Cider raspberry elderberry" in body
def test_the_long_and_short_labels_are_both_in_the_page(client, repo):
"""The long and the short label are both in the markup.
CSS picks which one shows - the long form on a desktop, the short
form on a phone. If only one is in the markup the other cannot be
shown at all, and if the CSS rule misses one of them the page shows
"Read the adviceAdvice".
This is a markup assertion, not a CSS assertion: it makes sure
both forms exist for the stylesheet to choose between.
"""
repo.users.insert_one({"username": "beer", "password": "beer_pass"})
client.post("/login", data={"username": "beer", "password": "beer_pass"},
follow_redirects=True)
seed_readings(repo, [("1", "Ale", "1042", "2026-01-01", False, "")])
doc = repo.readings.find_one({"beer_run_id": "1"})
repo.set_reading_advice(str(doc["_id"]), "Warm it to 21C and wait.", ok=True)
body = client.get("/brews").get_data(as_text=True)
# The heading has both forms.
assert '<span class="long">Brewing advice</span>' in body
assert '<span class="short">Advice</span>' in body
# The button has both forms.
assert '<span class="long">Read the advice</span>' in body
assert '<span class="short">Advice</span>' in body
def test_stale_advice_cannot_overwrite_a_comment_only_correction(repo):
"""A comment-only correction still invalidates the old advice.
The gravity alone is not enough to tell two generations apart - a
comment-only correction leaves the gravity unchanged. The editing
token is what tells them apart, because every edit writes a fresh
one.
"""
repo.readings.insert_one({
"_id": "readingone",
"beer_run_id": "1",
"date": "2026-01-01",
"sg": "1040",
"comment": "original comment",
"final_reading": "False",
"advice_token": "token-one",
})
# The slow thread holds the token it was given.
stale_token = "token-one"
# The brewer corrects only the comment.
repo.readings.update_one(
{"_id": "readingone"},
{"$set": {"comment": "corrected comment", "advice_token": "token-two"},
"$unset": {"advice": "", "advice_ok": "", "advice_at": ""}},
)
# The fresh advice lands first, carrying the new token.
assert repo.set_reading_advice(
"readingone", "FRESH ADVICE", ok=True, expected_token="token-two") == 1
# The slow thread lands afterwards, carrying the old token. It must
# be refused.
assert repo.set_reading_advice(
"readingone", "STALE ADVICE", ok=True, expected_token=stale_token) == 0
doc = repo.readings.find_one({"_id": "readingone"})
assert doc["advice"] == "FRESH ADVICE"
assert doc["comment"] == "corrected comment"
def test_stale_advice_cannot_overwrite_a_comment_edited_outside_the_app(repo):
"""The comment is a guard in its own right, not just the token.
The token covers every edit the application makes. A document
edited outside it - by a script, or by hand - keeps its empty
token, so a stale thread's write would still match on the gravity
alone. Matching the comment as well closes that.
"""
repo.readings.insert_one({
"_id": "outsideedit01",
"beer_run_id": "1",
"date": "2026-01-01",
"sg": "1040",
"comment": "original comment",
"final_reading": "False",
# No advice_token - a legacy reading.
})
# The advice is written for the original comment.
assert repo.set_reading_advice(
"outsideedit01", "FRESH ADVICE", ok=True,
expected_token="", expected_comment="original comment") == 1
# Somebody edits the comment directly in the database, leaving the
# token empty - so the token cannot tell the generations apart.
repo.readings.update_one(
{"_id": "outsideedit01"},
{"$set": {"comment": "edited outside the app"}},
)
# The slow thread lands. It must be refused, even though the token
# still matches.
assert repo.set_reading_advice(
"outsideedit01", "STALE ADVICE", ok=True,
expected_token="", expected_comment="original comment") == 0
doc = repo.readings.find_one({"_id": "outsideedit01"})
assert doc["advice"] == "FRESH ADVICE"
assert doc["comment"] == "edited outside the app"
def test_the_advice_pop_out_is_hidden_in_the_markup(client, repo):
"""The pop-out carries the `hidden` attribute.
A stale stylesheet - one that does not know about `.advice-pop` at
all - would otherwise leave the advice expanded inline in the table
cell, running down the column and off the screen. The `hidden`
attribute makes the markup fail-safe: the panel stays hidden until
the script opens it.
"""
repo.users.insert_one({"username": "beer", "password": "beer_pass"})
client.post("/login", data={"username": "beer", "password": "beer_pass"},
follow_redirects=True)
seed_readings(repo, [("1", "Ale", "1042", "2026-01-01", False, "")])
doc = repo.readings.find_one({"beer_run_id": "1"})
repo.set_reading_advice(str(doc["_id"]), "Warm it to 21C and wait.", ok=True)
body = client.get("/brews").get_data(as_text=True)
assert 'class="advice-pop"' in body
assert 'hidden' in body.split('class="advice-pop"')[1][:60]
def test_a_failed_advice_is_marked_as_such(client, repo):
"""A generation failure reads differently from real advice."""
repo.users.insert_one({"username": "beer", "password": "beer_pass"})
client.post("/login", data={"username": "beer", "password": "beer_pass"},
follow_redirects=True)
seed_readings(repo, [("1", "Ale", "1042", "2026-01-01", False, "")])
doc = repo.readings.find_one({"beer_run_id": "1"})
repo.set_reading_advice(str(doc["_id"]), "Advice could not be generated.",
ok=False)
response = client.get("/brews")
body = response.get_data(as_text=True)
# A failed generation is marked differently from real advice.
assert "advice-trigger--failed" in body
assert "Advice failed" in body
def test_advice_is_not_generated_in_tests(client, repo):
"""The suite must never reach out to a real model."""
from views.brews import advice_enabled
with client.application.app_context():
assert advice_enabled() is False
Reference in New Issue View Git Blame Copy Permalink