46 lines
1.4 KiB
Python
46 lines
1.4 KiB
Python
from mercurymesh.schema import get_ir_schemas, sample_payloads
|
|
from mercurymesh.bridge_ir import to_ir_market_signal, to_ir_aggregated_signal
|
|
from mercurymesh.adapters.venue_a import VenueAAdapter
|
|
from mercurymesh.adapters.venue_b import VenueBAdapter
|
|
|
|
|
|
def test_ir_schemas_present():
|
|
schemas = get_ir_schemas()
|
|
# Basic smoke checks for a few required IR models
|
|
assert "LocalMarketContext" in schemas
|
|
assert "AggregatedSignalIR" in schemas
|
|
assert "PlanDeltaIR" in schemas
|
|
|
|
|
|
def test_sample_payloads_shape():
|
|
samples = sample_payloads()
|
|
assert "LocalMarketContext" in samples
|
|
assert "AggregatedSignalIR" in samples
|
|
|
|
|
|
def test_adapter_to_ir_translation():
|
|
a = VenueAAdapter()
|
|
b = VenueBAdapter()
|
|
msa = a.extract_signal()
|
|
msb = b.extract_signal()
|
|
|
|
ira = to_ir_market_signal(msa)
|
|
irb = to_ir_market_signal(msb)
|
|
# Ensure canonical wrapper keys exist
|
|
assert "local_context" in ira
|
|
assert "local_context" in irb
|
|
|
|
# Build a simple AggregatedSignal-like dict for translation function
|
|
agg = {
|
|
"venues": [msa.venue_id, msb.venue_id],
|
|
"feature_vector": {**msa.features, **msb.features},
|
|
"privacy_budget_used": 0.05,
|
|
"nonce": "n-1",
|
|
"merkle_proof": None,
|
|
}
|
|
|
|
# Convert using the helper; ensure keys are present
|
|
out = to_ir_aggregated_signal(type("Agg", (), agg)())
|
|
assert "aggregated" in out
|
|
assert "feature_vector" in out["aggregated"]
|