box/tests/test_retention_archive_jobs.py

#!/usr/bin/env python3
"""test_retention_archive_jobs.py — Unit tests for Retention Piece 2: Job Archival & Pruning."""

import json
import os
import shutil
import sys
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch

REPO_ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(REPO_ROOT / "bin"))

import importlib.util
spec = importlib.util.spec_from_file_location("retention_archive_jobs", str(REPO_ROOT / "bin" / "retention-archive-jobs.py"))
raj = importlib.util.module_from_spec(spec)
spec.loader.exec_module(raj)


class TestRetentionArchiveJobs(unittest.TestCase):
    def setUp(self):
        self.test_dir = tempfile.mkdtemp()
        self.root = Path(self.test_dir)
        self.jobs_dir = self.root / "jobs"
        self.archive_dir = self.jobs_dir / "archive"
        self.jobs_dir.mkdir(parents=True)

    def tearDown(self):
        shutil.rmtree(self.test_dir, ignore_errors=True)

    def _create_job(self, name, schedule="manual", extra=None):
        data = {
            "name": name,
            "description": f"Test job {name}",
            "agent": "pip",
            "schedule": schedule,
            "timeout": 300,
        }
        if extra:
            data.update(extra)
        p = self.jobs_dir / f"{name}.json"
        p.write_text(json.dumps(data, indent=2))
        return p

    def test_core_protected_jobs(self):
        protected = raj.get_protected_jobs(self.root)
        self.assertIn("heartbeat", protected)
        self.assertIn("refine-system", protected)
        self.assertIn("canary-test", protected)

    def test_systemd_unit_protection(self):
        sdir = self.root / "systemd"
        sdir.mkdir()
        svc = sdir / "job-custom-worker.service"
        svc.write_text("[Service]\nExecStart=/home/super/Projects/NetVM/bin/job-dispatch.py custom-worker\n")

        protected = raj.get_protected_jobs(self.root)
        self.assertIn("custom-worker", protected)
        self.assertIn("systemd unit", protected["custom-worker"])

    def test_active_pipeline_protection(self):
        pipe_file = self.root / "pipelines.json"
        pipe_data = {
            "run-001": {
                "status": "running",
                "pipeline_name": "live-pipe-root",
                "steps": [{"job_name": "live-pipe-step2"}]
            },
            "run-002": {
                "status": "completed",
                "pipeline_name": "dead-pipe-root",
                "steps": [{"job_name": "dead-pipe-step"}]
            }
        }
        pipe_file.write_text(json.dumps(pipe_data))

        protected = raj.get_protected_jobs(self.root)
        self.assertIn("live-pipe-root", protected)
        self.assertIn("live-pipe-step2", protected)
        self.assertNotIn("dead-pipe-root", protected)

    def test_chain_target_protection(self):
        self._create_job("canary-test", schedule="manual", extra={"on_success": "canary-followup"})
        self._create_job("canary-followup", schedule="manual")

        protected = raj.get_protected_jobs(self.root)
        self.assertIn("canary-followup", protected)
        self.assertIn("linked by canary-test", protected["canary-followup"])

    def test_eligibility_criteria(self):
        manual_job = self._create_job("retired-task", schedule="manual")
        cron_job = self._create_job("hourly-task", schedule="*/5 * * * *")
        protected_job = self._create_job("heartbeat", schedule="manual")

        prot = {"heartbeat": "core baseline"}

        el1, _ = raj.is_eligible(manual_job, prot)
        self.assertTrue(el1)

        el2, _ = raj.is_eligible(cron_job, prot)
        self.assertFalse(el2)

        el3, _ = raj.is_eligible(protected_job, prot)
        self.assertFalse(el3)

    def test_archive_and_unarchive_roundtrip(self):
        self._create_job("auto-task-x1", schedule="manual")

        with patch.object(raj, "is_git_tracked", return_value=False):
            # Archive
            res_arch = raj.archive_job("auto-task-x1", dry_run=False, root_dir=self.root)
            self.assertEqual(res_arch["status"], "archived")
            self.assertFalse((self.jobs_dir / "auto-task-x1.json").exists())
            self.assertTrue((self.archive_dir / "auto-task-x1.json").exists())

            # Unarchive
            res_unarch = raj.unarchive_job("auto-task-x1", dry_run=False, root_dir=self.root)
            self.assertEqual(res_unarch["status"], "unarchived")
            self.assertTrue((self.jobs_dir / "auto-task-x1.json").exists())
            self.assertFalse((self.archive_dir / "auto-task-x1.json").exists())

    def test_refuse_archive_protected_job(self):
        self._create_job("heartbeat", schedule="manual")
        res = raj.archive_job("heartbeat", dry_run=False, root_dir=self.root)
        self.assertEqual(res["status"], "rejected")
        self.assertTrue((self.jobs_dir / "heartbeat.json").exists())

    def test_force_archive_protected_job(self):
        self._create_job("heartbeat", schedule="manual")
        with patch.object(raj, "is_git_tracked", return_value=False):
            res = raj.archive_job("heartbeat", dry_run=False, force=True, root_dir=self.root)
            self.assertEqual(res["status"], "archived")
            self.assertTrue((self.archive_dir / "heartbeat.json").exists())

    def test_scan_and_archive_dry_run(self):
        self._create_job("task-1", schedule="manual")
        self._create_job("task-2", schedule="*/10 * * * *")

        res = raj.scan_and_archive(dry_run=True, root_dir=self.root)
        self.assertEqual(res["eligible"], 1)
        self.assertEqual(res["archived"], 1)
        self.assertTrue((self.jobs_dir / "task-1.json").exists())
        self.assertFalse((self.archive_dir / "task-1.json").exists())

    def test_list_archived(self):
        self._create_job("archived-task", schedule="manual")
        with patch.object(raj, "is_git_tracked", return_value=False):
            raj.archive_job("archived-task", dry_run=False, root_dir=self.root)
            archived = raj.list_archived(root_dir=self.root)
            self.assertEqual(len(archived), 1)
            self.assertEqual(archived[0]["name"], "archived-task")


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