c3451560f5
- set_anchor self-guards the single-active-anchor invariant (archive prior active on re-inception) - append_record/record_series validate + normalize ISO-8601 dates (fail loud on malformed) - ordering test now inserts out-of-order; removed dead test imports/vars; clean Session import Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
59 lines
2.5 KiB
Python
59 lines
2.5 KiB
Python
import datetime as dt
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from fxhnt.adapters.persistence.forward_nav import ForwardNavRepo
|
|
|
|
|
|
def _repo(tmp_path):
|
|
repo = ForwardNavRepo(f"sqlite:///{tmp_path}/anchor.db")
|
|
repo.migrate()
|
|
return repo
|
|
|
|
|
|
def test_set_and_read_active_anchor(tmp_path):
|
|
repo = _repo(tmp_path)
|
|
at = dt.datetime(2026, 7, 8, 23, 30)
|
|
repo.set_anchor("multistrat", t0="2026-07-08", definition_hash="h1",
|
|
definition_version=1, params_json={"instruments": ["SPY"]},
|
|
record_mode="recompute", at=at)
|
|
a = repo.active_anchor("multistrat")
|
|
assert a is not None
|
|
assert (a.t0, a.definition_hash, a.record_mode, a.archived_at) == ("2026-07-08", "h1", "recompute", None)
|
|
|
|
|
|
def test_archive_then_reset_keeps_lineage_and_flips_active(tmp_path):
|
|
repo = _repo(tmp_path)
|
|
t1 = dt.datetime(2026, 7, 1, 23, 30)
|
|
t2 = dt.datetime(2026, 7, 8, 23, 30)
|
|
repo.set_anchor("multistrat", "2026-07-01", "h1", 1, {"n": 6}, "recompute", at=t1)
|
|
# re-inception: archive the old anchor, set a new one
|
|
repo.archive_anchor("multistrat", at=t2)
|
|
repo.set_anchor("multistrat", "2026-07-08", "h2", 1, {"n": 5}, "recompute", at=t2)
|
|
|
|
a = repo.active_anchor("multistrat")
|
|
assert a.definition_hash == "h2" and a.t0 == "2026-07-08" and a.archived_at is None
|
|
# lineage preserved: the old row still exists, archived
|
|
with Session(repo._engine) as s:
|
|
from fxhnt.adapters.persistence.cockpit_models import StrategyAnchorRow
|
|
rows = s.query(StrategyAnchorRow).filter_by(strategy_id="multistrat").all()
|
|
assert len(rows) == 2
|
|
assert sum(1 for r in rows if r.archived_at is None) == 1
|
|
|
|
|
|
def test_active_anchor_absent_returns_none(tmp_path):
|
|
assert _repo(tmp_path).active_anchor("nope") is None
|
|
|
|
|
|
def test_set_anchor_self_guards_single_active(tmp_path):
|
|
repo = _repo(tmp_path)
|
|
repo.set_anchor("m", "2026-07-01", "h1", 1, {"n": 6}, "recompute", at=dt.datetime(2026, 7, 1))
|
|
# re-inception WITHOUT an explicit archive_anchor call — set_anchor must archive the old active row
|
|
repo.set_anchor("m", "2026-07-08", "h2", 1, {"n": 5}, "recompute", at=dt.datetime(2026, 7, 8))
|
|
a = repo.active_anchor("m")
|
|
assert a.definition_hash == "h2" and a.t0 == "2026-07-08"
|
|
from fxhnt.adapters.persistence.cockpit_models import StrategyAnchorRow
|
|
with Session(repo._engine) as s:
|
|
rows = s.query(StrategyAnchorRow).filter_by(strategy_id="m").all()
|
|
assert len(rows) == 2 and sum(1 for r in rows if r.archived_at is None) == 1
|