import io
import os
from pathlib import Path
import shutil
import subprocess
import tarfile
import tempfile
import unittest


SCRIPTS = Path(__file__).resolve().parents[1] / "scripts"


@unittest.skipUnless(shutil.which("php") and shutil.which("rsync"), "requires PHP CLI and rsync")
class ConfigProtectionTest(unittest.TestCase):
    def setUp(self):
        self.temp = tempfile.TemporaryDirectory(prefix="cqhw-deploy-test-")
        self.addCleanup(self.temp.cleanup)
        self.root = Path(self.temp.name)
        self.live = self.root / "project"
        self.local = self.live / "application/config/local"
        self.local.mkdir(parents=True)
        self.env_text = "<?php define('TP_ENV','local'); define('BUSINESS','BUSINESS_VALUE');\n"
        (self.live / "env.php").write_text(self.env_text)
        (self.live / "env.php").chmod(0o600)
        (self.local / "config.php").write_text(
            "<?php return ['business'=>'zhct_cqhw', 'redis'=>['host'=>'127.0.0.1', 'port'=>6379, 'password'=>'fixture-only']];"
        )
        (self.local / "database.php").write_text(
            "<?php return ['hostname'=>'127.0.0.1', 'database'=>'zhct_cqhw'];"
        )
        (self.local / "config.php").chmod(0o600)
        (self.live / "public/dist").mkdir(parents=True)
        (self.live / "public/dist/index.html").write_text("old-live-page")
        self.package = {
            "env.php": "<?php define('TP_ENV','prod'); define('BUSINESS','zhct_sanquan');",
            "application/config/local/config.php": "<?php return ['business'=>'wrong-project'];",
            "application/config/local/database.php": "<?php return ['database'=>'wrong-project'];",
            "public/dist/index.html": "<script>var BUSINESS = 'zhct_cqhw';</script>",
            "application/new-code.php": "<?php echo 'updated';",
        }
        self.bin = self.root / "bin"
        self.bin.mkdir()
        for name, body in {
            "wget": '#!/bin/sh\ncp "$FIXTURE_TAR" "$FIXTURE_ROOT/zhct_207.tgz"\n',
            "chown": "#!/bin/sh\nexit 0\n",
        }.items():
            p = self.bin / name
            p.write_text(body)
            p.chmod(0o755)
        script = (SCRIPTS / "deploy_zhct_cqhw.sh").read_text()
        fixed = 'DIR="/workspace/wwwroot/zhct_cqhw"'
        self.assertEqual(script.count(fixed), 1)
        self.script = self.root / "deploy.sh"
        self.script.write_text(script.replace(fixed, 'DIR="$FIXTURE_ROOT"'))
        shutil.copy2(SCRIPTS / "check_cqhw_config.php", self.root)

    def run_deploy(self):
        package_path = self.root / "fixture.tgz"
        with tarfile.open(package_path, "w:gz") as archive:
            for name, content in self.package.items():
                raw = content.encode()
                info = tarfile.TarInfo(name)
                info.size = len(raw)
                archive.addfile(info, io.BytesIO(raw))
        env = dict(os.environ, FIXTURE_ROOT=str(self.root), FIXTURE_TAR=str(package_path),
                   CQHW_PHP_BIN=shutil.which("php"), PATH=str(self.bin) + os.pathsep + os.environ["PATH"])
        return subprocess.run(["bash", str(self.script), "207"], env=env, text=True,
                              capture_output=True, timeout=15)

    def assert_refused_without_live_change(self):
        before = {str(p.relative_to(self.live)): p.read_bytes() for p in self.live.rglob("*") if p.is_file()}
        result = self.run_deploy()
        self.assertNotEqual(result.returncode, 0, result.stdout + result.stderr)
        after = {str(p.relative_to(self.live)): p.read_bytes() for p in self.live.rglob("*") if p.is_file()}
        self.assertEqual(before, after)

    def test_preserves_config_content_permissions_and_backup(self):
        before = {p: (p.read_bytes(), p.stat().st_mode & 0o777) for p in [self.live / "env.php", *self.local.iterdir()]}
        result = self.run_deploy()
        self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
        for p, expected in before.items():
            self.assertEqual((p.read_bytes(), p.stat().st_mode & 0o777), expected)
        self.assertTrue((self.live / "application/new-code.php").exists())
        self.assertEqual((self.live / "public/dist/index.html").read_text(), self.package["public/dist/index.html"])
        backups = list((self.root / "config-backups").glob("*.tgz"))
        self.assertEqual(len(backups), 1)
        self.assertEqual(backups[0].stat().st_mode & 0o777, 0o600)
        with tarfile.open(backups[0]) as archive:
            self.assertEqual(archive.extractfile("env.php").read(), self.env_text.encode())

    def test_other_customer_frontend_refused(self):
        self.package["public/dist/index.html"] = "<script>var BUSINESS = 'zhct_sanquan';</script>"
        self.assert_refused_without_live_change()

    def test_existing_polluted_environment_refused(self):
        (self.live / "env.php").write_text(self.package["env.php"])
        self.assert_refused_without_live_change()

    def test_missing_local_database_refused(self):
        (self.local / "database.php").unlink()
        self.assert_refused_without_live_change()

    def test_remote_redis_refused(self):
        p = self.local / "config.php"
        p.write_text(p.read_text().replace("127.0.0.1", "10.2.203.117"))
        self.assert_refused_without_live_change()

    def test_other_database_refused(self):
        p = self.local / "database.php"
        p.write_text(p.read_text().replace("zhct_cqhw", "another_database"))
        self.assert_refused_without_live_change()

    def test_missing_frontend_refused(self):
        del self.package["public/dist/index.html"]
        self.assert_refused_without_live_change()

    def test_symlink_env_refused(self):
        (self.root / "linked-env.php").write_text(self.env_text)
        (self.live / "env.php").unlink()
        (self.live / "env.php").symlink_to(self.root / "linked-env.php")
        self.assert_refused_without_live_change()


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