🧪 Add Python test suite (pytest) — 55 tests across 5 modules

pytest.ini

@@ -0,0 +1,3 @@
+[pytest]
+testpaths = tests
+addopts = -v --tb=short

tests/__init__.py

@@ -0,0 +1 @@
+# ZeroSense test suite

tests/test_api.py

@@ -0,0 +1,142 @@
+"""Tests for FastAPI endpoints (api/main.py).
+
+Uses FastAPI TestClient — no server startup needed.
+"""
+import pytest
+from fastapi.testclient import TestClient
+from api.main import app
+
+
+@pytest.fixture(scope="module")
+def c():
+    return TestClient(app)
+
+
+# ── Root ──
+class TestRoot:
+    def test_200(self, c): assert c.get("/").status_code == 200
+    def test_project(self, c): assert c.get("/").json()["project"] == "ZeroSense"
+    def test_status(self, c): assert c.get("/").json()["status"] == "live"
+    def test_hackathon(self, c): assert "Stellar Hacks" in c.get("/").json()["hackathon"]
+
+
+# ── Generate Proof ──
+class TestGenerateProof:
+    def _req(self, c, pixels=None, task_id=None):
+        body = {
+            "robot_id": "robot-001",
+            "sensor_frames": [{"pixels": pixels or [0.8]*64, "frame_id": 0}],
+        }
+        if task_id:
+            body["task_id"] = task_id
+        return c.post("/generate-proof", json=body)
+
+    def test_200(self, c): assert self._req(c).status_code == 200
+    def test_has_task_id(self, c): assert "task_id" in self._req(c).json()
+    def test_has_proof_hex(self, c): assert "proof_hex" in self._req(c).json()
+    def test_has_confidence(self, c): assert "confidence" in self._req(c).json()
+    def test_has_action_label(self, c): assert "action_label" in self._req(c).json()
+    def test_has_input_hash(self, c): assert "input_hash" in self._req(c).json()
+    def test_confidence_range(self, c):
+        assert 0 <= self._req(c).json()["confidence"] <= 100
+    def test_action_label_valid(self, c):
+        assert self._req(c).json()["action_label"] in ("task_complete","obstacle_detected","incident")
+    def test_explicit_task_id(self, c):
+        r = self._req(c, task_id="mytask00000000000000000000000001")
+        assert r.json()["task_id"] == "mytask00000000000000000000000001"
+    def test_mock_proof_512_chars(self, c):
+        assert len(self._req(c).json()["proof_hex"]) == 512
+    def test_multi_frame(self, c):
+        r = c.post("/generate-proof", json={
+            "robot_id": "robot-001",
+            "sensor_frames": [
+                {"pixels": [0.8]*64, "frame_id": 0},
+                {"pixels": [0.9]*64, "frame_id": 1},
+            ]
+        })
+        assert r.status_code == 200
+
+
+# ── Verify Proof ──
+class TestVerifyProof:
+    def _verify(self, c, confidence=97, task_id="task_v_001"):
+        return c.post("/verify-proof", json={
+            "robot_id": "robot-001", "task_id": task_id,
+            "proof_hex": "0"*512, "model_hash": "mobilenet_v2_int8",
+            "confidence": confidence, "action_type": 0,
+        })
+
+    def test_200(self, c): assert self._verify(c).status_code == 200
+    def test_status_verified(self, c): assert self._verify(c).json()["status"] == "verified"
+    def test_auto_pay_high_conf(self, c): assert self._verify(c, 98, "task_v_002").json()["auto_payment"] is True
+    def test_no_auto_pay_low_conf(self, c): assert self._verify(c, 80, "task_v_003").json()["auto_payment"] is False
+    def test_has_stellar_tx(self, c): assert self._verify(c, task_id="task_v_004").json()["stellar_tx"] is not None
+
+
+# ── Claim Payment ──
+class TestClaimPayment:
+    def test_200(self, c):
+        assert c.post("/claim-payment", json={"task_id":"task_c_001","confidence":96}).status_code == 200
+    def test_paid_status(self, c):
+        assert c.post("/claim-payment", json={"task_id":"task_c_002","confidence":97}).json()["status"] == "paid"
+    def test_has_stellar_tx(self, c):
+        assert "stellar_tx" in c.post("/claim-payment", json={"task_id":"task_c_003","confidence":95}).json()
+
+
+# ── Insurance Claim ──
+class TestInsurance:
+    def test_200(self, c):
+        r = c.post("/file-insurance-claim", json={
+            "robot_id":"robot-001","incident_proof_hash":"abc"*21+"d","claim_amount":1_000_000
+        })
+        assert r.status_code == 200
+    def test_filed_status(self, c):
+        r = c.post("/file-insurance-claim", json={
+            "robot_id":"robot-001","incident_proof_hash":"def"*21+"g","claim_amount":500_000
+        })
+        assert r.json()["status"] == "filed"
+
+
+# ── Robot Identity ──
+class TestIdentity:
+    def test_200(self, c):
+        r = c.post("/robot/register-identity", json={
+            "robot_id":"robot-001","sensor_noise_sample":[0.01]*64
+        })
+        assert r.status_code == 200
+    def test_has_identity_hash(self, c):
+        r = c.post("/robot/register-identity", json={
+            "robot_id":"robot-002","sensor_noise_sample":[0.02]*64
+        })
+        assert len(r.json()["identity_hash"]) == 64
+    def test_different_robots_different_hashes(self, c):
+        r1 = c.post("/robot/register-identity", json={"robot_id":"rA","sensor_noise_sample":[0.01]*64})
+        r2 = c.post("/robot/register-identity", json={"robot_id":"rB","sensor_noise_sample":[0.99]*64})
+        assert r1.json()["identity_hash"] != r2.json()["identity_hash"]
+
+
+# ── Robot Status + Fleet ──
+class TestStatus:
+    def test_robot_status_200(self, c): assert c.get("/robot/robot-001/status").status_code == 200
+    def test_robot_has_fields(self, c):
+        d = c.get("/robot/robot-001/status").json()
+        assert "robot_id" in d and "zrep_balance" in d
+    def test_fleet_200(self, c): assert c.get("/fleet/report").status_code == 200
+    def test_fleet_has_fields(self, c):
+        d = c.get("/fleet/report").json()
+        assert "total_tasks_verified" in d and "guardian_status" in d
+
+
+# ── Guardian ──
+class TestGuardian:
+    def test_start_200(self, c): assert c.post("/guardian/start").status_code == 200
+    def test_start_status(self, c):
+        assert c.post("/guardian/start").json()["status"] in ("started","already_running")
+    def test_stop_200(self, c): assert c.post("/guardian/stop").status_code == 200
+    def test_stop_status(self, c): assert c.post("/guardian/stop").json()["status"] == "stopped"
+    def test_agents_count(self, c):
+        c.post("/guardian/stop")
+        r = c.post("/guardian/start")
+        if "agents" in r.json():
+            assert len(r.json()["agents"]) == 7
+        c.post("/guardian/stop")

