Owner calendar view + manual booking + owner cancel/reschedule (#20)

Adds an owner-authenticated weekly agenda (grouped by day, today highlighted)
with manual walk-in/phone booking creation, cancel, and reschedule -- all
routed through booking_api.py's create/cancel/reschedule logic (refactored
into shared helpers) so the EXCLUDE overlap constraint and confirmation
email stay on the single existing code path. Manual creation can skip the
opening-hours/min-notice/max-advance/buffer checks via an explicit override,
but never the overlap constraint itself.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
2026-08-03 17:16:19 +02:00
parent 59c35ce39f
commit 528a13ca7c
8 changed files with 678 additions and 56 deletions
+2
View File
@@ -25,12 +25,14 @@ from booking_api import bp as booking_bp
from public_booking import bp as public_booking_bp from public_booking import bp as public_booking_bp
from manage_booking import bp as manage_booking_bp from manage_booking import bp as manage_booking_bp
from owner_auth import bp as owner_auth_bp from owner_auth import bp as owner_auth_bp
from owner_booking import bp as owner_booking_bp
app = Flask(__name__, static_folder="static", static_url_path="") app = Flask(__name__, static_folder="static", static_url_path="")
app.register_blueprint(booking_bp) app.register_blueprint(booking_bp)
app.register_blueprint(public_booking_bp) app.register_blueprint(public_booking_bp)
app.register_blueprint(manage_booking_bp) app.register_blueprint(manage_booking_bp)
app.register_blueprint(owner_auth_bp) app.register_blueprint(owner_auth_bp)
app.register_blueprint(owner_booking_bp)
# Dedicated secret for the owner-login session cookie -- deliberately not # Dedicated secret for the owner-login session cookie -- deliberately not
# shared with CRM_API_TOKEN or BOOKING_TOKEN_SECRET (#19), same reasoning as # shared with CRM_API_TOKEN or BOOKING_TOKEN_SECRET (#19), same reasoning as
+133 -52
View File
@@ -96,6 +96,119 @@ def _available_slots(client_id, resource, tz_name, duration_minutes, date_from,
busy=busy) busy=busy)
class BookingRequestError(Exception):
"""Base for the errors create_booking_row/cancel_booking_row/
reschedule_booking_row raise -- kept distinct per case so each caller
(this module's JSON routes, owner_booking.py's session-authenticated
routes) can translate the same failure into its own response shape."""
class NotFound(BookingRequestError):
pass
class SlotUnavailable(BookingRequestError):
"""The requested slot violates a business rule (opening hours, min
notice, max advance, buffer) -- never raised when skip_availability_check
is set."""
class SlotTaken(BookingRequestError):
"""The Postgres EXCLUDE constraint rejected the write -- always checked,
override or not."""
class AlreadyCancelled(BookingRequestError):
pass
def create_booking_row(client_id, resource_id, service_id, start_time,
customer_name, customer_contact, source,
skip_availability_check=False):
"""Shared booking-creation path for the public API (#16/#17) and the
owner's manual-entry flow (#20). skip_availability_check bypasses only
the business-rule slot check (opening hours/min-notice/max-advance/
buffer) for an owner-entered walk-in/phone booking -- it never touches
bdb.create_booking's EXCLUDE-constraint check, which stays enforced
either way. Returns (booking, manage_token); raises NotFound/
SlotUnavailable/SlotTaken instead of building a response itself, so each
caller renders the failure its own way."""
resource = bdb.get_resource(client_id, resource_id)
service = bdb.get_service(client_id, service_id)
client = bdb.get_client(client_id)
if resource is None or service is None or client is None:
raise NotFound()
tz_name = _tz_name(client)
if not skip_availability_check:
day = _local_date(start_time, tz_name)
valid_starts = _available_slots(client_id, resource, tz_name,
service["duration_minutes"], day, day)
if start_time not in valid_starts:
raise SlotUnavailable()
end_time = start_time + timedelta(minutes=service["duration_minutes"])
status = "confirmed" if client.get("auto_confirm", True) else "pending"
try:
booking = bdb.create_booking(
client_id, resource_id, customer_name, customer_contact,
service["name"], start_time, end_time, source=source, status=status)
except bdb.BookingConflict:
raise SlotTaken() from None
token = _mint_manage_token(client_id, booking["booking_id"])
# #18: fires for every caller of this helper, public page (#17) and
# owner manual-entry (#20) included -- calling bdb.create_booking()
# directly would bypass it.
booking_mail.send_booking_confirmation(client, booking, token)
return booking, token
def cancel_booking_row(client_id, booking_id):
"""Shared cancel path for the customer's token-authenticated route
below and the owner's session-authenticated one (#20). client_id already
scopes the lookup, so an owner session can never cancel another
tenant's booking_id."""
existing = bdb.get_booking(client_id, booking_id)
if existing is None:
raise NotFound()
if existing["status"] == "cancelled":
raise AlreadyCancelled()
return bdb.update_booking(client_id, booking_id, status="cancelled")
def reschedule_booking_row(client_id, booking_id, new_start):
"""Shared reschedule path for the customer's token-authenticated route
below and the owner's session-authenticated one (#20). Unlike manual
creation, this never skips the business-rule slot check -- #20 only
calls out an override for creating a booking, not for moving one."""
existing = bdb.get_booking(client_id, booking_id)
if existing is None:
raise NotFound()
if existing["status"] == "cancelled":
raise AlreadyCancelled()
duration = existing["end_time"] - existing["start_time"]
new_end = new_start + duration
resource = bdb.get_resource(client_id, existing["resource_id"])
client = bdb.get_client(client_id)
if resource is None or client is None:
raise NotFound()
tz_name = _tz_name(client)
day = _local_date(new_start, tz_name)
valid_starts = _available_slots(client_id, resource, tz_name,
duration.total_seconds() // 60, day, day,
exclude_booking_id=booking_id)
if new_start not in valid_starts:
raise SlotUnavailable()
try:
return bdb.update_booking(client_id, booking_id, start_time=new_start,
end_time=new_end)
except bdb.BookingConflict:
raise SlotTaken() from None
class _BadDuration(ValueError): class _BadDuration(ValueError):
"""Raised by _resolve_duration_minutes on an unknown service_id or a """Raised by _resolve_duration_minutes on an unknown service_id or a
non-integer duration_minutes -- turned into a clean 4xx by slots().""" non-integer duration_minutes -- turned into a clean 4xx by slots()."""
@@ -171,34 +284,17 @@ def create_booking():
and body.get("customer_name") and body.get("customer_contact")): and body.get("customer_name") and body.get("customer_contact")):
return jsonify({"error": "client_id, resource_id, service_id, start_time, " return jsonify({"error": "client_id, resource_id, service_id, start_time, "
"customer_name, customer_contact are required"}), 400 "customer_name, customer_contact are required"}), 400
resource = bdb.get_resource(client_id, resource_id)
service = bdb.get_service(client_id, service_id)
client = bdb.get_client(client_id)
if resource is None or service is None or client is None:
return jsonify({"error": "not found"}), 404
tz_name = _tz_name(client)
day = _local_date(start_time, tz_name)
valid_starts = _available_slots(client_id, resource, tz_name,
service["duration_minutes"], day, day)
if start_time not in valid_starts:
return jsonify({"error": "that slot is no longer available"}), 409
end_time = start_time + timedelta(minutes=service["duration_minutes"])
status = "confirmed" if client.get("auto_confirm", True) else "pending"
try: try:
booking = bdb.create_booking( booking, token = create_booking_row(
client_id, resource_id, body["customer_name"], body["customer_contact"], client_id, resource_id, service_id, start_time,
service["name"], start_time, end_time, source="public", status=status) body["customer_name"], body["customer_contact"], source="public")
except bdb.BookingConflict: except NotFound:
return jsonify({"error": "not found"}), 404
except SlotUnavailable:
return jsonify({"error": "that slot is no longer available"}), 409
except SlotTaken:
return jsonify({"error": "that slot was just taken"}), 409 return jsonify({"error": "that slot was just taken"}), 409
token = _mint_manage_token(client_id, booking["booking_id"])
# #18: fires for every caller of this endpoint, public page (#17) included.
# A future ticket-6 owner-manual-entry flow only gets the confirmation
# email for free if it also creates bookings through this endpoint --
# calling bdb.create_booking() directly would bypass it.
booking_mail.send_booking_confirmation(client, booking, token)
return jsonify({"booking_id": booking["booking_id"], "status": booking["status"], return jsonify({"booking_id": booking["booking_id"], "status": booking["status"],
"token": token}), 201 "token": token}), 201
@@ -210,12 +306,12 @@ def cancel_booking():
if resolved is None: if resolved is None:
return jsonify({"error": "invalid or expired token"}), 400 return jsonify({"error": "invalid or expired token"}), 400
client_id, booking_id = resolved client_id, booking_id = resolved
existing = bdb.get_booking(client_id, booking_id) try:
if existing is None: updated = cancel_booking_row(client_id, booking_id)
except NotFound:
return jsonify({"error": "not found"}), 404 return jsonify({"error": "not found"}), 404
if existing["status"] == "cancelled": except AlreadyCancelled:
return jsonify({"error": "already cancelled"}), 409 return jsonify({"error": "already cancelled"}), 409
updated = bdb.update_booking(client_id, booking_id, status="cancelled")
return jsonify({"cancelled": updated["booking_id"]}) return jsonify({"cancelled": updated["booking_id"]})
@@ -230,30 +326,15 @@ def reschedule_booking():
if new_start is None: if new_start is None:
return jsonify({"error": "start_time is required"}), 400 return jsonify({"error": "start_time is required"}), 400
existing = bdb.get_booking(client_id, booking_id)
if existing is None:
return jsonify({"error": "not found"}), 404
if existing["status"] == "cancelled":
return jsonify({"error": "already cancelled"}), 409
duration = existing["end_time"] - existing["start_time"]
new_end = new_start + duration
resource = bdb.get_resource(client_id, existing["resource_id"])
client = bdb.get_client(client_id)
if resource is None or client is None:
return jsonify({"error": "not found"}), 404
tz_name = _tz_name(client)
day = _local_date(new_start, tz_name)
valid_starts = _available_slots(client_id, resource, tz_name,
duration.total_seconds() // 60, day, day,
exclude_booking_id=booking_id)
if new_start not in valid_starts:
return jsonify({"error": "that slot is no longer available"}), 409
try: try:
updated = bdb.update_booking(client_id, booking_id, start_time=new_start, updated = reschedule_booking_row(client_id, booking_id, new_start)
end_time=new_end) except NotFound:
except bdb.BookingConflict: return jsonify({"error": "not found"}), 404
except AlreadyCancelled:
return jsonify({"error": "already cancelled"}), 409
except SlotUnavailable:
return jsonify({"error": "that slot is no longer available"}), 409
except SlotTaken:
return jsonify({"error": "that slot was just taken"}), 409 return jsonify({"error": "that slot was just taken"}), 409
return jsonify({"booking_id": updated["booking_id"], return jsonify({"booking_id": updated["booking_id"],
"start_time": updated["start_time"].isoformat(), "start_time": updated["start_time"].isoformat(),
+25
View File
@@ -125,6 +125,18 @@ def list_active_resources(client_id):
return _list_active("resources", client_id) return _list_active("resources", client_id)
def list_resources(client_id):
"""All of client_id's resources, active or not -- unlike
list_active_resources, used where a name lookup must still resolve for a
booking made against a resource that's since been deactivated (the owner
agenda, #20)."""
with db.connect() as conn, conn.cursor() as cur:
cur.execute(
"SELECT * FROM resources WHERE client_id = %s ORDER BY name",
(client_id,))
return cur.fetchall()
# ---- bookings ---- # ---- bookings ----
_BOOKING_UPDATABLE = {"resource_id", "customer_name", "customer_contact", _BOOKING_UPDATABLE = {"resource_id", "customer_name", "customer_contact",
@@ -182,6 +194,19 @@ def list_bookings(client_id):
return cur.fetchall() return cur.fetchall()
def list_bookings_between(client_id, start, end):
"""Every one of client_id's bookings (any resource, any status --
including cancelled, so the owner agenda (#20) can still show a
cancelled slot rather than silently dropping it) whose start_time falls
in [start, end)."""
with db.connect() as conn, conn.cursor() as cur:
cur.execute(
"SELECT * FROM bookings WHERE client_id = %s AND start_time >= %s "
"AND start_time < %s ORDER BY start_time",
(client_id, start, end))
return cur.fetchall()
def list_active_bookings_for_resource(client_id, resource_id, start, end, def list_active_bookings_for_resource(client_id, resource_id, start, end,
exclude_booking_id=None): exclude_booking_id=None):
"""Bookings on client_id's own resource_id, not cancelled, that fall """Bookings on client_id's own resource_id, not cancelled, that fall
+2 -3
View File
@@ -53,9 +53,8 @@ def logout():
@bp.get("/") @bp.get("/")
@login_required @login_required
def dashboard(): def dashboard():
# Tickets 6/7/8 (calendar, manual booking, settings) build the real # Tickets 7/8 (settings, notify channel) still build on top of this
# dashboard content on top of this session; this is just the landing # session; the agenda/calendar (#20) now lives at owner_booking.agenda.
# page proving a login resolved to (and is scoped to) one client_id.
client = bdb.get_client(session["client_id"]) client = bdb.get_client(session["client_id"])
return render_template("owner/dashboard.html", client=client) return render_template("owner/dashboard.html", client=client)
+155
View File
@@ -0,0 +1,155 @@
"""Owner agenda: server-rendered week view, manual walk-in/phone bookings,
and owner-initiated cancel/reschedule (#20).
All routes are session-authenticated via owner_auth.login_required and
reuse booking_api.py's create/cancel/reschedule helpers, so the EXCLUDE
constraint (#16) and confirmation email (#18) stay on the single code path
those tickets already established -- this module never calls booking_db.py
directly for a mutation, only for the read-side agenda listing.
Manual creation passes skip_availability_check=True (the "override flag"
#20 asks for): it bypasses opening-hours/min-notice/max-advance/buffer, but
booking_api.create_booking_row still always goes through
booking_db.create_booking, so the Postgres EXCLUDE constraint -- the actual
double-booking guard -- is never bypassable, owner included.
"""
from datetime import datetime, timedelta, timezone
from zoneinfo import ZoneInfo
from flask import Blueprint, redirect, render_template, request, session, url_for
import booking_api as bapi
import booking_db as bdb
from owner_auth import login_required
bp = Blueprint("owner_booking", __name__, url_prefix="/owner")
def _tz(client):
return ZoneInfo((client or {}).get("timezone") or "Europe/Berlin")
def _week_start(value):
"""Monday (a date) of the week containing value (an ISO date string),
or of the current week if value is missing/unparseable."""
try:
d = datetime.fromisoformat(value).date() if value else datetime.now().date()
except ValueError:
d = datetime.now().date()
return d - timedelta(days=d.weekday())
def _parse_local_start(value, tz):
"""value is a <input type=datetime-local> string (e.g.
"2026-08-10T14:30"), naive and meant in the client's own timezone --
never UTC, since that's what an owner typing a time means."""
try:
naive = datetime.fromisoformat(value)
except (TypeError, ValueError):
return None
return naive.replace(tzinfo=tz).astimezone(timezone.utc)
def _agenda_redirect(monday, error=None):
return redirect(url_for("owner_booking.agenda", week=monday.isoformat(), error=error))
def _agenda_days(client_id, client, monday):
"""[(date, [booking, ...]), ...] for the 7 days starting at monday, each
booking augmented with resource_name/start_local for display."""
tz = _tz(client)
start_utc = datetime.combine(monday, datetime.min.time(), tzinfo=tz).astimezone(timezone.utc)
end_utc = datetime.combine(monday + timedelta(days=7), datetime.min.time(),
tzinfo=tz).astimezone(timezone.utc)
rows = bdb.list_bookings_between(client_id, start_utc, end_utc)
resource_names = {r["resource_id"]: r["name"] for r in bdb.list_resources(client_id)}
by_date = {monday + timedelta(days=i): [] for i in range(7)}
for row in rows:
local_date = row["start_time"].astimezone(tz).date()
if local_date not in by_date:
continue # a booking straddling the window edge in another tz
row = dict(row)
row["resource_name"] = resource_names.get(row["resource_id"], row["resource_id"])
row["start_local"] = row["start_time"].astimezone(tz)
by_date[local_date].append(row)
return sorted(by_date.items())
@bp.get("/agenda")
@login_required
def agenda():
client_id = session["client_id"]
client = bdb.get_client(client_id)
monday = _week_start(request.args.get("week"))
today = datetime.now(_tz(client)).date()
return render_template(
"owner/agenda.html", client=client, days=_agenda_days(client_id, client, monday),
week_start=monday, today=today,
prev_week=(monday - timedelta(days=7)).isoformat(),
next_week=(monday + timedelta(days=7)).isoformat(),
this_week=_week_start(None).isoformat(),
resources=bdb.list_active_resources(client_id),
services=bdb.list_active_services(client_id),
error=request.args.get("error"))
@bp.post("/bookings")
@login_required
def create_manual_booking():
client_id = session["client_id"]
client = bdb.get_client(client_id)
monday = _week_start(request.form.get("week"))
start_time = _parse_local_start(request.form.get("start_time"), _tz(client))
resource_id = request.form.get("resource_id")
service_id = request.form.get("service_id")
customer_name = (request.form.get("customer_name") or "").strip()
customer_contact = (request.form.get("customer_contact") or "").strip()
if not (start_time and resource_id and service_id and customer_name and customer_contact):
return _agenda_redirect(monday, error="missing_fields")
try:
bapi.create_booking_row(
client_id, resource_id, service_id, start_time, customer_name,
customer_contact, source="owner", skip_availability_check=True)
except bapi.NotFound:
return _agenda_redirect(monday, error="not_found")
except bapi.SlotTaken:
return _agenda_redirect(monday, error="slot_taken")
return _agenda_redirect(monday)
@bp.post("/bookings/<booking_id>/cancel")
@login_required
def cancel_manual_booking(booking_id):
client_id = session["client_id"]
monday = _week_start(request.form.get("week"))
try:
bapi.cancel_booking_row(client_id, booking_id)
except bapi.NotFound:
return _agenda_redirect(monday, error="not_found")
except bapi.AlreadyCancelled:
return _agenda_redirect(monday, error="already_cancelled")
return _agenda_redirect(monday)
@bp.post("/bookings/<booking_id>/reschedule")
@login_required
def reschedule_manual_booking(booking_id):
client_id = session["client_id"]
client = bdb.get_client(client_id)
monday = _week_start(request.form.get("week"))
new_start = _parse_local_start(request.form.get("start_time"), _tz(client))
if new_start is None:
return _agenda_redirect(monday, error="bad_time")
try:
bapi.reschedule_booking_row(client_id, booking_id, new_start)
except bapi.NotFound:
return _agenda_redirect(monday, error="not_found")
except bapi.AlreadyCancelled:
return _agenda_redirect(monday, error="already_cancelled")
except bapi.SlotUnavailable:
return _agenda_redirect(monday, error="unavailable")
except bapi.SlotTaken:
return _agenda_redirect(monday, error="slot_taken")
return _agenda_redirect(monday)
+150
View File
@@ -0,0 +1,150 @@
<!DOCTYPE html>
<html lang="de">
<head>
<meta charset="UTF-8" />
<meta name="viewport" content="width=device-width, initial-scale=1.0" />
<meta name="robots" content="noindex" />
<title>Kalender — {{ client.business_name if client else '' }}</title>
<style>
:root {
--brand: #0f8a7e;
--bg: #fff; --ink: #16302f; --muted: #5d716f; --line: #e3eae9;
--danger: #b3261e; --danger-bg: #fdecea;
}
* { box-sizing: border-box; }
body { margin: 0; font-family: system-ui, -apple-system, Segoe UI, Roboto, sans-serif;
background: var(--bg); color: var(--ink); padding: 20px; max-width: 760px; }
h1 { font-size: 1.2rem; margin: 0 0 6px; }
h2 { font-size: .95rem; margin: 0 0 8px; color: var(--muted); }
a { color: var(--brand); }
.nav { display: flex; align-items: center; justify-content: space-between; margin: 14px 0 20px; }
.nav a { text-decoration: none; font-size: .88rem; }
.day { border: 1px solid var(--line); border-radius: 10px; padding: 14px 16px; margin-bottom: 12px; }
.day.empty { color: var(--muted); }
.day.today { border-color: var(--brand); background: #f6faf9; }
.day.today h2 { color: var(--brand); }
table { width: 100%; border-collapse: collapse; font-size: .88rem; }
td, th { text-align: left; padding: 6px 4px; border-bottom: 1px solid var(--line); }
.status-cancelled { color: var(--muted); text-decoration: line-through; }
.status-pending { color: #9a6b00; }
form.inline { display: inline; }
input[type=text], input[type=datetime-local], select {
padding: 7px 9px; border: 1px solid var(--line); border-radius: 7px;
font: inherit; font-size: .85rem; }
.btn { background: var(--brand); color: #fff; border: 0; border-radius: 7px;
padding: 6px 12px; font: inherit; font-size: .82rem; font-weight: 600; cursor: pointer; }
.btn-danger { background: var(--danger); }
.btn-small { padding: 4px 9px; font-size: .78rem; }
.error { background: var(--danger-bg); color: var(--danger); border-radius: 9px;
padding: 10px 14px; font-size: .88rem; margin-bottom: 14px; }
.new-booking { border: 1px solid var(--line); border-radius: 10px; padding: 16px; margin-top: 24px; }
.new-booking .field { display: inline-block; margin: 0 8px 8px 0; }
label { display: block; font-size: .75rem; color: var(--muted); margin-bottom: 3px; }
</style>
</head>
<body>
<h1>{{ client.business_name if client else 'Kalender' }}</h1>
{% if error == "missing_fields" %}
<div class="error">Bitte alle Felder ausfüllen.</div>
{% elif error == "not_found" %}
<div class="error">Ressource, Leistung oder Buchung nicht gefunden.</div>
{% elif error == "slot_taken" %}
<div class="error">Dieser Zeitraum überschneidet sich mit einem bestehenden Termin.</div>
{% elif error == "unavailable" %}
<div class="error">Dieser Termin liegt außerhalb der verfügbaren Zeiten.</div>
{% elif error == "already_cancelled" %}
<div class="error">Diese Buchung wurde bereits storniert.</div>
{% elif error == "bad_time" %}
<div class="error">Bitte eine gültige Uhrzeit angeben.</div>
{% endif %}
<div class="nav">
<a href="{{ url_for('owner_booking.agenda', week=prev_week) }}">&larr; Vorherige Woche</a>
<strong>Woche ab {{ week_start.strftime('%d.%m.%Y') }}</strong>
<a href="{{ url_for('owner_booking.agenda', week=next_week) }}">Nächste Woche &rarr;</a>
</div>
{% if week_start.isoformat() != this_week %}
<p><a href="{{ url_for('owner_booking.agenda', week=this_week) }}">&uarr; Zu heute springen</a></p>
{% endif %}
{% for day, bookings in days %}
<div class="day {{ 'today' if day == today else '' }} {{ 'empty' if not bookings }}">
<h2>{{ day.strftime('%A, %d.%m.%Y') }}{{ ' — Heute' if day == today else '' }}</h2>
{% if not bookings %}
<p class="muted" style="margin:0;">Keine Buchungen.</p>
{% else %}
<table>
<thead><tr><th>Zeit</th><th>Kunde</th><th>Leistung</th><th>Ressource</th><th>Status</th><th></th></tr></thead>
<tbody>
{% for b in bookings %}
<tr class="status-{{ b.status }}">
<td>{{ b.start_local.strftime('%H:%M') }}</td>
<td>{{ b.customer_name }}</td>
<td>{{ b.service }}</td>
<td>{{ b.resource_name }}</td>
<td>{{ b.status }}</td>
<td>
{% if b.status != "cancelled" %}
<form class="inline" method="post"
action="{{ url_for('owner_booking.reschedule_manual_booking', booking_id=b.booking_id) }}">
<input type="hidden" name="week" value="{{ week_start.isoformat() }}" />
<input type="datetime-local" name="start_time" required />
<button type="submit" class="btn btn-small">Verschieben</button>
</form>
<form class="inline" method="post"
action="{{ url_for('owner_booking.cancel_manual_booking', booking_id=b.booking_id) }}">
<input type="hidden" name="week" value="{{ week_start.isoformat() }}" />
<button type="submit" class="btn btn-danger btn-small">Stornieren</button>
</form>
{% endif %}
</td>
</tr>
{% endfor %}
</tbody>
</table>
{% endif %}
</div>
{% endfor %}
<div class="new-booking">
<h2>Neue Buchung (Laufkundschaft / Telefon)</h2>
<form method="post" action="{{ url_for('owner_booking.create_manual_booking') }}">
<input type="hidden" name="week" value="{{ week_start.isoformat() }}" />
<div class="field">
<label>Ressource</label>
<select name="resource_id" required>
{% for r in resources %}
<option value="{{ r.resource_id }}">{{ r.name }}</option>
{% endfor %}
</select>
</div>
<div class="field">
<label>Leistung</label>
<select name="service_id" required>
{% for s in services %}
<option value="{{ s.service_id }}">{{ s.name }}</option>
{% endfor %}
</select>
</div>
<div class="field">
<label>Datum &amp; Uhrzeit</label>
<input type="datetime-local" name="start_time" required />
</div>
<div class="field">
<label>Name</label>
<input type="text" name="customer_name" required />
</div>
<div class="field">
<label>Kontakt</label>
<input type="text" name="customer_contact" required />
</div>
<div class="field" style="vertical-align:bottom;">
<button type="submit" class="btn">Buchen</button>
</div>
</form>
</div>
<p class="muted" style="margin-top:20px;"><a href="{{ url_for('owner_auth.dashboard') }}">Zurück</a></p>
</body>
</html>
@@ -23,7 +23,8 @@
<h1>{{ client.business_name if client else 'Mein Konto' }}</h1> <h1>{{ client.business_name if client else 'Mein Konto' }}</h1>
<div class="card"> <div class="card">
<p>Sie sind angemeldet.</p> <p>Sie sind angemeldet.</p>
<p class="muted">Kalender, Buchungen und Einstellungen folgen hier.</p> <p class="muted"><a href="{{ url_for('owner_booking.agenda') }}">Zum Kalender</a></p>
<p class="muted">Einstellungen folgen hier.</p>
</div> </div>
<p class="muted"><a href="{{ url_for('owner_auth.logout') }}">Abmelden</a></p> <p class="muted"><a href="{{ url_for('owner_auth.logout') }}">Abmelden</a></p>
</body> </body>
+209
View File
@@ -0,0 +1,209 @@
"""Flask test client / real-DB integration tests for the owner agenda,
manual booking, and owner-initiated cancel/reschedule (#20), per #14's
testing decision: assert on HTTP response + resulting DB state.
"""
from datetime import date, datetime, time, timedelta, timezone
import pytest
import booking_db as bdb
from app import app as flask_app
CLIENT_A = "C-TEST-OWNER-BOOKING-A"
CLIENT_B = "C-TEST-OWNER-BOOKING-B"
@pytest.fixture
def client():
flask_app.config["TESTING"] = True
flask_app.secret_key = "test-secret"
return flask_app.test_client()
def _next_monday(after):
d = after + timedelta(days=1)
while d.weekday() != 0:
d += timedelta(days=1)
return d
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) VALUES (%s, %s, %s) "
"ON CONFLICT (client_id) DO UPDATE SET timezone = EXCLUDED.timezone, "
"auto_confirm = EXCLUDED.auto_confirm",
(client_id, "Europe/Berlin", True))
conn.commit()
resource = bdb.create_resource(client_id, "Chair 1", **resource_kwargs)
bdb.set_resource_hours(client_id, resource["resource_id"], 0, time(9, 0), time(17, 0))
service = bdb.create_service(client_id, "Haircut", 60, price=25)
return resource, service
# ---- agenda view ----
def test_agenda_requires_login(client):
resp = client.get("/owner/agenda")
assert resp.status_code == 302
assert "/owner/login" in resp.headers["Location"]
def test_agenda_shows_only_own_clients_bookings(client):
resource_a, service_a = _setup(CLIENT_A, min_notice_minutes=0, max_advance_days=365)
resource_b, service_b = _setup(CLIENT_B, min_notice_minutes=0, max_advance_days=365)
day = _next_monday(date.today())
start = datetime.combine(day, time(10, 0), tzinfo=timezone(timedelta(hours=2)))
bdb.create_booking(CLIENT_A, resource_a["resource_id"], "Alice", "a@example.com",
"Haircut", start, start + timedelta(hours=1), source="public")
bdb.create_booking(CLIENT_B, resource_b["resource_id"], "Bob", "b@example.com",
"Haircut", start, start + timedelta(hours=1), source="public")
bdb.create_user(CLIENT_A, "owner@example.com", "correct horse")
client.post("/owner/login", data={"email": "owner@example.com", "password": "correct horse"})
resp = client.get("/owner/agenda", query_string={"week": day.isoformat()})
assert resp.status_code == 200
body = resp.get_data(as_text=True)
assert "Alice" in body
assert "Bob" not in body
# ---- manual booking ----
def test_owner_can_create_booking_outside_opening_hours(client):
resource, service = _setup(CLIENT_A)
bdb.create_user(CLIENT_A, "owner@example.com", "correct horse")
client.post("/owner/login", data={"email": "owner@example.com", "password": "correct horse"})
day = _next_monday(date.today())
# 20:00 local is outside the 09:00-17:00 hours configured by _setup.
outside_local = datetime.combine(day, time(20, 0)).strftime("%Y-%m-%dT%H:%M")
resp = client.post("/owner/bookings", data={
"resource_id": resource["resource_id"], "service_id": service["service_id"],
"start_time": outside_local, "customer_name": "Walk-in", "customer_contact": "n/a",
"week": day.isoformat()})
assert resp.status_code == 302
assert "error" not in resp.headers["Location"]
bookings = bdb.list_bookings(CLIENT_A)
assert len(bookings) == 1
assert bookings[0]["customer_name"] == "Walk-in"
assert bookings[0]["source"] == "owner"
def test_owner_cannot_double_book_the_same_resource(client):
resource, service = _setup(CLIENT_A)
bdb.create_user(CLIENT_A, "owner@example.com", "correct horse")
client.post("/owner/login", data={"email": "owner@example.com", "password": "correct horse"})
day = _next_monday(date.today())
local_time = datetime.combine(day, time(10, 0)).strftime("%Y-%m-%dT%H:%M")
first = client.post("/owner/bookings", data={
"resource_id": resource["resource_id"], "service_id": service["service_id"],
"start_time": local_time, "customer_name": "First", "customer_contact": "n/a",
"week": day.isoformat()})
assert "error" not in first.headers["Location"]
second = client.post("/owner/bookings", data={
"resource_id": resource["resource_id"], "service_id": service["service_id"],
"start_time": local_time, "customer_name": "Second", "customer_contact": "n/a",
"week": day.isoformat()})
assert "error=slot_taken" in second.headers["Location"]
bookings = bdb.list_bookings(CLIENT_A)
assert len(bookings) == 1
def test_owner_cannot_create_booking_for_another_tenants_resource(client):
resource_b, service_b = _setup(CLIENT_B)
bdb.create_user(CLIENT_A, "owner@example.com", "correct horse")
with bdb.db.connect() as conn, conn.cursor() as cur:
cur.execute(
"INSERT INTO clients (client_id, timezone, auto_confirm) VALUES (%s, %s, %s) "
"ON CONFLICT (client_id) DO NOTHING", (CLIENT_A, "Europe/Berlin", True))
conn.commit()
client.post("/owner/login", data={"email": "owner@example.com", "password": "correct horse"})
day = _next_monday(date.today())
local_time = datetime.combine(day, time(10, 0)).strftime("%Y-%m-%dT%H:%M")
resp = client.post("/owner/bookings", data={
"resource_id": resource_b["resource_id"], "service_id": service_b["service_id"],
"start_time": local_time, "customer_name": "Sneaky", "customer_contact": "n/a",
"week": day.isoformat()})
assert "error=not_found" in resp.headers["Location"]
assert bdb.list_bookings(CLIENT_A) == []
# ---- owner cancel/reschedule ----
def test_owner_can_cancel_own_booking(client):
resource, service = _setup(CLIENT_A)
booking = bdb.create_booking(
CLIENT_A, resource["resource_id"], "Dana", "d@example.com", "Haircut",
datetime.now(timezone.utc) + timedelta(days=2),
datetime.now(timezone.utc) + timedelta(days=2, hours=1))
bdb.create_user(CLIENT_A, "owner@example.com", "correct horse")
client.post("/owner/login", data={"email": "owner@example.com", "password": "correct horse"})
resp = client.post(f"/owner/bookings/{booking['booking_id']}/cancel", data={"week": ""})
assert resp.status_code == 302
assert "error" not in resp.headers["Location"]
assert bdb.get_booking(CLIENT_A, booking["booking_id"])["status"] == "cancelled"
def test_owner_cannot_cancel_another_tenants_booking(client):
resource_b, service_b = _setup(CLIENT_B)
booking = bdb.create_booking(
CLIENT_B, resource_b["resource_id"], "Eve", "e@example.com", "Haircut",
datetime.now(timezone.utc) + timedelta(days=2),
datetime.now(timezone.utc) + timedelta(days=2, hours=1))
with bdb.db.connect() as conn, conn.cursor() as cur:
cur.execute(
"INSERT INTO clients (client_id, timezone, auto_confirm) VALUES (%s, %s, %s) "
"ON CONFLICT (client_id) DO NOTHING", (CLIENT_A, "Europe/Berlin", True))
conn.commit()
bdb.create_user(CLIENT_A, "owner@example.com", "correct horse")
client.post("/owner/login", data={"email": "owner@example.com", "password": "correct horse"})
resp = client.post(f"/owner/bookings/{booking['booking_id']}/cancel", data={"week": ""})
assert "error=not_found" in resp.headers["Location"]
assert bdb.get_booking(CLIENT_B, booking["booking_id"])["status"] != "cancelled"
def test_owner_can_reschedule_own_booking(client):
resource, service = _setup(CLIENT_A, min_notice_minutes=0, max_advance_days=365)
day = _next_monday(date.today())
start = datetime.combine(day, time(9, 0), tzinfo=timezone(timedelta(hours=2)))
booking = bdb.create_booking(
CLIENT_A, resource["resource_id"], "Fay", "f@example.com", "Haircut",
start, start + timedelta(hours=1))
bdb.create_user(CLIENT_A, "owner@example.com", "correct horse")
client.post("/owner/login", data={"email": "owner@example.com", "password": "correct horse"})
new_local = datetime.combine(day, time(11, 0)).strftime("%Y-%m-%dT%H:%M")
resp = client.post(f"/owner/bookings/{booking['booking_id']}/reschedule",
data={"start_time": new_local, "week": day.isoformat()})
assert resp.status_code == 302
assert "error" not in resp.headers["Location"]
updated = bdb.get_booking(CLIENT_A, booking["booking_id"])
assert updated["start_time"].hour in (9, 10) # 11:00 Europe/Berlin -> 09:00/10:00 UTC
def test_owner_reschedule_into_occupied_slot_fails(client):
resource, service = _setup(CLIENT_A, min_notice_minutes=0, max_advance_days=365)
day = _next_monday(date.today())
slot_1 = datetime.combine(day, time(9, 0), tzinfo=timezone(timedelta(hours=2)))
slot_2 = datetime.combine(day, time(10, 0), tzinfo=timezone(timedelta(hours=2)))
bdb.create_booking(CLIENT_A, resource["resource_id"], "Gus", "g@example.com",
"Haircut", slot_1, slot_1 + timedelta(hours=1))
movable = bdb.create_booking(CLIENT_A, resource["resource_id"], "Hana", "h@example.com",
"Haircut", slot_2, slot_2 + timedelta(hours=1))
bdb.create_user(CLIENT_A, "owner@example.com", "correct horse")
client.post("/owner/login", data={"email": "owner@example.com", "password": "correct horse"})
new_local = datetime.combine(day, time(9, 0)).strftime("%Y-%m-%dT%H:%M")
resp = client.post(f"/owner/bookings/{movable['booking_id']}/reschedule",
data={"start_time": new_local, "week": day.isoformat()})
assert "error=unavailable" in resp.headers["Location"]
assert bdb.get_booking(CLIENT_A, movable["booking_id"])["start_time"] == slot_2