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_the_chart_library_is_loaded_exactly_once(logged_in):
    """Regression: the chart script was included both in the base template
    and again by each page, so every chart was drawn twice."""
    for path in ("/brews", "/active", "/history"):
        body = logged_in.get(path).data.decode()
        assert body.count("js/charts.js") == 1, path
        assert body.count("chart.umd.js") == 1, path


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)


# ------------------------------------------------- active beers and metadata #

def test_active_page_shows_only_unfinished_runs(logged_in, repo):
    seed_readings(repo, [
        ("1", "Finished Beer", "1050", "2026-01-01", False, ""),
        ("1", "Finished Beer", "1010", "2026-01-20", True, ""),
        ("2", "Still Brewing", "1045", "2026-02-01", False, ""),
    ])

    response = logged_in.get("/active")
    assert response.status_code == 200
    body = response.data.decode()
    assert "Still Brewing" in body
    assert "Finished Beer" not in body


def test_active_page_shows_the_tap_number(logged_in, repo):
    seed_readings(repo, [("2", "Still Brewing", "1045", "2026-02-01", False, "")])
    repo.upsert_run_metadata("2", tap_number=4)

    body = logged_in.get("/active").data.decode()
    assert "Tap 4" in body


def test_active_page_shows_the_projected_alcohol(logged_in, repo):
    """With two readings the first is the OG, so the estimate is meaningful."""
    seed_readings(repo, [
        ("2", "Still Brewing", "1050", "2026-02-01", False, ""),
        ("2", "Still Brewing", "1020", "2026-02-10", False, ""),
    ])

    body = logged_in.get("/active").data.decode()
    # 30 point drop = 3.94%
    assert "3.94%" in body


def test_active_page_shows_a_dash_when_no_og_is_known(logged_in, repo):
    """A single reading with no recorded OG has no reliable estimate."""
    seed_readings(repo, [("2", "Still Brewing", "1045", "2026-02-01", False, "")])

    body = logged_in.get("/active").data.decode()
    assert "needs 2nd reading" in body


def test_active_page_shows_the_latest_comment(logged_in, repo):
    seed_readings(repo, [
        ("2", "Still Brewing", "1050", "2026-02-01", False, "Hops in"),
        ("2", "Still Brewing", "1030", "2026-02-10", False, "Almost there"),
    ])

    body = logged_in.get("/active").data.decode()
    assert "Almost there" in body


def test_run_metadata_is_saved_and_surfaced(logged_in, repo):
    seed_readings(repo, [("8", "Pale Ale", "1045", "2026-03-01", False, "")])

    logged_in.post("/runs/8", data={
        "name": "Canonical Name",
        "tap_number": "3",
        "og_points": "1050",
        "notes": "Starter was a 50F mess.",
    }, follow_redirects=True)

    meta = repo.run_metadata("8")
    assert meta["name"] == "Canonical Name"
    assert meta["tap_number"] == 3
    assert meta["og_points"] == 1050
    assert meta["notes"] == "Starter was a 50F mess."

    # The canonical name must appear on the brew-process page and the active page.
    assert "Canonical Name" in logged_in.get("/brews").data.decode()
    assert "Canonical Name" in logged_in.get("/active").data.decode()


def test_run_metadata_can_be_cleared(logged_in, repo):
    seed_readings(repo, [("8", "Pale Ale", "1045", "2026-03-01", False, "")])
    repo.upsert_run_metadata("8", name="Set Name", tap_number=2)

    logged_in.post("/runs/8", data={"name": "", "tap_number": "", "og_points": "", "notes": ""},
                   follow_redirects=True)

    meta = repo.run_metadata("8")
    # The name was cleared, so it falls back to the reading-derived name.
    assert not meta.get("name")
    assert meta.get("tap_number") is None


def test_the_stored_name_overrides_the_drifting_reading_name(logged_in, repo):
    seed_readings(repo, [
        ("9", "Old Name", "1050", "2026-03-01", False, ""),
        ("9", "Newer Name", "1040", "2026-03-10", False, ""),
    ])
    repo.upsert_run_metadata("9", name="The Name I Want")

    body = logged_in.get("/brews").data.decode()
    assert "The Name I Want" in body


def test_next_free_tap_number_skips_the_ones_in_use(repo):
    assert repo.next_free_tap_number() == 1

    repo.upsert_run_metadata("1", tap_number=1)
    assert repo.next_free_tap_number() == 2

    repo.upsert_run_metadata("2", tap_number=2)
    assert repo.next_free_tap_number() == 3

    # A gap is filled before extending the range.
    repo.upsert_run_metadata("3", tap_number=4)
    assert repo.next_free_tap_number() == 3