tests/test_guardian.py

@@ -0,0 +1,148 @@
+"""Tests for ZeroSense Guardian v2 — all 7 autonomous agents."""
+import asyncio
+import pytest
+from unittest.mock import AsyncMock, MagicMock
+from api.agents.guardian import (
+    ZeroSenseGuardianV2, PaymentAgent, AnomalyAgent,
+    InsuranceAgent, ReputationAgent,
+)
+
+
+@pytest.fixture
+def stellar():
+    m = MagicMock()
+    m.get_recent_contract_events = AsyncMock(return_value=[])
+    m.claim_payment = AsyncMock(return_value="tx_mock_paid")
+    m.file_insurance_claim = AsyncMock(return_value=42)
+    return m
+
+@pytest.fixture
+def guardian(stellar):
+    return ZeroSenseGuardianV2(stellar_client=stellar)
+
+def run(coro):
+    return asyncio.get_event_loop().run_until_complete(coro)
+
+
+class TestGuardianInit:
+    def test_not_running(self, guardian): assert guardian.running is False
+    def test_has_7_agents(self, guardian): assert len(guardian.agents) == 7
+    def test_has_payment(self, guardian): assert "payment" in guardian.agents
+    def test_has_anomaly(self, guardian): assert "anomaly" in guardian.agents
+    def test_has_insurance(self, guardian): assert "insurance" in guardian.agents
+    def test_has_reputation(self, guardian): assert "reputation" in guardian.agents
+    def test_has_learning(self, guardian): assert "learning" in guardian.agents
+    def test_has_oracle(self, guardian): assert "oracle" in guardian.agents
+    def test_has_assistant(self, guardian): assert "assistant" in guardian.agents
+    def test_status_returns_all(self, guardian):
+        assert set(guardian.status().keys()) == {
+            "payment","anomaly","insurance","reputation","learning","oracle","assistant"
+        }
+    def test_initial_counts_zero(self, guardian):
+        for info in guardian.status().values():
+            assert info["processed"] == 0
+
+
+class TestPaymentAgent:
+    def test_init(self, stellar):
+        a = PaymentAgent(stellar)
+        assert a.threshold == 0.95 and len(a.paid_tasks) == 0
+
+    def test_no_events_no_pay(self, stellar):
+        a = PaymentAgent(stellar)
+        run(a.tick())
+        assert a.processed_count == 0
+
+    def test_high_conf_pays(self, stellar):
+        stellar.get_recent_contract_events = AsyncMock(
+            return_value=[{"task_id":"t001","confidence":97}]
+        )
+        a = PaymentAgent(stellar)
+        run(a.tick())
+        assert a.processed_count == 1
+        stellar.claim_payment.assert_called_once()
+
+    def test_low_conf_skips(self, stellar):
+        stellar.get_recent_contract_events = AsyncMock(
+            return_value=[{"task_id":"t002","confidence":80}]
+        )
+        a = PaymentAgent(stellar)
+        run(a.tick())
+        assert a.processed_count == 0
+        stellar.claim_payment.assert_not_called()
+
+    def test_no_duplicate_pay(self, stellar):
+        stellar.get_recent_contract_events = AsyncMock(
+            return_value=[{"task_id":"t003","confidence":98}]
+        )
+        a = PaymentAgent(stellar)
+        run(a.tick())
+        run(a.tick())   # second tick — same task
+        assert stellar.claim_payment.call_count == 1
+
+
+class TestAnomalyAgent:
+    def test_high_conf_no_count(self, stellar):
+        stellar.get_recent_contract_events = AsyncMock(
+            return_value=[{"robot_id":"r1","confidence":97}]
+        )
+        a = AnomalyAgent(stellar)
+        run(a.tick())
+        assert a.low_confidence_counts.get("r1", 0) == 0
+
+    def test_low_conf_increments(self, stellar):
+        stellar.get_recent_contract_events = AsyncMock(
+            return_value=[{"robot_id":"r1","confidence":70}]
+        )
+        a = AnomalyAgent(stellar)
+        run(a.tick())
+        assert a.low_confidence_counts["r1"] == 1
+
+    def test_killswitch_at_threshold(self, stellar):
+        stellar.get_recent_contract_events = AsyncMock(
+            return_value=[{"robot_id":"r-bad","confidence":50}]
+        )
+        a = AnomalyAgent(stellar, anomaly_threshold=3)
+        for _ in range(3):
+            run(a.tick())
+        assert a.processed_count >= 1
+
+    def test_good_reading_resets(self, stellar):
+        a = AnomalyAgent(stellar)
+        a.low_confidence_counts["r1"] = 2
+        stellar.get_recent_contract_events = AsyncMock(
+            return_value=[{"robot_id":"r1","confidence":98}]
+        )
+        run(a.tick())
+        assert a.low_confidence_counts["r1"] == 0
+
+
+class TestInsuranceAgent:
+    def test_files_claim(self, stellar):
+        stellar.get_recent_contract_events = AsyncMock(
+            return_value=[{"proof_hash":"deadbeef"*8,"robot_id":"r1","estimated_claim":1_000_000}]
+        )
+        a = InsuranceAgent(stellar)
+        run(a.tick())
+        assert a.processed_count == 1
+        stellar.file_insurance_claim.assert_called_once()
+
+    def test_no_duplicate_claim(self, stellar):
+        stellar.get_recent_contract_events = AsyncMock(
+            return_value=[{"proof_hash":"cafebabe"*8,"robot_id":"r1","estimated_claim":500_000}]
+        )
+        a = InsuranceAgent(stellar)
+        run(a.tick())
+        run(a.tick())
+        assert stellar.file_insurance_claim.call_count == 1
+
+
+class TestReputationAgent:
+    def test_processes_payment_event(self, stellar):
+        stellar.get_recent_contract_events = AsyncMock(
+            return_value=[{"task_id":"trep1","robot_id":"r1","confidence":97}]
+        )
+        a = ReputationAgent(stellar)
+        run(a.tick())
+        assert a.processed_count == 1
+        assert "trep1" in a.processed_tasks

