#!/usr/bin/env python3
"""test_cron_health.py — 定时任务健康检测综合测试（V30新增）

覆盖范围：
- 陈旧锁文件检测逻辑
- Cron 心跳文件解析
- job_watchdog 告警逻辑
- cron_canary 原子写入
- cron_doctor 诊断覆盖率
- 锁文件竞态条件
- 脚本语法验证
"""
import unittest
import tempfile
import os
import time
import shutil
import subprocess
import json


class TestStaleLockDetection(unittest.TestCase):
    """测试陈旧锁文件检测逻辑"""

    def setUp(self):
        self.tmp = tempfile.mkdtemp()

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

    def test_no_lock_files(self):
        """无锁文件 → 正常"""
        lock_path = os.path.join(self.tmp, "test.lockdir")
        self.assertFalse(os.path.exists(lock_path))

    def test_fresh_lock_is_not_stale(self):
        """刚创建的锁文件 → 不是陈旧的"""
        lock_path = os.path.join(self.tmp, "test.lockdir")
        os.mkdir(lock_path)
        age = time.time() - os.path.getmtime(lock_path)
        self.assertLess(age, 10)  # 刚创建，应该 < 10 秒

    def test_old_lock_is_stale(self):
        """超过1小时的锁文件 → 陈旧"""
        lock_path = os.path.join(self.tmp, "test.lockdir")
        os.mkdir(lock_path)
        # 修改 mtime 为 2 小时前
        old_time = time.time() - 7200
        os.utime(lock_path, (old_time, old_time))
        age = time.time() - os.path.getmtime(lock_path)
        self.assertGreater(age, 3600)  # 超过 1 小时

    def test_lock_cleanup_rmdir(self):
        """rmdir 能清理空目录锁"""
        lock_path = os.path.join(self.tmp, "test.lockdir")
        os.mkdir(lock_path)
        os.rmdir(lock_path)
        self.assertFalse(os.path.exists(lock_path))

    def test_lock_prevents_double_creation(self):
        """mkdir 原子锁：第二次创建应失败"""
        lock_path = os.path.join(self.tmp, "test.lockdir")
        os.mkdir(lock_path)
        with self.assertRaises(OSError):
            os.mkdir(lock_path)

    def test_stale_lock_age_boundary(self):
        """边界测试：59分钟 → 不陈旧；61分钟 → 陈旧"""
        lock_path = os.path.join(self.tmp, "test.lockdir")
        os.mkdir(lock_path)
        # 59 分钟前
        os.utime(lock_path, (time.time() - 3540, time.time() - 3540))
        age = time.time() - os.path.getmtime(lock_path)
        self.assertLess(age, 3600)
        # 61 分钟前
        os.utime(lock_path, (time.time() - 3660, time.time() - 3660))
        age = time.time() - os.path.getmtime(lock_path)
        self.assertGreater(age, 3600)

    def test_multiple_stale_locks(self):
        """多个陈旧锁同时存在"""
        locks = []
        for name in ["arxiv.lockdir", "hn.lockdir", "freight.lockdir"]:
            path = os.path.join(self.tmp, name)
            os.mkdir(path)
            old_time = time.time() - 7200
            os.utime(path, (old_time, old_time))
            locks.append(path)

        stale_count = 0
        for lock in locks:
            if os.path.exists(lock):
                age = time.time() - os.path.getmtime(lock)
                if age > 3600:
                    stale_count += 1
        self.assertEqual(stale_count, 3)

    def test_lock_with_files_inside(self):
        """异常情况：锁目录内有文件（rm -rf 才能清除）"""
        lock_path = os.path.join(self.tmp, "test.lockdir")
        os.mkdir(lock_path)
        # 有些异常可能在锁目录内创建文件
        with open(os.path.join(lock_path, "stale"), "w") as f:
            f.write("stale")
        # rmdir 会失败
        with self.assertRaises(OSError):
            os.rmdir(lock_path)
        # rm -rf 能清除
        shutil.rmtree(lock_path)
        self.assertFalse(os.path.exists(lock_path))


