From 739a477d3f37bd3b16942033c137810b9a4e6191 Mon Sep 17 00:00:00 2001 From: Benjamin Admin Date: Fri, 26 Jun 2026 09:52:46 +0200 Subject: [PATCH] feat(profile): CanonicalProductRegulatoryProfile convergence layer (types + mappers + tests) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ONE canonical product profile so the Go gap engine and the Python reasoning engine stop diverging ("SPS mit Remote Access" means the same everywhere). gap.ProductProfile LEADS; the reasoning ProductProfile becomes an adapter/DTO. Types + mappers only — no regulation logic, no Go changes, no UI, no new questions. - CanonicalProductRegulatoryProfile mirrors gap.ProductProfile + the Navigator gaps the audit found: economic-operator role, radio_module, generates_usage_data, lifecycle_phase, structured BOM (ProductComponent), safety-vs-security split, machine-vs-component + a forward-looking EnvironmentalImpact domain (wastewater/ air/chemicals triggers — fields only, no rules yet). - Mappers: from_product_wizard (lossless), from_company_profile (prefill incl. the machineBuilder block), to_gap_profile (emits the unchanged gap JSON shape), to_reasoning_profile (projects into the reasoning ProductProfile; AI stays delegated to ai-act/ucca). Only profile->reasoning is coupled; reasoning stays hermetic. - 10 tests = the 10 acceptance criteria incl. ProductWizard round-trip lossless, markets no longer forced ['EU'], and canonical->reasoning->discover_scope proving one semantic profile drives the engine. 33 tests green, mypy clean. Co-Authored-By: Claude Opus 4.7 --- .../compliance/profile/__init__.py | 38 ++++ .../compliance/profile/canonical.py | 158 +++++++++++++++ .../profile/from_company_profile.py | 59 ++++++ .../compliance/profile/from_product_wizard.py | 50 +++++ .../compliance/profile/to_gap.py | 41 ++++ .../compliance/profile/to_reasoning.py | 88 ++++++++ .../tests/test_profile_convergence.py | 188 ++++++++++++++++++ 7 files changed, 622 insertions(+) create mode 100644 backend-compliance/compliance/profile/__init__.py create mode 100644 backend-compliance/compliance/profile/canonical.py create mode 100644 backend-compliance/compliance/profile/from_company_profile.py create mode 100644 backend-compliance/compliance/profile/from_product_wizard.py create mode 100644 backend-compliance/compliance/profile/to_gap.py create mode 100644 backend-compliance/compliance/profile/to_reasoning.py create mode 100644 backend-compliance/tests/test_profile_convergence.py diff --git a/backend-compliance/compliance/profile/__init__.py b/backend-compliance/compliance/profile/__init__.py new file mode 100644 index 00000000..f78ea872 --- /dev/null +++ b/backend-compliance/compliance/profile/__init__.py @@ -0,0 +1,38 @@ +"""Product profile convergence layer. + +ONE canonical product profile (`CanonicalProductRegulatoryProfile`) that the Go +gap engine and the Python reasoning engine both project from — so "SPS mit +Remote Access" means the same thing everywhere. gap.ProductProfile leads; the +reasoning ProductProfile is an adapter/DTO. Types + mappers only — no regulation +logic, no UI, no new questions. +""" + +from __future__ import annotations + +from .canonical import ( + CanonicalLifecyclePhase, + CanonicalProductRegulatoryProfile, + CanonicalProductType, + ComponentKind, + EconomicOperatorRole, + EnvironmentalImpact, + ProductComponent, +) +from .from_company_profile import from_company_profile +from .from_product_wizard import from_product_wizard +from .to_gap import to_gap_profile +from .to_reasoning import to_reasoning_profile + +__all__ = [ + "CanonicalProductRegulatoryProfile", + "CanonicalProductType", + "EconomicOperatorRole", + "CanonicalLifecyclePhase", + "ComponentKind", + "ProductComponent", + "EnvironmentalImpact", + "from_product_wizard", + "from_company_profile", + "to_gap_profile", + "to_reasoning_profile", +] diff --git a/backend-compliance/compliance/profile/canonical.py b/backend-compliance/compliance/profile/canonical.py new file mode 100644 index 00000000..5614ccef --- /dev/null +++ b/backend-compliance/compliance/profile/canonical.py @@ -0,0 +1,158 @@ +"""CanonicalProductRegulatoryProfile — the single semantic product profile. + +Convergence layer (spec 2026-06-26): instead of letting the Go `gap.ProductProfile` +and the Python reasoning `ProductProfile` drift, ONE canonical type is the source +of truth. The Go gap engine LEADS (it carries real engine logic), so the canonical +mirrors gap's field names and adds the Navigator gaps the audit found missing +(economic-operator role, radio module, generates_usage_data, lifecycle phase, +structured BOM, safety-vs-security split, machine-vs-component) plus a +forward-looking Environmental-Impact domain. + +No regulation logic lives here — types only. Mappers live in sibling modules. +Python 3.9 compatible (no `|` unions). +""" + +from __future__ import annotations + +from enum import Enum +from typing import List, Optional + +from pydantic import BaseModel, Field + + +class CanonicalProductType(str, Enum): # mirrors gap.ProductType + SOFTWARE = "software" + HARDWARE = "hardware" + IOT = "iot" + SAAS = "saas" + EXCHANGE = "exchange" + MEDICAL_DEVICE = "medical_device" + MACHINERY = "machinery" + OTHER = "other" + + +class EconomicOperatorRole(str, Enum): # CE/CRA role — gap.ProductProfile has none + MANUFACTURER = "manufacturer" + IMPORTER = "importer" + DISTRIBUTOR = "distributor" + INTEGRATOR = "integrator" + OPERATOR = "operator" + SERVICE_PROVIDER = "service_provider" + + +class CanonicalLifecyclePhase(str, Enum): + DEVELOPMENT = "development" + PLACING_ON_MARKET = "placing_on_market" + OPERATION = "operation" + MAINTENANCE = "maintenance" + UPDATE = "update" + END_OF_LIFE = "end_of_life" + + +class ComponentKind(str, Enum): + MOTOR = "motor" + PUMP = "pump" + HEATING = "heating" + COOLING = "cooling" + CONTROLLER = "controller" + PLC = "plc" + HMI = "hmi" + SENSOR = "sensor" + ACTUATOR = "actuator" + CAMERA = "camera" + NETWORK_INTERFACE = "network_interface" + RADIO_MODULE = "radio_module" + CHEMICAL_DOSING = "chemical_dosing" + WATER_INLET = "water_inlet" + WASTEWATER_OUTLET = "wastewater_outlet" + BATTERY = "battery" + OTHER = "other" + + +class ProductComponent(BaseModel): + """One structured BOM node — these nodes are what later trigger domains.""" + + name: str + kind: ComponentKind = ComponentKind.OTHER + notes: Optional[str] = None + + +class EnvironmentalImpact(BaseModel): + """Forward-looking Umweltmedien-Trigger (own Navigator domain). + + No regulation logic consumes these yet — profile fields only, so the model + is not blind to wastewater/air/chemicals/waste questions when that domain + is wired later (AbwV/WRRL/REACH/CLP/IED/BImSchG ...). + """ + + discharges_to_wastewater: Optional[bool] = None + uses_cleaning_chemicals: Optional[bool] = None + supplies_chemicals: Optional[bool] = None + emits_to_air: Optional[bool] = None + uses_solvents: Optional[bool] = None + creates_waste: Optional[bool] = None + contains_restricted_substances: Optional[bool] = None + consumes_energy_or_water: Optional[bool] = None + has_cooling_or_spraying_water: Optional[bool] = None + + +class CanonicalProductRegulatoryProfile(BaseModel): + # --- identity --- + name: str = "" + description: str = "" + product_type: Optional[CanonicalProductType] = None + product_profile_id: Optional[str] = None + tenant_id: Optional[str] = None + iace_project_id: Optional[str] = None + + # --- gap-native lists --- + technologies: List[str] = Field(default_factory=list) + data_processing: List[str] = Field(default_factory=list) + markets: List[str] = Field(default_factory=list) # real list — never hardcoded ['EU'] + existing_certifications: List[str] = Field(default_factory=list) + applied_norms: List[str] = Field(default_factory=list) + + # --- gap-native product / IST-state booleans (tri-state: None = unknown) --- + connected_to_internet: Optional[bool] = None + has_software_updates: Optional[bool] = None + uses_ai: Optional[bool] = None + processes_personal_data: Optional[bool] = None + is_critical_infra_supplier: Optional[bool] = None + has_risk_assessment: Optional[bool] = None + has_technical_file: Optional[bool] = None + has_operating_manual: Optional[bool] = None + has_sbom: Optional[bool] = None + has_vuln_management: Optional[bool] = None + has_update_mechanism: Optional[bool] = None + has_incident_response: Optional[bool] = None + has_supply_chain_mgmt: Optional[bool] = None + ce_marking_since: Optional[str] = None + product_age: Optional[str] = None + + # --- NEW Navigator-gap fields (audit 2026-06-26) --- + economic_operator_role: Optional[EconomicOperatorRole] = None + has_radio_module: Optional[bool] = None + generates_usage_data: Optional[bool] = None + lifecycle_phase: Optional[CanonicalLifecyclePhase] = None + components: List[ProductComponent] = Field(default_factory=list) + has_safety_function: Optional[bool] = None + safety_function_description: Optional[str] = None + has_security_function: Optional[bool] = None # safety vs security split + has_remote_access: Optional[bool] = None + has_embedded_software: Optional[bool] = None + is_machine: Optional[bool] = None + is_component: Optional[bool] = None + is_spare_part: Optional[bool] = None + + # --- company / market context (NIS2 + scope; from company-profile) --- + b2b_or_b2c: Optional[str] = None + sector_industry: Optional[str] = None + company_size: Optional[str] = None + primary_jurisdiction: Optional[str] = None + + # --- AI context (classification stays delegated to ai-act/ucca) --- + ai_integration_type: List[str] = Field(default_factory=list) + human_oversight_level: Optional[str] = None + + # --- forward-looking environmental domain --- + environmental: EnvironmentalImpact = Field(default_factory=EnvironmentalImpact) diff --git a/backend-compliance/compliance/profile/from_company_profile.py b/backend-compliance/compliance/profile/from_company_profile.py new file mode 100644 index 00000000..06ac17c6 --- /dev/null +++ b/backend-compliance/compliance/profile/from_company_profile.py @@ -0,0 +1,59 @@ +"""company-profile -> CanonicalProductRegulatoryProfile (prefill, acceptance #2). + +Pulls master data (industry, business model, size, markets) and the conditional +`machine_builder` block (camelCase JSONB keys, defined frontend-side) so the user +re-answers nothing. The machineBuilder block is the richest product/safety/ +connectivity source — note it is industry-gated in the UI, so a prefill may find +it empty; that is fine (fields stay None = unknown). +""" + +from __future__ import annotations + +from typing import Any, Dict, List + +from .canonical import CanonicalProductRegulatoryProfile + +_EU_MEMBER_HINTS = {"DE", "AT", "FR", "IT", "NL", "LU", "LI", "EU", "EWR", "EEA", "DACH"} + + +def _markets(p: Dict[str, Any], mb: Dict[str, Any]) -> List[str]: + out: List[str] = [] + for source in (p.get("target_markets"), mb.get("exportMarkets"), [p.get("primary_jurisdiction")], [p.get("headquarters_country")]): + for m in source or []: + if m and m not in out: + out.append(m) + return out + + +def _is_machine(mb: Dict[str, Any]) -> Any: + types = mb.get("productTypes") + if types: + return True + return None + + +def from_company_profile(profile: Dict[str, Any]) -> CanonicalProductRegulatoryProfile: + p = profile + mb = p.get("machine_builder") or {} + contains_ai = mb.get("containsAI") + uses_ai = contains_ai if contains_ai is not None else p.get("uses_ai") + return CanonicalProductRegulatoryProfile( + description=mb.get("productDescription") or "", + sector_industry=p.get("industry") or None, + b2b_or_b2c=p.get("business_model") or None, + company_size=p.get("company_size") or None, + primary_jurisdiction=p.get("primary_jurisdiction") or None, + markets=_markets(p, mb), + uses_ai=uses_ai, + ai_integration_type=list(mb.get("aiIntegrationType") or []), + human_oversight_level=mb.get("humanOversightLevel") or None, + has_embedded_software=mb.get("containsFirmware"), + has_safety_function=mb.get("hasSafetyFunction"), + safety_function_description=mb.get("safetyFunctionDescription") or None, + has_remote_access=mb.get("hasRemoteAccess"), + connected_to_internet=mb.get("isNetworked"), + has_software_updates=mb.get("hasOTAUpdates"), + has_risk_assessment=mb.get("hasRiskAssessment"), + is_machine=_is_machine(mb), + is_critical_infra_supplier=mb.get("criticalSectorClients"), + ) diff --git a/backend-compliance/compliance/profile/from_product_wizard.py b/backend-compliance/compliance/profile/from_product_wizard.py new file mode 100644 index 00000000..3236fa20 --- /dev/null +++ b/backend-compliance/compliance/profile/from_product_wizard.py @@ -0,0 +1,50 @@ +"""ProductWizard payload -> CanonicalProductRegulatoryProfile (lossless). + +The gap-analysis ProductWizard POSTs exactly the gap.ProductProfile JSON shape +(see admin-compliance/.../ProductWizard.tsx handleSubmit). This mapper copies +every gap field verbatim so that `to_gap_profile(from_product_wizard(p))` +reproduces the gap subset of `p` byte-for-byte (acceptance #1). New Navigator +fields the wizard does not ask stay None. +""" + +from __future__ import annotations + +from typing import Any, Dict, Optional + +from .canonical import CanonicalProductRegulatoryProfile, CanonicalProductType + + +def _as_product_type(value: Any) -> Optional[CanonicalProductType]: + try: + return CanonicalProductType(value) + except ValueError: + return None + + +def from_product_wizard(payload: Dict[str, Any]) -> CanonicalProductRegulatoryProfile: + g = payload.get + return CanonicalProductRegulatoryProfile( + name=g("name", ""), + description=g("description", ""), + product_type=_as_product_type(g("product_type")), + technologies=list(g("technologies") or []), + data_processing=list(g("data_processing") or []), + markets=list(g("markets") or []), + existing_certifications=list(g("existing_certifications") or []), + applied_norms=list(g("applied_norms") or []), + connected_to_internet=g("connected_to_internet"), + has_software_updates=g("has_software_updates"), + uses_ai=g("uses_ai"), + processes_personal_data=g("processes_personal_data"), + is_critical_infra_supplier=g("is_critical_infra_supplier"), + has_risk_assessment=g("has_risk_assessment"), + has_technical_file=g("has_technical_file"), + has_operating_manual=g("has_operating_manual"), + has_sbom=g("has_sbom"), + has_vuln_management=g("has_vuln_management"), + has_update_mechanism=g("has_update_mechanism"), + has_incident_response=g("has_incident_response"), + has_supply_chain_mgmt=g("has_supply_chain_mgmt"), + ce_marking_since=g("ce_marking_since"), + product_age=g("product_age"), + ) diff --git a/backend-compliance/compliance/profile/to_gap.py b/backend-compliance/compliance/profile/to_gap.py new file mode 100644 index 00000000..d620be40 --- /dev/null +++ b/backend-compliance/compliance/profile/to_gap.py @@ -0,0 +1,41 @@ +"""CanonicalProductRegulatoryProfile -> gap.ProductProfile JSON shape. + +Emits exactly the keys the Go gap engine already consumes (gap/models.go json +tags), so the gap engine runs UNCHANGED — the canonical is a superset and gap is +its lossless projection. Canonical-only fields (role/radio/components/...) are +intentionally not emitted here; they reach the reasoning side via to_reasoning. +""" + +from __future__ import annotations + +from typing import Any, Dict + +from .canonical import CanonicalProductRegulatoryProfile + + +def to_gap_profile(c: CanonicalProductRegulatoryProfile) -> Dict[str, Any]: + return { + "name": c.name, + "description": c.description, + "product_type": c.product_type.value if c.product_type else "", + "technologies": list(c.technologies), + "data_processing": list(c.data_processing), + "markets": list(c.markets), + "existing_certifications": list(c.existing_certifications), + "applied_norms": list(c.applied_norms), + "connected_to_internet": bool(c.connected_to_internet), + "has_software_updates": bool(c.has_software_updates), + "uses_ai": bool(c.uses_ai), + "processes_personal_data": bool(c.processes_personal_data), + "is_critical_infra_supplier": bool(c.is_critical_infra_supplier), + "has_risk_assessment": bool(c.has_risk_assessment), + "has_technical_file": bool(c.has_technical_file), + "has_operating_manual": bool(c.has_operating_manual), + "has_sbom": bool(c.has_sbom), + "has_vuln_management": bool(c.has_vuln_management), + "has_update_mechanism": bool(c.has_update_mechanism), + "has_incident_response": bool(c.has_incident_response), + "has_supply_chain_mgmt": bool(c.has_supply_chain_mgmt), + "ce_marking_since": c.ce_marking_since if c.ce_marking_since is not None else "", + "product_age": c.product_age if c.product_age is not None else "", + } diff --git a/backend-compliance/compliance/profile/to_reasoning.py b/backend-compliance/compliance/profile/to_reasoning.py new file mode 100644 index 00000000..c588f3ca --- /dev/null +++ b/backend-compliance/compliance/profile/to_reasoning.py @@ -0,0 +1,88 @@ +"""CanonicalProductRegulatoryProfile -> reasoning ProductProfile (adapter/DTO). + +The reasoning engine stays the consumer, never the source of truth (spec): the +canonical leads, this projects it into the Python reasoning ProductProfile so the +Reasoning engine and the Go gap engine run off ONE semantic profile (acceptance +#10). AI classification is NOT done here — only `uses_ai` is forwarded; risk +classification stays delegated to ai-act/ucca (acceptance #3). + +This is the ONLY one-way coupling profile -> reasoning; reasoning never imports +profile, so the reasoning layer stays hermetic. +""" + +from __future__ import annotations + +from typing import List, Optional + +from compliance.reasoning.enums import ManufacturerRole, MarketModel, ProductLifecyclePhase +from compliance.reasoning.schemas import ProductProfile + +from .canonical import CanonicalProductRegulatoryProfile, CanonicalProductType + +_SOFTWARE_TYPES = {CanonicalProductType.SOFTWARE, CanonicalProductType.SAAS, CanonicalProductType.IOT} +_SOFTWARE_TECH = {"ai", "api", "database", "encryption", "ota_updates", "cloud", "blockchain"} +_EU_HINTS = {"DE", "AT", "FR", "IT", "NL", "LU", "LI", "EU", "EWR", "EEA", "DACH"} +_B2X = {"B2B": MarketModel.B2B, "B2C": MarketModel.B2C, "B2B_B2C": MarketModel.BOTH, "B2B2C": MarketModel.BOTH} + + +def _or_none(*values: Optional[bool]) -> Optional[bool]: + """True if any value is truthy; None if all are None/absent; else False.""" + if any(v is True for v in values): + return True + if all(v is None for v in values): + return None + return False + + +def _has_software(c: CanonicalProductRegulatoryProfile) -> Optional[bool]: + type_sig = True if c.product_type in _SOFTWARE_TYPES else None + tech_sig = True if (set(c.technologies) & _SOFTWARE_TECH) else None + return _or_none(c.has_embedded_software, c.has_software_updates, c.uses_ai, type_sig, tech_sig) + + +def _eu_market(markets: List[str]) -> Optional[bool]: + if not markets: + return None + return True if (set(markets) & _EU_HINTS) else False + + +def _has_radio(c: CanonicalProductRegulatoryProfile) -> Optional[bool]: + if c.has_radio_module is not None: + return c.has_radio_module + if any(comp.kind.value == "radio_module" for comp in c.components): + return True + return None + + +def to_reasoning_profile(c: CanonicalProductRegulatoryProfile) -> ProductProfile: + role = ManufacturerRole(c.economic_operator_role.value) if c.economic_operator_role else None + phase = ProductLifecyclePhase(c.lifecycle_phase.value) if c.lifecycle_phase else None + b2x = _B2X.get(c.b2b_or_b2c) if c.b2b_or_b2c else None + is_machine = c.is_machine if c.is_machine is not None else ( + True if c.product_type == CanonicalProductType.MACHINERY else None + ) + generates_data = c.generates_usage_data if c.generates_usage_data is not None else ( + True if "telemetry" in c.data_processing else None + ) + return ProductProfile( + product_name=c.name or "Produkt", + product_profile_id=c.product_profile_id, + manufacturer_role=role, + product_type=[c.product_type.value] if c.product_type else [], + has_software=_has_software(c), + has_embedded_software=c.has_embedded_software, + has_remote_access=c.has_remote_access, + has_cloud_connection=True if "cloud" in c.technologies else None, + has_ai_functionality=c.uses_ai, + has_radio_module=_has_radio(c), + has_safety_function=c.has_safety_function, + generates_usage_data=generates_data, + is_machine=is_machine, + is_component=c.is_component, + is_spare_part=c.is_spare_part, + eu_market=_eu_market(c.markets), + b2b_or_b2c=b2x, + lifecycle_phase=phase, + company_size=c.company_size, + sector=c.sector_industry, + ) diff --git a/backend-compliance/tests/test_profile_convergence.py b/backend-compliance/tests/test_profile_convergence.py new file mode 100644 index 00000000..e1ac5f9c --- /dev/null +++ b/backend-compliance/tests/test_profile_convergence.py @@ -0,0 +1,188 @@ +"""Tests for the Product Profile convergence layer. + +Covers the 10 acceptance criteria of the CanonicalProductRegulatoryProfile spec: +lossless ProductWizard mapping, company-profile prefill, AI stays delegated, +markets no longer hardcoded, and the new Navigator fields (role/radio/usage-data/ +lifecycle/BOM) plus one-semantic-profile across reasoning + gap. +""" + +from __future__ import annotations + +from compliance.profile import ( + CanonicalLifecyclePhase, + CanonicalProductRegulatoryProfile, + CanonicalProductType, + ComponentKind, + EconomicOperatorRole, + ProductComponent, + from_company_profile, + from_product_wizard, + to_gap_profile, + to_reasoning_profile, +) +from compliance.reasoning import discover_scope +from compliance.reasoning.enums import ManufacturerRole, ProductLifecyclePhase + +# A realistic ProductWizard payload — exactly the gap.ProductProfile JSON shape. +WIZARD = { + "name": "Industriespülmaschine", + "description": "vernetzte Spülmaschine", + "product_type": "machinery", + "technologies": ["cloud", "ota_updates", "sensor", "actuator"], + "data_processing": ["telemetry"], + "markets": ["EU"], + "connected_to_internet": True, + "has_software_updates": True, + "uses_ai": False, + "processes_personal_data": False, + "is_critical_infra_supplier": False, + "existing_certifications": ["CE"], + "applied_norms": ["ISO12100"], + "has_risk_assessment": True, + "has_technical_file": True, + "has_operating_manual": True, + "has_sbom": False, + "has_vuln_management": False, + "has_update_mechanism": True, + "has_incident_response": False, + "has_supply_chain_mgmt": False, + "ce_marking_since": "", + "product_age": "5", +} + +COMPANY = { + "company_name": "ACME Maschinen GmbH", + "industry": "Maschinenbau", + "business_model": "B2B", + "company_size": "medium", + "target_markets": ["DE", "EU"], + "primary_jurisdiction": "DE", + "headquarters_country": "DE", + "uses_ai": False, + "is_data_controller": True, + "machine_builder": { + "productDescription": "Industriespülmaschine", + "productTypes": ["special_machine"], + "containsSoftware": True, + "containsFirmware": True, + "containsAI": False, + "hasSafetyFunction": True, + "safetyFunctionDescription": "Türverriegelung", + "isNetworked": True, + "hasRemoteAccess": True, + "hasOTAUpdates": True, + "hasRiskAssessment": True, + "criticalSectorClients": False, + }, +} + + +# 1. ProductWizard data maps losslessly into the canonical and back to gap shape. +def test_product_wizard_lossless_roundtrip(): + canonical = from_product_wizard(WIZARD) + assert to_gap_profile(canonical) == WIZARD + + +# 2. company-profile can prefill the canonical profile. +def test_company_profile_prefill(): + c = from_company_profile(COMPANY) + assert c.sector_industry == "Maschinenbau" + assert c.b2b_or_b2c == "B2B" + assert c.company_size == "medium" + assert "DE" in c.markets and "EU" in c.markets + assert c.has_safety_function is True + assert c.has_remote_access is True + assert c.has_embedded_software is True + assert c.is_machine is True + assert c.description == "Industriespülmaschine" + + +# 3. AI-Act/ucca stays delegated — only uses_ai is forwarded, no risk classification. +def test_ai_classification_stays_delegated(): + c = CanonicalProductRegulatoryProfile(name="X", uses_ai=True) + rp = to_reasoning_profile(c) + assert rp.has_ai_functionality is True + assert not hasattr(rp, "ai_risk_category") # no AI classification produced here + + +# 4. markets are a real list, never hardcoded ['EU']. +def test_markets_not_hardcoded_eu(): + assert CanonicalProductRegulatoryProfile(name="X").markets == [] + c = from_product_wizard({**WIZARD, "markets": ["US", "JP", "CA"]}) + assert c.markets == ["US", "JP", "CA"] + assert to_gap_profile(c)["markets"] == ["US", "JP", "CA"] + assert to_reasoning_profile(c).eu_market is False # non-EU markets -> not EU + + +# 5. economic-operator role exists and maps to the reasoning role. +def test_economic_operator_role_exists(): + c = CanonicalProductRegulatoryProfile(name="X", economic_operator_role=EconomicOperatorRole.IMPORTER) + assert to_reasoning_profile(c).manufacturer_role == ManufacturerRole.IMPORTER + + +# 6. radio_module exists (direct + inferred from a BOM component). +def test_radio_module_exists(): + assert to_reasoning_profile(CanonicalProductRegulatoryProfile(name="X", has_radio_module=True)).has_radio_module is True + c = CanonicalProductRegulatoryProfile(name="X", components=[ProductComponent(name="WLAN", kind=ComponentKind.RADIO_MODULE)]) + assert to_reasoning_profile(c).has_radio_module is True + + +# 7. generates_usage_data exists (direct + inferred from telemetry). +def test_generates_usage_data_exists(): + c = CanonicalProductRegulatoryProfile(name="X", generates_usage_data=True) + assert to_reasoning_profile(c).generates_usage_data is True + inferred = from_product_wizard(WIZARD) # data_processing has telemetry + assert to_reasoning_profile(inferred).generates_usage_data is True + + +# 8. lifecycle_phase exists and maps. +def test_lifecycle_phase_exists(): + c = CanonicalProductRegulatoryProfile(name="X", lifecycle_phase=CanonicalLifecyclePhase.MAINTENANCE) + assert to_reasoning_profile(c).lifecycle_phase == ProductLifecyclePhase.MAINTENANCE + + +# 9. BOM components are structured. +def test_bom_components_structured(): + c = CanonicalProductRegulatoryProfile( + name="Spülmaschine", + components=[ + ProductComponent(name="Umwälzpumpe", kind=ComponentKind.PUMP), + ProductComponent(name="Heizung", kind=ComponentKind.HEATING), + ProductComponent(name="SPS", kind=ComponentKind.PLC), + ProductComponent(name="Abwasserablauf", kind=ComponentKind.WASTEWATER_OUTLET), + ], + ) + kinds = {comp.kind for comp in c.components} + assert ComponentKind.PLC in kinds and ComponentKind.WASTEWATER_OUTLET in kinds + + +# 10. reasoning engine + gap engine run off ONE semantic profile. +def test_one_semantic_profile_reasoning_and_gap(): + canonical = CanonicalProductRegulatoryProfile( + name="Industriespülmaschine", + product_type=CanonicalProductType.MACHINERY, + economic_operator_role=EconomicOperatorRole.MANUFACTURER, + markets=["EU", "DE"], + is_machine=True, + has_safety_function=True, + has_remote_access=True, + has_software_updates=True, + has_embedded_software=True, + technologies=["cloud", "ota_updates"], + ) + gap = to_gap_profile(canonical) + rp = to_reasoning_profile(canonical) + + # same facts, two projections + assert gap["markets"] == ["EU", "DE"] + assert rp.eu_market is True + assert rp.has_remote_access is True + assert rp.has_cloud_connection is True + assert rp.is_machine is True + assert rp.manufacturer_role == ManufacturerRole.MANUFACTURER + + # the projected reasoning profile actually drives the reasoning engine + scope = discover_scope(rp) + applicable = {r.regulation_id for r in scope.applicable_regulations} + assert "CRA" in applicable + assert "MaschinenVO" in applicable