from pathlib import Path
import base64
import os
import shlex
import shutil
import subprocess
import tempfile
import unittest


ROOT = Path(__file__).resolve().parents[1]


class LogMaintenanceArtifactTests(unittest.TestCase):
    BEE_LOGROTATE = ROOT / "deploy" / "logrotate" / "razgar-bee"
    APACHE_LOGROTATE = ROOT / "deploy" / "logrotate" / "apache2"
    PRUNER = ROOT / "deploy" / "maintenance" / "prune_bee_diagnostic_logs.sh"
    RAZGAR_LOGROTATE_SERVICE = ROOT / "deploy" / "systemd" / "razgar-logrotate.service"
    RAZGAR_LOGROTATE_TIMER = ROOT / "deploy" / "systemd" / "razgar-logrotate.timer"
    SERVICE = ROOT / "deploy" / "systemd" / "razgar-bee-log-maintenance.service"
    TIMER = ROOT / "deploy" / "systemd" / "razgar-bee-log-maintenance.timer"

    def test_all_versioned_artifacts_exist(self) -> None:
        for artifact in (
            self.BEE_LOGROTATE,
            self.APACHE_LOGROTATE,
            self.PRUNER,
            self.RAZGAR_LOGROTATE_SERVICE,
            self.RAZGAR_LOGROTATE_TIMER,
            self.SERVICE,
            self.TIMER,
        ):
            with self.subTest(artifact=artifact):
                self.assertTrue(artifact.is_file())

    def test_razgar_global_logrotate_units_are_versioned_source_files(self) -> None:
        units = (
            self.RAZGAR_LOGROTATE_SERVICE,
            self.RAZGAR_LOGROTATE_TIMER,
        )
        for unit in units:
            with self.subTest(unit=unit):
                self.assertTrue(unit.is_file())
        subprocess.run(
            [
                "git",
                "ls-files",
                "--error-unmatch",
                "--",
                *(str(unit.relative_to(ROOT)) for unit in units),
            ],
            cwd=ROOT,
            check=True,
            capture_output=True,
            text=True,
        )

    def test_obsolete_distribution_logrotate_timer_override_stays_absent(self) -> None:
        obsolete = (
            ROOT
            / "deploy"
            / "systemd"
            / "logrotate.timer.d"
            / "razgar-six-hourly.conf"
        )
        self.assertFalse(obsolete.exists())
        installer = (ROOT / "deploy" / "worker" / "install_worker.sh").read_text(
            encoding="utf-8"
        )
        self.assertNotIn("logrotate.timer.d/razgar-six-hourly.conf", installer)

    def test_bee_rotation_has_no_per_file_size_cap(self) -> None:
        config = self.BEE_LOGROTATE.read_text(encoding="utf-8")
        self.assertIn("/var/log/bee_logs/diagnostic-*.jsonl", config)
        self.assertIn("/var/log/bee_logs/test_output_*.txt", config)
        self.assertIn("/var/log/bee_logs/php-error.log", config)
        self.assertIn("hourly", config)
        self.assertIn("dateext", config)
        self.assertIn("compress", config)
        self.assertIn("rotate 28", config)
        self.assertRegex(config, r"(?im)^\s*su\s+www-data\s+www-data\s*$")
        self.assertNotRegex(config, r"(?im)^\s*(?:maxsize|minsize|size)\b")
        self.assertNotIn("WORKER_TEST_LOG_DIR", config)

    def test_bee_rotation_policy_parses_with_logrotate(self) -> None:
        policy = self.BEE_LOGROTATE.read_text(encoding="utf-8").replace(
            "    su www-data www-data\n", ""
        )
        with tempfile.TemporaryDirectory() as temporary:
            config = Path(temporary) / "razgar-bee"
            config.write_text(policy, encoding="utf-8")
            if os.name == "nt" and shutil.which("wsl.exe") is not None:
                config_path = config.resolve()
                config_path = (
                    f"/mnt/{config_path.drive[0].lower()}{config_path.as_posix()[2:]}"
                    if config_path.drive
                    else config_path.as_posix()
                )
                completed = subprocess.run(
                    [
                        "wsl.exe",
                        "-e",
                        "bash",
                        "-lc",
                        "logrotate -d -s /tmp/bee-logrotate-test.state "
                        + shlex.quote(config_path),
                    ],
                    text=True,
                    capture_output=True,
                    check=False,
                )
            elif shutil.which("logrotate") is not None:
                completed = subprocess.run(
                    [
                        "logrotate",
                        "-d",
                        "-s",
                        str(Path(temporary) / "state"),
                        str(config),
                    ],
                    text=True,
                    capture_output=True,
                    check=False,
                )
            else:
                self.skipTest("logrotate is unavailable")
        self.assertEqual(0, completed.returncode, completed.stderr)

    def test_apache_preserves_the_host_policy_verbatim_except_maxsize(self) -> None:
        config = self.APACHE_LOGROTATE.read_text(encoding="utf-8")
        expected = """/var/log/apache2/*.log {
\tdaily
\tmaxsize 128M
\tmissingok
\trotate 14
\tcompress
\tdelaycompress
\tnotifempty
\tcreate 640 root adm
\tsharedscripts
\tprerotate
\t\tif [ -d /etc/logrotate.d/httpd-prerotate ]; then
\t\t\trun-parts /etc/logrotate.d/httpd-prerotate
\t\tfi
\tendscript
\tpostrotate
\t\tif pgrep -f ^/usr/sbin/apache2 > /dev/null; then
\t\t\tinvoke-rc.d apache2 reload 2>&1 | logger -t apache2.logrotate
\t\tfi
\tendscript
}
"""
        self.assertEqual(expected, config)

    def test_custom_timer_has_one_persistent_six_hour_schedule(self) -> None:
        expected = "OnCalendar=*-*-* 00/6:00:00"
        source = self.TIMER.read_text(encoding="utf-8")
        non_empty = [
            line.strip()
            for line in source.splitlines()
            if line.strip().startswith("OnCalendar=")
            and line.strip() != "OnCalendar="
        ]
        self.assertEqual([expected], non_empty)
        self.assertIn("Persistent=true", source)
        self.assertNotIn("5min", source)
        self.assertNotIn("00:05", source)

    def test_razgar_global_logrotate_runs_the_global_config_every_six_hours(self) -> None:
        expected = "OnCalendar=*-*-* 00/6:00:00"
        service = self.RAZGAR_LOGROTATE_SERVICE.read_text(encoding="utf-8")
        self.assertIn("Type=oneshot", service)
        starts = [line.strip() for line in service.splitlines() if line.startswith("ExecStart=")]
        self.assertEqual(
            ["ExecStart=/usr/sbin/logrotate --wait-for-state-lock /etc/logrotate.conf"],
            starts,
        )
        source = self.RAZGAR_LOGROTATE_TIMER.read_text(encoding="utf-8")
        non_empty = [
            line.strip()
            for line in source.splitlines()
            if line.strip().startswith("OnCalendar=")
            and line.strip() != "OnCalendar="
        ]
        self.assertEqual([expected], non_empty)
        self.assertIn("Persistent=true", source)
        self.assertIn("RandomizedDelaySec=0", source)
        self.assertIn("Unit=razgar-logrotate.service", source)
        self.assertNotIn("5min", source)
        self.assertNotIn("00:05", source)

    def test_pruner_is_scoped_to_diagnostic_candidates_and_limits(self) -> None:
        source = self.PRUNER.read_text(encoding="utf-8")
        self.assertIn("/var/log/bee_logs", source)
        self.assertIn("diagnostic-*.jsonl*", source)
        self.assertIn("512 * 1024 * 1024", source)
        self.assertIn("10 * 1024 * 1024 * 1024", source)
        self.assertNotIn("WORKER_TEST_LOG_DIR", source)
        for forbidden in (
            "php-error",
            "/var/log/apache2",
            "/var/log/bot",
            "/var/log/redis",
            "/var/www",
            "/var/releases",
            "test_output_",
        ):
            self.assertNotIn(forbidden, source)
        self.assertRegex(source, r"find\s+-P\s+\"\$LOG_DIR\"")
        self.assertRegex(source, r"-name\s+'diagnostic-\*\.jsonl\*'")
        self.assertNotIn("rm -rf", source)
        self.assertNotIn("BEE_DIAGNOSTIC_LOG_DIR", source)
        self.assertNotIn("WORKER_TEST_LOG_DIR", source)
        self.assertIn('if [[ "${BASH_SOURCE[0]:-}" == "$0" ]]; then', source)
        self.assertIn('"/var/log/bee_logs"', source)
        self.assertIn('"$((512 * 1024 * 1024))"', source)
        self.assertIn('"$((10 * 1024 * 1024 * 1024))"', source)

    def _run_pruner(
        self,
        log_dir: Path,
        *,
        max_bytes: int | None = None,
        min_free_bytes: int | None = None,
    ) -> subprocess.CompletedProcess[str]:
        absolute_log_dir = log_dir.resolve()
        use_wsl = os.name == "nt" and shutil.which("wsl.exe") is not None
        if use_wsl and absolute_log_dir.drive:
            bash_log_dir = (
                f"/mnt/{absolute_log_dir.drive[0].lower()}"
                f"{absolute_log_dir.as_posix()[2:]}"
            )
        else:
            bash_log_dir = absolute_log_dir.as_posix()
        script = self.PRUNER.read_text(encoding="utf-8")
        arguments = [bash_log_dir]
        if max_bytes is not None or min_free_bytes is not None:
            arguments.extend(
                [
                    str(
                        max_bytes
                        if max_bytes is not None
                        else 512 * 1024 * 1024
                    ),
                    str(min_free_bytes if min_free_bytes is not None else 10 * 1024**3),
                ]
            )
        command = (
            script
            + "\nprune_bee_diagnostic_logs "
            + " ".join(shlex.quote(argument) for argument in arguments)
            + "\n"
        )
        payload = base64.b64encode(command.encode("utf-8")).decode("ascii")
        decode_command = f"printf '%s' '{payload}' | base64 -d | /bin/bash"
        shell_command = ["wsl.exe", "-e", "bash", "-c"] if use_wsl else ["bash", "-c"]
        return subprocess.run(
            [*shell_command, decode_command],
            text=True,
            capture_output=True,
            check=False,
        )

    def test_pruner_absent_directory_is_successful(self) -> None:
        with tempfile.TemporaryDirectory() as temporary:
            missing = Path(temporary) / "missing"
            completed = self._run_pruner(missing)
            self.assertEqual(0, completed.returncode, completed.stderr)
            self.assertFalse(missing.exists())

    def test_pruner_no_candidates_is_successful_and_untouched(self) -> None:
        with tempfile.TemporaryDirectory() as temporary:
            log_dir = Path(temporary)
            protected = {
                "php-error.log": "php",
                "test_output_42.txt": "legacy",
                "legacy.jsonl": "legacy-jsonl",
                "unrelated.log": "other",
                "unrelated.jsonl": "other-jsonl",
                "diagnostic-not-jsonl.txt": "nonmatching",
            }
            for name, contents in protected.items():
                (log_dir / name).write_text(contents, encoding="utf-8")
            completed = self._run_pruner(log_dir)
            self.assertEqual(0, completed.returncode, completed.stderr)
            self.assertEqual(
                protected,
                {
                    path.name: path.read_text(encoding="utf-8")
                    for path in log_dir.iterdir()
                },
            )

    def test_pruner_budget_deletes_only_oldest_diagnostic_candidates(self) -> None:
        with tempfile.TemporaryDirectory() as temporary:
            log_dir = Path(temporary)
            candidates = [
                ("diagnostic-old.jsonl.1", 10),
                ("diagnostic-middle.jsonl.gz", 20),
                ("diagnostic-new.jsonl", 30),
            ]
            for name, age in candidates:
                path = log_dir / name
                path.write_bytes(b"x" * 16)
                timestamp = 1_700_000_000 + age
                os.utime(path, (timestamp, timestamp))
            protected = {
                "php-error.log": "php",
                "test_output_42.txt": "legacy",
                "legacy.jsonl": "legacy-jsonl",
                "unrelated.log": "other",
                "unrelated.jsonl": "other-jsonl",
                "diagnostic-not-jsonl.txt": "nonmatching",
            }
            for name, contents in protected.items():
                (log_dir / name).write_text(contents, encoding="utf-8")

            completed = self._run_pruner(
                log_dir,
                max_bytes=16,
                min_free_bytes=0,
            )
            self.assertEqual(0, completed.returncode, completed.stderr)
            self.assertEqual(
                ["diagnostic-new.jsonl"],
                sorted(path.name for path in log_dir.glob("diagnostic-*.jsonl*")),
            )
            for name, contents in protected.items():
                self.assertEqual(contents, (log_dir / name).read_text(encoding="utf-8"))

    def test_pruner_free_pressure_removes_candidates_but_not_protected_logs(self) -> None:
        with tempfile.TemporaryDirectory() as temporary:
            log_dir = Path(temporary)
            candidate = log_dir / "diagnostic-free-pressure.jsonl"
            candidate.write_text("diagnostic", encoding="utf-8")
            protected = {
                "php-error.log": "php",
                "test_output_42.txt": "legacy",
                "legacy.jsonl": "legacy-jsonl",
                "unrelated.jsonl": "other-jsonl",
                "diagnostic-not-jsonl.txt": "nonmatching",
            }
            for name, contents in protected.items():
                (log_dir / name).write_text(contents, encoding="utf-8")
            completed = self._run_pruner(
                log_dir,
                max_bytes=512 * 1024 * 1024,
                min_free_bytes=10**18,
            )
            self.assertEqual(0, completed.returncode, completed.stderr)
            self.assertFalse(candidate.exists())
            for name, contents in protected.items():
                self.assertEqual(contents, (log_dir / name).read_text(encoding="utf-8"))

    def test_maintenance_service_waits_for_razgar_logrotate_then_prunes(self) -> None:
        service = self.SERVICE.read_text(encoding="utf-8")
        self.assertIn("Type=oneshot", service)
        self.assertIn("Requires=razgar-logrotate.service", service)
        self.assertIn("After=razgar-logrotate.service", service)
        self.assertNotIn("Wants=razgar-logrotate.service", service)
        self.assertNotIn("ConditionPathIsDirectory", service)
        starts = [line.strip() for line in service.splitlines() if line.startswith("ExecStart=")]
        self.assertEqual(
            ["ExecStart=/usr/local/sbin/prune_bee_diagnostic_logs.sh"],
            starts,
        )
        self.assertNotIn("/usr/sbin/logrotate", service)
        self.assertNotIn("/etc/logrotate.d/", service)
        self.assertNotIn("--state", service)
        self.assertNotIn("WORKER_TEST_LOG_DIR", service)

    def test_artifact_scripts_parse_as_bash(self) -> None:
        subprocess.run(
            ["bash", "-n"],
            input=self.PRUNER.read_bytes().replace(b"\r\n", b"\n"),
            check=True,
        )


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