diff --git a/backend-compliance/compliance/api/cra_assess_routes.py b/backend-compliance/compliance/api/cra_assess_routes.py index e6098a06..cd4b95b2 100644 --- a/backend-compliance/compliance/api/cra_assess_routes.py +++ b/backend-compliance/compliance/api/cra_assess_routes.py @@ -20,7 +20,7 @@ from compliance.services.cra_use_case_controls import enrich_findings_with_bread from compliance.services.cra_component_findings import findings_from_components from compliance.api.cra_annex_i_data import ANNEX_I_REQUIREMENTS, MEASURES, DEADLINES from compliance.api.cra_routes import _classify # reuse the deterministic Annex III/IV classifier -from compliance.api.machinery_reg_cyber import MACHINERY_REG_CYBER +from compliance.services.use_case_controls import UseCaseControlsService from database import SessionLocal from .tenant_utils import get_tenant_id @@ -142,6 +142,38 @@ _PATH_HINT = { "NOT_IN_SCOPE": "—", } +# Machinery Regulation 2023/1230 cyber-with-safety obligations come from the shared +# Controls-API (use_case=maschinen, atom-grain, license-clean) — NOT hardcoded. +# Cyber-relevant sub-topics -> guideline bucket. +_MACHINERY_SUBTOPICS = [ + ("sicherheitsanforderungen", "code"), + ("risikomanagement", "process"), + ("konformitaetsbewertung", "document"), +] + + +def _machinery_obligations(limit_per: int = 4) -> list: + """(bucket, guideline_item) tuples from use_case=maschinen. Best-effort.""" + out = [] + db = SessionLocal() + try: + svc = UseCaseControlsService(db) + for sub_topic, bucket in _MACHINERY_SUBTOPICS: + try: + res = svc.controls_for_use_case("maschinen", sub_topic=sub_topic, limit=limit_per) + except Exception: + continue + for c in res.get("controls", []): + out.append((bucket, { + "req_id": c.get("control_id"), "title": c.get("title"), "category": sub_topic, + "annex_anchor": c.get("source_regulation", "Maschinenverordnung (EU) 2023/1230"), + "severity": (c.get("severity") or "").upper(), "effort_days": None, + "measures": [], "source": "Maschinen-VO", + })) + finally: + db.close() + return out + @router.post("/readiness") async def readiness(body: ReadinessRequest): @@ -173,14 +205,11 @@ async def readiness(body: ReadinessRequest): # Machine/plant builders are ALSO hit by the new Machinery Regulation's # cyber-with-safety essential requirements (Annex III) — show the combination. if body.is_machinery: - regulations.append("Maschinen-VO 2023/1230") - for req in MACHINERY_REG_CYBER: - bucket = _GUIDELINE_BUCKET.get(req.get("evidence_type", "process"), "process") - groups[bucket].append({ - "req_id": req["req_id"], "title": req["title"], "category": req["category"], - "annex_anchor": req["annex_anchor"], "severity": req["severity"], - "effort_days": None, "measures": [], "source": "Maschinen-VO", - }) + machinery = _machinery_obligations() + if machinery: + regulations.append("Maschinen-VO 2023/1230") + for bucket, item in machinery: + groups[bucket].append(item) total_effort = sum(r["effort_days"] for g in groups.values() for r in g if r.get("effort_days")) return { "in_scope": in_scope, diff --git a/backend-compliance/tests/test_cra_readiness.py b/backend-compliance/tests/test_cra_readiness.py index 4e150b44..7a34b50e 100644 --- a/backend-compliance/tests/test_cra_readiness.py +++ b/backend-compliance/tests/test_cra_readiness.py @@ -31,11 +31,14 @@ def test_no_digital_element_not_in_scope(): assert d["counts"]["code"] == 0 -def test_machinery_adds_tagged_machinery_reg_obligations(): - d = client.post("/api/v1/cra/readiness", json={ - "intended_use": "App fuer Industrieanlagen", "connected_to_internet": True, "is_machinery": True}).json() - assert "Maschinen-VO 2023/1230" in d["regulations"] +def test_machinery_flag_does_not_break_assessment(): + # Machinery-Reg obligations come from the Controls-API (use_case=maschinen, DB) and + # are verified live, not here. Without a DB the endpoint must still return the CRA + # guideline (best-effort machinery fetch). + r = client.post("/api/v1/cra/readiness", json={ + "intended_use": "App fuer Industrieanlagen", "connected_to_internet": True, "is_machinery": True}) + assert r.status_code == 200 + d = r.json() + assert d["in_scope"] is True items = d["guideline"]["code"] + d["guideline"]["process"] + d["guideline"]["document"] - assert any(it["source"] == "Maschinen-VO" for it in items) - assert any(it["req_id"] == "MR-1.1.9" for it in items) assert any(it["source"] == "CRA" for it in items)