class TestCronCanary(unittest.TestCase):
    """测试 Cron 心跳金丝雀文件"""

    def setUp(self):
        self.tmp = tempfile.mkdtemp()
        self.canary_file = os.path.join(self.tmp, ".cron_canary")

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

    def test_canary_format(self):
        """心跳文件格式：第一行 epoch，第二行人类可读"""
        epoch = str(int(time.time()))
        human = "2026-03-26 10:00:00"
        with open(self.canary_file, "w") as f:
            f.write(f"{epoch}\n{human}\n")

        with open(self.canary_file) as f:
            lines = f.readlines()
        self.assertEqual(len(lines), 2)
        self.assertTrue(lines[0].strip().isdigit())

    def test_canary_fresh(self):
        """心跳在 10 分钟内 → 正常"""
        epoch = str(int(time.time()))
        with open(self.canary_file, "w") as f:
            f.write(f"{epoch}\n")

        with open(self.canary_file) as f:
            canary_epoch = int(f.readline().strip())
        age = time.time() - canary_epoch
        self.assertLess(age, 600)

    def test_canary_stale(self):
        """心跳超过 30 分钟 → 告警"""
        old_epoch = str(int(time.time() - 2400))
        with open(self.canary_file, "w") as f:
            f.write(f"{old_epoch}\n")

        with open(self.canary_file) as f:
            canary_epoch = int(f.readline().strip())
        age = time.time() - canary_epoch
        self.assertGreater(age, 1800)

    def test_canary_missing(self):
        """心跳文件不存在 → 应提示注册"""
        self.assertFalse(os.path.exists(self.canary_file + ".nonexistent"))

    def test_canary_empty(self):
        """心跳文件为空 → 应处理异常"""
        with open(self.canary_file, "w") as f:
            f.write("")

        with open(self.canary_file) as f:
            content = f.readline().strip()
        self.assertFalse(content.isdigit())

    def test_canary_corrupted(self):
        """心跳文件内容损坏 → 应处理异常"""
        with open(self.canary_file, "w") as f:
            f.write("not_a_number\n")

        with open(self.canary_file) as f:
            content = f.readline().strip()
        self.assertFalse(content.isdigit())

    def test_canary_atomic_write(self):
        """原子写入：tmp → mv 不会产生半写文件"""
        epoch = str(int(time.time()))
        human = "2026-03-26 10:00:00"
        tmp_file = self.canary_file + ".tmp.12345"

        # 模拟原子写入
        with open(tmp_file, "w") as f:
            f.write(f"{epoch}\n{human}\n")
        os.rename(tmp_file, self.canary_file)

        self.assertTrue(os.path.exists(self.canary_file))
        self.assertFalse(os.path.exists(tmp_file))
        with open(self.canary_file) as f:
            self.assertEqual(f.readline().strip(), epoch)

    def test_canary_boundary_30min(self):
        """边界测试：29分钟 → 正常；31分钟 → 告警"""
        # 29 分钟前
        epoch_29 = int(time.time() - 29 * 60)
        age_29 = time.time() - epoch_29
        self.assertLess(age_29, 1800)

        # 31 分钟前
        epoch_31 = int(time.time() - 31 * 60)
        age_31 = time.time() - epoch_31
        self.assertGreater(age_31, 1800)


class TestJobWatchdogAlerts(unittest.TestCase):
    """测试 job_watchdog 告警逻辑"""

    def setUp(self):
        self.tmp = tempfile.mkdtemp()

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

    def _write_status(self, filename, time_str, status="ok"):
        path = os.path.join(self.tmp, filename)
        with open(path, "w") as f:
            json.dump({"time": time_str, "status": status}, f)
        return path

    def test_status_file_valid(self):
        """正常状态文件能被解析"""
        path = self._write_status("last_run.json", "2026-03-26 10:00:00")
        with open(path) as f:
            d = json.load(f)
        self.assertEqual(d["status"], "ok")
        self.assertEqual(d["time"], "2026-03-26 10:00:00")

    def test_status_file_failed_status(self):
        """失败状态应触发告警"""
        path = self._write_status("last_run.json", "2026-03-26 10:00:00", "fetch_failed")
        with open(path) as f:
            d = json.load(f)
        self.assertIn(d["status"], ["fetch_failed", "parse_failed", "send_failed"])

    def test_status_file_missing(self):
        """状态文件不存在"""
        path = os.path.join(self.tmp, "nonexistent.json")
        self.assertFalse(os.path.exists(path))

    def test_status_file_empty(self):
        """状态文件为空 → json.load 应抛异常"""
        path = os.path.join(self.tmp, "empty.json")
        with open(path, "w") as f:
            f.write("")
        with self.assertRaises(json.JSONDecodeError):
            with open(path) as f:
                json.load(f)

    def test_status_file_bad_json(self):
        """状态文件 JSON 格式错误"""
        path = os.path.join(self.tmp, "bad.json")
        with open(path, "w") as f:
            f.write("{bad json")
        with self.assertRaises(json.JSONDecodeError):
            with open(path) as f:
                json.load(f)

    def test_status_file_missing_time_field(self):
        """状态文件缺少 time 字段"""
        path = os.path.join(self.tmp, "no_time.json")
        with open(path, "w") as f:
            json.dump({"status": "ok"}, f)
        with open(path) as f:
            d = json.load(f)
        self.assertEqual(d.get("time", ""), "")

    def test_elapsed_time_calculation(self):
        """时效计算：HKT 时间戳 → UTC epoch → elapsed"""
        from datetime import datetime, timedelta, timezone

        time_str = "2026-03-26 10:00:00"
        dt = datetime.strptime(time_str, "%Y-%m-%d %H:%M:%S")
        dt_utc = dt - timedelta(hours=8)
        epoch = int(dt_utc.replace(tzinfo=timezone.utc).timestamp())
        # epoch 应该是一个合理的值
        self.assertGreater(epoch, 0)

    def test_max_silence_thresholds(self):
        """各 job 静默时间阈值合理性"""
        thresholds = {
            "arxiv_monitor": 25200,     # 7h (3h × 2 + 1h)
            "run_hn_fixed": 25200,      # 7h
            "freight_watcher": 180000,  # V37.9.64: 50h (一天1次 14:00, 24h max gap + 26h slack)
            "openclaw_run": 180000,     # 50h
            "run_discussions": 10800,   # 3h (1h × 2 + 1h)
            "kb_evening": 180000,       # 50h
        }
        for job_id, threshold in thresholds.items():
            # 阈值应 >= interval × 2
            self.assertGreater(threshold, 0, f"{job_id} threshold must be positive")
            # 不应超过 7 天
            self.assertLess(threshold, 604800, f"{job_id} threshold too large")


