fix: block test-tagged orders after test→prod transition

VERSION

@@ -1 +1 @@
-1.2.4
+1.3.0

cloud/backend/app/event_status.py

@@ -69,6 +69,12 @@ def validate_status_transition(old: str, new: str) -> None:
         raise api_error("cannot_transition_status", status.HTTP_422_UNPROCESSABLE_CONTENT, old_status=old_n, new_status=new_n)
 
 
+def payload_is_stale_test(event: Event, payload: dict) -> bool:
+    """True when a Pi-synced test-mode order arrives after the event entered production."""
+    mode = str(payload.get("mode") or "").lower()
+    return mode == "test" and normalize_status(event.status) in {"prod", "archive"}
+
+
 def purge_event_operational_data(db: Session, event: Event) -> None:
     """Remove test orders/stats and reset stock when entering production."""
     event_id = event.id

cloud/backend/app/routers/edge.py

@@ -22,7 +22,7 @@
 from ..edge_operational_snapshot import build_operational_snapshot_for_events
 from ..event_cash_sessions import upsert_edge_cash_session
 from ..event_stats import resolve_sync_ordered_at
-from ..event_status import ORDER_ACCEPT_STATUSES, PI_VISIBLE_STATUSES, normalize_status
+from ..event_status import ORDER_ACCEPT_STATUSES, PI_VISIBLE_STATUSES, normalize_status, payload_is_stale_test
 from ..i18n.errors import api_error
 from ..models import (
     Appliance,
@@ -564,6 +564,9 @@ def submit_edge_order(
         raise api_error("event_status_does_not_accept_orders", status.HTTP_403_FORBIDDEN, status=ev_status)
 
     payload = body.payload or {}
+    if payload_is_stale_test(event, payload):
+        return EdgeOrderAck(server_order_id=0, duplicate=False)
+
     row = EdgeSubmittedOrder(
         client_order_id=body.client_order_id,
         appliance_id=ctx.appliance.id,
@@ -632,6 +635,9 @@ def submit_operational_chunk(
         raise api_error("event_not_found_for_organisation", status.HTTP_404_NOT_FOUND)
 
     payload = body.payload or {}
+    if payload_is_stale_test(event, payload):
+        return EdgeOperationalChunkAck(chunk_id=body.chunk_id, status="acked", accepted=0)
+
     entity_type = (body.entity_type or payload.get("entity_type") or "").strip().lower()
 
     row = EdgeSubmittedOrder(

cloud/backend/tests/test_events_crud.py

@@ -153,6 +153,24 @@ def test_status_test_to_prod_purges_all_operational_data():
     finally:
         db.close()
 
+    now = _utc_now()
+    stats = client.get(
+        f"/events/{event_id}/stats",
+        headers=headers,
+        params={
+            "from": (now - timedelta(hours=1)).isoformat(),
+            "to": (now + timedelta(hours=1)).isoformat(),
+        },
+    )
+    assert stats.status_code == 200, stats.text
+    assert stats.json()["totals"]["distinct_orders_count"] == 0
+    assert stats.json()["totals"]["line_cents"] == 0
+
+    transactions = client.get(f"/events/{event_id}/transactions", headers=headers)
+    assert transactions.status_code == 200, transactions.text
+    assert transactions.json()["items"] == []
+    assert transactions.json()["total"] == 0
+
 
 def test_create_event_and_status_transition():
     org_a_id, _ = _setup_two_tenants()

cloud/backend/tests/test_operational_mode_sync.py

@@ -0,0 +1,190 @@
+"""Reject test-tagged operational sync after event enters production."""
+
+from datetime import UTC, datetime, timedelta
+from uuid import uuid4
+
+import pytest
+from app.database import SessionLocal
+from app.event_status import payload_is_stale_test
+from app.main import app
+from app.models import (
+    Appliance,
+    ApplianceEdgeCredential,
+    ApplianceLending,
+    EdgeOrderItem,
+    EdgeSubmittedOrder,
+    Event,
+    HireCompany,
+    Organisation,
+)
+from app.security import get_password_hash
+from fastapi.testclient import TestClient
+
+from tests.helpers import country_id_by_code
+
+client = TestClient(app)
+
+
+def _edge_fixture(*, event_status: str = "prod") -> tuple[dict[str, str], int]:
+    suffix = uuid4().hex
+    db = SessionLocal()
+    try:
+        hc = HireCompany(name=f"Mode HC {suffix}")
+        db.add(hc)
+        db.flush()
+        org = Organisation(
+            name=f"Mode Org {suffix}",
+            country_id=country_id_by_code(db, "CH"),
+            hire_company_id=hc.id,
+            currency="CHF",
+        )
+        db.add(org)
+        db.flush()
+        now = datetime.now(UTC)
+        ev = Event(
+            name="Live",
+            status=event_status,
+            start=now - timedelta(hours=1),
+            end=now + timedelta(days=1),
+            organisation_id=org.id,
+        )
+        db.add(ev)
+        db.flush()
+        appliance = Appliance(hire_company_id=hc.id, type="server", name="Pi")
+        db.add(appliance)
+        db.flush()
+        today = now.date()
+        db.add(
+            ApplianceLending(
+                appliance_id=appliance.id,
+                organisation_id=org.id,
+                start_date=today,
+                end_date=today,
+                returned_at=None,
+            )
+        )
+        secret = f"secret-{suffix}"
+        cred = ApplianceEdgeCredential(
+            appliance_id=appliance.id,
+            edge_client_id=f"cid-{suffix}",
+            edge_secret_hash=get_password_hash(secret),
+            status="active",
+        )
+        db.add(cred)
+        db.commit()
+        return (
+            {
+                "X-Edge-Client-Id": cred.edge_client_id,
+                "X-Edge-Secret": secret,
+            },
+            ev.id,
+        )
+    finally:
+        db.close()
+
+
+@pytest.mark.parametrize(
+    ("event_status", "payload_mode", "expected"),
+    [
+        ("prod", "test", True),
+        ("archive", "test", True),
+        ("prod", "prod", False),
+        ("test", "test", False),
+        ("prod", "", False),
+        ("prod", None, False),
+    ],
+)
+def test_payload_is_stale_test(event_status, payload_mode, expected):
+    event = Event(status=event_status)
+    payload = {}
+    if payload_mode is not None:
+        payload["mode"] = payload_mode
+    assert payload_is_stale_test(event, payload) is expected
+
+
+def test_test_tagged_chunk_dropped_when_event_prod():
+    headers, event_id = _edge_fixture(event_status="prod")
+    chunk_id = f"chunk-test-{uuid4().hex}"
+    payload = {
+        "chunk_id": chunk_id,
+        "event_id": event_id,
+        "entity_type": "submission",
+        "payload": {
+            "mode": "test",
+            "client_order_id": "o-test-1",
+            "payment_status": "paid",
+            "lines": [{"article_id": 1, "qty": 1, "unit_cents": 500}],
+            "payments": [{"type": "cash", "amount_cents": 500}],
+            "local_order_id": 1,
+            "session_id": 1,
+        },
+    }
+    response = client.post("/edge/v1/sync/operational/chunk", headers=headers, json=payload)
+    assert response.status_code == 200, response.text
+    assert response.json()["accepted"] == 0
+
+    db = SessionLocal()
+    try:
+        assert db.query(EdgeSubmittedOrder).filter(EdgeSubmittedOrder.event_id == event_id).count() == 0
+        assert db.query(EdgeOrderItem).filter(EdgeOrderItem.event_id == event_id).count() == 0
+    finally:
+        db.close()
+
+
+def test_prod_tagged_chunk_accepted_when_event_prod():
+    headers, event_id = _edge_fixture(event_status="prod")
+    chunk_id = f"chunk-prod-{uuid4().hex}"
+    payload = {
+        "chunk_id": chunk_id,
+        "event_id": event_id,
+        "entity_type": "submission",
+        "payload": {
+            "mode": "prod",
+            "client_order_id": "o-prod-1",
+            "payment_status": "paid",
+            "lines": [{"article_id": 1, "qty": 1, "unit_cents": 500}],
+            "payments": [{"type": "cash", "amount_cents": 500}],
+            "local_order_id": 2,
+            "session_id": 2,
+        },
+    }
+    response = client.post("/edge/v1/sync/operational/chunk", headers=headers, json=payload)
+    assert response.status_code == 200, response.text
+    assert response.json()["accepted"] == 1
+
+    db = SessionLocal()
+    try:
+        assert db.query(EdgeSubmittedOrder).filter(EdgeSubmittedOrder.event_id == event_id).count() == 1
+        assert db.query(EdgeOrderItem).filter(EdgeOrderItem.event_id == event_id).count() == 1
+    finally:
+        db.close()
+
+
+def test_test_tagged_edge_order_dropped_when_event_prod():
+    headers, event_id = _edge_fixture(event_status="prod")
+    client_order_id = f"order-{uuid4().hex}"
+    payload = {
+        "client_order_id": client_order_id,
+        "event_id": event_id,
+        "payload": {
+            "mode": "test",
+            "client_order_id": client_order_id,
+            "payment_status": "open",
+            "lines": [{"article_id": 1, "qty": 1, "unit_cents": 500}],
+        },
+    }
+    response = client.post("/edge/v1/orders", headers=headers, json=payload)
+    assert response.status_code == 200, response.text
+    assert response.json()["server_order_id"] == 0
+    assert response.json()["duplicate"] is False
+
+    db = SessionLocal()
+    try:
+        assert (
+            db.query(EdgeSubmittedOrder)
+            .filter(EdgeSubmittedOrder.client_order_id == client_order_id)
+            .count()
+            == 0
+        )
+    finally:
+        db.close()

pi/backend/app/domain/kitchen_sync.py

@@ -8,6 +8,16 @@
 from sqlalchemy.orm import Session
 
 from ..models import KitchenTicket, KitchenTicketLine, LocalOrder, OutboxEntry
+from .sync_enqueue import stamp_operational_mode
+
+
+def _order_operational_mode(order: LocalOrder) -> str | None:
+    try:
+        payload = json.loads(order.payload_json or "{}")
+    except json.JSONDecodeError:
+        return None
+    mode = payload.get("mode")
+    return str(mode).lower() if mode else None
 
 
 def build_kitchen_tickets_payload(db: Session, order: LocalOrder) -> dict:
@@ -49,7 +59,10 @@ def build_kitchen_tickets_payload(db: Session, order: LocalOrder) -> dict:
 
 
 def enqueue_kitchen_tickets_sync(db: Session, order: LocalOrder) -> None:
-    payload = build_kitchen_tickets_payload(db, order)
+    payload = stamp_operational_mode(
+        build_kitchen_tickets_payload(db, order),
+        _order_operational_mode(order),
+    )
     for out in (
         db.query(OutboxEntry)
         .filter(

pi/backend/app/domain/sync_enqueue.py

@@ -10,14 +10,28 @@
 from ..models import OutboxEntry
 
 
+def event_mode_label(status: str | None) -> str:
+    return str(status or "config").lower()
+
+
+def stamp_operational_mode(payload: dict, mode: str | None) -> dict:
+    """Attach immutable event mode (test/prod) when the payload does not already have one."""
+    out = dict(payload)
+    if "mode" not in out and mode:
+        out["mode"] = event_mode_label(mode)
+    return out
+
+
 def enrich_payload_for_cloud_sync(
     payload: dict,
     *,
     local_order_id: int,
     session_id: int,
+    mode: str | None = None,
 ) -> dict:
     """Cloud mirror uses local_order_id / session_id for per-order counts in Umsatz."""
-    out = dict(payload)
+    effective_mode = mode or payload.get("mode")
+    out = stamp_operational_mode(payload, str(effective_mode) if effective_mode else None)
     out["local_order_id"] = int(local_order_id)
     out["session_id"] = int(session_id)
     return out

pi/backend/app/line_moves.py

@@ -8,7 +8,7 @@
 from sqlalchemy.orm import Session
 
 from .domain.sessions import ensure_order_session
-from .domain.sync_enqueue import enqueue_payload_sync, enrich_payload_for_cloud_sync
+from .domain.sync_enqueue import enqueue_payload_sync, enrich_payload_for_cloud_sync, event_mode_label
 from .models import CollectiveBill, LocalOrder
 from .order_line_utils import merge_lines_into_list, take_selections_from_orders
 
@@ -94,6 +94,7 @@ def append_lines_to_table(
             "payments": [],
             "payment_status": "open",
             "transferred": True,
+            "mode": event_mode_label(ev.get("status")),
         }
         session_id = ensure_order_session(
             db,
@@ -119,7 +120,12 @@ def append_lines_to_table(
             db,
             event_id=event_id,
             client_order_id=cid,
-            payload=enrich_payload_for_cloud_sync(payload, local_order_id=order.id, session_id=session_id),
+            payload=enrich_payload_for_cloud_sync(
+                payload,
+                local_order_id=order.id,
+                session_id=session_id,
+                mode=event_mode_label(ev.get("status")),
+            ),
         )
 
 
@@ -168,6 +174,7 @@ def append_lines_to_collective(
             "lines": stamped,
             "payments": [],
             "payment_status": "open",
+            "mode": event_mode_label(ev.get("status")),
         }
         session_id = ensure_order_session(
             db,
@@ -193,5 +200,10 @@ def append_lines_to_collective(
             db,
             event_id=event_id,
             client_order_id=cid,
-            payload=enrich_payload_for_cloud_sync(payload, local_order_id=order.id, session_id=session_id),
+            payload=enrich_payload_for_cloud_sync(
+                payload,
+                local_order_id=order.id,
+                session_id=session_id,
+                mode=event_mode_label(ev.get("status")),
+            ),
         )

pi/backend/app/routers/edge_common.py

@@ -469,6 +469,7 @@ def _sync_outbox_payload(db: Session, order: LocalOrder, payload: dict) -> None:
         payload,
         local_order_id=order.id,
         session_id=int(order.session_id),
+        mode=str(payload.get("mode") or "") or None,
     )
     cid = order.client_order_id
     for out in (