def test_history_page_shows_completed_brews(logged_in, repo):
    seed_readings(repo, [
        ("1", "Pale Ale", "1050", "2026-01-01", False, ""),
        ("1", "Pale Ale", "1010", "2026-01-20", True, "done"),
    ])

    response = logged_in.get("/history")
    assert response.status_code == 200
    body = response.data.decode()
    assert "Pale Ale" in body
    assert "5.25%" in body          # standard ABV for a 40 point drop
    assert "Legacy" in body         # the comparison table is present


def test_history_page_flags_an_implausible_reading(logged_in, repo):
    seed_readings(repo, [
        ("1", "Pale Ale", "1050", "2026-01-01", False, ""),
        ("1", "Pale Ale", "900", "2026-01-20", True, "typo?"),
    ])

    body = logged_in.get("/history").data.decode()
    assert "worth a second look" in body
    assert "typo?" in body


def test_partial_metadata_update_leaves_other_fields_alone(repo):
    """The backfill uses this: it must not wipe a tap number the operator
    has already set."""
    repo.upsert_run_metadata("1", name="A Name", tap_number=3, notes="Some notes")

    repo.upsert_run_metadata_partial("1", name="A Better Name")

    meta = repo.run_metadata("1")
    assert meta["name"] == "A Better Name"
    assert meta["tap_number"] == 3
    assert meta["notes"] == "Some notes"


def test_partial_metadata_update_ignores_none(repo):
    repo.upsert_run_metadata("1", name="A Name", tap_number=3)

    repo.upsert_run_metadata_partial("1", name=None, tap_number=None)

    meta = repo.run_metadata("1")
    assert meta["name"] == "A Name"
    assert meta["tap_number"] == 3


# ---------------------------------------------------------- security headers #

def test_security_headers_are_set(logged_in):
    """The app serves only its own vendored assets, so a strict CSP of
    'self' is enforceable."""
    response = logged_in.get("/brews")
    headers = response.headers

    assert headers.get("X-Content-Type-Options") == "nosniff"
    assert headers.get("X-Frame-Options") == "DENY"
    assert headers.get("Referrer-Policy") == "same-origin"

    csp = headers.get("Content-Security-Policy")
    assert csp is not None
    assert "default-src 'self'" in csp
    assert "script-src 'self'" in csp
    assert "object-src 'none'" in csp
    assert "frame-ancestors 'none'" in csp
    # No CDN is used, so nothing external should be allowed.
    assert "https://" not in csp
    # 'unsafe-inline' would defeat the point of the policy.
    assert "unsafe-inline" not in csp
    assert "unsafe-eval" not in csp


def test_security_headers_are_set_on_the_login_page(client):
    response = client.get("/login")
    assert response.headers.get("X-Content-Type-Options") == "nosniff"
    assert response.headers.get("Content-Security-Policy") is not None


# ------------------------------------------------------------------ indexes #

def test_the_repository_creates_its_indexes(repo):
    """The run metadata is always looked up by beer_run_id, and the readings
    by (beer_run_id, date) — both need an index."""
    index_names = repo.run_meta.index_information()
    assert "beer_run_id_1" in index_names

    reading_indexes = repo.readings.index_information()
    assert "run_date_1" in reading_indexes


def test_no_template_has_an_inline_script_or_style(client):
    """The strict CSP (script-src 'self'; style-src 'self') blocks inline
    scripts and style attributes, so the templates must have neither."""
    import glob
    import os
    import re

    template_dir = os.path.join(
        os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
        "src", "flask", "templates",
    )

    for path in glob.glob(os.path.join(template_dir, "*.html")):
        with open(path) as fh:
            body = fh.read()

        # An inline <script> is one with content rather than a src attribute.
        for match in re.finditer(r"<script(?![^>]*\bsrc=)[^>]*>(.*?)</script>",
                                 body, re.S):
            assert match.group(1).strip() == "", (
                f"{os.path.basename(path)} has an inline <script>, which the "
                "CSP blocks"
            )

        assert not re.search(r"\sstyle\s*=", body), (
            f"{os.path.basename(path)} has an inline style attribute, which "
            "the CSP blocks"
        )


# -------------------------------------------------------------------- API #

def test_api_runs_returns_json(logged_in, repo):
    seed_readings(repo, [
        ("1", "Pale Ale", "1050", "2026-01-01", False, ""),
        ("1", "Pale Ale", "1010", "2026-01-20", True, ""),
    ])

    response = logged_in.get("/api/runs")
    assert response.status_code == 200
    payload = response.get_json()
    assert len(payload) == 1
    assert payload[0]["abv"] == 5.25
    assert payload[0]["active"] is False


def test_api_run_detail_returns_json(logged_in, repo):
    seed_readings(repo, [("11", "Cider", "1045", "2026-04-01", False, "first")])

    response = logged_in.get("/api/runs/11")
    assert response.status_code == 200
    payload = response.get_json()
    assert payload["beer_run_id"] == "11"
    assert payload["name"] == "Cider"
    assert len(payload["readings"]) == 1


