52 lines
1.6 KiB
Python
52 lines
1.6 KiB
Python
import json
|
|
import sys
|
|
import os
|
|
|
|
# Ensure the src/ directory is on PYTHONPATH for tests when running from repo root
|
|
repo_root = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
|
|
src_dir = os.path.join(repo_root, "src")
|
|
if src_dir not in sys.path:
|
|
sys.path.insert(0, src_dir)
|
|
|
|
from exprove import LocalExecutionTask, SharedMarketContext, PlanDelta
|
|
from exprove.engine import compute_plan_delta
|
|
|
|
|
|
def test_local_execution_task_serialization():
|
|
t = LocalExecutionTask(
|
|
task_id="t1",
|
|
instrument="GOOG",
|
|
venue="VenueX",
|
|
objective="VWAP",
|
|
constraints={"latency": 10},
|
|
)
|
|
s = t.to_json()
|
|
assert isinstance(s, str)
|
|
d = json.loads(s)
|
|
assert d["task_id"] == "t1"
|
|
assert d["instrument"] == "GOOG"
|
|
|
|
|
|
def test_engine_deterministic_delta():
|
|
task = LocalExecutionTask(
|
|
task_id="t2",
|
|
instrument="MSFT",
|
|
venue="VenueY",
|
|
objective="VWAP",
|
|
constraints={"latency": 5},
|
|
)
|
|
ctx = SharedMarketContext(signals={"depth": 10}, version=2, contract_id="c-2")
|
|
delta1 = compute_plan_delta(task, ctx, author="engine")
|
|
delta2 = compute_plan_delta(task, ctx, author="engine")
|
|
assert delta1.delta == delta2.delta
|
|
assert delta1.timestamp <= delta2.timestamp # non-decreasing timestamp on re-run is fine
|
|
|
|
|
|
def test_attestation_and_auditlog_serialization():
|
|
from exprove.contracts import Attestation, AuditLog
|
|
a = Attestation(entry="delta committed", signer="issuer", timestamp=1.0, contract_id="c-3", version=1)
|
|
log = AuditLog()
|
|
log.add(a)
|
|
j = log.to_json()
|
|
assert isinstance(j, str)
|