import importlib import json import sys import types from datetime import datetime, timedelta, timezone import pytest @pytest.fixture def server(monkeypatch): started = [] monkeypatch.setattr("threading.Thread.start", lambda self: started.append(self)) sys.modules.pop("dashboard.server", None) module = importlib.import_module("dashboard.server") module._threads_started_during_import = started return module def test_server_import_does_not_start_scheduler_threads(server): assert server._threads_started_during_import == [] def test_frontend_uses_backtest_metric_values_contract(server): assert ".filter(d => d.metric_values && d.metric_values[metricKey] != null)" in server.DASHBOARD_HTML assert ".map(d => ({ date: d.date, value: d.metric_values[metricKey]" in server.DASHBOARD_HTML def test_dashboard_does_not_issue_duplicate_initial_backtest_request(server): assert "const br = await fetch('/api/backtest');" not in server.DASHBOARD_HTML assert server.DASHBOARD_HTML.count("fetch('/api/backtest?mode=' + currentMode)") == 1 def test_health_endpoints_distinguish_process_liveness_from_data_readiness(server, monkeypatch): assert server.health_live() == {"status": "ok"} monkeypatch.setattr(server, "load_cache", lambda: {}) unavailable = server.health_ready() assert unavailable.status_code == 503 monkeypatch.setattr( server, "load_cache", lambda: {"_scored": {"composite_score": 72, "scored_count": 8}}, ) assert server.health_ready() == { "status": "ready", "score": 72, "scored_metrics": 8, } def test_server_cache_and_history_use_reliable_persistence(server, monkeypatch, tmp_path): monkeypatch.setattr(server, "CACHE_PATH", str(tmp_path / "cache.json")) monkeypatch.setattr(server, "HISTORY_PATH", str(tmp_path / "scores.jsonl")) server.save_cache({"metric": {"value": 1}}) server.append_history({ "composite_score": 50, "scored_count": 1, "metrics": [{"key": "metric", "score": 5, "value": 1}], }) server.append_history({ "composite_score": 60, "scored_count": 1, "metrics": [{"key": "metric", "score": 6, "value": 2}], }) assert server.load_cache() == {"metric": {"value": 1}} assert len(server.load_history()) == 1 def test_partial_fast_scrape_preserves_last_known_good_metric(server, monkeypatch, tmp_path): cache_path = tmp_path / "cache.json" history_path = tmp_path / "scores.jsonl" now = datetime.now(timezone.utc) cache_path.write_text(json.dumps({ "price": { "price": 65000, "observed_at": (now - timedelta(minutes=15)).isoformat(), "source": "coingecko", "stale": False, "last_error": None, }, "puell_multiple": {"value": 1.2}, "_onchain_timestamp": now.isoformat(), })) monkeypatch.setattr(server, "CACHE_PATH", str(cache_path)) monkeypatch.setattr(server, "HISTORY_PATH", str(history_path)) monkeypatch.setattr(server.fear_greed, "fetch", lambda: {"value": 25}) monkeypatch.setattr(server.price, "fetch_current", lambda: (_ for _ in ()).throw(RuntimeError("price timeout"))) monkeypatch.setattr(server.price, "fetch_ath", lambda: {"ath": 70000}) monkeypatch.setattr(server.price, "fetch_historical", lambda: []) monkeypatch.setattr(server.engine, "score_all", lambda metrics: {"composite_score": 50, "scored_count": 1, "metrics": []}) monkeypatch.setattr(server.engine, "score_all_ml", lambda metrics: {"composite_score": 50, "scored_count": 1, "metrics": []}) fake_updater = types.ModuleType("scrapers.history_updater") fake_updater.update_history = lambda: None monkeypatch.setitem(sys.modules, "scrapers.history_updater", fake_updater) server.run_scrape() saved = json.loads(cache_path.read_text()) assert saved["price"]["price"] == 65000 assert saved["price"]["stale"] is True assert "price timeout" in saved["price"]["last_error"] assert saved["fear_greed"]["value"] == 25 assert saved["fear_greed"]["stale"] is False assert saved["fear_greed"]["source"] == "alternative.me" def test_expired_onchain_timestamp_triggers_real_refresh(server, monkeypatch, tmp_path): old = datetime.now(timezone.utc) - timedelta(hours=7) cache_path = tmp_path / "cache.json" cache_path.write_text(json.dumps({ "puell_multiple": {"value": 1.2}, "_onchain_timestamp": old.isoformat(), })) monkeypatch.setattr(server, "CACHE_PATH", str(cache_path)) monkeypatch.setattr(server, "HISTORY_PATH", str(tmp_path / "scores.jsonl")) monkeypatch.setattr(server.fear_greed, "fetch", lambda: {"value": 25}) monkeypatch.setattr(server.price, "fetch_current", lambda: {"price": 65000}) monkeypatch.setattr(server.price, "fetch_ath", lambda: {"ath": 70000}) monkeypatch.setattr(server.price, "fetch_historical", lambda: []) monkeypatch.setattr(server.engine, "score_all", lambda metrics: {"composite_score": 50, "scored_count": 1, "metrics": []}) monkeypatch.setattr(server.engine, "score_all_ml", lambda metrics: {"composite_score": 50, "scored_count": 1, "metrics": []}) calls = [] fake_lib = types.ModuleType("scrapers.lookintobitcoin") fake_lib.scrape_all = lambda: calls.append("lib") or {"puell_multiple": {"value": 0.8}} fake_coc = types.ModuleType("scrapers.checkonchain") fake_coc.scrape_all = lambda: calls.append("coc") or {"sopr": {"value": 0.99}} fake_updater = types.ModuleType("scrapers.history_updater") fake_updater.update_history = lambda: None import scrapers monkeypatch.setattr(scrapers, "lookintobitcoin", fake_lib, raising=False) monkeypatch.setattr(scrapers, "checkonchain", fake_coc, raising=False) monkeypatch.setitem(sys.modules, "scrapers.lookintobitcoin", fake_lib) monkeypatch.setitem(sys.modules, "scrapers.checkonchain", fake_coc) monkeypatch.setitem(sys.modules, "scrapers.history_updater", fake_updater) server.run_scrape() assert calls == ["lib", "coc"] saved = json.loads(cache_path.read_text()) assert saved["puell_multiple"]["value"] == 0.8 assert saved["puell_multiple"]["source"] == "lookintobitcoin" assert saved["sopr"]["source"] == "checkonchain" assert saved["_onchain_timestamp"] != old.isoformat() def test_refresh_job_is_reserved_before_thread_start(server, monkeypatch, tmp_path): from dashboard.jobs import JobRegistry registry = JobRegistry(tmp_path / "jobs.json") monkeypatch.setattr(server, "_jobs", registry, raising=False) monkeypatch.setattr(server, "_scraper_running", False) started = server.api_refresh(full=False) duplicate = server.api_refresh(full=False) assert started["job_id"] assert started["status"] == "queued" assert registry.get(started["job_id"])["status"] == "queued" assert duplicate.status_code == 409 def test_history_collection_has_job_id_and_job_scoped_progress(server, monkeypatch, tmp_path): from dashboard.jobs import JobRegistry registry = JobRegistry(tmp_path / "jobs.json") monkeypatch.setattr(server, "_jobs", registry, raising=False) started = server.api_backtest_collect() job = registry.get(started["job_id"]) assert started["status"] == "queued" assert job["kind"] == "history" assert job["progress"] == {"status": "starting", "current": "", "step": 0, "total": 0} assert server.api_job_status(started["job_id"])["id"] == started["job_id"]