hermes
6e1189e319
The round-2 review noted that `_upgrade_password` catches `PyMongoError` without logging it. A failed upgrade is harmless to the login itself - the user still gets in - but it leaves the password stored in plain text, and that is exactly the kind of thing an operator needs to know about. It is now logged as a warning with the username and the traceback. A test covers the path: the migration write is forced to fail, the login still succeeds, and the warning is emitted.
306 lines
10 KiB
Python
306 lines
10 KiB
Python
"""Tests for the web layer: login, the brew-process page and the API."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from conftest import seed_readings
|
|
|
|
|
|
@pytest.fixture()
|
|
def logged_in(client, repo):
|
|
"""A client with an authenticated session and a user in the database."""
|
|
repo.users.insert_one({"username": "beer", "password": "beer_pass"})
|
|
|
|
response = client.post("/login", data={"username": "beer", "password": "beer_pass"},
|
|
follow_redirects=True)
|
|
assert response.status_code == 200
|
|
return client
|
|
|
|
|
|
# ------------------------------------------------------------------ public #
|
|
|
|
def test_healthz_is_public(client):
|
|
response = client.get("/healthz")
|
|
assert response.status_code == 200
|
|
assert response.json == {"status": "ok"}
|
|
|
|
|
|
def test_readyz_checks_the_database(client):
|
|
response = client.get("/readyz")
|
|
assert response.status_code == 200
|
|
assert response.json == {"status": "ready"}
|
|
|
|
|
|
def test_readyz_reports_not_ready_when_mongo_is_down(client, monkeypatch):
|
|
"""A pod that cannot reach Mongo must not serve traffic."""
|
|
from pymongo.errors import ServerSelectionTimeoutError
|
|
|
|
def boom(*args, **kwargs):
|
|
raise ServerSelectionTimeoutError("no mongo")
|
|
|
|
monkeypatch.setattr("mongo.repository.BeerRepository.__init__",
|
|
lambda self, **kw: (_ for _ in ()).throw(ServerSelectionTimeoutError("no mongo")))
|
|
response = client.get("/readyz")
|
|
assert response.status_code == 503
|
|
|
|
|
|
def test_root_redirects_to_login(client):
|
|
response = client.get("/")
|
|
assert response.status_code == 302
|
|
assert "/login" in response.headers["Location"]
|
|
|
|
|
|
def test_login_page_renders(client):
|
|
response = client.get("/login")
|
|
assert response.status_code == 200
|
|
assert b"Brew Log" in response.data
|
|
|
|
|
|
def test_login_with_bad_credentials_is_rejected(client, repo):
|
|
repo.users.insert_one({"username": "beer", "password": "beer_pass"})
|
|
|
|
response = client.post("/login", data={"username": "beer", "password": "wrong"},
|
|
follow_redirects=True)
|
|
assert response.status_code == 200
|
|
assert b"not recognised" in response.data
|
|
|
|
|
|
def test_protected_pages_redirect_when_logged_out(client):
|
|
for path in ("/brews", "/runs/1"):
|
|
response = client.get(path)
|
|
assert response.status_code == 302, path
|
|
assert "/login" in response.headers["Location"], path
|
|
|
|
|
|
# ------------------------------------------------------------- brew pages #
|
|
|
|
def test_brews_page_lists_runs(logged_in, repo):
|
|
seed_readings(repo, [
|
|
("1", "Pale Ale", "1050", "2026-01-01", False, "start"),
|
|
("1", "Pale Ale", "1010", "2026-01-20", True, "Keghing"),
|
|
("2", "Ginger Beer", "1045", "2026-02-01", False, ""),
|
|
])
|
|
|
|
response = logged_in.get("/brews")
|
|
assert response.status_code == 200
|
|
body = response.data.decode()
|
|
assert "Pale Ale" in body
|
|
assert "Ginger Beer" in body
|
|
# The comment on the reading should be surfaced.
|
|
assert "Keghing" in body
|
|
|
|
|
|
def test_run_detail_page_renders(logged_in, repo):
|
|
seed_readings(repo, [
|
|
("1", "Pale Ale", "1050", "2026-01-01", False, "start"),
|
|
("1", "Pale Ale", "1010", "2026-01-20", True, "Keghing"),
|
|
])
|
|
|
|
response = logged_in.get("/runs/1")
|
|
assert response.status_code == 200
|
|
body = response.data.decode()
|
|
assert "Pale Ale" in body
|
|
assert "Keghing" in body
|
|
assert "5.25%" in body
|
|
|
|
|
|
# ------------------------------------------------------------- write paths #
|
|
|
|
def test_recording_a_reading_stores_it(logged_in, repo):
|
|
response = logged_in.post("/brews", data={
|
|
"beer_run_id": "5",
|
|
"beer_run_type": "Test Porter",
|
|
"Date": "2026-03-04",
|
|
"sg": "1045",
|
|
"comment": "Hops in",
|
|
}, follow_redirects=True)
|
|
assert response.status_code == 200
|
|
|
|
doc = repo.readings.find_one({"beer_run_id": "5"})
|
|
assert doc is not None
|
|
assert doc["sg"] == "1045"
|
|
assert doc["beer_type"] == "Test Porter"
|
|
assert doc["comment"] == "Hops in"
|
|
assert doc["final_reading"] == "False"
|
|
|
|
|
|
def test_recording_a_reading_twice_updates_in_place(logged_in, repo):
|
|
"""The unique key is (date, beer_run_id) — no duplicate rows."""
|
|
for sg in ("1045", "1035"):
|
|
logged_in.post("/brews", data={
|
|
"beer_run_id": "6",
|
|
"beer_run_type": "Test",
|
|
"Date": "2026-03-04",
|
|
"sg": sg,
|
|
})
|
|
|
|
count = repo.readings.count_documents({"beer_run_id": "6"})
|
|
doc = repo.readings.find_one({"beer_run_id": "6"})
|
|
assert count == 1
|
|
assert doc["sg"] == "1035"
|
|
|
|
|
|
def test_setting_a_final_reading_closes_the_run(logged_in, repo):
|
|
seed_readings(repo, [("7", "Ale", "1045", "2026-03-01", False, "")])
|
|
|
|
response = logged_in.post("/runs/7/final", data={"date": "2026-03-01"},
|
|
follow_redirects=True)
|
|
assert response.status_code == 200
|
|
|
|
doc = repo.readings.find_one({"beer_run_id": "7"})
|
|
assert doc["final_reading"] == "True"
|
|
|
|
|
|
def test_unsetting_a_final_reading_reactivates_the_run(logged_in, repo):
|
|
seed_readings(repo, [
|
|
("7", "Ale", "1045", "2026-03-01", False, ""),
|
|
("7", "Ale", "1010", "2026-03-20", True, ""),
|
|
])
|
|
|
|
logged_in.post("/runs/7/unfinal", data={"date": "2026-03-20"}, follow_redirects=True)
|
|
|
|
doc = repo.readings.find_one({"beer_run_id": "7", "date": "2026-03-20"})
|
|
assert doc["final_reading"] == "False"
|
|
|
|
|
|
# -------------------------------------------------------------------- API #
|
|
|
|
|
|
# ----------------------------------------------------------------- misc #
|
|
|
|
|
|
def test_static_assets_are_served(client):
|
|
for path in ("/static/js/app.js", "/static/css/app.css",
|
|
"/static/img/logo.svg"):
|
|
response = client.get(path)
|
|
assert response.status_code == 200, path
|
|
assert len(response.data) > 100, path
|
|
|
|
|
|
def test_logout_clears_the_session(logged_in):
|
|
response = logged_in.post("/logout", follow_redirects=True)
|
|
assert response.status_code == 200
|
|
# After logging out the protected page bounces back to the login page.
|
|
response = logged_in.get("/brews")
|
|
assert response.status_code == 302
|
|
assert "/login" in response.headers["Location"]
|
|
|
|
|
|
def test_out_of_range_reading_is_saved_but_warned(logged_in, repo):
|
|
response = logged_in.post("/brews", data={
|
|
"beer_run_id": "12",
|
|
"beer_run_type": "Odd",
|
|
"Date": "2026-05-01",
|
|
"sg": "950",
|
|
}, follow_redirects=True)
|
|
assert response.status_code == 200
|
|
assert b"outside the usual hydrometer range" in response.data
|
|
# It is still stored — the warning is advisory, not a rejection.
|
|
assert repo.readings.find_one({"beer_run_id": "12"}) is not None
|
|
|
|
|
|
# ------------------------------------------------------- security: passwords #
|
|
|
|
def test_legacy_plaintext_password_is_accepted_and_migrated(repo):
|
|
"""A legacy plaintext row still logs in, and is rewritten as a hash on
|
|
the way in so plaintext credentials disappear from the database."""
|
|
repo.users.insert_one({"username": "beer", "password": "beer_pass"})
|
|
|
|
assert repo.user_check("beer", "beer_pass") is True
|
|
|
|
doc = repo.users.find_one({"username": "beer"})
|
|
assert doc["password"] != "beer_pass"
|
|
assert "$" in doc["password"] # a werkzeug hash
|
|
assert "password_migrated_at" in doc
|
|
|
|
|
|
def test_a_failed_password_migration_still_lets_the_login_succeed(repo, caplog):
|
|
"""If the upgrade write fails, the login must still work — but the
|
|
failure is logged so the operator knows the password is still plaintext."""
|
|
import logging
|
|
|
|
from pymongo.errors import PyMongoError
|
|
|
|
repo.users.insert_one({"username": "beer", "password": "beer_pass"})
|
|
|
|
def boom(*args, **kwargs):
|
|
raise PyMongoError("write failed")
|
|
|
|
original = repo.users.update_one
|
|
repo.users.update_one = boom
|
|
|
|
with caplog.at_level(logging.WARNING):
|
|
assert repo.user_check("beer", "beer_pass") is True
|
|
|
|
repo.users.update_one = original
|
|
|
|
assert any("plain text" in rec.message for rec in caplog.records)
|
|
|
|
|
|
def test_hashed_password_verifies(repo):
|
|
from mongo.repository import hash_password
|
|
|
|
repo.users.insert_one({"username": "beer", "password": hash_password("beer_pass")})
|
|
|
|
assert repo.user_check("beer", "beer_pass") is True
|
|
assert repo.user_check("beer", "wrong") is False
|
|
|
|
|
|
def test_user_check_rejects_empty_credentials(repo):
|
|
repo.users.insert_one({"username": "beer", "password": "beer_pass"})
|
|
|
|
assert repo.user_check("", "") is False
|
|
assert repo.user_check("beer", "") is False
|
|
assert repo.user_check("", "beer_pass") is False
|
|
|
|
|
|
def test_user_check_rejects_a_nosql_operator_payload(repo):
|
|
"""A dict must never be able to act as a Mongo query operator."""
|
|
repo.users.insert_one({"username": "beer", "password": "beer_pass"})
|
|
|
|
assert repo.user_check({"$ne": ""}, "anything") is False
|
|
assert repo.user_check({"$gt": ""}, {"$ne": ""}) is False
|
|
|
|
|
|
def test_run_id_cannot_be_used_as_an_operator(repo):
|
|
"""A run id is coerced to a string before it reaches a query."""
|
|
from mongo.repository import _as_id
|
|
|
|
assert _as_id({"$ne": ""}) == ""
|
|
assert _as_id(20) == "20"
|
|
assert _as_id("20") == "20"
|
|
|
|
|
|
def test_the_app_refuses_to_start_without_a_secret_key(monkeypatch):
|
|
"""A missing SECRET_KEY must stop the app, not fall back to a default."""
|
|
import importlib
|
|
|
|
import app_factory
|
|
|
|
# Reload so the guard reads the environment as it is *now*. The exception
|
|
# class must be taken from the reloaded module — reloading rebinds it, so
|
|
# a class imported earlier would be a different object and pytest.raises
|
|
# would not match it.
|
|
monkeypatch.delenv("SECRET_KEY", raising=False)
|
|
importlib.reload(app_factory)
|
|
with pytest.raises(app_factory.ImproperlyConfigured):
|
|
app_factory.create_app()
|
|
|
|
monkeypatch.setenv("SECRET_KEY", "test-secret-key-not-a-real-secret")
|
|
importlib.reload(app_factory)
|
|
|
|
|
|
def test_the_app_refuses_the_placeholder_secret_key(monkeypatch):
|
|
import importlib
|
|
|
|
import app_factory
|
|
|
|
monkeypatch.setenv("SECRET_KEY", "change-me-in-the-deployment")
|
|
importlib.reload(app_factory)
|
|
with pytest.raises(app_factory.ImproperlyConfigured):
|
|
app_factory.create_app()
|
|
|
|
monkeypatch.setenv("SECRET_KEY", "test-secret-key-not-a-real-secret")
|
|
importlib.reload(app_factory)
|