"""Regression tests for the hardening pass (security, robustness, data-quality edge cases).""" from __future__ import annotations import uuid from datetime import UTC, datetime import pytest from sqlalchemy import delete, select from app.core.config import Settings, get_settings, insecure_default_secrets from app.core.db import SessionLocal from app.core.observability import UNMATCHED_ROUTE_LABEL from app.core.ratelimit import FailedAttemptLimiter from app.models.audit import AuditEvent from app.models.customer import Customer from app.models.data_quality import DataQualityIssue from app.services.data_quality import run_scan from tests.test_return import _activate_booking, _return_body def test_insecure_defaults_are_detected_only_for_relevant_secrets(): defaults = Settings(_env_file=None) assert "app_secret" in insecure_default_secrets(defaults) assert "mcp_hub_service_token" not in insecure_default_secrets(defaults) hardened = Settings( _env_file=None, app_secret="x" * 32, n8n_callback_token="c" * 32, mcp_hub_registration_enabled=True, ) assert insecure_default_secrets(hardened) == ["mcp_hub_service_token"] def test_production_refuses_to_start_with_placeholder_secrets(monkeypatch): monkeypatch.setenv("MOBILITYOPS_ENV", "production") monkeypatch.setenv("APP_SECRET", "replace-in-production") get_settings.cache_clear() try: with pytest.raises(RuntimeError, match="placeholder secrets"): get_settings() finally: get_settings.cache_clear() monkeypatch.undo() get_settings.cache_clear() # The cached settings the running app relies on must be intact afterwards. assert get_settings().mobilityops_env == "test" def test_failed_attempt_limiter_blocks_after_threshold_and_resets(): limiter = FailedAttemptLimiter(max_failures=3, window_seconds=60) for _ in range(3): assert limiter.retry_after_seconds("1.2.3.4") == 0 limiter.record_failure("1.2.3.4") assert limiter.retry_after_seconds("1.2.3.4") > 0 assert limiter.retry_after_seconds("5.6.7.8") == 0 limiter.reset("1.2.3.4") assert limiter.retry_after_seconds("1.2.3.4") == 0 def test_audit_export_accepts_naive_datetimes(ops_client): response = ops_client.get( "/api/v1/audit/export.csv", params={"occurred_from": "2026-01-01T00:00:00", "occurred_to": "2026-01-15T00:00:00"}, ) assert response.status_code == 200 def test_audit_list_rejects_non_uuid_correlation_id(ops_client): response = ops_client.get("/api/v1/audit", params={"correlation_id": "not-a-uuid"}) assert response.status_code == 422 def test_unmatched_routes_do_not_create_metric_series(client): probe = f"/api/v1/does-not-exist-{uuid.uuid4().hex}" assert client.get(probe).status_code == 404 metrics = client.get("/metrics").text assert probe not in metrics assert UNMATCHED_ROUTE_LABEL in metrics def test_bookings_page_for_unknown_vehicle_keeps_page_shape(ops_client): response = ops_client.get("/api/v1/bookings", params={"vehicle_ref": "MO-NOPE", "page": 1}) assert response.status_code == 200 assert response.json() == { "items": [], "page": 1, "page_size": 25, "total": 0, "total_pages": 1, } def test_return_idempotency_key_rejects_different_body(ops_client): booking_ref = _activate_booking("MO-011", start_odometer_km=30000) key = "test-return-fingerprint-001" first = ops_client.post( f"/api/v1/bookings/{booking_ref}/return", json=_return_body(end_odometer_km=30500), headers={"Idempotency-Key": key}, ) assert first.status_code == 201 replay = ops_client.post( f"/api/v1/bookings/{booking_ref}/return", json=_return_body(end_odometer_km=30500), headers={"Idempotency-Key": key}, ) assert replay.status_code == 201 mismatch = ops_client.post( f"/api/v1/bookings/{booking_ref}/return", json=_return_body(end_odometer_km=30999), headers={"Idempotency-Key": key}, ) assert mismatch.status_code == 409 assert mismatch.json()["error"]["code"] == "IDEMPOTENCY_KEY_REUSED" def test_return_callback_rejects_malformed_correlation_id(client): response = client.post( "/api/v1/integrations/n8n/return-callback", json={"follow_up": "cleaning", "correlation_id": "nope"}, headers={ "Idempotency-Key": str(uuid.uuid4()), "X-Service-Token": get_settings().n8n_callback_token, }, ) assert response.status_code == 422 def test_blocked_checkout_booking_can_be_cancelled(ops_client): window = {"starts_at": "2051-03-01T10:00:00Z", "ends_at": "2051-03-02T12:00:00Z"} available = ops_client.get("/api/v1/bookings/availability", params=window).json() vehicle_option = next(item for item in available if item["operational_status"] == "available") vehicle = ops_client.get(f"/api/v1/vehicles/{vehicle_option['public_ref']}").json() booking = ops_client.post( "/api/v1/bookings", json={ "customer_ref": "CUS-0002", "vehicle_ref": vehicle["public_ref"], "requirements_complete": True, **window, }, ).json() checkout = ops_client.post( f"/api/v1/bookings/{booking['public_ref']}/checkout", json={ "start_odometer_km": vehicle["odometer_km"], "fuel_level_percent": 80, "cleanliness_ok": True, "damage_reported": True, "technical_warning": False, }, ) assert checkout.status_code == 200 assert checkout.json()["booking_status"] == "blocked" cancelled = ops_client.post( f"/api/v1/bookings/{booking['public_ref']}/cancel", json={"reason": "Vehicle damaged at departure inspection"}, ) assert cancelled.status_code == 200 assert cancelled.json()["status"] == "cancelled" def test_scan_does_not_flag_anonymised_customers_as_missing_fields(ops_client): with SessionLocal() as db: customer = Customer( public_ref="CUS-ANON-SCAN", first_name="Anoniem", last_name="Klant", email=None, phone=None, postal_code=None, city=None, date_of_birth=None, anonymized_at=datetime.now(UTC), ) db.add(customer) db.commit() customer_id = customer.id try: with SessionLocal() as db: run_scan(db) db.commit() flagged = db.scalar( select(DataQualityIssue.id).where( DataQualityIssue.entity_type == "customer", DataQualityIssue.entity_id == customer_id, ) ) assert flagged is None finally: with SessionLocal() as db: db.execute(delete(DataQualityIssue).where(DataQualityIssue.entity_id == customer_id)) db.execute(delete(AuditEvent).where(AuditEvent.entity_id == customer_id)) db.execute(delete(Customer).where(Customer.id == customer_id)) db.commit()