hermes
21fa0d9a12
_as_doc_id returned None for a hostile id. That was guarded at the one call site, but the helper could still be misused into "$in needs an array" - and a caller who forgot the guard would have got an error rather than a safe no-op. It now returns an EMPTY LIST for anything that is not a reading id. An empty list is a valid $in argument that matches nothing, so the helper cannot produce a query error, cannot match every document, and cannot be interpreted as a query operator. Verified against the live database: a hostile id matches 0 documents and updates 0 documents. 164 tests.
541 lines
19 KiB
Python
541 lines
19 KiB
Python
"""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_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_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):
|
|
captured["reading"] = reading
|
|
captured["history"] = history
|
|
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"
|
|
)
|
|
|
|
|
|
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_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
|