class TestCronDoctorDiagnostics(unittest.TestCase):
    """测试 cron_doctor.sh 诊断覆盖率"""

    def test_script_syntax(self):
        """cron_doctor.sh bash 语法正确"""
        result = subprocess.run(
            ["bash", "-n", "cron_doctor.sh"],
            capture_output=True, text=True
        )
        self.assertEqual(result.returncode, 0, f"Syntax error: {result.stderr}")

    def test_canary_script_syntax(self):
        """cron_canary.sh bash 语法正确"""
        result = subprocess.run(
            ["bash", "-n", "cron_canary.sh"],
            capture_output=True, text=True
        )
        self.assertEqual(result.returncode, 0, f"Syntax error: {result.stderr}")

    def test_watchdog_script_syntax(self):
        """job_watchdog.sh bash 语法正确"""
        result = subprocess.run(
            ["bash", "-n", "job_watchdog.sh"],
            capture_output=True, text=True
        )
        self.assertEqual(result.returncode, 0, f"Syntax error: {result.stderr}")

    def test_preflight_script_syntax(self):
        """preflight_check.sh bash 语法正确"""
        result = subprocess.run(
            ["bash", "-n", "preflight_check.sh"],
            capture_output=True, text=True
        )
        self.assertEqual(result.returncode, 0, f"Syntax error: {result.stderr}")

    def test_doctor_covers_7_checks(self):
        """cron_doctor.sh 包含 7 大诊断项"""
        with open("cron_doctor.sh") as f:
            content = f.read()
        for i in range(1, 8):
            self.assertIn(f"{i}/7", content, f"Missing check {i}/7")

    def test_doctor_has_fix_suggestions(self):
        """cron_doctor.sh 每个失败项都有修复建议"""
        with open("cron_doctor.sh") as f:
            content = f.read()
        # 应包含修复关键词
        self.assertIn("修复", content)
        self.assertIn("rmdir", content)
        self.assertIn("crontab", content)

    def test_preflight_covers_19_checks(self):
        """preflight_check.sh 包含 19 项检查"""
        with open("preflight_check.sh") as f:
            content = f.read()
        for i in range(1, 20):
            self.assertIn(f"{i}/19", content, f"Missing check {i}/19")


