hermes
d0aef7a787
Two things I wanted to be sure of before this merges.
Stale advice
------------
The advice is written for the gravity that was there when it was
generated. If the brewer then edits the reading - to fix a typo, or to
mark it final - the old advice no longer describes it, and the page was
showing advice that contradicted the number next to it.
`upsert_reading` now clears the advice fields, so the stale advice goes
and a fresh one is generated (the form already regenerates on every save).
Only the advice fields are cleared - the brewer's own comment is never
touched.
The injection guard
-------------------
The reading's `_id` is an ObjectId, so the original `_as_id` coercion
(which turns run ids into plain strings, to stop a caller passing a dict
that Mongo would interpret as a query operator) did not apply - it turned
the ObjectId into an empty string and the update silently matched
nothing. `_as_doc_id` accepts only an ObjectId or a string that parses as
one.
Verified against a live injection attempt: `{"$ne": None}`, `{"$gt": ""}`,
a list, an int, None and a non-ObjectId string all return None and match
nothing. There is now a test that pins it, because it is a security
property that would otherwise only be covered by reading the code.
160 tests (up from 158).
385 lines
14 KiB
Python
385 lines
14 KiB
Python
"""Tests for the LLM brewing advice: the prompt, the storage and the display."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
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_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": " 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"}, [])
|
|
|
|
assert captured["body"]["max_tokens"] >= 2000
|
|
|
|
|
|
def test_the_advice_is_truncated():
|
|
"""A runaway reply must not fill the reading document."""
|
|
long_text = "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_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"
|
|
|
|
# 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_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)
|
|
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
|