jiangchang-platform-kit/tests/test_runtime_diagnostics.py

# -*- coding: utf-8 -*-
"""runtime_diagnostics 公共 Runtime / media-assets 诊断测试。"""
from __future__ import annotations

import json
from pathlib import Path
from unittest.mock import patch

import pytest

import jiangchang_skill_core
from jiangchang_skill_core import media_assets as ma
from jiangchang_skill_core.runtime_diagnostics import (
    RuntimeDiagnostics,
    RuntimeIssue,
    collect_runtime_diagnostics,
    format_runtime_health_lines,
    is_jiangchang_skill_core_from_skill_tree,
    runtime_diagnostics_dict,
    version_ge,
)


def test_version_ge_post_release() -> None:
    assert version_ge("1.0.10.post23", "1.0.10") is True
    assert version_ge("1.0.10", "1.0.10") is True
    assert version_ge("1.0.9.post19", "1.0.10") is False
    assert version_ge("1.0.10.post23", "1.0.10.post21") is True
    assert version_ge("1.0.10.post21", "1.0.10.post23") is False
    assert version_ge("1.0.10", "1.0.10.post1") is False
    assert version_ge("1.0.11", "1.0.10.post99") is True
    assert version_ge("1.0.11", "1.0.10") is True


def test_is_jiangchang_skill_core_from_skill_tree(tmp_path: Path) -> None:
    skill_root = tmp_path / "my-skill"
    inside = skill_root / "scripts" / "jiangchang_skill_core" / "__init__.py"
    inside.parent.mkdir(parents=True)
    inside.write_text("# vendored\n", encoding="utf-8")
    outside = tmp_path / "site-packages" / "jiangchang_skill_core" / "__init__.py"
    outside.parent.mkdir(parents=True)
    outside.write_text("# shared\n", encoding="utf-8")

    assert is_jiangchang_skill_core_from_skill_tree(
        skill_root=skill_root,
        core_file=inside,
    ) is True
    assert is_jiangchang_skill_core_from_skill_tree(
        skill_root=skill_root,
        core_file=outside,
    ) is False