tests/test_inference.py

@@ -0,0 +1,73 @@
+"""Tests for AI inference engine (model/inference.py).
+
+All 10 tests run in mock mode — no ONNX file required.
+"""
+import sys, os
+sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))
+
+import pytest
+from model.inference import RobotInferenceEngine, ACTION_LABELS
+
+
+@pytest.fixture
+def engine():
+    return RobotInferenceEngine()
+
+
+class TestRobotInferenceEngine:
+
+    def test_engine_initializes(self, engine):
+        assert engine is not None
+
+    def test_model_hash_set(self, engine):
+        assert engine.model_hash is not None
+        assert len(engine.model_hash) > 0
+
+    def test_run_inference_returns_triple(self, engine):
+        frames = [[0.5] * 64]
+        result = engine.run_inference(frames)
+        assert len(result) == 3
+
+    def test_action_in_valid_range(self, engine):
+        action, _, _ = engine.run_inference([[0.5] * 64])
+        assert action in (0, 1, 2)
+
+    def test_confidence_in_valid_range(self, engine):
+        _, confidence, _ = engine.run_inference([[0.5] * 64])
+        assert 0 <= confidence <= 100
+
+    def test_input_hash_is_sha256(self, engine):
+        _, _, h = engine.run_inference([[0.5] * 64])
+        assert len(h) == 64
+
+    def test_high_avg_gives_task_complete(self, engine):
+        action, confidence, _ = engine.run_inference([[0.9] * 64])
+        assert action == 0   # task_complete
+        assert confidence >= 90
+
+    def test_mid_avg_gives_obstacle(self, engine):
+        action, _, _ = engine.run_inference([[0.55] * 64])
+        assert action == 1   # obstacle_detected
+
+    def test_low_avg_gives_low_confidence(self, engine):
+        _, confidence, _ = engine.run_inference([[0.2] * 64])
+        assert confidence < 80
+
+    def test_empty_frames_returns_incident(self, engine):
+        action, _, _ = engine.run_inference([])
+        assert action == 2   # incident
+
+    def test_hash_is_deterministic(self, engine):
+        _, _, h1 = engine.run_inference([[0.5] * 64])
+        _, _, h2 = engine.run_inference([[0.5] * 64])
+        assert h1 == h2
+
+    def test_different_inputs_different_hashes(self, engine):
+        _, _, h1 = engine.run_inference([[0.1] * 64])
+        _, _, h2 = engine.run_inference([[0.9] * 64])
+        assert h1 != h2
+
+    def test_action_labels_complete(self):
+        assert ACTION_LABELS[0] == "task_complete"
+        assert ACTION_LABELS[1] == "obstacle_detected"
+        assert ACTION_LABELS[2] == "incident"