def test_api_run_detail_404s_for_an_unknown_run(logged_in):
    response = logged_in.get("/api/runs/does-not-exist")
    assert response.status_code == 404


def test_the_api_is_protected(client):
    """The charts fetch from the API, so it must not be public."""
    for path in ("/api/runs", "/api/runs/1"):
        response = client.get(path)
        assert response.status_code == 302, path
        assert "/login" in response.headers["Location"], path


def test_api_runs_includes_the_canonical_name_and_tap(logged_in, repo):
    seed_readings(repo, [("12", "Drifting Name", "1045", "2026-05-01", False, "")])
    repo.upsert_run_metadata("12", name="Canonical", tap_number=7)

    payload = logged_in.get("/api/runs").get_json()
    entry = next(r for r in payload if r["beer_run_id"] == "12")
    assert entry["canonical_name"] == "Canonical"
    assert entry["tap_number"] == 7


def test_the_months_chart_includes_the_gaps(logged_in, repo):
    """Months with no brews must appear with a zero, or the chart spaces the
    busy months evenly and reads as continuous activity."""
    seed_readings(repo, [
        ("1", "Spring", "1050", "2026-01-05", False, ""),
        ("1", "Spring", "1010", "2026-01-20", True, ""),
        ("2", "Autumn", "1050", "2026-06-05", False, ""),
        ("2", "Autumn", "1010", "2026-06-20", True, ""),
    ])

    from views.analysis import _month_range

    months = _month_range("2026-01", "2026-06")
    assert months == ["2026-01", "2026-02", "2026-03", "2026-04",
                      "2026-05", "2026-06"]


def test_the_months_chart_counts_zero_for_a_quiet_month(logged_in, repo):
    """The chart is fed by the API, so the quiet months are visible there."""
    seed_readings(repo, [
        ("1", "Spring", "1050", "2026-01-05", False, ""),
        ("1", "Spring", "1010", "2026-01-20", True, ""),
        ("2", "Autumn", "1050", "2026-06-05", False, ""),
        ("2", "Autumn", "1010", "2026-06-20", True, ""),
    ])

    import json
    import re

    body = logged_in.get("/history").data.decode()
    # The page embeds the month labels for the chart.
    assert "2026-06" in body


def test_month_range_covers_a_year_boundary():
    from views.analysis import _month_range

    assert _month_range("2025-11", "2026-02") == [
        "2025-11", "2025-12", "2026-01", "2026-02",
    ]
    assert _month_range("2026-05", "2026-05") == ["2026-05"]


# ------------------------------------------------------------- deep analysis #

def test_the_analysis_page_renders(logged_in, repo):
    seed_readings(repo, [
        ("1", "Pale Ale", "1050", "2026-01-01", False, ""),
        ("1", "Pale Ale", "1010", "2026-01-20", True, "done"),
        ("2", "Ginger Beer", "1045", "2026-02-01", False, ""),
        ("2", "Ginger Beer", "1015", "2026-02-15", True, "done"),
    ])

    response = logged_in.get("/analysis")
    assert response.status_code == 200
    body = response.data.decode()
    assert "Pale Ale" in body
    assert "Ginger Beer" in body
    # The attenuation figures and the summary sentence.
    assert "80.0%" in body
    assert "attenuation" in body.lower()


def test_the_analysis_page_excludes_active_runs(logged_in, repo):
    """An unfinished brew has no final reading, so it is not comparable."""
    seed_readings(repo, [
        ("1", "Finished", "1050", "2026-01-01", False, ""),
        ("1", "Finished", "1010", "2026-01-20", True, ""),
        ("2", "Still Going", "1045", "2026-02-01", False, ""),
    ])

    body = logged_in.get("/analysis").data.decode()
    assert "Finished" in body
    assert "Still Going" not in body


def test_the_analysis_page_needs_at_least_one_finished_brew(logged_in, repo):
    seed_readings(repo, [("1", "Just Started", "1045", "2026-02-01", False, "")])

    response = logged_in.get("/analysis")
    assert response.status_code == 200
    assert b"No finished brews" in response.data


def test_the_analysis_page_is_protected(client):
    response = client.get("/analysis")
    assert response.status_code == 302
    assert "/login" in response.headers["Location"]


def test_the_api_exposes_the_attenuation_and_curve(logged_in, repo):
    """The charts fetch the curve data from the API."""
    seed_readings(repo, [
        ("1", "Pale Ale", "1050", "2026-01-01", False, ""),
        ("1", "Pale Ale", "1010", "2026-01-20", True, ""),
    ])

    payload = logged_in.get("/api/runs").get_json()
    entry = payload[0]
    assert entry["attenuation"] == 80.0
    assert len(entry["curve"]) == 2
    assert entry["curve"][0]["progress"] == 0.0
    assert entry["curve"][1]["progress"] == 1.0