smb-online/backoffice/app/tests/test_owner_auth.py

"""Flask test client / real-DB integration tests for owner login, password
reset, and the operator-facing owner-accounts endpoints (#19), per #14's
testing decision: assert on HTTP response + resulting session/DB state.
"""
import pytest

import app as app_module
import booking_db as bdb
from app import app as flask_app

CLIENT_A = "C-TEST-OWNER-A"
CLIENT_B = "C-TEST-OWNER-B"


@pytest.fixture
def client(monkeypatch):
    flask_app.config["TESTING"] = True
    flask_app.secret_key = "test-secret"
    monkeypatch.setattr(app_module, "CRM_TOKEN", "test-crm-token")
    return flask_app.test_client()


def _insert_client(client_id):
    with bdb.db.connect() as conn, conn.cursor() as cur:
        cur.execute(
            "INSERT INTO clients (client_id, business_name) VALUES (%s, %s) "
            "ON CONFLICT (client_id) DO NOTHING",
            (client_id, "Café " + client_id))
        conn.commit()


# ---- login ----

def test_login_succeeds_and_scopes_session_to_client_id(client):
    _insert_client(CLIENT_A)
    bdb.create_user(CLIENT_A, "owner@example.com", "correct horse")

    resp = client.post("/owner/login", data={
        "email": "owner@example.com", "password": "correct horse"})
    assert resp.status_code == 302
    assert resp.headers["Location"].endswith("/owner/")

    with client.session_transaction() as sess:
        assert sess["client_id"] == CLIENT_A

    dashboard = client.get("/owner/")
    assert dashboard.status_code == 200
    assert CLIENT_A in dashboard.get_data(as_text=True)


def test_login_rejects_wrong_password(client):
    _insert_client(CLIENT_A)
    bdb.create_user(CLIENT_A, "owner@example.com", "correct horse")

    resp = client.post("/owner/login", data={
        "email": "owner@example.com", "password": "wrong"})
    assert resp.status_code == 401
    with client.session_transaction() as sess:
        assert "client_id" not in sess


def test_login_rejects_unknown_email(client):
    resp = client.post("/owner/login", data={
        "email": "nobody@example.com", "password": "whatever"})
    assert resp.status_code == 401


def test_dashboard_requires_login(client):
    resp = client.get("/owner/")
    assert resp.status_code == 302
    assert "/owner/login" in resp.headers["Location"]


def test_logout_clears_session(client):
    _insert_client(CLIENT_A)
    bdb.create_user(CLIENT_A, "owner@example.com", "correct horse")
    client.post("/owner/login", data={
        "email": "owner@example.com", "password": "correct horse"})
    client.get("/owner/logout")
    with client.session_transaction() as sess:
        assert "client_id" not in sess
    assert client.get("/owner/").status_code == 302


# ---- self-service password reset ----

def test_forgot_password_gives_same_response_for_unknown_email(client):
    known = client.post("/owner/forgot-password", data={"email": "nobody@example.com"})
    assert known.status_code == 200
    _insert_client(CLIENT_A)
    bdb.create_user(CLIENT_A, "owner@example.com", "old-password")
    real = client.post("/owner/forgot-password", data={"email": "owner@example.com"})
    assert real.status_code == 200
    assert known.get_data() == real.get_data()


def test_forgot_password_creates_a_usable_reset_token(client):
    _insert_client(CLIENT_A)
    u = bdb.create_user(CLIENT_A, "owner@example.com", "old-password")
    client.post("/owner/forgot-password", data={"email": "owner@example.com"})

    with bdb.db.connect() as conn, conn.cursor() as cur:
        cur.execute(
            "SELECT token FROM password_reset_tokens WHERE user_id = %s", (u["user_id"],))
        token = cur.fetchone()["token"]

    resp = client.post(f"/owner/reset-password/{token}", data={"password": "new-password"})
    assert resp.status_code == 200
    assert "geändert" in resp.get_data(as_text=True).lower()

    refreshed = bdb.get_user(u["user_id"])
    assert bdb.verify_password(refreshed, "new-password")
    assert not bdb.verify_password(refreshed, "old-password")


def test_reset_password_token_is_single_use(client):
    u = bdb.create_user(CLIENT_A, "owner@example.com", "old-password")
    tok = bdb.create_password_reset_token(u["user_id"])

    first = client.post(f"/owner/reset-password/{tok['token']}", data={"password": "new-password"})
    assert first.status_code == 200
    second = client.post(f"/owner/reset-password/{tok['token']}", data={"password": "another-one"})
    assert second.status_code == 400


def test_reset_password_rejects_short_password(client):
    u = bdb.create_user(CLIENT_A, "owner@example.com", "old-password")
    tok = bdb.create_password_reset_token(u["user_id"])
    resp = client.post(f"/owner/reset-password/{tok['token']}", data={"password": "short"})
    assert resp.status_code == 400
    refreshed = bdb.get_user(u["user_id"])
    assert bdb.verify_password(refreshed, "old-password")


def test_reset_password_rejects_bad_token(client):
    resp = client.post("/owner/reset-password/not-a-real-token", data={"password": "new-password"})
    assert resp.status_code == 400


# ---- operator-facing endpoints ----

def test_list_owner_users_requires_crm_token(client):
    assert client.get("/api/owner_users").status_code == 403


def test_list_owner_users_returns_rows_across_clients(client):
    _insert_client(CLIENT_A)
    _insert_client(CLIENT_B)
    bdb.create_user(CLIENT_A, "a@example.com", "pw12345")
    bdb.create_user(CLIENT_B, "b@example.com", "pw12345")

    resp = client.get("/api/owner_users", headers={"X-CRM-Token": "test-crm-token"})
    assert resp.status_code == 200
    emails = {r["email"] for r in resp.get_json()["rows"]}
    assert {"a@example.com", "b@example.com"} <= emails


def test_operator_triggered_reset_requires_crm_token(client):
    u = bdb.create_user(CLIENT_A, "owner@example.com", "pw12345")
    resp = client.post(f"/api/owner_users/{u['user_id']}/reset-password")
    assert resp.status_code == 403


def test_operator_triggered_reset_creates_token_for_existing_user(client):
    _insert_client(CLIENT_A)
    u = bdb.create_user(CLIENT_A, "owner@example.com", "pw12345")

    resp = client.post(f"/api/owner_users/{u['user_id']}/reset-password",
                        headers={"X-CRM-Token": "test-crm-token"})
    assert resp.status_code == 200

    with bdb.db.connect() as conn, conn.cursor() as cur:
        cur.execute(
            "SELECT count(*) AS n FROM password_reset_tokens WHERE user_id = %s",
            (u["user_id"],))
        assert cur.fetchone()["n"] == 1


def test_operator_triggered_reset_404s_for_unknown_user(client):
    resp = client.post("/api/owner_users/U-does-not-exist/reset-password",
                        headers={"X-CRM-Token": "test-crm-token"})
    assert resp.status_code == 404