55 lines
3.0 KiB
Python
55 lines
3.0 KiB
Python
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from unittest.mock import Mock, call, patch
|
|
|
|
from copienator import CliError, EvaluationWorkspace, ExitCode, atomic_write_json
|
|
from copienator.commands.batch_status import check_evaluation_jobs, main
|
|
|
|
|
|
class BatchReadinessTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.temp.cleanup)
|
|
self.workspace = EvaluationWorkspace(Path(self.temp.name))
|
|
self.client = Mock()
|
|
|
|
def manifest(self, jobs):
|
|
atomic_write_json(self.workspace.batch_jobs_file, {"jobs": jobs})
|
|
|
|
def test_only_recorded_jobs_are_checked_and_all_must_have_results(self):
|
|
self.manifest({"flash": {"name": "batches/flash"}, "pro": {"name": "batches/pro"}})
|
|
succeeded = SimpleNamespace(state="JOB_STATE_SUCCEEDED",
|
|
dest=SimpleNamespace(file_name="files/result"))
|
|
for state in ("JOB_STATE_PENDING", "JOB_STATE_RUNNING", "JOB_STATE_FAILED",
|
|
"JOB_STATE_CANCELLED", "JOB_STATE_EXPIRED", "UNKNOWN", "JOB_STATE_SUCCEEDED"):
|
|
with self.subTest(state=state):
|
|
self.client.reset_mock()
|
|
self.client.batches.get.side_effect = [succeeded, SimpleNamespace(
|
|
state=SimpleNamespace(name=state), dest=SimpleNamespace(file_name="files/pro"))]
|
|
result = check_evaluation_jobs(self.workspace, client=self.client)
|
|
self.assertEqual(result, ExitCode.SUCCESS if state == "JOB_STATE_SUCCEEDED" else ExitCode.PARTIAL)
|
|
self.assertEqual(self.client.batches.get.call_args_list,
|
|
[call(name="batches/flash"), call(name="batches/pro")])
|
|
self.client.batches.list.assert_not_called()
|
|
self.client.files.download.assert_not_called()
|
|
self.client.batches.get.side_effect = [succeeded, SimpleNamespace(
|
|
state="JOB_STATE_SUCCEEDED", dest=None)]
|
|
self.assertEqual(check_evaluation_jobs(self.workspace, client=self.client), ExitCode.PARTIAL)
|
|
|
|
def test_missing_empty_and_invalid_manifest_cannot_report_success(self):
|
|
self.assertEqual(check_evaluation_jobs(self.workspace, client=self.client), ExitCode.PARTIAL)
|
|
self.manifest({})
|
|
self.assertEqual(check_evaluation_jobs(self.workspace, client=self.client), ExitCode.PARTIAL)
|
|
for jobs in ([], {"flash": {}}, {"flash": {"name": ""}}, {"flash": None}):
|
|
self.manifest(jobs)
|
|
with self.assertRaises(CliError):
|
|
check_evaluation_jobs(self.workspace, client=self.client)
|
|
self.client.batches.get.assert_not_called()
|
|
|
|
def test_cli_returns_readiness_code_for_selected_evaluation(self):
|
|
with patch("copienator.commands.batch_status.check_evaluation_jobs", return_value=ExitCode.PARTIAL) as check:
|
|
self.assertEqual(main(["--evaluation", str(self.workspace.root)]), ExitCode.PARTIAL)
|
|
self.assertEqual(check.call_args.args[0].root, self.workspace.root)
|