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