tests/test_simulation.py

@@ -0,0 +1,54 @@
+"""Tests for warehouse robot simulation (simulation/robot_sim.py)."""
+import math
+import pytest
+from simulation.robot_sim import WarehouseRobot
+
+
+@pytest.fixture
+def robot():
+    return WarehouseRobot(robot_id="test-robot-001")
+
+
+class TestWarehouseRobot:
+    def test_init(self, robot):
+        assert robot.robot_id == "test-robot-001"
+        assert robot.position == [0.0, 0.0]
+        assert robot.task_count == 0
+
+    def test_sensor_frame_64_values(self, robot):
+        assert len(robot.capture_sensor_frame()) == 64
+
+    def test_sensor_values_in_range(self, robot):
+        for v in robot.capture_sensor_frame():
+            assert 0.0 <= v <= 1.0
+
+    def test_generate_task_data_num_frames(self, robot):
+        assert len(robot.generate_task_sensor_data(num_frames=5)) == 5
+
+    def test_generate_task_data_frame_shape(self, robot):
+        for frame in robot.generate_task_sensor_data(num_frames=3):
+            assert len(frame) == 64
+
+    def test_navigate_already_at_target(self, robot):
+        robot.position = [0.0, 0.0]
+        assert robot.navigate_to_task((0.05, 0.05)) is True
+
+    def test_navigate_moves_toward_target(self, robot):
+        robot.position = [0.0, 0.0]
+        robot.navigate_to_task((5.0, 0.0))
+        assert robot.position[0] > 0.0
+
+    def test_navigate_eventually_reaches(self, robot):
+        robot.position = [0.0, 0.0]
+        reached = any(robot.navigate_to_task((1.0, 0.0)) for _ in range(100))
+        assert reached
+
+    def test_heading_updates(self, robot):
+        robot.position = [0.0, 0.0]
+        robot.navigate_to_task((0.0, 5.0))
+        assert abs(robot.heading - math.pi / 2) < 0.01
+
+    def test_frames_have_noise(self, robot):
+        f1, f2 = robot.capture_sensor_frame(), robot.capture_sensor_frame()
+        diffs = sum(1 for a, b in zip(f1, f2) if abs(a-b) > 0.001)
+        assert diffs > 0

