100 lines
3.2 KiB
Python
100 lines
3.2 KiB
Python
import json
|
|
import threading
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
import pytest
|
|
|
|
from dashboard.persistence import (
|
|
append_daily_jsonl,
|
|
atomic_write_json,
|
|
load_jsonl_tail,
|
|
merge_observation,
|
|
onchain_refresh_due,
|
|
)
|
|
|
|
|
|
def test_atomic_write_json_remains_readable_under_concurrent_writers(tmp_path):
|
|
path = tmp_path / "cache.json"
|
|
|
|
threads = [
|
|
threading.Thread(target=atomic_write_json, args=(path, {"writer": i, "values": list(range(100))}))
|
|
for i in range(12)
|
|
]
|
|
for thread in threads:
|
|
thread.start()
|
|
for thread in threads:
|
|
thread.join()
|
|
|
|
saved = json.loads(path.read_text())
|
|
assert saved["writer"] in range(12)
|
|
assert saved["values"] == list(range(100))
|
|
assert not list(tmp_path.glob(".cache.json.*.tmp"))
|
|
|
|
|
|
def test_append_daily_jsonl_writes_at_most_one_entry_per_utc_day(tmp_path):
|
|
path = tmp_path / "scores.jsonl"
|
|
first = {"timestamp": "2026-07-26T01:00:00+00:00", "score": 10}
|
|
duplicate_day = {"timestamp": "2026-07-26T23:59:00+00:00", "score": 20}
|
|
next_day = {"timestamp": "2026-07-27T00:01:00+00:00", "score": 30}
|
|
|
|
assert append_daily_jsonl(path, first) is True
|
|
assert append_daily_jsonl(path, duplicate_day) is False
|
|
assert append_daily_jsonl(path, next_day) is True
|
|
|
|
assert load_jsonl_tail(path, limit=90) == [first, next_day]
|
|
|
|
|
|
def test_load_jsonl_tail_is_bounded_and_ignores_malformed_lines(tmp_path):
|
|
path = tmp_path / "scores.jsonl"
|
|
path.write_text("".join(json.dumps({"n": i}) + "\n" for i in range(200)) + "partial{")
|
|
|
|
assert load_jsonl_tail(path, limit=3, chunk_size=64) == [{"n": 197}, {"n": 198}, {"n": 199}]
|
|
|
|
|
|
def test_merge_observation_preserves_last_known_good_with_stale_metadata():
|
|
old = {
|
|
"value": 1.25,
|
|
"observed_at": "2026-07-25T12:00:00+00:00",
|
|
"source": "lookintobitcoin",
|
|
"stale": False,
|
|
"last_error": None,
|
|
}
|
|
|
|
merged = merge_observation(old, None, source="lookintobitcoin", error="timeout")
|
|
|
|
assert merged == {
|
|
"value": 1.25,
|
|
"observed_at": "2026-07-25T12:00:00+00:00",
|
|
"source": "lookintobitcoin",
|
|
"stale": True,
|
|
"last_error": "timeout",
|
|
}
|
|
|
|
|
|
def test_merge_observation_records_metadata_for_fresh_value():
|
|
observed_at = "2026-07-26T12:00:00+00:00"
|
|
|
|
merged = merge_observation(
|
|
{"value": 1.0}, {"value": 2.0, "trend": "up"},
|
|
source="checkonchain", observed_at=observed_at,
|
|
)
|
|
|
|
assert merged["value"] == 2.0
|
|
assert merged["trend"] == "up"
|
|
assert merged["observed_at"] == observed_at
|
|
assert merged["source"] == "checkonchain"
|
|
assert merged["stale"] is False
|
|
assert merged["last_error"] is None
|
|
|
|
|
|
@pytest.mark.parametrize("timestamp", [None, "", "not-a-time"])
|
|
def test_onchain_refresh_due_when_timestamp_is_missing_or_invalid(timestamp):
|
|
assert onchain_refresh_due(timestamp, now=datetime(2026, 7, 26, tzinfo=timezone.utc)) is True
|
|
|
|
|
|
def test_onchain_refresh_due_after_ttl():
|
|
now = datetime(2026, 7, 26, 12, tzinfo=timezone.utc)
|
|
|
|
assert onchain_refresh_due((now - timedelta(hours=5)).isoformat(), now=now, ttl_seconds=21600) is False
|
|
assert onchain_refresh_due((now - timedelta(hours=7)).isoformat(), now=now, ttl_seconds=21600) is True
|