refactor(cra): readiness fetches Machinery-Reg obligations from use_case=maschinen

Follow-up to the machinery_reg_cyber.py removal: the readiness endpoint now pulls
Machinery Regulation 2023/1230 cyber-with-safety obligations from the shared
Controls-API (use_case=maschinen), tagged "Maschinen-VO", best-effort.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

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")
+            machinery = _machinery_obligations()
-            for req in MACHINERY_REG_CYBER:
+            if machinery:
-                bucket = _GUIDELINE_BUCKET.get(req.get("evidence_type", "process"), "process")
+                regulations.append("Maschinen-VO 2023/1230")
-                groups[bucket].append({
+                for bucket, item in machinery:
-                    "req_id": req["req_id"], "title": req["title"], "category": req["category"],
+                    groups[bucket].append(item)
-                    "annex_anchor": req["annex_anchor"], "severity": req["severity"],
-                    "effort_days": None, "measures": [], "source": "Maschinen-VO",
-                })
     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,

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():
+def test_machinery_flag_does_not_break_assessment():
-    d = client.post("/api/v1/cra/readiness", json={
+    # Machinery-Reg obligations come from the Controls-API (use_case=maschinen, DB) and
-        "intended_use": "App fuer Industrieanlagen", "connected_to_internet": True, "is_machinery": True}).json()
+    # are verified live, not here. Without a DB the endpoint must still return the CRA
-    assert "Maschinen-VO 2023/1230" in d["regulations"]
+    # 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)