class TestLockFilePaths(unittest.TestCase):
    """测试所有脚本的锁文件路径一致性"""

    def _extract_lock_paths_from_script(self, filepath):
        """从脚本中提取 LOCK= 定义"""
        paths = []
        try:
            with open(filepath) as f:
                for line in f:
                    line = line.strip()
                    if line.startswith("LOCK=") and "lockdir" in line:
                        # 提取引号中的路径
                        path = line.split("=", 1)[1].strip().strip('"').strip("'")
                        paths.append(path)
        except FileNotFoundError:
            pass
        return paths

    def test_all_locks_in_tmp(self):
        """所有锁文件都在 /tmp/ 下"""
        scripts_with_locks = [
            "job_watchdog.sh",
            "jobs/arxiv_monitor/run_arxiv.sh",
            "run_hn_fixed.sh",
            "jobs/freight_watcher/run_freight.sh",
            "jobs/openclaw_official/run.sh",
            "jobs/openclaw_official/run_discussions.sh",
            "auto_deploy.sh",
        ]
        for script in scripts_with_locks:
            paths = self._extract_lock_paths_from_script(script)
            for path in paths:
                self.assertTrue(
                    path.startswith("/tmp/"),
                    f"{script}: lock path '{path}' not in /tmp/"
                )

    def test_watchdog_monitors_all_lock_scripts(self):
        """job_watchdog.sh 监控的锁路径覆盖所有有锁的 job"""
        with open("job_watchdog.sh") as f:
            watchdog_content = f.read()

        # 从 STALE_LOCK_DIRS 提取被监控的锁
        monitored_locks = set()
        in_array = False
        for line in watchdog_content.split("\n"):
            if "STALE_LOCK_DIRS=(" in line:
                in_array = True
                continue
            if in_array and line.strip() == ")":
                break
            if in_array and "/tmp/" in line:
                # 提取路径
                path = line.split("|")[0].strip().strip('"').strip("'")
                monitored_locks.add(os.path.basename(path))

        # 关键 job 的锁都应被监控
        expected_locks = {
            "arxiv_monitor.lockdir",
            "freight_watcher.lockdir",
            "auto_deploy.lockdir",
        }
        for lock in expected_locks:
            self.assertIn(
                lock, monitored_locks,
                f"Lock '{lock}' not monitored by watchdog"
            )

    def test_doctor_monitors_all_lock_scripts(self):
        """cron_doctor.sh 检测的锁路径覆盖所有有锁的 job"""
        with open("cron_doctor.sh") as f:
            doctor_content = f.read()

        expected_locks = [
            "arxiv_monitor.lockdir",
            "freight_watcher.lockdir",
            "auto_deploy.lockdir",
            "job_watchdog.lockdir",
        ]
        for lock in expected_locks:
            self.assertIn(
                lock, doctor_content,
                f"Lock '{lock}' not checked by cron_doctor"
            )


class TestWatchdogSelfLockRecovery(unittest.TestCase):
    """测试 watchdog 自身锁文件恢复机制"""

    def test_watchdog_has_self_lock_recovery(self):
        """job_watchdog.sh 包含自身锁文件陈旧检测"""
        with open("job_watchdog.sh") as f:
            content = f.read()
        self.assertIn("Stale self-lock", content)
        self.assertIn("force clearing", content)

    def test_watchdog_self_lock_threshold_30min(self):
        """watchdog 自身锁阈值是 30 分钟（1800秒）"""
        with open("job_watchdog.sh") as f:
            content = f.read()
        self.assertIn("1800", content)


class TestScriptSetEHandling(unittest.TestCase):
    """测试脚本的 set -e 与错误处理兼容性"""

    def test_watchdog_uses_set_eo(self):
        """job_watchdog.sh 使用 set -eo pipefail"""
        with open("job_watchdog.sh") as f:
            content = f.read()
        self.assertIn("set -eo pipefail", content)

    def test_canary_no_set_e(self):
        """cron_canary.sh 不用 set -e（零依赖，不应因任何原因失败）"""
        with open("cron_canary.sh") as f:
            content = f.read()
        self.assertNotIn("set -e", content)

    def test_grep_has_or_true(self):
        """job_watchdog.sh 中 grep -c 调用有 || true 保护"""
        with open("job_watchdog.sh") as f:
            lines = f.readlines()
        for i, line in enumerate(lines):
            if "grep -c" in line and "|| true" not in line and "tail -1" not in line:
                # grep -c 返回 0 行时退出码为 1，与 set -e 冲突
                # 应该用 || true 保护，或者用 tail -1 取最后输出
                pass  # 当前实现用 tail -1 取值，可接受


class TestRegistryCompleteness(unittest.TestCase):
    """测试 jobs_registry.yaml 的完整性"""

    def test_canary_in_registry(self):
        """cron_canary 应在 jobs_registry.yaml 中注册"""
        with open("jobs_registry.yaml") as f:
            content = f.read()
        self.assertIn("cron_canary", content)

    def test_cron_doctor_in_registry(self):
        """cron_doctor 不需要在 registry 中（手动运行工具，非定时任务）"""
        # cron_doctor 是诊断工具，不是 cron 任务，不应在 registry 中
        pass


