| """ |
| Full system integration test — kernel, public feeds, engines, gateway. |
| Run after both servers are up (gateway :3000, SRC API :8080). |
| """ |
|
|
| from __future__ import annotations |
|
|
| import asyncio |
| import json |
| import sys |
| import time |
| from dataclasses import dataclass, field |
| from typing import Any, Dict, List |
|
|
| import httpx |
| import numpy as np |
|
|
| SRC_URL = "http://localhost:8080" |
| GW_URL = "http://localhost:3000" |
| GW_TOKEN = "pilot-console-primary-jwt-fallback-x992z" |
|
|
|
|
| @dataclass |
| class TestResult: |
| name: str |
| passed: bool |
| detail: str = "" |
| data: Dict[str, Any] = field(default_factory=dict) |
|
|
|
|
| results: List[TestResult] = [] |
|
|
|
|
| def record(name: str, passed: bool, detail: str = "", **data: Any) -> None: |
| results.append(TestResult(name, passed, detail, dict(data))) |
| status = "PASS" if passed else "FAIL" |
| print(f" [{status}] {name}" + (f" — {detail}" if detail else "")) |
|
|
|
|
| async def wait_for(url: str, path: str = "/health", timeout: float = 30.0) -> bool: |
| deadline = time.time() + timeout |
| async with httpx.AsyncClient() as client: |
| while time.time() < deadline: |
| try: |
| r = await client.get(f"{url}{path}", timeout=3.0) |
| if r.status_code == 200: |
| return True |
| except Exception: |
| pass |
| await asyncio.sleep(1.0) |
| return False |
|
|
|
|
| async def test_src_health() -> None: |
| async with httpx.AsyncClient(timeout=10.0) as c: |
| r = await c.get(f"{SRC_URL}/health") |
| d = r.json() |
| record("SRC API health", r.status_code == 200 and d.get("status") == "ok", json.dumps(d)) |
|
|
|
|
| async def test_gateway_health() -> None: |
| async with httpx.AsyncClient(timeout=10.0) as c: |
| r = await c.get(f"{GW_URL}/v1/health") |
| d = r.json() |
| record("Gateway health", r.status_code == 200 and d.get("status") == "ok", json.dumps(d)) |
|
|
|
|
| async def test_kernel_examples() -> None: |
| async with httpx.AsyncClient(timeout=15.0) as c: |
| r = await c.get(f"{SRC_URL}/api/examples/worked") |
| d = r.json() |
| ok = r.status_code == 200 and len(d) == 5 |
| record("Worked examples (5 forcing functions)", ok, f"modes={list(d.keys())}") |
|
|
|
|
| async def test_benchmarks() -> None: |
| async with httpx.AsyncClient(timeout=30.0) as c: |
| r = await c.get(f"{SRC_URL}/api/benchmarks") |
| d = r.json() |
| n = d.get("summary", {}).get("operators_tested", 0) |
| record("20-operator benchmark", r.status_code == 200 and n == 20, f"mean_gain={d.get('summary',{}).get('mean_gain_pct')}%") |
|
|
|
|
| async def test_live_feed() -> None: |
| async with httpx.AsyncClient(timeout=30.0) as c: |
| r = await c.post(f"{SRC_URL}/api/feed/live", json={"latitude": 38.627, "longitude": -90.199, "hours": 1}) |
| d = r.json() |
| snap = d.get("snapshot", {}) |
| ok = r.status_code == 200 and "Q_t" in snap and "DT" in snap |
| record("Live public sensor feed", ok, f"source={snap.get('source')}, Q_t={snap.get('Q_t')}") |
|
|
|
|
| async def test_feed_series_pipeline() -> None: |
| async with httpx.AsyncClient(timeout=60.0) as c: |
| r = await c.post(f"{SRC_URL}/api/feed/series", json={"latitude": 38.627, "longitude": -90.199, "hours": 12}) |
| d = r.json() |
| engines = d.get("primal_bridge", {}) |
| atlas = d.get("atlas_fusion", {}) |
| ok = ( |
| r.status_code == 200 |
| and d.get("points", 0) >= 1 |
| and "trust_anchor_sha512" in d |
| and "coherence_retention" in d |
| ) |
| record( |
| "Full feed→Dx→Atlas→Primal pipeline", |
| ok, |
| f"points={d.get('points')}, coherence={round(d.get('coherence_retention',0),3)}", |
| engines_available=d.get("primal_bridge", {}).get("ewic_control") is not None, |
| atlas_converged=atlas.get("converged"), |
| ) |
|
|
|
|
| async def test_compute_dx_meta() -> None: |
| q = [1.0, 2.5, -0.3, 1.8, 0.5] |
| async with httpx.AsyncClient(timeout=15.0) as c: |
| rdx = await c.post(f"{SRC_URL}/api/compute/dx", json={"q_values": q, "a": 1.0, "dt": 1.0}) |
| rmeta = await c.post(f"{SRC_URL}/api/compute/meta", json={"q_values": q, "b": 0.091, "dt": 1.0}) |
| ok = rdx.status_code == 200 and rmeta.status_code == 200 |
| record("Dx + O(f) compute endpoints", ok, f"dx={rdx.json().get('dx_final')}, of={rmeta.json().get('of_final')}") |
|
|
|
|
| async def test_gateway_psi_rpo() -> None: |
| headers = {"Authorization": f"Bearer {GW_TOKEN}"} |
| async with httpx.AsyncClient(timeout=20.0) as c: |
| psi = await c.post(f"{GW_URL}/api/compute/psi", headers=headers, json={"t": 3.5, "x0": 150.0}) |
| rpo = await c.post(f"{GW_URL}/api/compute/rpo", headers=headers, json={"n": 30, "x_init": 120.0}) |
| ok = psi.status_code == 200 and rpo.status_code == 200 and psi.json().get("status") == "STABLE" |
| record("Gateway psi + rpo offload", ok, f"psi={psi.json().get('psi')}, converged={rpo.json().get('converged')}") |
|
|
|
|
| async def test_gateway_src_offload() -> None: |
| headers = {"Authorization": f"Bearer {GW_TOKEN}"} |
| q = [1.95, 4.25, -0.05, 2.1, -1.3, 0.8] |
| async with httpx.AsyncClient(timeout=20.0) as c: |
| r = await c.post( |
| f"{GW_URL}/api/compute/src", |
| headers=headers, |
| json={"q_values": q, "a": 1.0, "b": 0.091, "mu": 0.16905, "dt": 3600.0}, |
| ) |
| d = r.json() |
| ok = r.status_code == 200 and d.get("status") == "SYMBOLIC_RECURSION_STABLE" |
| record("Gateway SRC offload endpoint", ok, f"dx_final={d.get('dx_final')}, converged={d.get('converged')}") |
|
|
|
|
| async def test_gateway_stk() -> None: |
| async with httpx.AsyncClient(timeout=60.0) as c: |
| r = await c.post(f"{GW_URL}/api/physics/stk/simulate", json={"steps": 500, "dt": 0.05}) |
| d = r.json() |
| ok = r.status_code == 200 and "trustAnchor" in d |
| record("Gateway STK simulation", ok, f"finalState={d.get('finalState')}, anchor={str(d.get('trustAnchor',''))[:16]}...") |
|
|
|
|
| async def test_src_gateway_bridge() -> None: |
| async with httpx.AsyncClient(timeout=20.0) as c: |
| r = await c.get(f"{SRC_URL}/api/gateway/status") |
| d = r.json() |
| record("SRC→Gateway status bridge", d.get("connected") is True, json.dumps(d.get("health", {}))) |
|
|
|
|
| async def test_src_gateway_offload() -> None: |
| async with httpx.AsyncClient(timeout=60.0) as c: |
| r = await c.post(f"{SRC_URL}/api/gateway/offload", params={"t": 3.5, "x0": 150.0}) |
| d = r.json() |
| ok = r.status_code == 200 and "psi" in d |
| record("SRC gateway offload proxy", ok, f"psi_status={d.get('psi',{}).get('status')}") |
|
|
|
|
| async def test_primal_engines_direct() -> None: |
| from symbolic_recursion.integrations.primal_bridge import PrimalBridge |
| from symbolic_recursion.integrations.atlas_bridge import AtlasBridge |
| from symbolic_recursion.kernel import SymbolicRecursionKernel |
|
|
| bridge = PrimalBridge() |
| avail = bridge.engines_available |
| q = np.array([1.0, 2.0, -0.5, 1.5, 0.3]) |
| result = bridge.process_q_series(q.tolist(), dt=1.0) |
| dx = SymbolicRecursionKernel().integrate_series(q, dt=1.0) |
| atlas = AtlasBridge().fuse_with_symbolic_recursion(q, dx) |
| ok = avail["primal_trading"] and avail["primallang_physics"] and "physics" in result |
| record( |
| "Direct primal-trading + primallang physics", |
| ok, |
| f"ewic={result.get('ewic_control')}, eigenfreq={result.get('physics',{}).get('eigenfrequency_hz')}", |
| atlas_offset=atlas.get("offset_from_attractor"), |
| ) |
|
|
|
|
| async def test_unit_tests() -> None: |
| import subprocess |
| proc = subprocess.run( |
| [sys.executable, "-m", "pytest", "tests/", "-q"], |
| cwd=r"C:\Users\stlta\symbolic-recursion-coherence", |
| capture_output=True, |
| text=True, |
| ) |
| ok = proc.returncode == 0 |
| record("Unit test suite", ok, proc.stdout.strip().split("\n")[-1] if proc.stdout else proc.stderr[:80]) |
|
|
|
|
| async def main() -> int: |
| print("=" * 60) |
| print("SYMBOLIC RECURSION COHERENCE — FULL INTEGRATION TEST") |
| print("=" * 60) |
|
|
| print("\n[1/3] Waiting for services...") |
| src_up = await wait_for(SRC_URL) |
| gw_up = await wait_for(GW_URL, "/v1/health") |
| record("SRC API reachable", src_up, SRC_URL) |
| record("Gateway reachable", gw_up, GW_URL) |
| if not src_up: |
| print("\nABORT: Start SRC API with: python -m symbolic_recursion.cli serve") |
| return 1 |
|
|
| print("\n[2/3] Running integration tests...") |
| await test_unit_tests() |
| await test_src_health() |
| if gw_up: |
| await test_gateway_health() |
| await test_kernel_examples() |
| await test_benchmarks() |
| await test_compute_dx_meta() |
| await test_live_feed() |
| await test_feed_series_pipeline() |
| await test_primal_engines_direct() |
| if gw_up: |
| await test_gateway_psi_rpo() |
| await test_gateway_src_offload() |
| await test_gateway_stk() |
| await test_src_gateway_bridge() |
| await test_src_gateway_offload() |
| else: |
| record("Gateway tests", False, "skipped — gateway not running") |
|
|
| print("\n[3/3] Summary") |
| passed = sum(1 for r in results if r.passed) |
| failed = sum(1 for r in results if not r.passed) |
| print(f" Total: {len(results)} | Passed: {passed} | Failed: {failed}") |
| print("=" * 60) |
|
|
| report_path = r"C:\Users\stlta\symbolic-recursion-coherence\integration_report.json" |
| with open(report_path, "w") as f: |
| json.dump( |
| [{"name": r.name, "passed": r.passed, "detail": r.detail, "data": r.data} for r in results], |
| f, |
| indent=2, |
| ) |
| print(f"Report saved: {report_path}") |
| return 0 if failed == 0 else 1 |
|
|
|
|
| if __name__ == "__main__": |
| raise SystemExit(asyncio.run(main())) |
