386 lines
14 KiB
Python
386 lines
14 KiB
Python
"""Lightweight integration tests for cadCAD state management (src/cadcad/).
|
|
|
|
These tests verify that the cadCAD state infrastructure (create_initial_state,
|
|
bootstrap_system, sync_metrics, extract_metrics) works correctly without
|
|
running full multi-step simulations.
|
|
|
|
Note: src/cadcad/config.py imports pandas (not installed in test env), so
|
|
bootstrap_system logic is replicated inline here using only the underlying
|
|
primitives, which are the same code paths config.py calls.
|
|
"""
|
|
|
|
import pytest
|
|
import numpy as np
|
|
|
|
from src.cadcad.state import (
|
|
MycoFiState,
|
|
create_initial_state,
|
|
extract_metrics,
|
|
sync_metrics,
|
|
)
|
|
from src.primitives.risk_tranching import (
|
|
TrancheParams, deposit_collateral, mint_tranche,
|
|
)
|
|
from src.primitives.conviction import (
|
|
ConvictionParams, Proposal, Voter, stake as cv_stake,
|
|
)
|
|
from src.crosschain.hub_spoke import simulate_deposit
|
|
|
|
|
|
# ---------- Inline bootstrap helper (mirrors config.bootstrap_system) ----------
|
|
|
|
def bootstrap_state(
|
|
state: dict,
|
|
initial_deposits: dict | None = None,
|
|
initial_tranche_mints: dict | None = None,
|
|
n_voters: int = 5,
|
|
) -> dict:
|
|
"""Replicate bootstrap_system logic without importing config.py (pandas)."""
|
|
s: MycoFiState = state["mycofi"]
|
|
|
|
if initial_deposits and s.crosschain:
|
|
total_usd = 0.0
|
|
for chain, assets in initial_deposits.items():
|
|
for asset_sym, qty in assets.items():
|
|
simulate_deposit(s.crosschain, chain, asset_sym, qty, 0.0)
|
|
spoke = s.crosschain.hub.spokes[chain]
|
|
for a in spoke.accepted_assets:
|
|
if a.symbol == asset_sym:
|
|
total_usd += qty * a.price
|
|
break
|
|
s.crosschain.hub.process_messages(0.0)
|
|
if s.tranche_system:
|
|
deposit_collateral(s.tranche_system, total_usd)
|
|
|
|
if initial_tranche_mints and s.tranche_system:
|
|
for tranche, amount in initial_tranche_mints.items():
|
|
mint_tranche(s.tranche_system, tranche, amount)
|
|
|
|
if s.myco_system and s.crosschain:
|
|
total_value = s.crosschain.hub.total_collateral_usd
|
|
if total_value > 0:
|
|
n = s.myco_system.config.n_reserve_assets
|
|
amounts = np.full(n, total_value / n)
|
|
s.myco_system.deposit(amounts, 0.0)
|
|
|
|
if s.governance:
|
|
for i in range(n_voters):
|
|
voter_id = f"voter_{i}"
|
|
holdings = float(np.random.lognormal(mean=np.log(5000), sigma=1.0))
|
|
s.governance.voters[voter_id] = Voter(
|
|
id=voter_id,
|
|
holdings=holdings,
|
|
sentiment=float(np.random.uniform(0.3, 0.9)),
|
|
)
|
|
s.governance.total_supply = sum(
|
|
v.holdings for v in s.governance.voters.values()
|
|
)
|
|
|
|
sync_metrics(s)
|
|
return state
|
|
|
|
|
|
# ---------- create_initial_state ----------
|
|
|
|
class TestCreateInitialState:
|
|
def test_returns_dict_with_mycofi_key(self):
|
|
state = create_initial_state()
|
|
assert "mycofi" in state
|
|
|
|
def test_mycofi_is_mycofi_state(self):
|
|
state = create_initial_state()
|
|
assert isinstance(state["mycofi"], MycoFiState)
|
|
|
|
def test_myco_system_is_populated(self):
|
|
state = create_initial_state()
|
|
assert state["mycofi"].myco_system is not None
|
|
|
|
def test_tranche_system_is_populated(self):
|
|
state = create_initial_state()
|
|
assert state["mycofi"].tranche_system is not None
|
|
|
|
def test_crosschain_is_populated(self):
|
|
state = create_initial_state()
|
|
assert state["mycofi"].crosschain is not None
|
|
|
|
def test_governance_is_populated(self):
|
|
state = create_initial_state()
|
|
assert state["mycofi"].governance is not None
|
|
|
|
def test_crosschain_has_five_chains(self):
|
|
state = create_initial_state()
|
|
s = state["mycofi"]
|
|
assert len(s.crosschain.hub.spokes) == 5
|
|
|
|
def test_total_chains_reflects_crosschain(self):
|
|
state = create_initial_state()
|
|
s = state["mycofi"]
|
|
assert s.total_chains == len(s.crosschain.hub.spokes)
|
|
|
|
def test_custom_tranche_params(self):
|
|
tp = TrancheParams(senior_collateral_ratio=2.0)
|
|
state = create_initial_state(tranche_params=tp)
|
|
assert state["mycofi"].tranche_system.params.senior_collateral_ratio == 2.0
|
|
|
|
def test_custom_conviction_params(self):
|
|
cp = ConvictionParams(alpha=0.8, beta=0.3)
|
|
state = create_initial_state(conviction_params=cp)
|
|
assert state["mycofi"].governance.params.alpha == 0.8
|
|
assert state["mycofi"].governance.params.beta == 0.3
|
|
|
|
def test_initial_aggregate_fields_are_zero(self):
|
|
state = create_initial_state()
|
|
s = state["mycofi"]
|
|
assert s.time == 0.0
|
|
assert s.total_collateral_usd == 0.0
|
|
|
|
|
|
# ---------- bootstrap_system (inline implementation) ----------
|
|
|
|
class TestBootstrapSystem:
|
|
def test_returns_dict_with_mycofi_key(self):
|
|
state = create_initial_state()
|
|
result = bootstrap_state(state)
|
|
assert "mycofi" in result
|
|
|
|
def test_populates_governance_voters(self):
|
|
state = create_initial_state()
|
|
bootstrap_state(state, n_voters=10)
|
|
assert len(state["mycofi"].governance.voters) == 10
|
|
|
|
def test_governance_voters_have_holdings(self):
|
|
state = create_initial_state()
|
|
bootstrap_state(state, n_voters=5)
|
|
for voter in state["mycofi"].governance.voters.values():
|
|
assert voter.holdings > 0.0
|
|
|
|
def test_governance_total_supply_updated(self):
|
|
state = create_initial_state()
|
|
bootstrap_state(state, n_voters=5)
|
|
s = state["mycofi"]
|
|
expected_supply = sum(v.holdings for v in s.governance.voters.values())
|
|
assert s.governance.total_supply == pytest.approx(expected_supply)
|
|
|
|
def test_with_initial_deposits_seeds_crosschain(self):
|
|
state = create_initial_state()
|
|
deposits = {"ethereum": {"stETH": 50.0}}
|
|
bootstrap_state(state, initial_deposits=deposits, n_voters=3)
|
|
spoke = state["mycofi"].crosschain.hub.spokes["ethereum"]
|
|
assert spoke.balances.get("stETH", 0.0) == pytest.approx(50.0)
|
|
|
|
def test_with_initial_deposits_seeds_tranche_collateral(self):
|
|
state = create_initial_state()
|
|
deposits = {"ethereum": {"stETH": 100.0}}
|
|
bootstrap_state(state, initial_deposits=deposits, n_voters=3)
|
|
s = state["mycofi"]
|
|
assert s.tranche_system.total_collateral > 0.0
|
|
|
|
def test_with_initial_tranche_mints(self):
|
|
state = create_initial_state()
|
|
deposits = {"ethereum": {"stETH": 1000.0}}
|
|
mints = {"senior": 100_000.0, "mezzanine": 50_000.0}
|
|
bootstrap_state(state, initial_deposits=deposits,
|
|
initial_tranche_mints=mints, n_voters=3)
|
|
s = state["mycofi"]
|
|
# Verify no crash and state is internally consistent
|
|
assert s.tranche_system.senior.supply >= 0.0
|
|
assert s.tranche_system.mezzanine.supply >= 0.0
|
|
|
|
def test_default_bootstrap_no_crash(self):
|
|
state = create_initial_state()
|
|
bootstrap_state(state) # Should not raise
|
|
assert "mycofi" in state
|
|
|
|
def test_sync_metrics_called_by_bootstrap(self):
|
|
state = create_initial_state()
|
|
deposits = {"ethereum": {"stETH": 100.0}}
|
|
bootstrap_state(state, initial_deposits=deposits, n_voters=5)
|
|
s = state["mycofi"]
|
|
assert s.total_chains == 5
|
|
|
|
|
|
# ---------- sync_metrics ----------
|
|
|
|
class TestSyncMetrics:
|
|
def _make_bootstrapped_state(self) -> MycoFiState:
|
|
state = create_initial_state()
|
|
deposits = {"ethereum": {"stETH": 100.0}, "base": {"cbETH": 50.0}}
|
|
mints = {"senior": 50_000.0, "mezzanine": 20_000.0, "junior": 10_000.0}
|
|
bootstrap_state(state, initial_deposits=deposits,
|
|
initial_tranche_mints=mints, n_voters=5)
|
|
return state["mycofi"]
|
|
|
|
def test_sync_returns_state(self):
|
|
s = self._make_bootstrapped_state()
|
|
result = sync_metrics(s)
|
|
assert result is s
|
|
|
|
def test_sync_updates_senior_supply(self):
|
|
s = self._make_bootstrapped_state()
|
|
s.tranche_system.senior.supply = 999.0
|
|
sync_metrics(s)
|
|
assert s.senior_supply == pytest.approx(999.0)
|
|
|
|
def test_sync_updates_mezzanine_supply(self):
|
|
s = self._make_bootstrapped_state()
|
|
s.tranche_system.mezzanine.supply = 888.0
|
|
sync_metrics(s)
|
|
assert s.mezzanine_supply == pytest.approx(888.0)
|
|
|
|
def test_sync_updates_junior_supply(self):
|
|
s = self._make_bootstrapped_state()
|
|
s.tranche_system.junior.supply = 777.0
|
|
sync_metrics(s)
|
|
assert s.junior_supply == pytest.approx(777.0)
|
|
|
|
def test_sync_updates_collateral_ratios(self):
|
|
s = self._make_bootstrapped_state()
|
|
sync_metrics(s)
|
|
assert s.senior_cr == pytest.approx(s.tranche_system.senior.collateral_ratio)
|
|
assert s.mezzanine_cr == pytest.approx(s.tranche_system.mezzanine.collateral_ratio)
|
|
|
|
def test_sync_updates_system_collateral_ratio(self):
|
|
s = self._make_bootstrapped_state()
|
|
sync_metrics(s)
|
|
assert s.system_collateral_ratio == pytest.approx(
|
|
s.tranche_system.system_collateral_ratio
|
|
)
|
|
|
|
def test_sync_updates_total_chains(self):
|
|
s = self._make_bootstrapped_state()
|
|
sync_metrics(s)
|
|
assert s.total_chains == len(s.crosschain.hub.spokes)
|
|
|
|
def test_sync_updates_total_collateral_usd(self):
|
|
s = self._make_bootstrapped_state()
|
|
sync_metrics(s)
|
|
assert s.total_collateral_usd == pytest.approx(s.crosschain.hub.total_collateral_usd)
|
|
|
|
def test_sync_updates_governance_epoch(self):
|
|
s = self._make_bootstrapped_state()
|
|
s.governance.epoch = 42
|
|
sync_metrics(s)
|
|
assert s.governance_epoch == 42
|
|
|
|
def test_sync_updates_proposals_passed(self):
|
|
s = self._make_bootstrapped_state()
|
|
dummy = Proposal(id="x", title="done", status="passed")
|
|
s.governance.passed_proposals.append(dummy)
|
|
sync_metrics(s)
|
|
assert s.proposals_passed == 1
|
|
|
|
def test_sync_updates_total_staked(self):
|
|
s = self._make_bootstrapped_state()
|
|
prop = Proposal(id="p1", title="Test", funds_requested=0.01)
|
|
s.governance.proposals["p1"] = prop
|
|
v = list(s.governance.voters.values())[0]
|
|
cv_stake(s.governance, v.id, "p1", min(100.0, v.holdings))
|
|
sync_metrics(s)
|
|
assert s.total_staked > 0.0
|
|
|
|
def test_sync_updates_total_yield(self):
|
|
s = self._make_bootstrapped_state()
|
|
s.crosschain.total_yield_generated = 999.99
|
|
sync_metrics(s)
|
|
assert s.total_yield == pytest.approx(999.99)
|
|
|
|
def test_sync_with_none_subsystems_is_safe(self):
|
|
s = MycoFiState()
|
|
sync_metrics(s) # Should not raise
|
|
|
|
|
|
# ---------- extract_metrics ----------
|
|
|
|
class TestExtractMetrics:
|
|
def _make_synced_state(self) -> MycoFiState:
|
|
state = create_initial_state()
|
|
deposits = {"ethereum": {"stETH": 100.0}}
|
|
bootstrap_state(state, initial_deposits=deposits, n_voters=5)
|
|
s = state["mycofi"]
|
|
sync_metrics(s)
|
|
return s
|
|
|
|
def test_returns_dict(self):
|
|
s = self._make_synced_state()
|
|
m = extract_metrics(s)
|
|
assert isinstance(m, dict)
|
|
|
|
def test_contains_time(self):
|
|
s = self._make_synced_state()
|
|
m = extract_metrics(s)
|
|
assert "time" in m
|
|
|
|
def test_contains_total_supply(self):
|
|
s = self._make_synced_state()
|
|
m = extract_metrics(s)
|
|
assert "total_supply" in m
|
|
|
|
def test_contains_total_collateral_usd(self):
|
|
s = self._make_synced_state()
|
|
m = extract_metrics(s)
|
|
assert "total_collateral_usd" in m
|
|
|
|
def test_contains_system_cr(self):
|
|
s = self._make_synced_state()
|
|
m = extract_metrics(s)
|
|
assert "system_cr" in m
|
|
|
|
def test_contains_myco_price(self):
|
|
s = self._make_synced_state()
|
|
m = extract_metrics(s)
|
|
assert "myco_price" in m
|
|
|
|
def test_contains_tranche_fields(self):
|
|
s = self._make_synced_state()
|
|
m = extract_metrics(s)
|
|
for field_name in ["senior_supply", "senior_cr", "mezzanine_supply",
|
|
"mezzanine_cr", "junior_supply", "junior_cr"]:
|
|
assert field_name in m, f"Missing field: {field_name}"
|
|
|
|
def test_contains_crosschain_fields(self):
|
|
s = self._make_synced_state()
|
|
m = extract_metrics(s)
|
|
assert "total_chains" in m
|
|
assert "total_yield" in m
|
|
assert "ccip_messages" in m
|
|
|
|
def test_contains_governance_fields(self):
|
|
s = self._make_synced_state()
|
|
m = extract_metrics(s)
|
|
assert "governance_epoch" in m
|
|
assert "total_staked" in m
|
|
assert "proposals_passed" in m
|
|
|
|
def test_per_chain_collateral_included(self):
|
|
s = self._make_synced_state()
|
|
m = extract_metrics(s)
|
|
for chain in ["ethereum", "arbitrum", "optimism", "base", "polygon"]:
|
|
assert f"collateral_{chain}" in m, f"Missing per-chain field: collateral_{chain}"
|
|
|
|
def test_values_match_state(self):
|
|
s = self._make_synced_state()
|
|
m = extract_metrics(s)
|
|
assert m["senior_supply"] == pytest.approx(s.senior_supply)
|
|
assert m["mezzanine_supply"] == pytest.approx(s.mezzanine_supply)
|
|
assert m["junior_supply"] == pytest.approx(s.junior_supply)
|
|
assert m["governance_epoch"] == s.governance_epoch
|
|
assert m["total_chains"] == s.total_chains
|
|
|
|
def test_per_chain_values_match_spokes(self):
|
|
s = self._make_synced_state()
|
|
m = extract_metrics(s)
|
|
for chain, spoke in s.crosschain.hub.spokes.items():
|
|
assert m[f"collateral_{chain}"] == pytest.approx(spoke.total_value_usd)
|
|
|
|
def test_all_values_numeric(self):
|
|
state = create_initial_state()
|
|
deposits = {"ethereum": {"stETH": 500.0}}
|
|
mints = {"senior": 100_000.0, "mezzanine": 50_000.0, "junior": 20_000.0}
|
|
bootstrap_state(state, initial_deposits=deposits,
|
|
initial_tranche_mints=mints, n_voters=3)
|
|
s = state["mycofi"]
|
|
sync_metrics(s)
|
|
m = extract_metrics(s)
|
|
for key, value in m.items():
|
|
assert isinstance(value, (int, float)), f"Non-numeric metric: {key}={value}"
|