79 lines
2.9 KiB
Python
79 lines
2.9 KiB
Python
import sys
|
|
import os
|
|
import pytest
|
|
|
|
# Ensure the src layout is on PYTHONPATH for tests without installing the package
|
|
ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
|
|
SRC = os.path.join(ROOT, "src")
|
|
if SRC not in sys.path:
|
|
sys.path.insert(0, SRC)
|
|
|
|
from idea34_openpassmarket_privacy_preserving.core import LocalProblem, PerformanceSignal, PrivacyBudget, aggregate_signals
|
|
from idea34_openpassmarket_privacy_preserving import PassSpec, verify_invariants
|
|
from idea34_openpassmarket_privacy_preserving.registry import RegistryWrapper
|
|
|
|
|
|
def test_privacy_budget_consume():
|
|
budget = PrivacyBudget(budget=5.0, leakage_model="Laplace")
|
|
assert budget.consume(2.0) is True
|
|
assert budget.used == 2.0
|
|
# Exceeding budget should fail gracefully
|
|
assert budget.consume(4.0) is False
|
|
assert budget.used == 2.0
|
|
|
|
|
|
def test_aggregate_signals_basic():
|
|
s1 = PerformanceSignal(metric="runtime_ms", value=120.0, unit="ms", privacy_tag="v1", version=1)
|
|
s2 = PerformanceSignal(metric="runtime_ms", value=110.0, unit="ms", privacy_tag="v1", version=1)
|
|
budget = PrivacyBudget(budget=10.0, leakage_model="Laplace")
|
|
merged = aggregate_signals([s1, s2], budget)
|
|
assert merged.metric == "aggregate_runtime_ms" or merged.metric == "aggregate_runtime"
|
|
assert merged.value == pytest.approx((120.0 + 110.0) / 2.0)
|
|
|
|
|
|
def test_verify_invariants_valid_and_invalid():
|
|
lp = LocalProblem(
|
|
problem_id="p1",
|
|
code_region="def f(): pass",
|
|
inlining_decisions={},
|
|
loop_tiling_params={},
|
|
vectorization_hints={},
|
|
constraints={},
|
|
)
|
|
spec = PassSpec(pass_id="pass_01", language="Python", target_IR="IR1", transform_type="mutate")
|
|
budget = PrivacyBudget(budget=5.0, leakage_model="Laplace")
|
|
# Valid case
|
|
assert verify_invariants(lp, spec, budget) is True
|
|
# Invalid: empty code region should fail
|
|
lp_bad = LocalProblem(
|
|
problem_id="p2",
|
|
code_region="",
|
|
inlining_decisions={},
|
|
loop_tiling_params={},
|
|
vectorization_hints={},
|
|
constraints={},
|
|
)
|
|
assert verify_invariants(lp_bad, spec, budget) is False
|
|
|
|
|
|
def test_registry_register_and_conformance():
|
|
reg = RegistryWrapper()
|
|
adapter_id = "adapter_foo"
|
|
info = {
|
|
"supported_languages": "Python, C",
|
|
"contract_version": "v1",
|
|
"supported_features": "pass_spec,performance_signal",
|
|
}
|
|
# Register and retrieve
|
|
reg.register(adapter_id, info)
|
|
got = reg.get(adapter_id)
|
|
assert got is not None
|
|
assert got.get("contract_version") == "v1"
|
|
|
|
# Conformance checks
|
|
assert reg.conforms(adapter_id, required_keys=["supported_languages", "contract_version"]) is True
|
|
# Missing key should fail
|
|
assert reg.conforms(adapter_id, required_keys=["nonexistent_key"]) is False
|
|
# Non-registered adapter should return False
|
|
assert reg.conforms("no_such_adapter") is False
|