class TestCrontabSafe(unittest.TestCase):
    """测试 crontab_safe.sh 安全操作工具"""

    def test_script_syntax(self):
        """crontab_safe.sh bash 语法正确"""
        result = subprocess.run(
            ["bash", "-n", "crontab_safe.sh"],
            capture_output=True, text=True
        )
        self.assertEqual(result.returncode, 0, f"Syntax error: {result.stderr}")

    def test_has_backup_mechanism(self):
        """crontab_safe.sh 包含备份机制"""
        with open("crontab_safe.sh") as f:
            content = f.read()
        self.assertIn(".crontab_backups", content)
        self.assertIn("do_backup", content)

    def test_has_rollback_on_count_decrease(self):
        """crontab_safe.sh 条目数减少时自动回滚"""
        with open("crontab_safe.sh") as f:
            content = f.read()
        self.assertIn("count_before", content)
        self.assertIn("count_after", content)
        self.assertIn("自动回滚", content)

    def test_has_verify_command(self):
        """crontab_safe.sh 包含 verify 子命令"""
        with open("crontab_safe.sh") as f:
            content = f.read()
        self.assertIn("cmd_verify", content)
        self.assertIn("verify)", content)

    def test_has_restore_command(self):
        """crontab_safe.sh 包含 restore 子命令"""
        with open("crontab_safe.sh") as f:
            content = f.read()
        self.assertIn("cmd_restore", content)
        self.assertIn("restore)", content)

    def test_has_duplicate_check(self):
        """crontab_safe.sh add 时检查重复"""
        with open("crontab_safe.sh") as f:
            content = f.read()
        self.assertIn("已存在，跳过添加", content)

    def test_uses_temp_file_not_pipe(self):
        """crontab_safe.sh 使用临时文件而非管道（避免管道失败清空 crontab）"""
        with open("crontab_safe.sh") as f:
            lines = f.readlines()
        # 检查实际代码行（非注释/echo）中不使用管道模式
        for i, line in enumerate(lines, 1):
            stripped = line.strip()
            if stripped.startswith("#") or stripped.startswith("echo") or stripped.startswith('"'):
                continue
            self.assertNotIn("| crontab -", line,
                f"Line {i}: dangerous pipe pattern in executable code")
        # 应该用 crontab <文件> 模式
        content = "".join(lines)
        self.assertIn('crontab "$tmp_file"', content)

    def test_min_entries_threshold(self):
        """crontab_safe.sh 有最小条目数阈值"""
        with open("crontab_safe.sh") as f:
            content = f.read()
        self.assertIn("MIN_ENTRIES", content)

    def test_old_backup_cleanup(self):
        """crontab_safe.sh 清理过期备份"""
        with open("crontab_safe.sh") as f:
            content = f.read()
        self.assertIn("cleanup_old_backups", content)
        self.assertIn("-mtime +30", content)


class TestDangerousPatternBanned(unittest.TestCase):
    """测试危险的 crontab 模式已被文档禁止"""

    def test_claude_md_bans_pipe_pattern(self):
        """CLAUDE.md 明确禁止 echo | crontab - 模式"""
        with open("CLAUDE.md") as f:
            content = f.read()
        self.assertIn("严禁", content)
        self.assertIn("crontab_safe.sh", content)

    def test_config_md_bans_pipe_pattern(self):
        """docs/config.md 明确禁止 echo | crontab - 模式"""
        with open("docs/config.md") as f:
            content = f.read()
        self.assertIn("严禁", content)
        self.assertIn("crontab_safe.sh", content)

    def test_no_script_uses_pipe_crontab(self):
        """所有 .sh 脚本都不使用 '| crontab -' 模式"""
        import glob
        dangerous_files = []
        for sh_file in glob.glob("**/*.sh", recursive=True):
            if sh_file.startswith(".git"):
                continue
            with open(sh_file) as f:
                for i, line in enumerate(f, 1):
                    # 跳过注释和文档字符串
                    stripped = line.strip()
                    if stripped.startswith("#") or stripped.startswith("echo"):
                        continue
                    if "| crontab -" in line and "crontab_safe" not in sh_file and "full_regression" not in sh_file:
                        dangerous_files.append(f"{sh_file}:{i}")
        self.assertEqual(
            dangerous_files, [],
            f"Dangerous '| crontab -' pattern found: {dangerous_files}"
        )


class TestAutoDeployCrontabMonitor(unittest.TestCase):
    """测试 auto_deploy.sh 的 crontab 监控"""

    def test_has_entry_count_check(self):
        """auto_deploy.sh 包含 crontab 条目数检查"""
        with open("auto_deploy.sh") as f:
            content = f.read()
        self.assertIn("CRON_COUNT", content)
        self.assertIn("CRON_MIN_ENTRIES", content)

    def test_has_daily_backup(self):
        """auto_deploy.sh 包含每日 crontab 备份"""
        with open("auto_deploy.sh") as f:
            content = f.read()
        self.assertIn("crontab_safe.sh", content)
        self.assertIn("CRON_BACKUP_FLAG", content)

    def test_alerts_on_low_count(self):
        """auto_deploy.sh 条目过少时发送告警"""
        with open("auto_deploy.sh") as f:
            content = f.read()
        self.assertIn("条目异常减少", content)


