airfRANS-model-exploration/tests/test_remote_smoke.py

52 lines
1.9 KiB
Python
Raw Permalink Normal View History

from __future__ import annotations
import json
import sys
import tempfile
import types
import unittest
from pathlib import Path
from unittest.mock import patch
from airfrans_frontier.remote.smoke import run_smoke_training
class SmokeTrainingFailureTests(unittest.TestCase):
def test_pre_checkpoint_training_error_writes_terminal_failure_report(self) -> None:
fake_loop = types.ModuleType("airfrans_frontier.training.loop")
def fail_train(*_: object, **__: object) -> object:
raise RuntimeError("hub commit rate limited")
fake_loop.train_from_config_path = fail_train
fake_torch = types.ModuleType("torch")
fake_torch.__version__ = "fake"
fake_torch.version = types.SimpleNamespace(cuda=None)
fake_torch.cuda = types.SimpleNamespace(
is_available=lambda: False,
get_device_name=lambda _index: None,
)
with tempfile.TemporaryDirectory() as tmp, patch.dict(
sys.modules,
{
"airfrans_frontier.training.loop": fake_loop,
"torch": fake_torch,
},
):
artifact_dir = Path(tmp)
with self.assertRaisesRegex(RuntimeError, "hub commit rate limited"):
run_smoke_training("missing-config.toml", artifact_dir=artifact_dir, run_id="smoke-fail")
report = json.loads((artifact_dir / "failure_report.json").read_text())
self.assertEqual(report["run_id"], "smoke-fail")
self.assertEqual(report["error_type"], "RuntimeError")
self.assertEqual(report["error_message"], "hub commit rate limited")
verification = json.loads((artifact_dir / "verification_report.json").read_text())
self.assertTrue(verification["ok"])
self.assertEqual(verification["checks"]["terminal_artifact"], "failure_report.json")
if __name__ == "__main__":
unittest.main()