def test_collect_runtime_diagnostics_json_serializable(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    data_root = tmp_path / "data"
    env = {
        "CLAW_DATA_ROOT": str(data_root),
        "OPENCLAW_RECORD_VIDEO": "0",
    }
    media_root = data_root / "shared" / "media-assets"
    (media_root / "music").mkdir(parents=True)

    def _fake_probe_media_assets(data_root_arg=None, env_arg=None):
        return ma.MediaAssetsStatus(
            root=media_root,
            exists=True,
            ready=False,
            source="local",
            warnings=["ffmpeg_missing"],
            ffmpeg_path=None,
            music_dir=media_root / "music",
            music_ready=False,
        )

    monkeypatch.setattr(
        "jiangchang_skill_core.runtime_diagnostics.probe_media_assets",
        _fake_probe_media_assets,
    )
    monkeypatch.setattr(
        "jiangchang_skill_core.runtime_diagnostics.probe_background_music",
        lambda *a, **k: {
            "music_root": str(media_root / "music"),
            "mp3_count": 0,
            "usable_count": 0,
            "sample_path": None,
            "issue": "background_music_dir_empty",
        },
    )
    monkeypatch.setattr(
        "jiangchang_skill_core.runtime_diagnostics._platform_kit_version",
        lambda: "1.0.10",
    )

    diag = collect_runtime_diagnostics(
        skill_slug="my-skill",
        platform_kit_min_version="1.0.9",
        env=env,
    )
    payload = runtime_diagnostics_dict(diag)
    json.dumps(payload)
    assert payload["skill_slug"] == "my-skill"
    assert payload["resolved_data_root"] == str(data_root)
    assert payload["platform_kit_version_ok"] is True


def test_platform_kit_min_version_none_skips_version_issue(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    env = {"CLAW_DATA_ROOT": str(tmp_path / "data")}
    monkeypatch.setattr(
        "jiangchang_skill_core.runtime_diagnostics.probe_media_assets",
        lambda *a, **k: ma.MediaAssetsStatus(
            root=tmp_path,
            exists=False,
            ready=False,
            source="local",
        ),
    )
    monkeypatch.setattr(
        "jiangchang_skill_core.runtime_diagnostics.probe_background_music",
        lambda *a, **k: {
            "music_root": None,
            "mp3_count": 0,
            "usable_count": 0,
            "sample_path": None,
            "issue": "music_dir_missing",
        },
    )
    monkeypatch.setattr(
        "jiangchang_skill_core.runtime_diagnostics._platform_kit_version",
        lambda: "1.0.8",
    )

    diag = collect_runtime_diagnostics(skill_slug="my-skill", env=env)
    assert diag.platform_kit_version_ok is None
    assert "platform_kit_version_low" not in diag.issue_codes()


def test_platform_kit_missing_records_issue(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    env = {"CLAW_DATA_ROOT": str(tmp_path / "data")}
    monkeypatch.setattr(
        "jiangchang_skill_core.runtime_diagnostics.probe_media_assets",
        lambda *a, **k: ma.MediaAssetsStatus(
            root=tmp_path,
            exists=False,
            ready=False,
            source="local",
        ),
    )
    monkeypatch.setattr(
        "jiangchang_skill_core.runtime_diagnostics.probe_background_music",
        lambda *a, **k: {
            "music_root": None,
            "mp3_count": 0,
            "usable_count": 0,
            "sample_path": None,
            "issue": None,
        },
    )
    monkeypatch.setattr(
        "jiangchang_skill_core.runtime_diagnostics._platform_kit_version",
        lambda: None,
    )

    diag = collect_runtime_diagnostics(
        skill_slug="my-skill",
        platform_kit_min_version="1.0.10",
        env=env,
    )
    assert "platform_kit_not_installed" in diag.issue_codes()


def test_record_video_ffmpeg_and_music_warnings_not_fatal(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    media_root = tmp_path / "shared" / "media-assets"
    env = {"CLAW_DATA_ROOT": str(tmp_path), "OPENCLAW_RECORD_VIDEO": "1"}
    monkeypatch.setattr(
        "jiangchang_skill_core.runtime_diagnostics.probe_media_assets",
        lambda *a, **k: ma.MediaAssetsStatus(
            root=media_root,
            exists=True,
            ready=False,
            source="local",
            warnings=["ffmpeg_missing", "music_dir_missing"],
            ffmpeg_path=None,
            music_dir=None,
        ),
    )
    monkeypatch.setattr(
        "jiangchang_skill_core.runtime_diagnostics.probe_background_music",
        lambda *a, **k: {
            "music_root": None,
            "mp3_count": 0,
            "usable_count": 0,
            "sample_path": None,
            "issue": "music_dir_missing",
        },
    )
    monkeypatch.setattr(
        "jiangchang_skill_core.runtime_diagnostics._platform_kit_version",
        lambda: "1.0.10",
    )

    diag = collect_runtime_diagnostics(skill_slug="my-skill", env=env)
    codes = diag.issue_codes()
    assert "ffmpeg_unavailable" in codes
    assert "background_music_unavailable" in codes
    assert diag.has_fatal_issues is False
    ffmpeg_issue = next(i for i in diag.issues if i.code == "ffmpeg_unavailable")
    assert ffmpeg_issue.severity == "warning"


def test_skill_tree_core_issue_when_skill_root_given(tmp_path: Path) -> None:
    skill_root = tmp_path / "my-skill"
    fake_core = skill_root / "scripts" / "jiangchang_skill_core" / "__init__.py"
    fake_core.parent.mkdir(parents=True)
    fake_core.write_text("# vendored\n", encoding="utf-8")

    env = {"CLAW_DATA_ROOT": str(tmp_path / "data")}
    with patch.object(jiangchang_skill_core, "__file__", str(fake_core)):
        diag = collect_runtime_diagnostics(
            skill_slug="my-skill",
            skill_root=skill_root,
            record_video=False,
            env=env,
        )
    assert "jiangchang_skill_core_loaded_from_skill_tree" in diag.issue_codes()


def test_format_runtime_health_lines_core_fields(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    env = {"CLAW_DATA_ROOT": str(tmp_path / "data")}
    monkeypatch.setattr(
        "jiangchang_skill_core.runtime_diagnostics.probe_media_assets",
        lambda *a, **k: ma.MediaAssetsStatus(
            root=tmp_path,
            exists=False,
            ready=False,
            source="local",
        ),
    )
    monkeypatch.setattr(
        "jiangchang_skill_core.runtime_diagnostics.probe_background_music",
        lambda *a, **k: {
            "music_root": None,
            "mp3_count": 0,
            "usable_count": 0,
            "sample_path": None,
            "issue": None,
        },
    )
    monkeypatch.setattr(
        "jiangchang_skill_core.runtime_diagnostics._platform_kit_version",
        lambda: "1.0.10",
    )

    diag = collect_runtime_diagnostics(skill_slug="my-skill", env=env)
    text = "\n".join(format_runtime_health_lines(diag))
    for marker in (
        "python_executable:",
        "platform_kit_version:",
        "resolved_data_root:",
        "media_assets_root:",
        "ffmpeg_available:",
        "background_music_mp3_count:",
    ):
        assert marker in text


def test_background_music_lfs_and_size_rules(tmp_path: Path) -> None:
    music = tmp_path / "music"
    music.mkdir()
    (music / "tiny.mp3").write_bytes(b"x" * 512)
    (music / "lfs.mp3").write_text(
        "version https://git-lfs.github.com/spec/v1\noid sha256:abc\nsize 123\n"
    )
    (music / "ok.mp3").write_bytes(b"x" * 1024)

    probe = ma.probe_background_music(tmp_path)
    assert probe["mp3_count"] == 3
    assert probe["usable_count"] == 1
    assert probe["sample_path"] == str(music / "ok.mp3")


def test_runtime_diagnostics_frozen_dataclass() -> None:
    diag = RuntimeDiagnostics(
        skill_slug="x",
        python_executable="python",
        platform_kit_version="1.0.10",
        platform_kit_min_version=None,
        platform_kit_version_ok=None,
        jiangchang_skill_core_file=None,
        claw_data_root=None,
        jiangchang_data_root=None,
        resolved_data_root="/tmp",
        media_assets_root="/tmp/media",
        ffmpeg_available=False,
        ffmpeg_path=None,
        background_music_mp3_count=0,
        background_music_usable_count=0,
        background_music_issue=None,
        background_music_sample_path=None,
        record_video_enabled=False,
        issues=(RuntimeIssue(code="ffmpeg_unavailable", message="no ffmpeg"),),
    )
    assert diag.has_fatal_issues is False
    assert diag.issue_codes() == ["ffmpeg_unavailable"]


def test_format_runtime_health_lines_issue_separator_ascii() -> None:
    diag = RuntimeDiagnostics(
        skill_slug="x",
        python_executable="python",
        platform_kit_version="1.0.11",
        platform_kit_min_version=None,
        platform_kit_version_ok=None,
        jiangchang_skill_core_file=None,
        claw_data_root=None,
        jiangchang_data_root=None,
        resolved_data_root="/tmp",
        media_assets_root="/tmp/media",
        ffmpeg_available=False,
        ffmpeg_path=None,
        background_music_mp3_count=0,
        background_music_usable_count=0,
        background_music_issue=None,
        background_music_sample_path=None,
        record_video_enabled=False,
        issues=(RuntimeIssue(code="ffmpeg_unavailable", message="no ffmpeg"),),
    )
    lines = format_runtime_health_lines(diag)
    issue_lines = [line for line in lines if line.startswith("runtime_issue[")]
    assert len(issue_lines) == 1
    assert " - " in issue_lines[0]
    assert "\u2014" not in issue_lines[0]
    assert "\u2013" not in issue_lines[0]