25d5da78ef
All checks were successful
CI / go-lint (push) Has been skipped
CI / python-lint (push) Has been skipped
CI / nodejs-lint (push) Has been skipped
CI / test-go-ai-compliance (push) Successful in 34s
CI / test-python-backend-compliance (push) Successful in 32s
CI / test-python-document-crawler (push) Successful in 21s
CI / test-python-dsms-gateway (push) Successful in 17s
Paket A — RAG Proxy: - NEU: admin-compliance/app/api/sdk/v1/rag/[[...path]]/route.ts → Proxy zu ai-compliance-sdk:8090, GET+POST, UUID-Validierung - UPDATE: rag/page.tsx — setTimeout Mock → echte API-Calls GET /regulations → dynamische suggestedQuestions POST /search → Qdrant-Ergebnisse mit score, title, reference Paket B — Security-Backlog + Quality: - NEU: migrations/014_security_backlog.sql + 015_quality.sql - NEU: compliance/api/security_backlog_routes.py — CRUD + Stats - NEU: compliance/api/quality_routes.py — Metrics + Tests CRUD + Stats - UPDATE: security-backlog/page.tsx — mockItems → API - UPDATE: quality/page.tsx — mockMetrics/mockTests → API - UPDATE: compliance/api/__init__.py — Router-Registrierung - NEU: tests/test_security_backlog_routes.py (48 Tests — 48/48 bestanden) - NEU: tests/test_quality_routes.py (67 Tests — 67/67 bestanden) Paket C — Notfallplan Incidents + Templates: - NEU: migrations/016_notfallplan_incidents.sql compliance_notfallplan_incidents + compliance_notfallplan_templates - UPDATE: notfallplan_routes.py — GET/POST/PUT/DELETE für /incidents + /templates - UPDATE: notfallplan/page.tsx — Incidents-Tab + Templates-Tab → API - UPDATE: tests/test_notfallplan_routes.py (+76 neue Tests — alle bestanden) Paket D — Loeschfristen localStorage → API: - NEU: migrations/017_loeschfristen.sql (JSONB: legal_holds, storage_locations, ...) - NEU: compliance/api/loeschfristen_routes.py — CRUD + Stats + Status-Update - UPDATE: loeschfristen/page.tsx — vollständige localStorage → API Migration createNewPolicy → POST (API-UUID als id), deletePolicy → DELETE, handleSaveAndClose → PUT, adoptGeneratedPolicies → POST je Policy apiToPolicy() + policyToPayload() Mapper, saving-State für Buttons - NEU: tests/test_loeschfristen_routes.py (58 Tests — alle bestanden) Gesamt: 253 neue Tests, alle bestanden (48 + 67 + 76 + 58 + bestehende) Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
631 lines
24 KiB
Python
631 lines
24 KiB
Python
"""Tests for Loeschfristen routes and schemas (loeschfristen_routes.py)."""
|
|
|
|
import pytest
|
|
from unittest.mock import MagicMock
|
|
from fastapi.testclient import TestClient
|
|
from fastapi import FastAPI
|
|
from datetime import datetime
|
|
|
|
from compliance.api.loeschfristen_routes import (
|
|
LoeschfristCreate,
|
|
LoeschfristUpdate,
|
|
StatusUpdate,
|
|
_row_to_dict,
|
|
_get_tenant_id,
|
|
DEFAULT_TENANT_ID,
|
|
JSONB_FIELDS,
|
|
router,
|
|
)
|
|
|
|
app = FastAPI()
|
|
app.include_router(router)
|
|
client = TestClient(app)
|
|
|
|
DEFAULT_TENANT = DEFAULT_TENANT_ID # "9282a473-5c95-4b3a-bf78-0ecc0ec71d3e"
|
|
POLICY_ID = "ffffffff-0001-0001-0001-000000000001"
|
|
UNKNOWN_ID = "aaaaaaaa-9999-9999-9999-999999999999"
|
|
|
|
|
|
# =============================================================================
|
|
# Helpers
|
|
# =============================================================================
|
|
|
|
def make_policy_row(overrides=None):
|
|
data = {
|
|
"id": POLICY_ID,
|
|
"tenant_id": DEFAULT_TENANT,
|
|
"policy_id": "LF-2024-001",
|
|
"data_object_name": "Kundendaten",
|
|
"description": "Kundendaten Loeschfrist",
|
|
"affected_groups": [],
|
|
"data_categories": [],
|
|
"primary_purpose": "Vertrag",
|
|
"deletion_trigger": "PURPOSE_END",
|
|
"retention_driver": "HGB_257",
|
|
"retention_driver_detail": None,
|
|
"retention_duration": 10,
|
|
"retention_unit": "YEARS",
|
|
"retention_description": None,
|
|
"start_event": None,
|
|
"has_active_legal_hold": False,
|
|
"legal_holds": [],
|
|
"storage_locations": [],
|
|
"deletion_method": "MANUAL_REVIEW_DELETE",
|
|
"deletion_method_detail": None,
|
|
"responsible_role": None,
|
|
"responsible_person": None,
|
|
"release_process": None,
|
|
"linked_vvt_activity_ids": [],
|
|
"status": "DRAFT",
|
|
"last_review_date": None,
|
|
"next_review_date": None,
|
|
"review_interval": "ANNUAL",
|
|
"tags": [],
|
|
"created_at": datetime(2024, 1, 1),
|
|
"updated_at": datetime(2024, 1, 1),
|
|
}
|
|
if overrides:
|
|
data.update(overrides)
|
|
row = MagicMock()
|
|
row._mapping = data
|
|
return row
|
|
|
|
|
|
def make_stats_row(overrides=None):
|
|
data = {
|
|
"total": 0,
|
|
"active": 0,
|
|
"draft": 0,
|
|
"review_needed": 0,
|
|
"archived": 0,
|
|
"legal_holds_count": 0,
|
|
"overdue_reviews": 0,
|
|
}
|
|
if overrides:
|
|
data.update(overrides)
|
|
row = MagicMock()
|
|
row._mapping = data
|
|
return row
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_db():
|
|
from classroom_engine.database import get_db
|
|
db = MagicMock()
|
|
app.dependency_overrides[get_db] = lambda: db
|
|
yield db
|
|
app.dependency_overrides.clear()
|
|
|
|
|
|
# =============================================================================
|
|
# Helper / Utility Tests
|
|
# =============================================================================
|
|
|
|
class TestRowToDict:
|
|
def test_converts_datetime_to_isoformat(self):
|
|
row = make_policy_row({"created_at": datetime(2024, 6, 1, 12, 0, 0)})
|
|
result = _row_to_dict(row)
|
|
assert result["created_at"] == "2024-06-01T12:00:00"
|
|
|
|
def test_converts_none_datetime_remains_none(self):
|
|
row = make_policy_row({"next_review_date": None})
|
|
result = _row_to_dict(row)
|
|
assert result["next_review_date"] is None
|
|
|
|
def test_preserves_string_values(self):
|
|
row = make_policy_row({"data_object_name": "Mitarbeiterdaten"})
|
|
result = _row_to_dict(row)
|
|
assert result["data_object_name"] == "Mitarbeiterdaten"
|
|
|
|
def test_preserves_list_values(self):
|
|
row = make_policy_row({"tags": ["dsgvo", "hgb"]})
|
|
result = _row_to_dict(row)
|
|
assert result["tags"] == ["dsgvo", "hgb"]
|
|
|
|
def test_preserves_int_values(self):
|
|
row = make_policy_row({"retention_duration": 7})
|
|
result = _row_to_dict(row)
|
|
assert result["retention_duration"] == 7
|
|
|
|
|
|
class TestGetTenantId:
|
|
def test_valid_uuid_is_returned(self):
|
|
assert _get_tenant_id("9282a473-5c95-4b3a-bf78-0ecc0ec71d3e") == "9282a473-5c95-4b3a-bf78-0ecc0ec71d3e"
|
|
|
|
def test_invalid_uuid_returns_default(self):
|
|
assert _get_tenant_id("not-a-uuid") == DEFAULT_TENANT_ID
|
|
|
|
def test_none_returns_default(self):
|
|
assert _get_tenant_id(None) == DEFAULT_TENANT_ID
|
|
|
|
|
|
class TestJsonbFields:
|
|
def test_jsonb_fields_set(self):
|
|
expected = {"affected_groups", "data_categories", "legal_holds",
|
|
"storage_locations", "linked_vvt_activity_ids", "tags"}
|
|
assert JSONB_FIELDS == expected
|
|
|
|
|
|
# =============================================================================
|
|
# Schema Tests — LoeschfristCreate
|
|
# =============================================================================
|
|
|
|
class TestLoeschfristCreate:
|
|
def test_minimal_requires_data_object_name(self):
|
|
obj = LoeschfristCreate(data_object_name="Kundendaten")
|
|
assert obj.data_object_name == "Kundendaten"
|
|
assert obj.deletion_trigger == "PURPOSE_END"
|
|
assert obj.status == "DRAFT"
|
|
assert obj.has_active_legal_hold is False
|
|
|
|
def test_full_object(self):
|
|
obj = LoeschfristCreate(
|
|
data_object_name="Mitarbeiterdaten",
|
|
description="HR-Daten",
|
|
primary_purpose="Arbeitsvertrag",
|
|
retention_driver="AO_147",
|
|
retention_duration=6,
|
|
retention_unit="YEARS",
|
|
status="ACTIVE",
|
|
tags=["hr", "personal"],
|
|
data_categories=["name", "address"],
|
|
)
|
|
assert obj.retention_duration == 6
|
|
assert obj.retention_unit == "YEARS"
|
|
assert obj.status == "ACTIVE"
|
|
assert len(obj.tags) == 2
|
|
|
|
def test_missing_data_object_name_raises_validation_error(self):
|
|
import pydantic
|
|
with pytest.raises(pydantic.ValidationError):
|
|
LoeschfristCreate()
|
|
|
|
def test_optional_fields_default_to_none(self):
|
|
obj = LoeschfristCreate(data_object_name="Test")
|
|
assert obj.description is None
|
|
assert obj.retention_duration is None
|
|
assert obj.responsible_role is None
|
|
assert obj.policy_id is None
|
|
|
|
|
|
class TestLoeschfristUpdate:
|
|
def test_empty_update(self):
|
|
obj = LoeschfristUpdate()
|
|
data = obj.model_dump(exclude_unset=True)
|
|
assert data == {}
|
|
|
|
def test_partial_update_status(self):
|
|
obj = LoeschfristUpdate(status="ACTIVE")
|
|
data = obj.model_dump(exclude_unset=True)
|
|
assert data == {"status": "ACTIVE"}
|
|
|
|
def test_partial_update_multiple_fields(self):
|
|
obj = LoeschfristUpdate(
|
|
data_object_name="Neuer Name",
|
|
retention_duration=5,
|
|
retention_unit="YEARS",
|
|
)
|
|
data = obj.model_dump(exclude_unset=True)
|
|
assert data["data_object_name"] == "Neuer Name"
|
|
assert data["retention_duration"] == 5
|
|
assert "status" not in data
|
|
|
|
|
|
class TestStatusUpdateSchema:
|
|
def test_valid_status(self):
|
|
obj = StatusUpdate(status="ACTIVE")
|
|
assert obj.status == "ACTIVE"
|
|
|
|
def test_all_valid_statuses(self):
|
|
for s in ("DRAFT", "ACTIVE", "REVIEW_NEEDED", "ARCHIVED"):
|
|
obj = StatusUpdate(status=s)
|
|
assert obj.status == s
|
|
|
|
|
|
# =============================================================================
|
|
# GET /loeschfristen — List
|
|
# =============================================================================
|
|
|
|
class TestListLoeschfristen:
|
|
def test_list_returns_empty(self, mock_db):
|
|
mock_db.execute.return_value.fetchone.return_value = MagicMock(__getitem__=lambda s, i: 0)
|
|
mock_db.execute.return_value.fetchall.return_value = []
|
|
# Two execute calls: COUNT then SELECT
|
|
count_result = MagicMock()
|
|
count_result.__getitem__ = lambda s, i: 0
|
|
list_result = MagicMock()
|
|
list_result.fetchall.return_value = []
|
|
mock_db.execute.side_effect = [
|
|
MagicMock(fetchone=lambda: count_result),
|
|
list_result,
|
|
]
|
|
resp = client.get("/loeschfristen")
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
assert "policies" in body
|
|
assert "total" in body
|
|
|
|
def test_list_returns_policies(self, mock_db):
|
|
count_result = MagicMock()
|
|
count_result.__getitem__ = lambda s, i: 1
|
|
policy_row = make_policy_row()
|
|
list_result = MagicMock()
|
|
list_result.fetchall.return_value = [policy_row]
|
|
mock_db.execute.side_effect = [
|
|
MagicMock(fetchone=lambda: count_result),
|
|
list_result,
|
|
]
|
|
resp = client.get("/loeschfristen")
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
assert len(body["policies"]) == 1
|
|
assert body["policies"][0]["data_object_name"] == "Kundendaten"
|
|
|
|
def test_list_filter_by_status(self, mock_db):
|
|
count_result = MagicMock()
|
|
count_result.__getitem__ = lambda s, i: 0
|
|
list_result = MagicMock()
|
|
list_result.fetchall.return_value = []
|
|
mock_db.execute.side_effect = [
|
|
MagicMock(fetchone=lambda: count_result),
|
|
list_result,
|
|
]
|
|
resp = client.get("/loeschfristen?status=ACTIVE")
|
|
assert resp.status_code == 200
|
|
|
|
def test_list_filter_by_retention_driver(self, mock_db):
|
|
count_result = MagicMock()
|
|
count_result.__getitem__ = lambda s, i: 0
|
|
list_result = MagicMock()
|
|
list_result.fetchall.return_value = []
|
|
mock_db.execute.side_effect = [
|
|
MagicMock(fetchone=lambda: count_result),
|
|
list_result,
|
|
]
|
|
resp = client.get("/loeschfristen?retention_driver=HGB_257")
|
|
assert resp.status_code == 200
|
|
|
|
def test_list_filter_by_search(self, mock_db):
|
|
count_result = MagicMock()
|
|
count_result.__getitem__ = lambda s, i: 0
|
|
list_result = MagicMock()
|
|
list_result.fetchall.return_value = []
|
|
mock_db.execute.side_effect = [
|
|
MagicMock(fetchone=lambda: count_result),
|
|
list_result,
|
|
]
|
|
resp = client.get("/loeschfristen?search=Kunden")
|
|
assert resp.status_code == 200
|
|
|
|
def test_list_uses_default_tenant(self, mock_db):
|
|
count_result = MagicMock()
|
|
count_result.__getitem__ = lambda s, i: 0
|
|
list_result = MagicMock()
|
|
list_result.fetchall.return_value = []
|
|
mock_db.execute.side_effect = [
|
|
MagicMock(fetchone=lambda: count_result),
|
|
list_result,
|
|
]
|
|
client.get("/loeschfristen")
|
|
# First call is the COUNT query
|
|
first_call_params = mock_db.execute.call_args_list[0][0][1]
|
|
assert first_call_params["tenant_id"] == DEFAULT_TENANT
|
|
|
|
def test_list_pagination_params(self, mock_db):
|
|
count_result = MagicMock()
|
|
count_result.__getitem__ = lambda s, i: 0
|
|
list_result = MagicMock()
|
|
list_result.fetchall.return_value = []
|
|
mock_db.execute.side_effect = [
|
|
MagicMock(fetchone=lambda: count_result),
|
|
list_result,
|
|
]
|
|
resp = client.get("/loeschfristen?limit=10&offset=20")
|
|
assert resp.status_code == 200
|
|
|
|
|
|
# =============================================================================
|
|
# GET /loeschfristen/stats
|
|
# =============================================================================
|
|
|
|
class TestGetLoeschfristenStats:
|
|
def test_stats_returns_all_keys(self, mock_db):
|
|
stats_row = make_stats_row()
|
|
mock_db.execute.return_value.fetchone.return_value = stats_row
|
|
resp = client.get("/loeschfristen/stats")
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
for key in ("total", "active", "draft", "review_needed", "archived",
|
|
"legal_holds_count", "overdue_reviews"):
|
|
assert key in body, f"Missing key: {key}"
|
|
|
|
def test_stats_returns_correct_counts(self, mock_db):
|
|
stats_row = make_stats_row({
|
|
"total": 10,
|
|
"active": 4,
|
|
"draft": 3,
|
|
"review_needed": 2,
|
|
"archived": 1,
|
|
"legal_holds_count": 1,
|
|
"overdue_reviews": 0,
|
|
})
|
|
mock_db.execute.return_value.fetchone.return_value = stats_row
|
|
resp = client.get("/loeschfristen/stats")
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
assert body["total"] == 10
|
|
assert body["active"] == 4
|
|
assert body["draft"] == 3
|
|
assert body["review_needed"] == 2
|
|
assert body["archived"] == 1
|
|
assert body["legal_holds_count"] == 1
|
|
|
|
def test_stats_all_zeros_when_no_data(self, mock_db):
|
|
stats_row = make_stats_row()
|
|
mock_db.execute.return_value.fetchone.return_value = stats_row
|
|
resp = client.get("/loeschfristen/stats")
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
for key in ("total", "active", "draft", "review_needed", "archived",
|
|
"legal_holds_count", "overdue_reviews"):
|
|
assert body[key] == 0
|
|
|
|
def test_stats_uses_default_tenant(self, mock_db):
|
|
stats_row = make_stats_row()
|
|
mock_db.execute.return_value.fetchone.return_value = stats_row
|
|
client.get("/loeschfristen/stats")
|
|
call_params = mock_db.execute.call_args[0][1]
|
|
assert call_params["tenant_id"] == DEFAULT_TENANT
|
|
|
|
def test_stats_with_valid_tenant_header(self, mock_db):
|
|
stats_row = make_stats_row()
|
|
mock_db.execute.return_value.fetchone.return_value = stats_row
|
|
client.get(
|
|
"/loeschfristen/stats",
|
|
headers={"x-tenant-id": "9282a473-5c95-4b3a-bf78-0ecc0ec71d3e"},
|
|
)
|
|
call_params = mock_db.execute.call_args[0][1]
|
|
assert call_params["tenant_id"] == "9282a473-5c95-4b3a-bf78-0ecc0ec71d3e"
|
|
|
|
|
|
# =============================================================================
|
|
# POST /loeschfristen
|
|
# =============================================================================
|
|
|
|
class TestCreateLoeschfrist:
|
|
def test_create_minimal(self, mock_db):
|
|
row = make_policy_row()
|
|
mock_db.execute.return_value.fetchone.return_value = row
|
|
resp = client.post("/loeschfristen", json={"data_object_name": "Kundendaten"})
|
|
assert resp.status_code == 201
|
|
assert resp.json()["data_object_name"] == "Kundendaten"
|
|
|
|
def test_create_missing_data_object_name_returns_422(self, mock_db):
|
|
resp = client.post("/loeschfristen", json={"description": "No name"})
|
|
assert resp.status_code == 422
|
|
|
|
def test_create_full_payload(self, mock_db):
|
|
row = make_policy_row({
|
|
"data_object_name": "Mitarbeiterdaten",
|
|
"status": "ACTIVE",
|
|
"retention_duration": 6,
|
|
"retention_unit": "YEARS",
|
|
})
|
|
mock_db.execute.return_value.fetchone.return_value = row
|
|
resp = client.post("/loeschfristen", json={
|
|
"data_object_name": "Mitarbeiterdaten",
|
|
"description": "HR-Datensatz",
|
|
"retention_driver": "AO_147",
|
|
"retention_duration": 6,
|
|
"retention_unit": "YEARS",
|
|
"status": "ACTIVE",
|
|
})
|
|
assert resp.status_code == 201
|
|
data = resp.json()
|
|
assert data["status"] == "ACTIVE"
|
|
assert data["retention_duration"] == 6
|
|
|
|
def test_create_commits(self, mock_db):
|
|
row = make_policy_row()
|
|
mock_db.execute.return_value.fetchone.return_value = row
|
|
client.post("/loeschfristen", json={"data_object_name": "X"})
|
|
mock_db.commit.assert_called_once()
|
|
|
|
def test_create_uses_default_tenant(self, mock_db):
|
|
row = make_policy_row()
|
|
mock_db.execute.return_value.fetchone.return_value = row
|
|
client.post("/loeschfristen", json={"data_object_name": "X"})
|
|
call_params = mock_db.execute.call_args[0][1]
|
|
assert call_params["tenant_id"] == DEFAULT_TENANT
|
|
|
|
def test_create_jsonb_fields_are_json_encoded(self, mock_db):
|
|
row = make_policy_row()
|
|
mock_db.execute.return_value.fetchone.return_value = row
|
|
client.post("/loeschfristen", json={
|
|
"data_object_name": "Test",
|
|
"tags": ["a", "b"],
|
|
"data_categories": ["name"],
|
|
})
|
|
call_params = mock_db.execute.call_args[0][1]
|
|
import json
|
|
# JSONB fields must be JSON strings for CAST
|
|
assert json.loads(call_params["tags"]) == ["a", "b"]
|
|
assert json.loads(call_params["data_categories"]) == ["name"]
|
|
|
|
|
|
# =============================================================================
|
|
# GET /loeschfristen/{id}
|
|
# =============================================================================
|
|
|
|
class TestGetLoeschfrist:
|
|
def test_get_existing_policy(self, mock_db):
|
|
row = make_policy_row()
|
|
mock_db.execute.return_value.fetchone.return_value = row
|
|
resp = client.get(f"/loeschfristen/{POLICY_ID}")
|
|
assert resp.status_code == 200
|
|
assert resp.json()["data_object_name"] == "Kundendaten"
|
|
|
|
def test_get_not_found_returns_404(self, mock_db):
|
|
mock_db.execute.return_value.fetchone.return_value = None
|
|
resp = client.get(f"/loeschfristen/{UNKNOWN_ID}")
|
|
assert resp.status_code == 404
|
|
|
|
def test_get_passes_id_and_tenant(self, mock_db):
|
|
row = make_policy_row()
|
|
mock_db.execute.return_value.fetchone.return_value = row
|
|
client.get(f"/loeschfristen/{POLICY_ID}")
|
|
call_params = mock_db.execute.call_args[0][1]
|
|
assert call_params["id"] == POLICY_ID
|
|
assert call_params["tenant_id"] == DEFAULT_TENANT
|
|
|
|
def test_get_all_fields_present(self, mock_db):
|
|
row = make_policy_row()
|
|
mock_db.execute.return_value.fetchone.return_value = row
|
|
resp = client.get(f"/loeschfristen/{POLICY_ID}")
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
for field in ("id", "tenant_id", "data_object_name", "status",
|
|
"retention_duration", "retention_unit", "created_at", "updated_at"):
|
|
assert field in body, f"Missing field: {field}"
|
|
|
|
|
|
# =============================================================================
|
|
# PUT /loeschfristen/{id}
|
|
# =============================================================================
|
|
|
|
class TestUpdateLoeschfrist:
|
|
def test_update_success(self, mock_db):
|
|
updated_row = make_policy_row({"data_object_name": "Neuer Name"})
|
|
mock_db.execute.return_value.fetchone.return_value = updated_row
|
|
resp = client.put(f"/loeschfristen/{POLICY_ID}", json={"data_object_name": "Neuer Name"})
|
|
assert resp.status_code == 200
|
|
assert resp.json()["data_object_name"] == "Neuer Name"
|
|
|
|
def test_update_not_found_returns_404(self, mock_db):
|
|
mock_db.execute.return_value.fetchone.return_value = None
|
|
resp = client.put(f"/loeschfristen/{UNKNOWN_ID}", json={"data_object_name": "X"})
|
|
assert resp.status_code == 404
|
|
|
|
def test_update_empty_body_returns_400(self, mock_db):
|
|
resp = client.put(f"/loeschfristen/{POLICY_ID}", json={})
|
|
assert resp.status_code == 400
|
|
|
|
def test_update_jsonb_field(self, mock_db):
|
|
updated_row = make_policy_row({"tags": ["urgent"]})
|
|
mock_db.execute.return_value.fetchone.return_value = updated_row
|
|
resp = client.put(f"/loeschfristen/{POLICY_ID}", json={"tags": ["urgent"]})
|
|
assert resp.status_code == 200
|
|
call_params = mock_db.execute.call_args[0][1]
|
|
import json
|
|
assert json.loads(call_params["tags"]) == ["urgent"]
|
|
|
|
def test_update_sets_updated_at(self, mock_db):
|
|
updated_row = make_policy_row()
|
|
mock_db.execute.return_value.fetchone.return_value = updated_row
|
|
client.put(f"/loeschfristen/{POLICY_ID}", json={"status": "ACTIVE"})
|
|
call_params = mock_db.execute.call_args[0][1]
|
|
assert "updated_at" in call_params
|
|
|
|
def test_update_commits(self, mock_db):
|
|
updated_row = make_policy_row()
|
|
mock_db.execute.return_value.fetchone.return_value = updated_row
|
|
client.put(f"/loeschfristen/{POLICY_ID}", json={"status": "ACTIVE"})
|
|
mock_db.commit.assert_called_once()
|
|
|
|
|
|
# =============================================================================
|
|
# PUT /loeschfristen/{id}/status
|
|
# =============================================================================
|
|
|
|
class TestUpdateLoeschfristStatus:
|
|
def test_valid_status_active(self, mock_db):
|
|
updated_row = make_policy_row({"status": "ACTIVE"})
|
|
mock_db.execute.return_value.fetchone.return_value = updated_row
|
|
resp = client.put(f"/loeschfristen/{POLICY_ID}/status", json={"status": "ACTIVE"})
|
|
assert resp.status_code == 200
|
|
assert resp.json()["status"] == "ACTIVE"
|
|
|
|
def test_valid_status_archived(self, mock_db):
|
|
updated_row = make_policy_row({"status": "ARCHIVED"})
|
|
mock_db.execute.return_value.fetchone.return_value = updated_row
|
|
resp = client.put(f"/loeschfristen/{POLICY_ID}/status", json={"status": "ARCHIVED"})
|
|
assert resp.status_code == 200
|
|
|
|
def test_valid_status_review_needed(self, mock_db):
|
|
updated_row = make_policy_row({"status": "REVIEW_NEEDED"})
|
|
mock_db.execute.return_value.fetchone.return_value = updated_row
|
|
resp = client.put(f"/loeschfristen/{POLICY_ID}/status", json={"status": "REVIEW_NEEDED"})
|
|
assert resp.status_code == 200
|
|
|
|
def test_valid_status_draft(self, mock_db):
|
|
updated_row = make_policy_row({"status": "DRAFT"})
|
|
mock_db.execute.return_value.fetchone.return_value = updated_row
|
|
resp = client.put(f"/loeschfristen/{POLICY_ID}/status", json={"status": "DRAFT"})
|
|
assert resp.status_code == 200
|
|
|
|
def test_invalid_status_returns_400(self, mock_db):
|
|
resp = client.put(f"/loeschfristen/{POLICY_ID}/status", json={"status": "INVALID_STATUS"})
|
|
assert resp.status_code == 400
|
|
|
|
def test_status_not_found_returns_404(self, mock_db):
|
|
mock_db.execute.return_value.fetchone.return_value = None
|
|
resp = client.put(f"/loeschfristen/{UNKNOWN_ID}/status", json={"status": "ACTIVE"})
|
|
assert resp.status_code == 404
|
|
|
|
def test_status_update_commits(self, mock_db):
|
|
updated_row = make_policy_row()
|
|
mock_db.execute.return_value.fetchone.return_value = updated_row
|
|
client.put(f"/loeschfristen/{POLICY_ID}/status", json={"status": "ACTIVE"})
|
|
mock_db.commit.assert_called_once()
|
|
|
|
def test_status_update_passes_correct_params(self, mock_db):
|
|
updated_row = make_policy_row({"status": "ACTIVE"})
|
|
mock_db.execute.return_value.fetchone.return_value = updated_row
|
|
client.put(f"/loeschfristen/{POLICY_ID}/status", json={"status": "ACTIVE"})
|
|
call_params = mock_db.execute.call_args[0][1]
|
|
assert call_params["status"] == "ACTIVE"
|
|
assert call_params["id"] == POLICY_ID
|
|
assert call_params["tenant_id"] == DEFAULT_TENANT
|
|
|
|
|
|
# =============================================================================
|
|
# DELETE /loeschfristen/{id}
|
|
# =============================================================================
|
|
|
|
class TestDeleteLoeschfrist:
|
|
def test_delete_success_returns_204(self, mock_db):
|
|
mock_db.execute.return_value.rowcount = 1
|
|
resp = client.delete(f"/loeschfristen/{POLICY_ID}")
|
|
assert resp.status_code == 204
|
|
|
|
def test_delete_not_found_returns_404(self, mock_db):
|
|
mock_db.execute.return_value.rowcount = 0
|
|
resp = client.delete(f"/loeschfristen/{UNKNOWN_ID}")
|
|
assert resp.status_code == 404
|
|
|
|
def test_delete_commits(self, mock_db):
|
|
mock_db.execute.return_value.rowcount = 1
|
|
client.delete(f"/loeschfristen/{POLICY_ID}")
|
|
mock_db.commit.assert_called_once()
|
|
|
|
def test_delete_commits_before_rowcount_check(self, mock_db):
|
|
mock_db.execute.return_value.rowcount = 0
|
|
client.delete(f"/loeschfristen/{UNKNOWN_ID}")
|
|
mock_db.commit.assert_called_once()
|
|
|
|
def test_delete_passes_correct_id_and_tenant(self, mock_db):
|
|
mock_db.execute.return_value.rowcount = 1
|
|
client.delete(f"/loeschfristen/{POLICY_ID}")
|
|
call_params = mock_db.execute.call_args[0][1]
|
|
assert call_params["id"] == POLICY_ID
|
|
assert call_params["tenant_id"] == DEFAULT_TENANT
|
|
|
|
def test_delete_with_custom_tenant(self, mock_db):
|
|
mock_db.execute.return_value.rowcount = 1
|
|
client.delete(
|
|
f"/loeschfristen/{POLICY_ID}",
|
|
headers={"x-tenant-id": "9282a473-5c95-4b3a-bf78-0ecc0ec71d3e"},
|
|
)
|
|
call_params = mock_db.execute.call_args[0][1]
|
|
assert call_params["tenant_id"] == "9282a473-5c95-4b3a-bf78-0ecc0ec71d3e"
|