7d9e209ffb
Move manhuang/shenzhou off the default hot path (side cache + bg refresh), add collect_timing and infra/runtime/flywheel pulse layers, fix clone zsh→sleep mapping, watch single-instance lock, SPA lite/omit/section load path with gzip JSON, and bump dashboard schema to v2.
143 lines
4.6 KiB
Python
143 lines
4.6 KiB
Python
"""dashboard 性能路径：旁路缓存、section 映射、紧凑 JSON、SSH 闸常量。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
|
||
SRC = Path(__file__).parent.parent / "src"
|
||
sys.path.insert(0, str(SRC))
|
||
|
||
|
||
def test_json_dump_compact():
|
||
from joy.dashboard.collect import _json_dump
|
||
|
||
s = _json_dump({"a": 1, "b": [2, 3]})
|
||
assert "\n" not in s
|
||
assert s == '{"a":1,"b":[2,3]}'
|
||
assert json.loads(s) == {"a": 1, "b": [2, 3]}
|
||
pretty = json.dumps({"a": 1, "b": [2, 3]}, indent=2)
|
||
assert len(s) < len(pretty)
|
||
|
||
|
||
def test_manhuang_side_cache_roundtrip(tmp_path, monkeypatch):
|
||
from joy.dashboard import collect as c
|
||
|
||
side = tmp_path / "mh.json"
|
||
monkeypatch.setattr(c, "MANHUANG_CACHE_FILE", str(side))
|
||
monkeypatch.setattr(c, "MANHUANG_TTL_SEC", 600)
|
||
|
||
assert c._load_manhuang_side() is None
|
||
assert c.manhuang_needs_refresh() is True
|
||
|
||
c._save_manhuang_side({"ok": True, "pulse": {"x": 1}})
|
||
loaded = c._load_manhuang_side()
|
||
assert loaded is not None
|
||
assert loaded["ok"] is True
|
||
assert loaded.get("_source") == "side_cache"
|
||
assert c.manhuang_needs_refresh() is False
|
||
|
||
|
||
def test_include_manhuang_env(monkeypatch):
|
||
from joy.dashboard.collect import _include_manhuang_inline
|
||
|
||
monkeypatch.delenv("JOY_DASHBOARD_INCLUDE_MANHUANG", raising=False)
|
||
assert _include_manhuang_inline(None) is False
|
||
assert _include_manhuang_inline(True) is True
|
||
assert _include_manhuang_inline(False) is False
|
||
monkeypatch.setenv("JOY_DASHBOARD_INCLUDE_MANHUANG", "1")
|
||
assert _include_manhuang_inline(None) is True
|
||
|
||
|
||
def test_dashboard_sections_map():
|
||
from joy.commands.web import DASHBOARD_SECTIONS, _dashboard_core_subset
|
||
|
||
assert "manhuang" in DASHBOARD_SECTIONS
|
||
assert "shenzhou" in DASHBOARD_SECTIONS
|
||
core = _dashboard_core_subset({
|
||
"ts": "t", "pulse": {"status": "healthy"}, "manhuang": {"huge": True}, "vms": {},
|
||
})
|
||
assert "pulse" in core
|
||
assert "manhuang" not in core
|
||
assert "vms" not in core
|
||
|
||
|
||
def test_ticket_ttl_reduced():
|
||
from joy.dashboard.shenzhou import TICKET_TTL_SEC
|
||
|
||
assert TICKET_TTL_SEC <= 120
|
||
assert TICKET_TTL_SEC >= 30
|
||
|
||
|
||
def test_ssh_concurrency_positive():
|
||
from joy.dashboard.collect import _SSH_CONCURRENCY
|
||
|
||
assert 1 <= _SSH_CONCURRENCY <= 64
|
||
|
||
|
||
def test_vm_layer_colors_infra_green_even_if_journal_stale():
|
||
from joy.dashboard.collect import _vm_layer_colors, _composite_health_grid
|
||
|
||
d = {
|
||
"reachable": True,
|
||
"work_session": "work",
|
||
"clone": "pi",
|
||
"journal": {"age_hours": 200},
|
||
}
|
||
layers = _vm_layer_colors(d)
|
||
assert layers["infra"] == "green"
|
||
assert layers["runtime"] == "green"
|
||
assert layers["flywheel"] == "yellow"
|
||
# 卡片总色：journal 债不单独染黄
|
||
assert _composite_health_grid({"devops": layers})["devops"] == "green"
|
||
|
||
|
||
def test_vm_layer_offline_red():
|
||
from joy.dashboard.collect import _vm_layer_colors
|
||
|
||
layers = _vm_layer_colors({"reachable": False, "work_session": "OFFLINE"})
|
||
assert layers["infra"] == "red"
|
||
assert layers["runtime"] == "red"
|
||
|
||
|
||
def test_side_stub_and_shenzhou_include(monkeypatch):
|
||
from joy.dashboard.collect import _side_stub, _is_side_stub, _include_shenzhou_inline
|
||
|
||
stub = _side_stub("shenzhou", {"ok": True, "view": "x"})
|
||
assert _is_side_stub(stub)
|
||
assert not _is_side_stub({"ok": True, "view": "x", "hosts": [], "tickets": {}, "a": 1, "b": 2, "c": 3, "d": 4, "e": 5})
|
||
monkeypatch.delenv("JOY_DASHBOARD_INCLUDE_SHENZHOU", raising=False)
|
||
assert _include_shenzhou_inline(None) is False
|
||
monkeypatch.setenv("JOY_DASHBOARD_INCLUDE_SHENZHOU", "1")
|
||
assert _include_shenzhou_inline(None) is True
|
||
|
||
|
||
def test_map_clone_cmd_zsh_is_sleep():
|
||
from joy.dashboard.collect import _map_clone_cmd
|
||
|
||
assert _map_clone_cmd("node") == "pi"
|
||
assert _map_clone_cmd("bash") == "sleep"
|
||
assert _map_clone_cmd("zsh") == "sleep"
|
||
assert _map_clone_cmd("sh") == "sleep"
|
||
assert _map_clone_cmd("") == "—"
|
||
assert _map_clone_cmd("python3") == "python3"
|
||
|
||
|
||
def test_watch_lock_exclusive(tmp_path, monkeypatch):
|
||
from joy.dashboard import collect as c
|
||
|
||
lock = tmp_path / "watch.lock"
|
||
monkeypatch.setattr(c, "WATCH_LOCK_FILE", str(lock))
|
||
fp1, ok1, msg1 = c.acquire_watch_lock()
|
||
assert ok1, msg1
|
||
fp2, ok2, msg2 = c.acquire_watch_lock()
|
||
assert ok2 is False
|
||
assert "pid" in msg2 or "lock" in msg2.lower()
|
||
c.release_watch_lock(fp1)
|
||
# 释放后应能再拿（文件可能仍在，但 flock 已解）
|
||
fp3, ok3, msg3 = c.acquire_watch_lock()
|
||
assert ok3, msg3
|
||
c.release_watch_lock(fp3)
|
