kneron_model_converter/tests/workers/test_precheck.py

"""Unit tests for platform/operator compatibility pre-check.

這些測試刻意不 import onnx / ktc（本機環境沒有 toolchain），只透過 mock 的
evaluator backend 驗證 pre-check 的判斷邏輯。
"""

from pathlib import Path
import sys

import pytest

ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(ROOT))

from services.backends.precheck import (
    UnsupportedOperatorError,
    build_error_message,
    find_unsupported_operators,
    run_precheck,
)


class _FakeEvaluator:
    """可注入的假 evaluator：回傳固定字串，或在呼叫時 raise。"""

    def __init__(self, *, report=None, raises=None):
        self._report = report
        self._raises = raises
        self.calls = []

    def evaluate(self, onnx_path, **kwargs):
        self.calls.append((onnx_path, kwargs))
        if self._raises is not None:
            raise self._raises
        return self._report


# --------------------------------------------------------------------------- #
# find_unsupported_operators
# --------------------------------------------------------------------------- #

def test_find_unsupported_empty_report_returns_empty():
    assert find_unsupported_operators("") == []
    assert find_unsupported_operators(None) == []


def test_find_unsupported_clean_report_returns_empty():
    report = "kdp520/FPS: 1200, cpu_node: None, docker_version: 1.0"
    assert find_unsupported_operators(report) == []


def test_find_unsupported_detects_empty_node_op():
    report = "compiler: creating an EmptyNode instance for op_type: Softmax"
    assert find_unsupported_operators(report) == ["Softmax"]


def test_find_unsupported_detects_unimplemented_feature_with_bracket_op():
    report = "UnimplementedFeature: undefined CPU op [Softmax]"
    assert find_unsupported_operators(report) == ["Softmax"]


def test_find_unsupported_detects_hw_not_support_with_op():
    report = "HW_NOT_SUPPORT: Softmax"
    assert find_unsupported_operators(report) == ["Softmax"]


def test_find_unsupported_signal_without_op_returns_placeholder():
    # 有明確訊號但抽不到 op 名稱 → 回占位符，讓呼叫端仍能 fail-fast。
    report = "kdp520/ERROR, HW not support"
    assert find_unsupported_operators(report) == ["<unknown>"]


def test_find_unsupported_does_not_false_positive_on_supported_wording():
    # "int16 is not supported in 520" 這種 bitwidth 說明不應被誤判為 op 不支援。
    report = "note: int16 is not supported in 520; datapath int8 ok"
    assert find_unsupported_operators(report) == []


def test_find_unsupported_detects_multiple_ops_across_markers():
    # 模型含多個不支援 op，且分散在不同 marker pattern（EmptyNode + HW_NOT_SUPPORT）。
    # 應回傳兩者、順序穩定（依報告出現順序）。
    report = (
        "compiler: creating an EmptyNode instance for op_type: Softmax\n"
        "HW_NOT_SUPPORT: LSTM"
    )
    assert find_unsupported_operators(report) == ["Softmax", "LSTM"]


def test_find_unsupported_dedups_repeated_op():
    # 同一個不支援 op 在報告中出現多次（多個節點）→ 只回一次（precheck.py 的 dedup）。
    report = (
        "creating an EmptyNode instance for op_type: Softmax\n"
        "creating an EmptyNode instance for op_type: Softmax\n"
        "HW_NOT_SUPPORT: Softmax"
    )
    assert find_unsupported_operators(report) == ["Softmax"]


def test_run_precheck_raises_lists_multiple_ops_in_message():
    # 多個不支援 op → 擋掉時錯誤訊息應同時列出（去重後）。
    evaluator = _FakeEvaluator(
        report=(
            "creating an EmptyNode instance for op_type: Softmax\n"
            "HW_NOT_SUPPORT: LSTM\n"
            "creating an EmptyNode instance for op_type: Softmax"
        )
    )
    with pytest.raises(UnsupportedOperatorError) as exc:
        run_precheck(
            "/tmp/out.onnx",
            model_id=6,
            version="v1",
            platform="520",
            evaluator=evaluator,
        )
    msg = str(exc.value)
    assert "Softmax" in msg
    assert "LSTM" in msg
    assert "520" in msg
    # 去重：Softmax 只出現一次
    assert msg.count("Softmax") == 1


# --------------------------------------------------------------------------- #
# build_error_message
# --------------------------------------------------------------------------- #

def test_build_error_message_with_op_names():
    msg = build_error_message("520", ["Softmax"])
    assert "520" in msg
    assert "Softmax" in msg
    # 帶有引導使用者的補救建議
    assert "720" in msg or "730" in msg


def test_build_error_message_with_unknown_op():
    msg = build_error_message("520", ["<unknown>"])
    assert "520" in msg
    assert "operator" in msg


# --------------------------------------------------------------------------- #
# run_precheck
# --------------------------------------------------------------------------- #

def test_run_precheck_raises_on_unsupported_report():
    evaluator = _FakeEvaluator(
        report="compiler: creating an EmptyNode instance for op_type: Softmax"
    )
    with pytest.raises(UnsupportedOperatorError) as exc:
        run_precheck(
            "/tmp/out.onnx",
            model_id=1,
            version="v1",
            platform="520",
            evaluator=evaluator,
        )
    assert "Softmax" in str(exc.value)
    assert "520" in str(exc.value)
    assert len(evaluator.calls) == 1


def test_run_precheck_passes_on_clean_report():
    evaluator = _FakeEvaluator(report="kdp720/FPS: 900, cpu_node: None")
    # 不應 raise
    run_precheck(
        "/tmp/out.onnx",
        model_id=2,
        version="v1",
        platform="720",
        evaluator=evaluator,
    )
    assert len(evaluator.calls) == 1


def test_run_precheck_raises_when_evaluate_raises_with_signal():
    # evaluate() 直接 raise，但例外訊息含明確不支援訊號 → 應轉成清楚錯誤。
    evaluator = _FakeEvaluator(
        raises=AssertionError("UnimplementedFeature: undefined CPU op [Softmax]")
    )
    with pytest.raises(UnsupportedOperatorError) as exc:
        run_precheck(
            "/tmp/out.onnx",
            model_id=3,
            version="v1",
            platform="520",
            evaluator=evaluator,
        )
    assert "Softmax" in str(exc.value)


def test_run_precheck_passes_when_evaluate_raises_without_signal():
    # evaluate() raise，但訊息不含不支援訊號（transient / 環境問題）→ 放行不誤擋。
    evaluator = _FakeEvaluator(
        raises=RuntimeError("Quantization model generation failed. See above message.")
    )
    # 不應 raise
    run_precheck(
        "/tmp/out.onnx",
        model_id=4,
        version="v1",
        platform="530",
        evaluator=evaluator,
    )


def test_run_precheck_passes_target_platform_to_evaluator():
    evaluator = _FakeEvaluator(report="")
    run_precheck(
        "/tmp/out.onnx",
        model_id=5,
        version="abc",
        platform="730",
        evaluator=evaluator,
    )
    _path, kwargs = evaluator.calls[0]
    assert kwargs["platform"] == "730"
    assert kwargs["model_id"] == 5
    assert kwargs["version"] == "abc"