93 lines
3.8 KiB
Python
93 lines
3.8 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
from idea180_credimesh_federated_privacy import (
|
|
CrediMeshOrchestrator,
|
|
GraphOfContractsRegistry,
|
|
Identity,
|
|
LocalUnderwritingProblem,
|
|
SQLiteGovernanceLedger,
|
|
create_identity,
|
|
)
|
|
from idea180_credimesh_federated_privacy.adapters import IncomeVerificationAdapter, PropertyAppraisalAdapter
|
|
from idea180_credimesh_federated_privacy.contracts import ConformanceHarness
|
|
from idea180_credimesh_federated_privacy.models import PrivacyBudget
|
|
|
|
|
|
def test_contract_registry_tracks_versions_and_digest() -> None:
|
|
registry = GraphOfContractsRegistry()
|
|
income = IncomeVerificationAdapter().contract
|
|
appraisal = PropertyAppraisalAdapter().contract
|
|
registry.register(income)
|
|
registry.register(appraisal)
|
|
registry.connect(income.key, appraisal.key)
|
|
|
|
assert registry.list_versions("income_verification") == ["1.0"]
|
|
assert registry.digest()
|
|
|
|
|
|
def test_adapters_emit_privacy_minimized_signals() -> None:
|
|
adapter = IncomeVerificationAdapter()
|
|
budget = PrivacyBudget(max_bytes=4096, max_fields=32)
|
|
signal = adapter.run(
|
|
"borrower-1",
|
|
[
|
|
{"gross_monthly_income": 8500, "employment_months": 36, "employer_stability_score": 8.5},
|
|
{"gross_monthly_income": 9100, "employment_months": 42, "employer_stability_score": 9.0},
|
|
{"gross_monthly_income": 8900, "employment_months": 40, "employer_stability_score": 8.8},
|
|
],
|
|
budget,
|
|
)
|
|
|
|
assert set(signal.payload) == {"verified_monthly_income", "employment_stability", "income_confidence"}
|
|
assert signal.provenance_hash
|
|
|
|
|
|
def test_orchestrator_produces_deterministic_plan_and_audit_log(tmp_path: Path) -> None:
|
|
ledger_path = tmp_path / "ledger.sqlite3"
|
|
identity = create_identity("lender-a")
|
|
problem = LocalUnderwritingProblem(
|
|
borrower_id="borrower-99",
|
|
lender_id="lender-a",
|
|
requested_amount=420000,
|
|
property_value=540000,
|
|
annual_income=186000,
|
|
monthly_obligations=750,
|
|
)
|
|
|
|
with SQLiteGovernanceLedger(ledger_path) as ledger:
|
|
orchestrator = CrediMeshOrchestrator(identity=identity, ledger=ledger, harness=ConformanceHarness())
|
|
result_one = orchestrator.evaluate(
|
|
problem,
|
|
[
|
|
{"gross_monthly_income": 15400, "employment_months": 48, "employer_stability_score": 9.0},
|
|
{"gross_monthly_income": 15600, "employment_months": 51, "employer_stability_score": 8.9},
|
|
{"gross_monthly_income": 15500, "employment_months": 50, "employer_stability_score": 9.1},
|
|
],
|
|
{"appraised_value": 560000, "confidence": 0.88, "comparables_count": 7, "days_on_market": 22},
|
|
)
|
|
result_two = orchestrator.evaluate(
|
|
problem,
|
|
[
|
|
{"gross_monthly_income": 15400, "employment_months": 48, "employer_stability_score": 9.0},
|
|
{"gross_monthly_income": 15600, "employment_months": 51, "employer_stability_score": 8.9},
|
|
{"gross_monthly_income": 15500, "employment_months": 50, "employer_stability_score": 9.1},
|
|
],
|
|
{"appraised_value": 560000, "confidence": 0.88, "comparables_count": 7, "days_on_market": 22},
|
|
)
|
|
|
|
assert result_one.plan.loan_amount == result_two.plan.loan_amount
|
|
assert result_one.replay_token == result_two.replay_token
|
|
audit_entries = ledger.list_audit_entries()
|
|
assert len(audit_entries) == 2
|
|
assert audit_entries[0].signature
|
|
identity.verify_audit_entry(audit_entries[0])
|
|
|
|
|
|
def test_ledger_persists_contracts_and_entries(tmp_path: Path) -> None:
|
|
ledger_path = tmp_path / "ledger.sqlite3"
|
|
with SQLiteGovernanceLedger(ledger_path) as ledger:
|
|
ledger.store_contract("example@1.0", {"foo": "bar"})
|
|
assert ledger.list_contracts()["example@1.0"]["foo"] == "bar"
|