beer_data/tests/test_app.py

"""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_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)