mivanchenko
16500c4392
Adds a session-authenticated /owner/settings blueprint for services CRUD (create/edit/deactivate), per-resource opening hours + min-notice/max-advance/ buffer, and client-level auto_confirm/notify_channel — all scoped to the logged-in owner's own client_id. Extends booking_db.py with the missing tenant-scoped update_service/update_resource/update_client writes, mirroring the existing update_booking allowlist pattern. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
248 lines
10 KiB
Python
248 lines
10 KiB
Python
"""Flask test client / real-DB integration tests for the owner settings
|
|
page: services CRUD, per-resource hours/notice/buffer, auto_confirm, and
|
|
notify_channel (#21). Same testing decision as #20: assert on HTTP response
|
|
+ resulting DB state, real Postgres.
|
|
"""
|
|
from datetime import date, time, timedelta
|
|
|
|
import pytest
|
|
|
|
import booking_db as bdb
|
|
from app import app as flask_app
|
|
|
|
CLIENT_A = "C-TEST-OWNER-SETTINGS-A"
|
|
CLIENT_B = "C-TEST-OWNER-SETTINGS-B"
|
|
|
|
|
|
@pytest.fixture
|
|
def client():
|
|
flask_app.config["TESTING"] = True
|
|
flask_app.secret_key = "test-secret"
|
|
return flask_app.test_client()
|
|
|
|
|
|
def _setup(client_id=CLIENT_A, **resource_kwargs):
|
|
with bdb.db.connect() as conn, conn.cursor() as cur:
|
|
cur.execute(
|
|
"INSERT INTO clients (client_id, timezone, auto_confirm, notify_channel) "
|
|
"VALUES (%s, %s, %s, %s) ON CONFLICT (client_id) DO UPDATE SET "
|
|
"timezone = EXCLUDED.timezone, auto_confirm = EXCLUDED.auto_confirm, "
|
|
"notify_channel = EXCLUDED.notify_channel",
|
|
(client_id, "Europe/Berlin", True, None))
|
|
conn.commit()
|
|
resource = bdb.create_resource(client_id, "Chair 1", **resource_kwargs)
|
|
service = bdb.create_service(client_id, "Haircut", 60, price=25)
|
|
return resource, service
|
|
|
|
|
|
def _login(client, client_id, email="owner@example.com", password="correct horse"):
|
|
bdb.create_user(client_id, email, password)
|
|
client.post("/owner/login", data={"email": email, "password": password})
|
|
|
|
|
|
# ---- auth gate ----
|
|
|
|
def test_settings_requires_login(client):
|
|
resp = client.get("/owner/settings")
|
|
assert resp.status_code == 302
|
|
assert "/owner/login" in resp.headers["Location"]
|
|
|
|
|
|
# ---- services ----
|
|
|
|
def test_owner_can_create_service(client):
|
|
_setup(CLIENT_A)
|
|
_login(client, CLIENT_A)
|
|
resp = client.post("/owner/settings/services", data={
|
|
"name": "Coloring", "duration_minutes": "90", "price": "60"})
|
|
assert resp.status_code == 302
|
|
assert "error" not in resp.headers["Location"]
|
|
names = {s["name"] for s in bdb.list_services(CLIENT_A)}
|
|
assert "Coloring" in names
|
|
|
|
|
|
def test_create_service_rejects_missing_fields(client):
|
|
_setup(CLIENT_A)
|
|
_login(client, CLIENT_A)
|
|
resp = client.post("/owner/settings/services", data={"name": "", "duration_minutes": "90"})
|
|
assert "error=invalid_service" in resp.headers["Location"]
|
|
assert len(bdb.list_services(CLIENT_A)) == 1 # only the seed service from _setup
|
|
|
|
|
|
def test_owner_can_update_and_deactivate_service(client):
|
|
resource, service = _setup(CLIENT_A)
|
|
_login(client, CLIENT_A)
|
|
resp = client.post(f"/owner/settings/services/{service['service_id']}", data={
|
|
"name": "Haircut Deluxe", "duration_minutes": "45", "price": "30"})
|
|
assert "error" not in resp.headers["Location"]
|
|
updated = bdb.get_service(CLIENT_A, service["service_id"])
|
|
assert updated["name"] == "Haircut Deluxe"
|
|
assert updated["duration_minutes"] == 45
|
|
assert updated["active"] is False # checkbox omitted from form data == unchecked
|
|
|
|
resp = client.post(f"/owner/settings/services/{service['service_id']}", data={
|
|
"name": "Haircut Deluxe", "duration_minutes": "45", "price": "30", "active": "on"})
|
|
assert bdb.get_service(CLIENT_A, service["service_id"])["active"] is True
|
|
|
|
|
|
def test_owner_cannot_update_another_tenants_service(client):
|
|
resource_b, service_b = _setup(CLIENT_B)
|
|
_setup(CLIENT_A)
|
|
_login(client, CLIENT_A)
|
|
resp = client.post(f"/owner/settings/services/{service_b['service_id']}", data={
|
|
"name": "Hijacked", "duration_minutes": "30", "active": "on"})
|
|
assert "error=not_found" in resp.headers["Location"]
|
|
assert bdb.get_service(CLIENT_B, service_b["service_id"])["name"] == "Haircut"
|
|
|
|
|
|
def test_deactivated_service_disappears_from_public_picker(client):
|
|
resource, service = _setup(CLIENT_A)
|
|
_login(client, CLIENT_A)
|
|
client.post(f"/owner/settings/services/{service['service_id']}", data={
|
|
"name": "Haircut", "duration_minutes": "60", "price": "25"})
|
|
assert bdb.list_active_services(CLIENT_A) == []
|
|
assert len(bdb.list_services(CLIENT_A)) == 1
|
|
|
|
|
|
# ---- resource availability config ----
|
|
|
|
def test_owner_can_update_resource_notice_and_buffer(client):
|
|
resource, service = _setup(CLIENT_A)
|
|
_login(client, CLIENT_A)
|
|
resp = client.post(f"/owner/settings/resources/{resource['resource_id']}", data={
|
|
"min_notice_minutes": "120", "max_advance_days": "14", "buffer_minutes": "15"})
|
|
assert "error" not in resp.headers["Location"]
|
|
updated = bdb.get_resource(CLIENT_A, resource["resource_id"])
|
|
assert updated["min_notice_minutes"] == 120
|
|
assert updated["max_advance_days"] == 14
|
|
assert updated["buffer_minutes"] == 15
|
|
|
|
|
|
def test_update_resource_rejects_negative_values(client):
|
|
resource, service = _setup(CLIENT_A)
|
|
_login(client, CLIENT_A)
|
|
resp = client.post(f"/owner/settings/resources/{resource['resource_id']}", data={
|
|
"min_notice_minutes": "-1", "max_advance_days": "14", "buffer_minutes": "15"})
|
|
assert "error=invalid_resource" in resp.headers["Location"]
|
|
|
|
|
|
def test_owner_cannot_update_another_tenants_resource(client):
|
|
resource_b, service_b = _setup(CLIENT_B)
|
|
_setup(CLIENT_A)
|
|
_login(client, CLIENT_A)
|
|
resp = client.post(f"/owner/settings/resources/{resource_b['resource_id']}", data={
|
|
"min_notice_minutes": "0", "max_advance_days": "1", "buffer_minutes": "0"})
|
|
assert "error=not_found" in resp.headers["Location"]
|
|
assert bdb.get_resource(CLIENT_B, resource_b["resource_id"])["max_advance_days"] != 1
|
|
|
|
|
|
def test_owner_can_set_opening_hours(client):
|
|
resource, service = _setup(CLIENT_A)
|
|
_login(client, CLIENT_A)
|
|
form = {"opens_at_0": "09:00", "closes_at_0": "17:00", "closed_1": "on"}
|
|
for weekday in range(2, 7):
|
|
form[f"closed_{weekday}"] = "on"
|
|
resp = client.post(
|
|
f"/owner/settings/resources/{resource['resource_id']}/hours", data=form)
|
|
assert "error" not in resp.headers["Location"]
|
|
hours = bdb.get_resource_hours(CLIENT_A, resource["resource_id"])
|
|
assert hours[0] == (time(9, 0), time(17, 0))
|
|
assert 1 not in hours
|
|
|
|
|
|
def test_set_opening_hours_rejects_end_before_start(client):
|
|
resource, service = _setup(CLIENT_A)
|
|
_login(client, CLIENT_A)
|
|
form = {"opens_at_0": "17:00", "closes_at_0": "09:00"}
|
|
for weekday in range(1, 7):
|
|
form[f"closed_{weekday}"] = "on"
|
|
resp = client.post(
|
|
f"/owner/settings/resources/{resource['resource_id']}/hours", data=form)
|
|
assert "error=invalid_hours" in resp.headers["Location"]
|
|
|
|
|
|
def test_owner_cannot_set_hours_for_another_tenants_resource(client):
|
|
resource_b, service_b = _setup(CLIENT_B)
|
|
_setup(CLIENT_A)
|
|
_login(client, CLIENT_A)
|
|
form = {"opens_at_0": "09:00", "closes_at_0": "17:00"}
|
|
for weekday in range(1, 7):
|
|
form[f"closed_{weekday}"] = "on"
|
|
resp = client.post(
|
|
f"/owner/settings/resources/{resource_b['resource_id']}/hours", data=form)
|
|
assert "error=not_found" in resp.headers["Location"]
|
|
assert bdb.get_resource_hours(CLIENT_B, resource_b["resource_id"]) == {}
|
|
|
|
|
|
def test_availability_change_reflected_in_slot_generation(client):
|
|
resource, service = _setup(CLIENT_A, min_notice_minutes=0, max_advance_days=365)
|
|
_login(client, CLIENT_A)
|
|
form = {"opens_at_0": "09:00", "closes_at_0": "10:00"}
|
|
for weekday in range(1, 7):
|
|
form[f"closed_{weekday}"] = "on"
|
|
client.post(f"/owner/settings/resources/{resource['resource_id']}/hours", data=form)
|
|
updated_hours = bdb.get_resource_hours(CLIENT_A, resource["resource_id"])
|
|
assert updated_hours[0] == (time(9, 0), time(10, 0))
|
|
|
|
|
|
# ---- client-level booking settings ----
|
|
|
|
def test_owner_can_toggle_auto_confirm(client):
|
|
_setup(CLIENT_A)
|
|
_login(client, CLIENT_A)
|
|
resp = client.post("/owner/settings/client", data={"notify_channel": ""})
|
|
assert "error" not in resp.headers["Location"]
|
|
assert bdb.get_client(CLIENT_A)["auto_confirm"] is False
|
|
|
|
resp = client.post("/owner/settings/client", data={
|
|
"auto_confirm": "on", "notify_channel": ""})
|
|
assert bdb.get_client(CLIENT_A)["auto_confirm"] is True
|
|
|
|
|
|
def test_owner_can_set_notify_channel_to_telegram(client):
|
|
_setup(CLIENT_A)
|
|
_login(client, CLIENT_A)
|
|
resp = client.post("/owner/settings/client", data={
|
|
"auto_confirm": "on", "notify_channel": "telegram"})
|
|
assert "error" not in resp.headers["Location"]
|
|
assert bdb.get_client(CLIENT_A)["notify_channel"] == "telegram"
|
|
|
|
|
|
def test_invalid_notify_channel_is_rejected(client):
|
|
_setup(CLIENT_A)
|
|
_login(client, CLIENT_A)
|
|
resp = client.post("/owner/settings/client", data={
|
|
"auto_confirm": "on", "notify_channel": "sms"})
|
|
assert "error=invalid_notify_channel" in resp.headers["Location"]
|
|
assert bdb.get_client(CLIENT_A)["notify_channel"] is None
|
|
|
|
|
|
def _next_monday(after):
|
|
d = after + timedelta(days=1)
|
|
while d.weekday() != 0:
|
|
d += timedelta(days=1)
|
|
return d
|
|
|
|
|
|
def test_auto_confirm_toggle_drives_new_public_booking_status(client):
|
|
resource, service = _setup(CLIENT_A, min_notice_minutes=0, max_advance_days=365)
|
|
_login(client, CLIENT_A)
|
|
day = _next_monday(date.today())
|
|
form = {"opens_at_0": "09:00", "closes_at_0": "17:00"}
|
|
for weekday in range(1, 7):
|
|
form[f"closed_{weekday}"] = "on"
|
|
client.post(f"/owner/settings/resources/{resource['resource_id']}/hours", data=form)
|
|
client.post("/owner/settings/client", data={"notify_channel": ""}) # auto_confirm off
|
|
assert bdb.get_client(CLIENT_A)["auto_confirm"] is False
|
|
|
|
slots = client.get("/api/booking/slots", query_string={
|
|
"client_id": CLIENT_A, "resource_id": resource["resource_id"],
|
|
"service_id": service["service_id"],
|
|
"date_from": day.isoformat(), "date_to": day.isoformat()}).get_json()["slots"]
|
|
resp = client.post("/api/booking", json={
|
|
"client_id": CLIENT_A, "resource_id": resource["resource_id"],
|
|
"service_id": service["service_id"], "start_time": slots[0],
|
|
"customer_name": "Ivy", "customer_contact": "ivy@example.com"})
|
|
assert resp.status_code == 201
|
|
assert resp.get_json()["status"] == "pending"
|