submmit/tests/test_gpu_strategy.py

from __future__ import annotations

import sys
import tempfile
import unittest
from collections import Counter
from datetime import datetime, timedelta, timezone
from pathlib import Path


PACKAGE_DIR = Path(__file__).resolve().parents[1] / "modelhub_submmit_api"
if str(PACKAGE_DIR) in sys.path:
    sys.path.remove(str(PACKAGE_DIR))
sys.path.insert(0, str(PACKAGE_DIR))

from gpu_strategy import (  # noqa: E402
    ALL_SUPPORTED,
    LONG_TERM,
    RECENT,
    GPUStrategyManager,
    build_strategy_snapshot,
    choose_next_category,
)


def make_history(gpu: str, success: int, failure: int, *, start: datetime) -> list[dict]:
    records: list[dict] = []
    for index in range(success + failure):
        passed = index < success
        records.append(
            {
                "taskId": f"{gpu}-{index}",
                "gpuType": gpu,
                "status": "success",
                "verifyResult": 1 if passed else -1,
                "updateTime": (start + timedelta(seconds=index)).isoformat(),
            }
        )
    return records


class HistoryClient:
    def __init__(self, tasks: list[dict]) -> None:
        self.tasks = tasks
        self.list_calls = 0

    def list_tasks(self, **_kwargs) -> list[dict]:  # noqa: ANN003
        self.list_calls += 1
        return list(self.tasks)


class GPUStrategyTests(unittest.TestCase):
    def test_long_term_ranking_rejects_tiny_high_rate_sample(self) -> None:
        start = datetime(2026, 1, 1, tzinfo=timezone.utc)
        supported = ["tiny", "best", "second", "third", "bad"]
        tasks = [
            *make_history("tiny", 8, 0, start=start),
            *make_history("best", 40, 160, start=start),
            *make_history("second", 30, 270, start=start),
            *make_history("third", 45, 455, start=start),
            *make_history("bad", 5, 495, start=start),
        ]

        snapshot = build_strategy_snapshot(
            tasks,
            supported_gpus=supported,
            long_term_min_samples=100,
        )

        self.assertEqual(["best", "second", "third"], snapshot["longTermGpus"])
        self.assertNotIn("tiny", snapshot["longTermGpus"])

    def test_weighted_category_planner_is_exact_over_200_accepts(self) -> None:
        counts = {LONG_TERM: 0, ALL_SUPPORTED: 0, RECENT: 0}
        for _ in range(200):
            category = choose_next_category(counts)
            counts[category] += 1

        self.assertEqual({LONG_TERM: 100, ALL_SUPPORTED: 60, RECENT: 40}, counts)

    def test_candidate_order_uses_50_30_20_for_first_strategy_generation(self) -> None:
        supported = ["gpu-a", "gpu-b", "gpu-c", "gpu-d"]
        snapshot = build_strategy_snapshot([], supported_gpus=supported)
        manager = GPUStrategyManager("unused.json")
        manager.state = snapshot
        candidates = [
            {"repoId": f"owner/model-{model}", "targetGpu": gpu}
            for model in range(250)
            for gpu in supported
        ]

        ordered = manager.order_candidates(candidates)
        categories = Counter(candidate["strategyCategory"] for candidate in ordered[:200])

        self.assertEqual(100, categories[LONG_TERM])
        self.assertEqual(60, categories[ALL_SUPPORTED])
        self.assertEqual(40, categories[RECENT])

    def test_refresh_happens_only_after_200_accepted_submissions(self) -> None:
        with tempfile.TemporaryDirectory() as temporary_dir:
            path = Path(temporary_dir) / "strategy.json"
            client = HistoryClient([])
            manager = GPUStrategyManager(path, refresh_submissions=200)
            manager.prepare(client, supported_gpus=["gpu-a", "gpu-b", "gpu-c"])
            self.assertEqual(1, client.list_calls)

            manager.record_accepted([{"strategyCategory": LONG_TERM}] * 199)
            self.assertEqual(1, manager.submissions_until_refresh)
            manager.prepare(client, supported_gpus=["gpu-a", "gpu-b", "gpu-c"])
            self.assertEqual(1, client.list_calls)

            manager.record_accepted([{"strategyCategory": RECENT}])
            self.assertEqual(0, manager.submissions_until_refresh)
            manager.prepare(client, supported_gpus=["gpu-a", "gpu-b", "gpu-c"])
            self.assertEqual(2, client.list_calls)
            self.assertEqual(1, manager.state["generation"])
            self.assertEqual(200, manager.state["acceptedTotal"])
            self.assertEqual(0, manager.state["acceptedSinceRefresh"])


if __name__ == "__main__":
    unittest.main()