Newer
Older
TIASJudgeModel / tests / remote / test_report.py
"""GPU 一覧の表示のテスト(SPEC_03「表示」).

各 GPU の状態・使い方の型と使用率の平均/最大・計算プロセスごとの所有ユーザー(自分は「自分」)・
経過時間・SM 使用率の最大が表示されること.
"""

from __future__ import annotations

from probe_helpers import PMON_HEADER, gpu_line, probe_text, sample_lines

from tias_judge_model.remote.config import GpuFreeThreshold, ServerConfig
from tias_judge_model.remote.gpu import ServerProbe, parse_probe_output, select_gpu
from tias_judge_model.remote.report import format_server_report, format_shared_warning

THRESHOLD = GpuFreeThreshold(
    max_memory_used_mib=1024, max_utilization_percent=80, observe_sec=10, sample_interval_ms=500
)
UUID0 = "GPU-0000"
SERVER = ServerConfig(
    name="Newton",
    address="192.168.23.14",
    user="hasegawa",
    priority=1,
    remote_dir="~/work/TIASJudgeModel",
    data_dir="~/data/TIASJudgeModel",
    data_ready=True,
)


def _probe() -> ServerProbe:
    text = probe_text(
        [gpu_line(0, UUID0, used=500, util=0)],
        # 平均 21%・最大 74%(断続的に使用)
        samples=sample_lines(0, [0, 74, 10, 0], memory=500),
        pmon=[
            *PMON_HEADER,
            "    0        100     C     74      1      -      -      -      -    python",
            "    0        200     C      5      1      -      -      -      -    python",
        ],
        apps=[f"{UUID0}, 100, 250", f"{UUID0}, 200, 250"],
        owners=["100 hasegawa 99720", "200 ando 45"],
    )
    return ServerProbe(SERVER, parse_probe_output(text, THRESHOLD))


def test_report_shows_state_pattern_users_elapsed_and_sm() -> None:
    report = format_server_report([_probe()], show_jobs=False)
    assert "Newton" in report
    assert "空き(利用者あり)" in report
    assert "断続的に使用" in report
    assert "平均 21%" in report
    assert "最大 74%" in report
    assert "自分" in report
    assert "ando" in report
    assert "27.7 時間" in report
    assert "SM 最大 74%" in report


def test_report_marks_unreachable_server() -> None:
    probe = ServerProbe(SERVER, None, "Connection timed out")
    report = format_server_report([probe], show_jobs=True)
    assert "Newton" in report
    assert "接続不可" in report


def test_shared_selection_warns_with_user_names() -> None:
    selection = select_gpu([_probe()])
    assert selection is not None
    warning = format_shared_warning(selection)
    assert warning is not None
    assert "ando" in warning