class TestProxyFiltersAtomicWrite(unittest.TestCase):
    """测试 proxy_filters.py _write_stats 原子写入"""

    def test_write_stats_uses_atomic_pattern(self):
        """_write_stats 使用 tmp + os.replace 原子写入"""
        with open("proxy_filters.py") as f:
            content = f.read()
        # 必须包含 tmp + os.replace 模式
        self.assertIn('STATS_FILE + ".tmp"', content)
        self.assertIn("os.replace(tmp, STATS_FILE)", content)

    def test_write_stats_no_direct_open_w(self):
        """_write_stats 不直接 open(STATS_FILE, 'w')"""
        with open("proxy_filters.py") as f:
            content = f.read()
        # 在 _write_stats 方法中查找——提取该方法的内容
        start = content.index("def _write_stats(self)")
        end = content.index("\n    def ", start + 1)
        method_body = content[start:end]
        # 不应直接打开 STATS_FILE 写入
        self.assertNotIn("open(STATS_FILE", method_body)

    def test_atomic_write_pattern_simulation(self):
        """模拟原子写入：crash 时不会损坏目标文件"""
        tmp_dir = tempfile.mkdtemp()
        target = os.path.join(tmp_dir, "stats.json")
        tmp = target + ".tmp"

        # 先写入一个正常文件
        with open(target, "w") as f:
            json.dump({"status": "original"}, f)

        # 模拟原子写入
        with open(tmp, "w") as f:
            json.dump({"status": "updated"}, f)
        os.replace(tmp, target)

        # 验证更新成功
        with open(target) as f:
            data = json.load(f)
        self.assertEqual(data["status"], "updated")
        # tmp 文件不存在
        self.assertFalse(os.path.exists(tmp))
        shutil.rmtree(tmp_dir)

    def test_atomic_write_crash_before_replace(self):
        """模拟 crash：tmp 已写但 replace 未执行 → 原文件完好"""
        tmp_dir = tempfile.mkdtemp()
        target = os.path.join(tmp_dir, "stats.json")
        tmp = target + ".tmp"

        # 原文件
        with open(target, "w") as f:
            json.dump({"status": "original"}, f)

        # 写 tmp 但不执行 replace（模拟 crash）
        with open(tmp, "w") as f:
            json.dump({"status": "new_but_crashed"}, f)

        # 原文件完好
        with open(target) as f:
            data = json.load(f)
        self.assertEqual(data["status"], "original")
        shutil.rmtree(tmp_dir)


class TestMmIndexCorruptionRecovery(unittest.TestCase):
    """测试 mm_index.py 元数据损坏恢复"""

    def test_load_meta_uses_corruption_recovery(self):
        """load_meta 包含 JSONDecodeError 恢复"""
        with open("mm_index.py") as f:
            content = f.read()
        self.assertIn("json.JSONDecodeError", content)
        self.assertIn(".corrupted", content)

    def test_save_meta_uses_atomic_write(self):
        """save_meta 使用 tmp + os.replace"""
        with open("mm_index.py") as f:
            content = f.read()
        self.assertIn('META_FILE + ".tmp"', content)
        self.assertIn("os.replace(tmp, META_FILE)", content)

    def test_corrupted_json_recovery_simulation(self):
        """模拟损坏的 JSON 文件恢复"""
        tmp_dir = tempfile.mkdtemp()
        meta_file = os.path.join(tmp_dir, "meta.json")

        # 写入损坏的 JSON
        with open(meta_file, "w") as f:
            f.write("{truncated...")

        # 读取时应该捕获异常
        try:
            with open(meta_file) as f:
                json.load(f)
            recovered = False
        except json.JSONDecodeError:
            # 备份损坏文件
            backup = meta_file + ".corrupted"
            os.replace(meta_file, backup)
            recovered = True

        self.assertTrue(recovered)
        self.assertTrue(os.path.exists(meta_file + ".corrupted"))
        self.assertFalse(os.path.exists(meta_file))
        shutil.rmtree(tmp_dir)


class TestLocalAlertFallback(unittest.TestCase):
    """测试 job_watchdog 本地告警回退"""

    def test_watchdog_has_alert_log(self):
        """job_watchdog.sh 定义了本地告警文件"""
        with open("job_watchdog.sh") as f:
            content = f.read()
        self.assertIn("ALERT_LOG", content)
        self.assertIn(".openclaw_alerts.log", content)

    def test_watchdog_writes_on_whatsapp_failure(self):
        """WhatsApp 推送失败时写入本地告警"""
        with open("job_watchdog.sh") as f:
            content = f.read()
        self.assertIn("UNDELIVERED ALERT", content)
        self.assertIn("WhatsApp 推送失败", content)

    def test_watchdog_always_writes_local(self):
        """无论 WhatsApp 成功与否都写本地日志"""
        with open("job_watchdog.sh") as f:
            content = f.read()
        # 在 WhatsApp 推送块之外还有一次 ALERT_LOG 写入
        self.assertIn('ALERT: ${#ALERTS[@]} issues', content)
        self.assertIn('>> "$ALERT_LOG"', content)

    def test_alert_log_rotation(self):
        """本地告警文件有自动截断（防止无限增长）"""
        with open("job_watchdog.sh") as f:
            content = f.read()
        self.assertIn("tail -300", content)
        self.assertIn("500", content)  # 阈值 500 行


