195 lines
8.2 KiB
Python
195 lines
8.2 KiB
Python
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: (_ for _ in ()).throw(RuntimeError("ATH timeout")))
|
|
monkeypatch.setattr(server.price, "fetch_historical", lambda: (_ for _ in ()).throw(RuntimeError("history timeout")))
|
|
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_onchain_sources_fail_independently(server, monkeypatch):
|
|
fake_lib = types.ModuleType("scrapers.lookintobitcoin")
|
|
setattr(fake_lib, "scrape_all", lambda: (_ for _ in ()).throw(RuntimeError("LIB down")))
|
|
fake_coc = types.ModuleType("scrapers.checkonchain")
|
|
setattr(fake_coc, "scrape_all", lambda: {"sopr": {"value": 0.99}})
|
|
monkeypatch.setitem(sys.modules, "scrapers.lookintobitcoin", fake_lib)
|
|
monkeypatch.setitem(sys.modules, "scrapers.checkonchain", fake_coc)
|
|
|
|
observations, errors, successful_sources = server._scrape_onchain_sources()
|
|
|
|
assert observations["sopr"]["value"] == 0.99
|
|
assert successful_sources == 1
|
|
assert any("LookIntoBitcoin" in error for error in errors)
|
|
|
|
|
|
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"]
|