tests/test_stellar_client.py

@@ -0,0 +1,65 @@
+"""Tests for Stellar SDK client (api/stellar/client.py).
+
+Runs in mock/dev mode — no real Stellar keys needed.
+"""
+import asyncio
+import pytest
+from api.stellar.client import StellarClient
+
+
+@pytest.fixture
+def client():
+    return StellarClient(secret_key="", network="testnet")
+
+
+class TestStellarClientInit:
+
+    def test_client_initializes(self, client):
+        assert client is not None
+        assert client.network == "testnet"
+
+    def test_testnet_urls(self, client):
+        assert "testnet" in client.horizon_url
+        assert "testnet" in client.soroban_url
+
+    def test_public_key_length(self, client):
+        assert len(client.keypair.public_key) == 56
+
+    def test_public_key_starts_with_G(self, client):
+        assert client.keypair.public_key.startswith("G")
+
+
+class TestStellarClientAsync:
+
+    def _run(self, coro):
+        return asyncio.get_event_loop().run_until_complete(coro)
+
+    def test_verify_proof_returns_tx(self, client):
+        tx = self._run(client.verify_proof_on_chain(
+            proof_hex="00" * 256, robot_id="robot-001",
+            task_id="task_test_verify", model_hash="mobilenet_v2_int8",
+            confidence=97, action_type=0,
+        ))
+        assert "verified" in tx
+
+    def test_claim_payment_returns_tx(self, client):
+        tx = self._run(client.claim_payment(task_id="task_pay_001", confidence=95))
+        assert "paid" in tx
+
+    def test_insurance_claim_returns_id(self, client):
+        cid = self._run(client.file_insurance_claim(
+            robot_id="robot-001", proof_hash="abc" * 20, claim_amount=1_000_000
+        ))
+        assert isinstance(cid, int) and cid >= 1
+
+    def test_zrep_balance_returns_int(self, client):
+        b = self._run(client.get_zrep_balance("robot-001"))
+        assert isinstance(b, int)
+
+    def test_mint_soulbound_returns_tx(self, client):
+        tx = self._run(client.mint_soulbound_identity_token("robot-001", "deadbeef" * 8))
+        assert "identity" in tx
+
+    def test_events_returns_list(self, client):
+        events = self._run(client.get_recent_contract_events("RobotActionVerified"))
+        assert isinstance(events, list)