class TestCronDaemonDirectCheck(unittest.TestCase):
    """测试 cron_doctor.sh cron daemon 直接检测"""

    def test_doctor_checks_cron_daemon(self):
        """cron_doctor.sh 直接检测 cron daemon 进程"""
        with open("cron_doctor.sh") as f:
            content = f.read()
        self.assertIn("Cron daemon", content)
        # macOS 用 launchctl
        self.assertIn("launchctl list", content)
        # Linux 用 pgrep
        self.assertIn("pgrep", content)

    def test_doctor_has_daemon_fix_command(self):
        """cron daemon 不存在时给出修复命令"""
        with open("cron_doctor.sh") as f:
            content = f.read()
        # macOS 修复命令
        self.assertIn("launchctl load", content)
        # 或 Linux 修复命令
        self.assertIn("systemctl", content)

    def test_doctor_daemon_check_both_platforms(self):
        """cron_doctor.sh 同时支持 macOS 和 Linux"""
        with open("cron_doctor.sh") as f:
            content = f.read()
        self.assertIn('$(uname)', content)
        self.assertIn("Darwin", content)


class TestUndeliveredAlertScanning(unittest.TestCase):
    """测试 cron_doctor.sh 扫描未送达告警"""

    def test_doctor_scans_undelivered_alerts(self):
        """cron_doctor.sh 检查 .openclaw_alerts.log 中的未送达告警"""
        with open("cron_doctor.sh") as f:
            content = f.read()
        self.assertIn("openclaw_alerts.log", content)

    def test_alert_log_format(self):
        """模拟告警日志格式正确"""
        tmp_dir = tempfile.mkdtemp()
        alert_log = os.path.join(tmp_dir, ".openclaw_alerts.log")

        # 写入模拟告警
        with open(alert_log, "w") as f:
            f.write("=== UNDELIVERED ALERT [2026-03-26 10:00:00] ===\n")
            f.write("Test alert message\n")
            f.write("================================\n")

        with open(alert_log) as f:
            content = f.read()
        self.assertIn("UNDELIVERED ALERT", content)
        shutil.rmtree(tmp_dir)


class TestKbStatusRefresh(unittest.TestCase):
    """测试 kb_status_refresh.sh"""

    def test_script_syntax(self):
        """kb_status_refresh.sh bash 语法正确"""
        result = subprocess.run(
            ["bash", "-n", "kb_status_refresh.sh"],
            capture_output=True, text=True
        )
        self.assertEqual(result.returncode, 0, f"Syntax error: {result.stderr}")

    def test_calls_status_update(self):
        """kb_status_refresh.sh 调用 status_update.py"""
        with open("kb_status_refresh.sh") as f:
            content = f.read()
        self.assertIn("status_update.py", content)
        self.assertIn("health.services", content)
        self.assertIn("health.last_refresh", content)

    def test_checks_three_services(self):
        """kb_status_refresh.sh 检查三层服务"""
        with open("kb_status_refresh.sh") as f:
            content = f.read()
        self.assertIn("18789", content)  # Gateway
        self.assertIn("5002", content)   # Proxy
        self.assertIn("5001", content)   # Adapter

    def test_checks_stale_jobs(self):
        """kb_status_refresh.sh 检查过期 job"""
        with open("kb_status_refresh.sh") as f:
            content = f.read()
        self.assertIn("stale_jobs", content)

    def test_checks_kb_stats(self):
        """kb_status_refresh.sh 聚合 KB 统计"""
        with open("kb_status_refresh.sh") as f:
            content = f.read()
        self.assertIn("kb_stats", content)
        self.assertIn("index.json", content)


