diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml new file mode 100644 index 00000000..eb72f4ea --- /dev/null +++ b/.github/workflows/test.yml @@ -0,0 +1,36 @@ +name: Tests + +on: + pull_request: + push: + branches: [main] + +jobs: + test: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + + - uses: actions/setup-python@v5 + with: + python-version: "3.12" + + - name: Install uv + uses: astral-sh/setup-uv@v4 + + - name: Install dependencies + run: uv sync --extra dev + + - name: Lint + run: uv run ruff check src/crusades/core/ src/crusades/storage/ src/crusades/api/ tests/ + + - name: Test with coverage + run: uv run pytest tests/ --cov=crusades.core --cov=crusades.storage --cov=crusades.api --cov-report=xml --cov-report=term -v + + - name: Upload coverage + if: github.event_name == 'pull_request' + uses: orgoro/coverage@v3.2 + with: + coverageFile: coverage.xml + token: ${{ secrets.GITHUB_TOKEN }} + thresholdAll: 60 diff --git a/docker/compose.yml b/docker/compose.yml index e6e272a2..2c23d52c 100644 --- a/docker/compose.yml +++ b/docker/compose.yml @@ -99,4 +99,27 @@ services: - WATCHTOWER_INCLUDE_RESTARTING=true # For private GitHub Container Registry - REPO_USER=${GITHUB_USER:-} - - REPO_PASSWORD=${GITHUB_TOKEN:-} \ No newline at end of file + - REPO_PASSWORD=${GITHUB_TOKEN:-} + + # Health monitor - probes validator API and alerts via Discord + health-monitor: + image: python:3.12-slim + container_name: templar-crusades-health-monitor + profiles: + - monitoring + restart: unless-stopped + + volumes: + - ../scripts:/app/scripts:ro + + environment: + - VALIDATOR_URL=http://validator:8080 + - DISCORD_WEBHOOK_URL=${DISCORD_WEBHOOK_URL:-} + - CHECK_INTERVAL_SECONDS=${CHECK_INTERVAL_SECONDS:-60} + - STALE_THRESHOLD_SECONDS=${STALE_THRESHOLD_SECONDS:-3600} + + command: ["python", "/app/scripts/health_monitor.py"] + + depends_on: + validator: + condition: service_started diff --git a/pyproject.toml b/pyproject.toml index e737d0a0..e2b87b41 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -49,6 +49,7 @@ dev = [ "pytest>=7.4.0", "pytest-asyncio>=0.23.0", "pytest-cov>=4.1.0", + "pytest-mock>=3.12.0", "mypy>=1.8.0", "ruff>=0.1.0", ] diff --git a/scripts/health_monitor.py b/scripts/health_monitor.py new file mode 100644 index 00000000..dca888f6 --- /dev/null +++ b/scripts/health_monitor.py @@ -0,0 +1,159 @@ +#!/usr/bin/env python3 +"""Health monitor for crusades validator with Discord alerting. + +Probes the validator's /health endpoint on a configurable interval. +Posts to a Discord webhook when the validator is unreachable, returning +non-200, or stale (zero evaluations in the staleness window). + +Sends recovery notifications when the validator returns to healthy after +a failure period. + +Configuration via environment variables: + VALIDATOR_URL - Base URL of the validator API (default: http://localhost:8080) + DISCORD_WEBHOOK_URL - Discord webhook for alerts (optional; omit for log-only mode) + CHECK_INTERVAL_SECONDS - Seconds between health checks (default: 60) + STALE_THRESHOLD_SECONDS - Seconds of zero evaluations before alerting (default: 3600) + ALERT_COOLDOWN_SECONDS - Minimum seconds between repeat alerts (default: 300) +""" + +import json +import logging +import os +import sys +import time +from datetime import UTC, datetime +from urllib.error import URLError +from urllib.request import Request, urlopen + +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s | %(levelname)s | %(message)s", +) +logger = logging.getLogger("health_monitor") + +VALIDATOR_URL = os.getenv("VALIDATOR_URL", "http://localhost:8080").rstrip("/") +DISCORD_WEBHOOK_URL = os.getenv("DISCORD_WEBHOOK_URL", "") +CHECK_INTERVAL = int(os.getenv("CHECK_INTERVAL_SECONDS", "60")) +STALE_THRESHOLD = int(os.getenv("STALE_THRESHOLD_SECONDS", "3600")) +ALERT_COOLDOWN = int(os.getenv("ALERT_COOLDOWN_SECONDS", "300")) + + +def probe_health() -> dict: + """Probe the validator /health endpoint. + + Returns: + Parsed JSON response, or a dict with 'error' key on failure. + """ + url = f"{VALIDATOR_URL}/health" + try: + req = Request(url, method="GET") + with urlopen(req, timeout=10) as resp: + if resp.status != 200: + return {"error": f"HTTP {resp.status}"} + return json.loads(resp.read().decode()) + except URLError as e: + return {"error": f"Unreachable: {e.reason}"} + except Exception as e: + return {"error": str(e)} + + +def post_discord(content: str, color: int = 0xFF0000) -> None: + """Post an embed to Discord webhook. Fails silently if no webhook configured.""" + if not DISCORD_WEBHOOK_URL: + return + + payload = json.dumps({ + "embeds": [{ + "title": "Crusades Validator Health Alert", + "description": content, + "color": color, + "timestamp": datetime.now(UTC).isoformat(), + }] + }).encode() + + try: + req = Request( + DISCORD_WEBHOOK_URL, + data=payload, + headers={"Content-Type": "application/json"}, + method="POST", + ) + with urlopen(req, timeout=10): + pass + except Exception as e: + logger.error(f"Discord webhook failed: {e}") + + +def evaluate_health(data: dict) -> tuple[bool, str]: + """Evaluate health response. + + Returns: + (is_healthy, reason) + """ + if "error" in data: + return False, data["error"] + + status = data.get("status", "unknown") + if status == "unhealthy": + return False, "Validator reports unhealthy (DB disconnected)" + + if status == "degraded": + return False, ( + f"Validator is degraded: 0 evaluations in last hour, " + f"queue_depth={data.get('queue_depth', '?')}" + ) + + return True, "healthy" + + +def run() -> None: + """Main monitoring loop.""" + logger.info(f"Monitoring {VALIDATOR_URL}/health every {CHECK_INTERVAL}s") + if not DISCORD_WEBHOOK_URL: + logger.warning("DISCORD_WEBHOOK_URL not set - running in log-only mode") + + is_alerting = False + last_alert_time = 0.0 + + while True: + data = probe_health() + healthy, reason = evaluate_health(data) + + if healthy: + if is_alerting: + # Recovery + logger.info("Validator recovered") + post_discord( + f"Validator at `{VALIDATOR_URL}` has **recovered**.\n" + f"Status: `{data.get('status', 'healthy')}`\n" + f"Evaluations/1h: `{data.get('evaluations_1h', 'N/A')}`", + color=0x00FF00, + ) + is_alerting = False + else: + logger.info(f"OK | evals_1h={data.get('evaluations_1h', '?')} " + f"queue={data.get('queue_depth', '?')}") + else: + now = time.time() + if not is_alerting or (now - last_alert_time) >= ALERT_COOLDOWN: + logger.error(f"ALERT | {reason}") + post_discord( + f"Validator at `{VALIDATOR_URL}` is **down or degraded**.\n" + f"Reason: `{reason}`\n" + f"Checked at: `{datetime.now(UTC).isoformat()}`", + color=0xFF0000, + ) + is_alerting = True + last_alert_time = now + else: + logger.warning(f"Still failing: {reason} (cooldown active)") + + time.sleep(CHECK_INTERVAL) + + +if __name__ == "__main__": + try: + run() + except KeyboardInterrupt: + logger.info("Monitor stopped") + sys.exit(0) diff --git a/src/crusades/api/server.py b/src/crusades/api/server.py index 459508e4..f758f853 100644 --- a/src/crusades/api/server.py +++ b/src/crusades/api/server.py @@ -123,8 +123,33 @@ def create_app(api_key: str | None = None) -> FastAPI: @app.get("/health") async def health_check(): - """Health check endpoint.""" - return {"status": "healthy"} + """Health check endpoint with staleness detection. + + Returns DB connectivity, last evaluation time, and queue depth + so external probes can detect a running-but-stale validator. + """ + try: + client = get_db_client() + if client is None: + return {"status": "healthy", "db_connected": False} + + validator_status = client.get_validator_status() + queue = client.get_queue_stats() + + evaluations_1h = validator_status.get("evaluations_completed_1h", 0) + status = "healthy" if evaluations_1h > 0 else "degraded" + + return { + "status": status, + "db_connected": True, + "evaluations_1h": evaluations_1h, + "queue_depth": queue.get("queued_count", 0), + "current_evaluation": validator_status.get("current_evaluation"), + "validator_status": validator_status.get("status", "unknown"), + } + except Exception: + logger.exception("Health check failed") + return {"status": "unhealthy", "db_connected": False} # ============================================================ diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 00000000..4478a9c8 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,124 @@ +"""Shared test fixtures for crusades test suite.""" + +import uuid +from datetime import UTC, datetime, timedelta + +import pytest +import pytest_asyncio + +from crusades.core.protocols import SubmissionStatus +from crusades.storage.database import Database +from crusades.storage.models import ( + EvaluationModel, + SubmissionModel, +) + + +@pytest_asyncio.fixture +async def db(): + """Create an in-memory async Database instance with tables initialized.""" + database = Database(url="sqlite+aiosqlite:///:memory:") + await database.initialize() + yield database + await database.close() + + +@pytest_asyncio.fixture +async def seeded_db(db: Database): + """Database seeded with submissions and evaluations in various states.""" + async with db.session_factory() as session: + now = datetime.now(UTC) + + # Finished submission with evaluations (high score) + sub1 = SubmissionModel( + submission_id="sub-finished-1", + miner_hotkey="hotkey_alice", + miner_uid=1, + code_hash="hash_abc", + bucket_path="https://example.com/train1.py", + spec_version=19, + status=SubmissionStatus.FINISHED, + final_score=65.5, + code_content="def inner_steps(): pass", + created_at=now - timedelta(hours=2), + ) + session.add(sub1) + + eval1 = EvaluationModel( + evaluation_id=str(uuid.uuid4()), + submission_id="sub-finished-1", + evaluator_hotkey="validator_1", + mfu=65.5, + tokens_per_second=1200.0, + total_tokens=24000, + wall_time_seconds=20.0, + success=True, + created_at=now - timedelta(hours=2), + ) + session.add(eval1) + + # Finished submission (lower score, older) + sub2 = SubmissionModel( + submission_id="sub-finished-2", + miner_hotkey="hotkey_bob", + miner_uid=2, + code_hash="hash_def", + bucket_path="https://example.com/train2.py", + spec_version=19, + status=SubmissionStatus.FINISHED, + final_score=55.0, + created_at=now - timedelta(hours=5), + ) + session.add(sub2) + + eval2 = EvaluationModel( + evaluation_id=str(uuid.uuid4()), + submission_id="sub-finished-2", + evaluator_hotkey="validator_1", + mfu=55.0, + tokens_per_second=1000.0, + total_tokens=20000, + wall_time_seconds=20.0, + success=True, + created_at=now - timedelta(hours=5), + ) + session.add(eval2) + + # Pending submission + sub3 = SubmissionModel( + submission_id="sub-pending-1", + miner_hotkey="hotkey_carol", + miner_uid=3, + code_hash="hash_ghi", + bucket_path="https://example.com/train3.py", + spec_version=19, + status=SubmissionStatus.PENDING, + created_at=now - timedelta(minutes=10), + ) + session.add(sub3) + + # Failed submission + sub4 = SubmissionModel( + submission_id="sub-failed-1", + miner_hotkey="hotkey_dave", + miner_uid=4, + code_hash="hash_jkl", + bucket_path="https://example.com/train4.py", + spec_version=19, + status=SubmissionStatus.FAILED_VALIDATION, + error_message="SYNTAX_ERROR: invalid syntax", + created_at=now - timedelta(hours=1), + ) + session.add(sub4) + + await session.commit() + + return db + + +@pytest.fixture +def mock_db_client(): + """Return a MockClient for API tests that don't need real DB.""" + from crusades.tui.client import MockClient + + return MockClient() diff --git a/tests/test_api.py b/tests/test_api.py new file mode 100644 index 00000000..2ebb1d90 --- /dev/null +++ b/tests/test_api.py @@ -0,0 +1,231 @@ +"""Integration tests for FastAPI endpoints using MockClient.""" + +from unittest.mock import patch + +import pytest +from fastapi.testclient import TestClient + +import crusades.api.server as server_module +from crusades.tui.client import MockClient + + +@pytest.fixture +def client(): + """TestClient using the module-level app with MockClient injected.""" + mock = MockClient() + app = server_module.app + # Ensure no API key is required + app.state.api_key = None + with patch.object(server_module, "_db_client", mock): + with TestClient(app) as c: + yield c + + +@pytest.fixture +def authed_client(): + """TestClient with API key required.""" + mock = MockClient() + app = server_module.app + app.state.api_key = "test-secret-key" + with patch.object(server_module, "_db_client", mock): + with TestClient(app) as c: + yield c + + +class TestHealthEndpoint: + """Health check returns structured status.""" + + def test_health_returns_200(self, client): + resp = client.get("/health") + assert resp.status_code == 200 + data = resp.json() + assert "status" in data + assert "db_connected" in data + + def test_health_with_mock_client(self, client): + resp = client.get("/health") + data = resp.json() + assert data["db_connected"] is True + + def test_health_full_response_shape(self, client): + """Healthy response includes all expected keys.""" + resp = client.get("/health") + data = resp.json() + assert data["status"] == "healthy" + assert data["db_connected"] is True + assert "evaluations_1h" in data + assert "queue_depth" in data + assert "current_evaluation" in data + assert "validator_status" in data + # MockClient returns evaluations_completed_1h=18 → healthy + assert data["evaluations_1h"] == 18 + + def test_health_degraded_when_zero_evaluations(self): + """Status is 'degraded' when evaluations_completed_1h is 0.""" + + class StaleClient(MockClient): + def get_validator_status(self): + return { + "status": "running", + "evaluations_completed_1h": 0, + "current_evaluation": None, + } + + app = server_module.app + app.state.api_key = None + with patch.object(server_module, "_db_client", StaleClient()): + with TestClient(app) as c: + resp = c.get("/health") + data = resp.json() + assert data["status"] == "degraded" + assert data["db_connected"] is True + assert data["evaluations_1h"] == 0 + + def test_health_unhealthy_on_exception(self): + """Status is 'unhealthy' when client methods raise.""" + + class BrokenClient(MockClient): + def get_validator_status(self): + raise RuntimeError("DB gone") + + app = server_module.app + app.state.api_key = None + with patch.object(server_module, "_db_client", BrokenClient()): + with TestClient(app) as c: + resp = c.get("/health") + data = resp.json() + assert data["status"] == "unhealthy" + assert data["db_connected"] is False + + def test_health_db_unavailable(self): + """When db client is None, returns healthy with db_connected=False.""" + app = server_module.app + app.state.api_key = None + with patch.object(server_module, "_db_client", None): + with patch.object(server_module, "get_db_client", return_value=None): + with TestClient(app) as c: + resp = c.get("/health") + data = resp.json() + assert data["status"] == "healthy" + assert data["db_connected"] is False + + +class TestStatsEndpoints: + """Stats endpoints return expected structures.""" + + def test_overview(self, client): + resp = client.get("/api/stats/overview") + assert resp.status_code == 200 + data = resp.json() + assert "submissions_24h" in data + assert "current_top_score" in data + assert "active_miners" in data + + def test_validator_status(self, client): + resp = client.get("/api/stats/validator") + assert resp.status_code == 200 + data = resp.json() + assert "status" in data + assert "evaluations_completed_1h" in data + + def test_recent_submissions(self, client): + resp = client.get("/api/stats/recent") + assert resp.status_code == 200 + data = resp.json() + assert isinstance(data, list) + + def test_recent_submissions_limit(self, client): + resp = client.get("/api/stats/recent?limit=5") + assert resp.status_code == 200 + data = resp.json() + assert len(data) <= 5 + + def test_history(self, client): + resp = client.get("/api/stats/history") + assert resp.status_code == 200 + assert isinstance(resp.json(), list) + + def test_history_limit(self, client): + resp = client.get("/api/stats/history?limit=10") + assert resp.status_code == 200 + + def test_queue_stats(self, client): + resp = client.get("/api/stats/queue") + assert resp.status_code == 200 + data = resp.json() + assert "queued_count" in data + assert "running_count" in data + assert "finished_count" in data + + def test_threshold(self, client): + resp = client.get("/api/stats/threshold") + assert resp.status_code == 200 + data = resp.json() + assert "current_threshold" in data + assert "decayed_threshold" in data + + +class TestLeaderboard: + """Leaderboard endpoint.""" + + def test_leaderboard_default(self, client): + resp = client.get("/leaderboard") + assert resp.status_code == 200 + data = resp.json() + assert isinstance(data, list) + + def test_leaderboard_with_limit(self, client): + resp = client.get("/leaderboard?limit=5") + assert resp.status_code == 200 + data = resp.json() + assert len(data) <= 5 + + +class TestSubmissionEndpoints: + """Submission detail endpoints.""" + + def test_submission_detail(self, client): + # MockClient returns data for any submission_id + resp = client.get("/api/submissions/test-sub-1") + assert resp.status_code == 200 + + def test_submission_not_found(self): + """When get_submission returns empty dict, endpoint returns 404.""" + + class EmptyClient(MockClient): + def get_submission(self, submission_id): + return {} + + app = server_module.app + app.state.api_key = None + with patch.object(server_module, "_db_client", EmptyClient()): + with TestClient(app) as c: + resp = c.get("/api/submissions/nonexistent") + assert resp.status_code == 404 + + def test_submission_evaluations(self, client): + resp = client.get("/api/submissions/test-sub-1/evaluations") + assert resp.status_code == 200 + assert isinstance(resp.json(), list) + + def test_submission_code(self, client): + resp = client.get("/api/submissions/test-sub-1/code") + assert resp.status_code == 200 + data = resp.json() + assert "code" in data + + +class TestAuthentication: + """API key authentication.""" + + def test_no_key_returns_401(self, authed_client): + resp = authed_client.get("/health") + assert resp.status_code == 401 + + def test_wrong_key_returns_401(self, authed_client): + resp = authed_client.get("/health", headers={"X-API-Key": "wrong"}) + assert resp.status_code == 401 + + def test_correct_key_returns_200(self, authed_client): + resp = authed_client.get("/health", headers={"X-API-Key": "test-secret-key"}) + assert resp.status_code == 200 diff --git a/tests/test_config.py b/tests/test_config.py new file mode 100644 index 00000000..799c5c80 --- /dev/null +++ b/tests/test_config.py @@ -0,0 +1,104 @@ +"""Tests for configuration loading.""" + +import json +from pathlib import Path + +import pytest +from pydantic import ValidationError + +from crusades.config import HParams, StorageConfig + + +class TestHParamsLoad: + """HParams loading from JSON.""" + + def test_load_from_project_hparams(self): + """Loads real hparams.json from repository.""" + hparams = HParams.load() + assert hparams.netuid == 3 + assert hparams.evaluation_runs >= 1 + assert hparams.eval_timeout > 0 + assert hparams.benchmark_model_name != "" + + def test_load_from_explicit_path(self, tmp_path: Path): + """Loads from an explicit path.""" + data = { + "netuid": 99, + "burn_rate": 0.05, + "burn_uid": 0, + "evaluation_runs": 2, + "eval_steps": 20, + "eval_timeout": 3600, + "benchmark_model_name": "test/model", + "benchmark_dataset_name": "test/data", + "benchmark_dataset_split": "train", + "benchmark_data_samples": 100, + "benchmark_master_seed": 42, + "benchmark_sequence_length": 1024, + "benchmark_batch_size": 16, + "set_weights_interval_blocks": 50, + "reveal_blocks": 10, + "min_blocks_between_commits": 5, + "block_time": 12, + } + path = tmp_path / "hparams.json" + path.write_text(json.dumps(data)) + + hparams = HParams.load(path) + assert hparams.netuid == 99 + assert hparams.burn_rate == 0.05 + + def test_missing_file_raises(self, tmp_path: Path): + with pytest.raises(FileNotFoundError): + HParams.load(tmp_path / "nonexistent.json") + + def test_missing_required_fields_raises(self, tmp_path: Path): + path = tmp_path / "hparams.json" + path.write_text(json.dumps({"netuid": 1})) + with pytest.raises(ValidationError): + HParams.load(path) + + def test_burn_rate_bounds(self, tmp_path: Path): + """burn_rate must be in [0.0, 1.0].""" + data = { + "netuid": 1, + "burn_rate": 1.5, # invalid + "burn_uid": 0, + "evaluation_runs": 2, + "eval_steps": 20, + "eval_timeout": 3600, + "benchmark_model_name": "m", + "benchmark_dataset_name": "d", + "benchmark_dataset_split": "train", + "benchmark_data_samples": 100, + "benchmark_master_seed": 42, + "benchmark_sequence_length": 1024, + "benchmark_batch_size": 16, + "set_weights_interval_blocks": 50, + "reveal_blocks": 10, + "min_blocks_between_commits": 5, + "block_time": 12, + } + path = tmp_path / "hparams.json" + path.write_text(json.dumps(data)) + with pytest.raises(ValidationError): + HParams.load(path) + + +class TestNestedConfigs: + """Nested config defaults.""" + + def test_storage_config_default(self): + sc = StorageConfig() + assert "sqlite" in sc.database_url + + def test_mfu_config_from_hparams(self): + hparams = HParams.load() + assert hparams.mfu.gpu_peak_tflops > 0 + assert hparams.mfu.max_plausible_mfu > hparams.mfu.min_mfu + + def test_adaptive_threshold_config(self): + hparams = HParams.load() + assert hparams.adaptive_threshold.base_threshold > 0 + assert 0 < hparams.adaptive_threshold.decay_percent < 1 + assert hparams.adaptive_threshold.decay_interval_blocks > 0 diff --git a/tests/test_database.py b/tests/test_database.py new file mode 100644 index 00000000..b437b8e5 --- /dev/null +++ b/tests/test_database.py @@ -0,0 +1,349 @@ +"""Tests for async Database operations.""" + +import uuid + +import pytest + +from crusades.core.protocols import SubmissionStatus +from crusades.storage.database import Database +from crusades.storage.models import ( + EvaluationModel, + SubmissionModel, +) + + +class TestInitialize: + """Database initialization and idempotency.""" + + async def test_initialize_creates_tables(self, db: Database): + """Tables exist after initialize.""" + async with db.engine.connect() as conn: + from sqlalchemy import inspect as sa_inspect + + table_names = await conn.run_sync( + lambda sync_conn: sa_inspect(sync_conn).get_table_names() + ) + assert "submissions" in table_names + assert "evaluations" in table_names + assert "validator_state" in table_names + + async def test_initialize_is_idempotent(self, db: Database): + """Calling initialize twice does not raise.""" + await db.initialize() # second call + + +class TestSubmissionCRUD: + """Submission save, get, update operations.""" + + async def test_save_and_get(self, db: Database): + sub = SubmissionModel( + submission_id="test-sub-1", + miner_hotkey="hk1", + miner_uid=1, + code_hash="hash1", + bucket_path="https://example.com/1", + spec_version=19, + ) + await db.save_submission(sub) + + retrieved = await db.get_submission("test-sub-1") + assert retrieved is not None + assert retrieved.miner_hotkey == "hk1" + assert retrieved.status == SubmissionStatus.PENDING + + async def test_get_nonexistent_returns_none(self, db: Database): + result = await db.get_submission("nonexistent") + assert result is None + + async def test_update_status(self, db: Database): + sub = SubmissionModel( + submission_id="test-status", + miner_hotkey="hk1", + miner_uid=1, + code_hash="h", + bucket_path="https://example.com/x", + spec_version=19, + ) + await db.save_submission(sub) + + await db.update_submission_status( + "test-status", + SubmissionStatus.FAILED_VALIDATION, + error_message="bad syntax", + ) + updated = await db.get_submission("test-status") + assert updated.status == SubmissionStatus.FAILED_VALIDATION + assert updated.error_message == "bad syntax" + + async def test_update_score_sets_finished(self, db: Database): + sub = SubmissionModel( + submission_id="test-score", + miner_hotkey="hk1", + miner_uid=1, + code_hash="h", + bucket_path="https://example.com/x", + spec_version=19, + status=SubmissionStatus.EVALUATING, + ) + await db.save_submission(sub) + + await db.update_submission_score("test-score", 72.5) + updated = await db.get_submission("test-score") + assert updated.final_score == 72.5 + assert updated.status == SubmissionStatus.FINISHED + + async def test_update_code_content(self, db: Database): + sub = SubmissionModel( + submission_id="test-code", + miner_hotkey="hk1", + miner_uid=1, + code_hash="h", + bucket_path="https://example.com/x", + spec_version=19, + ) + await db.save_submission(sub) + + await db.update_submission_code("test-code", "def inner_steps(): pass") + code = await db.get_submission_code("test-code") + assert code == "def inner_steps(): pass" + + async def test_get_pending_submissions(self, seeded_db: Database): + pending = await seeded_db.get_pending_submissions(spec_version=19) + assert len(pending) == 1 + assert pending[0].submission_id == "sub-pending-1" + + async def test_get_latest_submission_by_hotkey(self, seeded_db: Database): + latest = await seeded_db.get_latest_submission_by_hotkey("hotkey_alice") + assert latest is not None + assert latest.submission_id == "sub-finished-1" + + async def test_get_latest_submission_by_hotkey_none(self, db: Database): + latest = await db.get_latest_submission_by_hotkey("nonexistent") + assert latest is None + + +class TestEvaluationCRUD: + """Evaluation save and retrieval.""" + + async def test_save_and_get_evaluations(self, db: Database): + sub = SubmissionModel( + submission_id="eval-sub", + miner_hotkey="hk1", + miner_uid=1, + code_hash="h", + bucket_path="https://example.com/x", + spec_version=19, + ) + await db.save_submission(sub) + + ev = EvaluationModel( + evaluation_id=str(uuid.uuid4()), + submission_id="eval-sub", + evaluator_hotkey="val1", + mfu=60.0, + tokens_per_second=1000.0, + total_tokens=20000, + wall_time_seconds=20.0, + success=True, + ) + await db.save_evaluation(ev) + + evals = await db.get_evaluations("eval-sub") + assert len(evals) == 1 + assert evals[0].mfu == 60.0 + + async def test_count_evaluations(self, seeded_db: Database): + count = await seeded_db.count_evaluations("sub-finished-1") + assert count == 1 + + +class TestLeaderboard: + """Leaderboard with threshold logic.""" + + async def test_get_top_submission(self, seeded_db: Database): + top = await seeded_db.get_top_submission(spec_version=19) + assert top is not None + assert top.final_score == 65.5 + + async def test_get_leaderboard_winner_default_threshold(self, seeded_db: Database): + """With default 1% threshold, higher-score submission wins.""" + winner = await seeded_db.get_leaderboard_winner(threshold=0.01, spec_version=19) + assert winner is not None + # sub-finished-2 (55.0) created first, sub-finished-1 (65.5) beats it by >1% + assert winner.submission_id == "sub-finished-1" + + async def test_get_leaderboard_winner_high_threshold(self, seeded_db: Database): + """With very high threshold, first submission holds rank 1.""" + winner = await seeded_db.get_leaderboard_winner(threshold=0.50, spec_version=19) + assert winner is not None + # 65.5 > 55.0 * 1.50 = 82.5? No. So sub-finished-2 stays at #1. + assert winner.submission_id == "sub-finished-2" + + async def test_get_leaderboard_no_submissions(self, db: Database): + winner = await db.get_leaderboard_winner(threshold=0.01) + assert winner is None + + async def test_get_leaderboard_ordered(self, seeded_db: Database): + board = await seeded_db.get_leaderboard(limit=10, spec_version=19, threshold=0.01) + assert len(board) == 2 + # Winner at position 0 + assert board[0].final_score >= board[1].final_score + + +class TestValidatorState: + """Key-value validator state persistence.""" + + async def test_set_and_get(self, db: Database): + await db.set_validator_state("last_block", "12345") + val = await db.get_validator_state("last_block") + assert val == "12345" + + async def test_get_nonexistent(self, db: Database): + val = await db.get_validator_state("missing_key") + assert val is None + + async def test_upsert(self, db: Database): + await db.set_validator_state("key", "v1") + await db.set_validator_state("key", "v2") + val = await db.get_validator_state("key") + assert val == "v2" + + +class TestPaymentVerification: + """Payment recording and double-spend prevention.""" + + async def test_record_and_check(self, db: Database): + sub = SubmissionModel( + submission_id="pay-sub", + miner_hotkey="hk1", + miner_uid=1, + code_hash="h", + bucket_path="https://example.com/x", + spec_version=19, + ) + await db.save_submission(sub) + + await db.record_verified_payment( + submission_id="pay-sub", + miner_hotkey="hk1", + miner_coldkey="ck1", + block_hash="0xabc", + extrinsic_index=0, + amount_rao=100_000_000, + ) + assert await db.is_payment_used("0xabc", 0) is True + assert await db.is_payment_used("0xabc", 1) is False + + async def test_duplicate_payment_raises(self, db: Database): + sub1 = SubmissionModel( + submission_id="pay-dup-1", + miner_hotkey="hk1", + miner_uid=1, + code_hash="h", + bucket_path="https://example.com/x", + spec_version=19, + ) + sub2 = SubmissionModel( + submission_id="pay-dup-2", + miner_hotkey="hk2", + miner_uid=2, + code_hash="h2", + bucket_path="https://example.com/y", + spec_version=19, + ) + await db.save_submission(sub1) + await db.save_submission(sub2) + + await db.record_verified_payment( + submission_id="pay-dup-1", + miner_hotkey="hk1", + miner_coldkey="ck1", + block_hash="0xdup", + extrinsic_index=0, + amount_rao=100_000_000, + ) + with pytest.raises(ValueError, match="already claimed"): + await db.record_verified_payment( + submission_id="pay-dup-2", + miner_hotkey="hk2", + miner_coldkey="ck2", + block_hash="0xdup", + extrinsic_index=0, + amount_rao=100_000_000, + ) + + +class TestAdaptiveThreshold: + """Adaptive threshold decay and update math.""" + + async def test_no_state_returns_base(self, db: Database): + threshold = await db.get_adaptive_threshold( + current_block=1000, + base_threshold=0.01, + ) + assert threshold == 0.01 + + async def test_update_threshold_first_submission(self, db: Database): + new_thresh = await db.update_adaptive_threshold( + new_score=60.0, + old_score=0.0, + current_block=100, + base_threshold=0.01, + ) + # old_score=0 -> improvement=base_threshold -> threshold=base_threshold + assert new_thresh == 0.01 + + async def test_update_threshold_improvement(self, db: Database): + # First leader at 50% + await db.update_adaptive_threshold( + new_score=50.0, old_score=0.0, current_block=100, base_threshold=0.01 + ) + # New leader at 60% -> 20% improvement + new_thresh = await db.update_adaptive_threshold( + new_score=60.0, old_score=50.0, current_block=200, base_threshold=0.01 + ) + expected = (60.0 - 50.0) / 50.0 # 0.20 + assert abs(new_thresh - expected) < 1e-9 + + async def test_threshold_decays_over_blocks(self, db: Database): + # Set a threshold of 20% at block 100 + await db.update_adaptive_threshold( + new_score=60.0, old_score=50.0, current_block=100, base_threshold=0.01 + ) + + # Get threshold at block 200 (100 blocks = 1 decay step at default interval) + decayed = await db.get_adaptive_threshold( + current_block=200, + base_threshold=0.01, + decay_percent=0.05, + decay_interval_blocks=100, + ) + # Expected: 0.01 + (0.20 - 0.01) * (0.95)^1 = 0.01 + 0.1805 = 0.1905 + expected = 0.01 + (0.20 - 0.01) * 0.95 + assert abs(decayed - expected) < 1e-6 + + async def test_threshold_never_below_base(self, db: Database): + await db.update_adaptive_threshold( + new_score=50.1, old_score=50.0, current_block=100, base_threshold=0.01 + ) + # After many decay steps + decayed = await db.get_adaptive_threshold( + current_block=100_000, + base_threshold=0.01, + decay_percent=0.05, + decay_interval_blocks=100, + ) + assert decayed >= 0.01 + + async def test_zero_decay_interval_returns_current(self, db: Database): + await db.update_adaptive_threshold( + new_score=60.0, old_score=50.0, current_block=100, base_threshold=0.01 + ) + result = await db.get_adaptive_threshold( + current_block=200, + base_threshold=0.01, + decay_percent=0.05, + decay_interval_blocks=0, + ) + # With interval=0, returns current_threshold unchanged + assert result == pytest.approx(0.20, abs=1e-6) diff --git a/tests/test_exceptions.py b/tests/test_exceptions.py new file mode 100644 index 00000000..56c542a3 --- /dev/null +++ b/tests/test_exceptions.py @@ -0,0 +1,85 @@ +"""Tests for EvaluationErrorCode classification methods.""" + +import pytest + +from crusades.core.exceptions import EvaluationError, EvaluationErrorCode + +# Every member must be classified by exactly the expected predicates. +# Table-driven: exhaustive coverage of the enum surface. + +VERIFICATION_CODES = { + EvaluationErrorCode.INSUFFICIENT_TRAINABLE_PARAMS, + EvaluationErrorCode.INSUFFICIENT_PARAMS_CHANGED, + EvaluationErrorCode.GRADIENT_COVERAGE_FAILED, + EvaluationErrorCode.GRADIENT_RELATIVE_ERROR_FAILED, + EvaluationErrorCode.LOSS_MISMATCH, + EvaluationErrorCode.TOKEN_COUNT_MISMATCH, + EvaluationErrorCode.NO_GRADIENTS_CAPTURED, + EvaluationErrorCode.MISSING_LOGITS, + EvaluationErrorCode.INVALID_LOGITS_SHAPE, + EvaluationErrorCode.SEQUENCE_TRUNCATION, + EvaluationErrorCode.WEIGHT_MISMATCH, +} + +FATAL_CODES = { + EvaluationErrorCode.NO_CODE, + EvaluationErrorCode.SYNTAX_ERROR, + EvaluationErrorCode.MISSING_INNER_STEPS, + EvaluationErrorCode.INVALID_RETURN_TYPE, + EvaluationErrorCode.INSUFFICIENT_TRAINABLE_PARAMS, + EvaluationErrorCode.NO_GRADIENTS_CAPTURED, + EvaluationErrorCode.MISSING_LOGITS, + EvaluationErrorCode.INVALID_LOGITS_SHAPE, + EvaluationErrorCode.SEQUENCE_TRUNCATION, +} + +INFRASTRUCTURE_CODES = { + EvaluationErrorCode.MODEL_LOAD_FAILED, + EvaluationErrorCode.DATA_LOAD_FAILED, + EvaluationErrorCode.DOCKER_FAILED, + EvaluationErrorCode.TIMEOUT, +} + + +@pytest.mark.parametrize("code", list(EvaluationErrorCode)) +class TestErrorCodeClassification: + """Each error code is tested against all three classifiers.""" + + def test_is_verification_failure(self, code: EvaluationErrorCode): + expected = code in VERIFICATION_CODES + assert EvaluationErrorCode.is_verification_failure(code) is expected, ( + f"{code}: expected is_verification_failure={expected}" + ) + + def test_is_fatal(self, code: EvaluationErrorCode): + expected = code in FATAL_CODES + assert EvaluationErrorCode.is_fatal(code) is expected, ( + f"{code}: expected is_fatal={expected}" + ) + + def test_is_miner_fault(self, code: EvaluationErrorCode): + expected = code not in INFRASTRUCTURE_CODES + assert EvaluationErrorCode.is_miner_fault(code) is expected, ( + f"{code}: expected is_miner_fault={expected}" + ) + + +def test_evaluation_error_carries_code(): + """EvaluationError preserves message and code.""" + err = EvaluationError("boom", EvaluationErrorCode.TIMEOUT) + assert str(err) == "boom" + assert err.code == EvaluationErrorCode.TIMEOUT + assert err.message == "boom" + + +def test_evaluation_error_default_code(): + """EvaluationError defaults to UNKNOWN.""" + err = EvaluationError("unknown failure") + assert err.code == EvaluationErrorCode.UNKNOWN + + +def test_error_code_values_are_strings(): + """All error codes are valid StrEnum members.""" + for code in EvaluationErrorCode: + assert isinstance(code.value, str) + assert code.value == str(code) diff --git a/tests/test_health_monitor.py b/tests/test_health_monitor.py new file mode 100644 index 00000000..1c0a9c3c --- /dev/null +++ b/tests/test_health_monitor.py @@ -0,0 +1,110 @@ +"""Tests for scripts/health_monitor.py.""" + +import json +import sys +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest + +# Add scripts directory to path so we can import health_monitor +sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "scripts")) +import health_monitor + + +class TestEvaluateHealth: + """Table-driven tests for evaluate_health — all four return paths.""" + + @pytest.mark.parametrize( + "data, expected_healthy, expected_reason_contains", + [ + # Error key present → unhealthy + ({"error": "Unreachable: connection refused"}, False, "Unreachable"), + # Status unhealthy → unhealthy + ({"status": "unhealthy"}, False, "DB disconnected"), + # Status degraded → unhealthy with queue info + ( + {"status": "degraded", "queue_depth": 12}, + False, + "0 evaluations", + ), + # Healthy → healthy + ( + {"status": "healthy", "evaluations_1h": 5, "queue_depth": 0}, + True, + "healthy", + ), + ], + ids=["error", "unhealthy", "degraded", "healthy"], + ) + def test_evaluate_health(self, data, expected_healthy, expected_reason_contains): + is_healthy, reason = health_monitor.evaluate_health(data) + assert is_healthy is expected_healthy + assert expected_reason_contains in reason + + +class TestPostDiscord: + """Discord webhook posting.""" + + def test_noop_when_no_webhook_url(self): + """post_discord is a no-op when DISCORD_WEBHOOK_URL is empty.""" + with patch.object(health_monitor, "DISCORD_WEBHOOK_URL", ""): + # Should return without error and without making any network calls + health_monitor.post_discord("test alert") + + def test_posts_embed_to_webhook(self): + """post_discord sends a JSON embed to the configured webhook.""" + fake_url = "https://discord.com/api/webhooks/fake" + with patch.object(health_monitor, "DISCORD_WEBHOOK_URL", fake_url): + with patch.object(health_monitor, "urlopen") as mock_urlopen: + mock_urlopen.return_value.__enter__ = MagicMock() + mock_urlopen.return_value.__exit__ = MagicMock(return_value=False) + + health_monitor.post_discord("validator down", color=0xFF0000) + + mock_urlopen.assert_called_once() + req = mock_urlopen.call_args[0][0] + assert req.full_url == fake_url + body = json.loads(req.data.decode()) + assert body["embeds"][0]["description"] == "validator down" + assert body["embeds"][0]["color"] == 0xFF0000 + + +class TestProbeHealth: + """probe_health I/O logic.""" + + def test_returns_parsed_json_on_success(self): + """Successful probe returns parsed JSON from the endpoint.""" + fake_body = json.dumps({"status": "healthy", "evaluations_1h": 5}).encode() + mock_resp = MagicMock() + mock_resp.status = 200 + mock_resp.read.return_value = fake_body + mock_resp.__enter__ = MagicMock(return_value=mock_resp) + mock_resp.__exit__ = MagicMock(return_value=False) + + with patch.object(health_monitor, "urlopen", return_value=mock_resp): + result = health_monitor.probe_health() + + assert result == {"status": "healthy", "evaluations_1h": 5} + + def test_returns_error_on_url_error(self): + """URLError produces a dict with 'error' key.""" + from urllib.error import URLError + + with patch.object( + health_monitor, "urlopen", side_effect=URLError("connection refused") + ): + result = health_monitor.probe_health() + + assert "error" in result + assert "Unreachable" in result["error"] + + def test_returns_error_on_generic_exception(self): + """Any other exception produces a dict with 'error' key.""" + with patch.object( + health_monitor, "urlopen", side_effect=ValueError("bad data") + ): + result = health_monitor.probe_health() + + assert "error" in result + assert "bad data" in result["error"] diff --git a/tests/test_models.py b/tests/test_models.py new file mode 100644 index 00000000..18dd122c --- /dev/null +++ b/tests/test_models.py @@ -0,0 +1,110 @@ +"""Tests for SQLAlchemy ORM models and SubmissionStatus enum. + +Note: SQLAlchemy `default` callables only fire on flush/commit, not on +construction. Tests that need defaults use the database fixture. +""" + +import uuid + +from crusades.core.protocols import SubmissionStatus +from crusades.storage.database import Database +from crusades.storage.models import ( + Base, + EvaluationModel, + SubmissionModel, +) + + +class TestSubmissionStatus: + """SubmissionStatus enum values match expected strings.""" + + def test_all_statuses_defined(self): + expected = { + "pending", "validating", "evaluating", + "finished", "failed_validation", "failed_evaluation", "error", + } + actual = {s.value for s in SubmissionStatus} + assert actual == expected + + def test_str_enum_behavior(self): + assert str(SubmissionStatus.PENDING) == "pending" + assert SubmissionStatus.FINISHED == "finished" + + +class TestSubmissionModelViaDB: + """Submission model defaults verified through DB round-trip.""" + + async def test_default_status_is_pending(self, db: Database): + sub = SubmissionModel( + submission_id="defaults-test", + miner_hotkey="hk", + miner_uid=1, + code_hash="abc", + bucket_path="https://example.com/train.py", + spec_version=19, + ) + await db.save_submission(sub) + retrieved = await db.get_submission("defaults-test") + assert retrieved.status == SubmissionStatus.PENDING + + async def test_default_payment_verified_false(self, db: Database): + sub = SubmissionModel( + submission_id="pay-defaults", + miner_hotkey="hk", + miner_uid=1, + code_hash="abc", + bucket_path="https://example.com/train.py", + spec_version=19, + ) + await db.save_submission(sub) + retrieved = await db.get_submission("pay-defaults") + assert retrieved.payment_verified is False + + async def test_optional_fields_are_none(self, db: Database): + sub = SubmissionModel( + submission_id="opt-fields", + miner_hotkey="hk", + miner_uid=1, + code_hash="abc", + bucket_path="https://example.com/train.py", + spec_version=19, + ) + await db.save_submission(sub) + retrieved = await db.get_submission("opt-fields") + assert retrieved.final_score is None + assert retrieved.error_message is None + assert retrieved.code_content is None + + async def test_evaluation_mfu_default(self, db: Database): + sub = SubmissionModel( + submission_id="eval-mfu-test", + miner_hotkey="hk", + miner_uid=1, + code_hash="abc", + bucket_path="https://example.com/train.py", + spec_version=19, + ) + await db.save_submission(sub) + + ev = EvaluationModel( + evaluation_id=str(uuid.uuid4()), + submission_id="eval-mfu-test", + evaluator_hotkey="val-1", + tokens_per_second=100.0, + total_tokens=1000, + wall_time_seconds=10.0, + success=True, + ) + await db.save_evaluation(ev) + evals = await db.get_evaluations("eval-mfu-test") + assert evals[0].mfu == 0.0 + + +class TestMetadataTableNames: + """All models register their expected table names.""" + + def test_table_names(self): + table_names = {t.name for t in Base.metadata.sorted_tables} + expected = {"submissions", "evaluations", "validator_state", + "adaptive_threshold", "verified_payments"} + assert expected.issubset(table_names) diff --git a/uv.lock b/uv.lock index 7911780b..ea0a25aa 100644 --- a/uv.lock +++ b/uv.lock @@ -2305,6 +2305,18 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/ee/49/1377b49de7d0c1ce41292161ea0f721913fa8722c19fb9c1e3aa0367eecb/pytest_cov-7.0.0-py3-none-any.whl", hash = "sha256:3b8e9558b16cc1479da72058bdecf8073661c7f57f7d3c5f22a1c23507f2d861", size = 22424, upload-time = "2025-09-09T10:57:00.695Z" }, ] +[[package]] +name = "pytest-mock" +version = "3.15.1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "pytest" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/68/14/eb014d26be205d38ad5ad20d9a80f7d201472e08167f0bb4361e251084a9/pytest_mock-3.15.1.tar.gz", hash = "sha256:1849a238f6f396da19762269de72cb1814ab44416fa73a8686deac10b0d87a0f", size = 34036, upload-time = "2025-09-16T16:37:27.081Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/5a/cc/06253936f4a7fa2e0f48dfe6d851d9c56df896a9ab09ac019d70b760619c/pytest_mock-3.15.1-py3-none-any.whl", hash = "sha256:0a25e2eb88fe5168d535041d09a4529a188176ae608a6d249ee65abc0949630d", size = 10095, upload-time = "2025-09-16T16:37:25.734Z" }, +] + [[package]] name = "python-dateutil" version = "2.9.0.post0" @@ -2809,6 +2821,7 @@ dev = [ { name = "pytest" }, { name = "pytest-asyncio" }, { name = "pytest-cov" }, + { name = "pytest-mock" }, { name = "ruff" }, ] @@ -2830,6 +2843,7 @@ requires-dist = [ { name = "pytest", marker = "extra == 'dev'", specifier = ">=7.4.0" }, { name = "pytest-asyncio", marker = "extra == 'dev'", specifier = ">=0.23.0" }, { name = "pytest-cov", marker = "extra == 'dev'", specifier = ">=4.1.0" }, + { name = "pytest-mock", marker = "extra == 'dev'", specifier = ">=3.12.0" }, { name = "python-logging-loki", specifier = ">=0.3.1" }, { name = "rich", specifier = ">=13.0.0" }, { name = "ruff", marker = "extra == 'dev'", specifier = ">=0.1.0" },