"""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}"