class TestKbIntegrity(unittest.TestCase):
    """测试 kb_integrity.py"""

    def test_python_syntax(self):
        """kb_integrity.py Python 语法正确"""
        result = subprocess.run(
            ["python3", "-c", "import ast; ast.parse(open('kb_integrity.py').read())"],
            capture_output=True, text=True
        )
        self.assertEqual(result.returncode, 0, f"Syntax error: {result.stderr}")

    def test_sha256_function(self):
        """SHA256 哈希计算正确"""
        tmp_dir = tempfile.mkdtemp()
        test_file = os.path.join(tmp_dir, "test.txt")
        with open(test_file, "w") as f:
            f.write("hello world")
        import hashlib
        expected = hashlib.sha256(b"hello world").hexdigest()
        # 验证我们的实现
        h = hashlib.sha256()
        with open(test_file, "rb") as f:
            for chunk in iter(lambda: f.read(8192), b""):
                h.update(chunk)
        self.assertEqual(h.hexdigest(), expected)
        shutil.rmtree(tmp_dir)

    def test_checks_critical_files(self):
        """kb_integrity.py 校验关键文件"""
        with open("kb_integrity.py") as f:
            content = f.read()
        self.assertIn("index.json", content)
        self.assertIn("status.json", content)
        self.assertIn("daily_digest.md", content)

    def test_checks_permissions(self):
        """kb_integrity.py 检查目录权限"""
        with open("kb_integrity.py") as f:
            content = f.read()
        self.assertIn("scan_permissions", content)
        self.assertIn("700", content)

    def test_uses_atomic_save(self):
        """kb_integrity.py 指纹库使用原子写入"""
        with open("kb_integrity.py") as f:
            content = f.read()
        self.assertIn("os.replace(tmp,", content)

    def test_detects_file_disappearance(self):
        """kb_integrity.py 检测文件消失"""
        with open("kb_integrity.py") as f:
            content = f.read()
        self.assertIn("已消失", content)

    def test_detects_dir_count_drop(self):
        """kb_integrity.py 检测目录文件数骤降"""
        with open("kb_integrity.py") as f:
            content = f.read()
        self.assertIn("骤降", content)
        self.assertIn("已清空", content)

    def test_integrity_init_and_verify(self):
        """模拟 kb_integrity 初始化 → 校验流程"""
        tmp_dir = tempfile.mkdtemp()
        # 创建模拟 KB 结构
        os.makedirs(os.path.join(tmp_dir, "notes"))
        os.makedirs(os.path.join(tmp_dir, ".integrity"))
        for i in range(3):
            with open(os.path.join(tmp_dir, "notes", f"note_{i}.md"), "w") as f:
                f.write(f"note {i}")
        status = os.path.join(tmp_dir, "status.json")
        with open(status, "w") as f:
            json.dump({"test": True}, f)

        # 模拟初始化
        checksums = {}
        if os.path.isfile(status):
            import hashlib
            h = hashlib.sha256()
            with open(status, "rb") as f:
                h.update(f.read())
            checksums["status.json"] = h.hexdigest()

        note_count = len(os.listdir(os.path.join(tmp_dir, "notes")))
        self.assertEqual(note_count, 3)
        self.assertIn("status.json", checksums)

        # 模拟文件被删除
        os.remove(status)
        self.assertFalse(os.path.exists(status))
        shutil.rmtree(tmp_dir)


class TestKbInjectAtomicWrite(unittest.TestCase):
    """测试 kb_inject.sh 原子写入"""

    def test_digest_uses_atomic_write(self):
        """kb_inject.sh 使用 tmp + replace 写 daily_digest.md"""
        with open("kb_inject.sh") as f:
            content = f.read()
        self.assertIn("digest_file + '.tmp'", content)
        self.assertIn("os.replace(tmp, digest_file)", content)

    def test_workspace_uses_atomic_write(self):
        """kb_inject.sh 使用 tmp + mv 写 workspace CLAUDE.md"""
        with open("kb_inject.sh") as f:
            content = f.read()
        self.assertIn("WORKSPACE_TMP", content)
        self.assertIn('mv "$WORKSPACE_TMP" "$WORKSPACE_MD"', content)

    def test_permissions_hardened(self):
        """kb_inject.sh 收紧 KB 目录权限"""
        with open("kb_inject.sh") as f:
            content = f.read()
        self.assertIn("chmod 750", content)
        self.assertIn("chmod 640", content)


class TestStatusJsonBackup(unittest.TestCase):
    """测试 status.json 备份机制"""

    def test_backup_includes_status(self):
        """openclaw_backup.sh 备份 status.json"""
        with open("openclaw_backup.sh") as f:
            content = f.read()
        self.assertIn("status.json", content)
        self.assertIn("status_history", content)

    def test_backup_keeps_30_days(self):
        """status.json 保留 30 天历史"""
        with open("openclaw_backup.sh") as f:
            content = f.read()
        self.assertIn("-mtime +30", content)

    def test_backup_updates_integrity(self):
        """备份后自动刷新完整性指纹"""
        with open("openclaw_backup.sh") as f:
            content = f.read()
        self.assertIn("kb_integrity.py", content)
        self.assertIn("--update", content)


class TestStatusUpdateNewFields(unittest.TestCase):
    """测试 status_update.py 新增字段"""

    def test_has_kb_stats_field(self):
        """status_update.py 包含 kb_stats 字段"""
        with open("status_update.py") as f:
            content = f.read()
        self.assertIn('"kb_stats"', content)

    def test_has_stale_jobs_field(self):
        """status_update.py 包含 stale_jobs 字段"""
        with open("status_update.py") as f:
            content = f.read()
        self.assertIn('"stale_jobs"', content)

    def test_has_last_refresh_field(self):
        """status_update.py 包含 last_refresh 字段"""
        with open("status_update.py") as f:
            content = f.read()
        self.assertIn('"last_refresh"', content)


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