98 lines
3.7 KiB
Python
98 lines
3.7 KiB
Python
import unittest
|
|
from datetime import date
|
|
from pathlib import Path
|
|
|
|
import collect_dy_persona_to_bitable as dy_worker
|
|
import collect_jd_persona_to_bitable as jd_worker
|
|
import collect_persona_to_bitable as tm_worker
|
|
import run_daily_persona
|
|
|
|
|
|
class PersonaLauncherTests(unittest.TestCase):
|
|
def test_any_platform_failure_makes_launcher_fail(self):
|
|
self.assertEqual(run_daily_persona.combined_exit_code({"tm": 0, "dy": 2, "jd": 0}), 2)
|
|
|
|
def test_all_platforms_success_makes_launcher_success(self):
|
|
self.assertEqual(run_daily_persona.combined_exit_code({"tm": 0, "dy": 0, "jd": 0}), 0)
|
|
|
|
def test_partial_platform_results_are_accepted_when_real_rows_were_written(self):
|
|
results = {"tm": 2, "dy": 2, "jd": 2}
|
|
summaries = {
|
|
"tm": {"ok": 31, "skipped": 0, "failed": 4, "total": 35},
|
|
"dy": {"ok": 2, "skipped": 0, "failed": 30, "total": 31},
|
|
"jd": {"ok": 18, "skipped": 0, "failed": 13, "total": 31},
|
|
}
|
|
|
|
self.assertEqual(run_daily_persona.combined_exit_code(results, summaries), 0)
|
|
|
|
def test_partial_platform_without_real_rows_still_fails(self):
|
|
results = {"tm": 2, "dy": 2, "jd": 2}
|
|
summaries = {
|
|
"tm": {"ok": 31},
|
|
"dy": {"ok": 0},
|
|
"jd": {"ok": 18},
|
|
}
|
|
|
|
self.assertEqual(run_daily_persona.combined_exit_code(results, summaries), 2)
|
|
|
|
def test_result_summary_is_scoped_to_current_log_segment(self):
|
|
from tempfile import TemporaryDirectory
|
|
|
|
with TemporaryDirectory() as directory:
|
|
path = Path(directory) / "persona.log"
|
|
path.write_text(
|
|
"[DONE] ok=0 permission_skipped=0 failed=31 total=31\n",
|
|
encoding="utf-8",
|
|
)
|
|
offset = path.stat().st_size
|
|
with path.open("a", encoding="utf-8") as stream:
|
|
stream.write(
|
|
"[DONE] ok=18 permission_skipped=0 failed=13 total=31\n"
|
|
)
|
|
|
|
self.assertEqual(
|
|
run_daily_persona.read_result_summary(path, offset),
|
|
{"ok": 18, "skipped": 0, "failed": 13, "total": 31},
|
|
)
|
|
|
|
def test_launcher_passes_business_date_to_each_worker(self):
|
|
command = run_daily_persona.build_child_command(
|
|
Path("collect_persona_to_bitable.py"),
|
|
date(2026, 8, 4),
|
|
)
|
|
|
|
self.assertEqual(command[-2:], ["--date", "2026-08-04"])
|
|
|
|
def test_workers_fail_when_permission_is_skipped(self):
|
|
results = [{"status": "ok"}, {"status": "permission_skipped"}]
|
|
|
|
for worker in (tm_worker, dy_worker, jd_worker):
|
|
with self.subTest(worker=worker.__name__):
|
|
self.assertEqual(worker.combined_result_exit_code(results), 2)
|
|
|
|
def test_workers_fail_when_no_styles_were_processed(self):
|
|
for worker in (tm_worker, dy_worker, jd_worker):
|
|
with self.subTest(worker=worker.__name__):
|
|
self.assertEqual(worker.combined_result_exit_code([]), 2)
|
|
|
|
def test_workers_succeed_only_when_all_styles_are_successful(self):
|
|
results = [{"status": "ok"}, {"status": "dry-run"}]
|
|
|
|
for worker in (tm_worker, dy_worker, jd_worker):
|
|
with self.subTest(worker=worker.__name__):
|
|
self.assertEqual(worker.combined_result_exit_code(results), 0)
|
|
|
|
def test_workers_fail_when_collector_itself_failed(self):
|
|
results = [{"status": "ok"}]
|
|
|
|
for worker in (tm_worker, dy_worker, jd_worker):
|
|
with self.subTest(worker=worker.__name__):
|
|
self.assertEqual(
|
|
worker.combined_result_exit_code(results, collection_ok=False),
|
|
2,
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|