import unittest
import os
import subprocess
from pathlib import Path
from tempfile import TemporaryDirectory
from unittest.mock import patch

from deploy_tool.deployer import DeploymentConfig, DeploymentStep, build_deployment_plan, run_steps, resolve_process_command
import deploy_tool.deployer as deployer_module
from deploy_tool.scanner import ProjectComponent, ScanResult


class DeployerTests(unittest.TestCase):
    def test_extracts_backend_port_from_java_system_property(self):
        extract_port = getattr(deployer_module, "_extract_server_port", lambda command: None)

        port = extract_port(["java", "-Dserver.port=9090", "-jar", "app.jar"])

        self.assertEqual(port, 9090)

    def test_rejects_backend_port_outside_valid_tcp_range(self):
        for value in ("0", "-1", "65536"):
            with self.subTest(value=value), self.assertRaisesRegex(ValueError, "1 到 65535"):
                deployer_module._extract_server_port(["java", f"-Dserver.port={value}", "-jar", "app.jar"])

    def test_parses_ipv4_and_ipv6_listeners_for_configured_port(self):
        parse_listeners = getattr(deployer_module, "_parse_windows_listening_pids", lambda output, port: set())
        output = """
          TCP    0.0.0.0:8080           0.0.0.0:0              LISTENING       27544
          TCP    [::]:8080              [::]:0                 LISTENING       27544
          TCP    127.0.0.1:8080         127.0.0.1:52100        ESTABLISHED     30000
          TCP    0.0.0.0:9090           0.0.0.0:0              LISTENING       31000
        """

        pids = parse_listeners(output, 8080)

        self.assertEqual(pids, {27544})

    def test_port_release_waits_until_no_interface_has_a_listener(self):
        listener_snapshots = iter(({27544}, set()))

        with (
            patch.object(deployer_module, "_find_listening_pids", side_effect=lambda port: next(listener_snapshots)) as find_listeners,
            patch.object(deployer_module, "_is_port_open", return_value=False),
            patch.object(deployer_module.time, "sleep", return_value=None),
        ):
            released = deployer_module._wait_until_port_closed(8080)

        self.assertTrue(released)
        self.assertEqual(find_listeners.call_count, 2)

    def test_replaces_all_processes_listening_on_backend_port(self):
        replace_listeners = getattr(deployer_module, "_replace_port_listeners", lambda *args, **kwargs: False)
        terminated = []
        events = []

        released = replace_listeners(
            8080,
            events.append,
            find_listeners=lambda port: {27544, 23780},
            terminate=lambda pid: terminated.append(pid),
            wait_until_closed=lambda port: True,
        )

        self.assertTrue(released)
        self.assertEqual(set(terminated), {27544, 23780})
        self.assertTrue(any("8080" in event and "27544" in event for event in events))

    def test_port_replacement_skips_pid_removed_by_an_earlier_tree_kill(self):
        replace_listeners = getattr(deployer_module, "_replace_port_listeners", lambda *args, **kwargs: False)
        active_pids = {23780, 27544}
        terminated = []

        def terminate_tree(pid):
            terminated.append(pid)
            active_pids.clear()

        released = replace_listeners(
            8080,
            lambda value: None,
            find_listeners=lambda port: set(active_pids),
            terminate=terminate_tree,
            wait_until_closed=lambda port: True,
        )

        self.assertTrue(released)
        self.assertEqual(terminated, [23780])

    def test_backend_restart_fails_before_spawn_when_port_is_not_released(self):
        restart_backend = getattr(deployer_module, "_restart_backend", lambda *args, **kwargs: 0)
        spawned = []

        with TemporaryDirectory() as temp:
            exit_code = restart_backend(
                ["java", "-Dserver.port=8080", "-jar", "app.jar"],
                Path(temp),
                lambda value: None,
                replace_listeners=lambda port, on_output: False,
                popen=lambda *args, **kwargs: spawned.append(args),
            )

        self.assertEqual(exit_code, 1)
        self.assertEqual(spawned, [])

    def test_backend_restart_returns_child_exit_code_when_java_exits_early(self):
        restart_backend = getattr(deployer_module, "_restart_backend", lambda *args, **kwargs: 0)

        class ExitedProcess:
            pid = 23780

            def poll(self):
                return 2

        with TemporaryDirectory() as temp:
            exit_code = restart_backend(
                ["java", "-Dserver.port=65530", "-jar", "app.jar"],
                Path(temp),
                lambda value: None,
                replace_listeners=lambda port, on_output: True,
                popen=lambda *args, **kwargs: ExitedProcess(),
            )

        self.assertEqual(exit_code, 2)

    def test_backend_restart_reports_success_only_after_port_is_reachable(self):
        restart_backend = getattr(deployer_module, "_restart_backend", lambda *args, **kwargs: 0)
        checked_ports = []
        logs = []

        class RunningProcess:
            pid = 27544

            def poll(self):
                return None

        def port_is_open(port):
            checked_ports.append(port)
            return True

        with (
            TemporaryDirectory() as temp,
            patch.object(deployer_module, "_is_port_open", side_effect=port_is_open),
            patch.object(deployer_module, "_find_listening_pids", return_value={27544}),
        ):
            exit_code = restart_backend(
                ["java", "-Dserver.port=65530", "-jar", "app.jar"],
                Path(temp),
                logs.append,
                replace_listeners=lambda port, on_output: True,
                popen=lambda *args, **kwargs: RunningProcess(),
            )

        self.assertEqual(exit_code, 0)
        self.assertEqual(checked_ports, [65530])
        self.assertTrue(any("27544" in line and "启动成功" in line for line in logs))

    def test_backend_restart_rejects_reachable_port_owned_by_another_process(self):
        restart_backend = getattr(deployer_module, "_restart_backend", lambda *args, **kwargs: 0)
        terminated = []

        class RunningProcess:
            pid = 27544

            def poll(self):
                return None

        with (
            TemporaryDirectory() as temp,
            patch.object(deployer_module, "_is_port_open", return_value=True),
            patch.object(deployer_module, "_find_listening_pids", return_value={31000}),
            patch.object(deployer_module, "_terminate_process_tree", side_effect=lambda pid: terminated.append(pid)),
            patch.object(deployer_module.time, "monotonic", side_effect=[0.0, 0.0, 61.0]),
        ):
            exit_code = restart_backend(
                ["java", "-Dserver.port=65530", "-jar", "app.jar"],
                Path(temp),
                lambda value: None,
                replace_listeners=lambda port, on_output: True,
                popen=lambda *args, **kwargs: RunningProcess(),
            )

        self.assertEqual(exit_code, 1)
        self.assertEqual(terminated, [27544])

    def test_backend_restart_terminates_child_after_startup_timeout(self):
        restart_backend = getattr(deployer_module, "_restart_backend", lambda *args, **kwargs: 0)
        terminated = []

        class RunningProcess:
            pid = 27544

            def poll(self):
                return None

        with (
            TemporaryDirectory() as temp,
            patch.object(deployer_module, "_is_port_open", return_value=False),
            patch.object(deployer_module, "_terminate_process_tree", side_effect=lambda pid: terminated.append(pid)),
            patch.object(deployer_module.time, "monotonic", side_effect=[0.0, 61.0]),
        ):
            exit_code = restart_backend(
                ["java", "-Dserver.port=65530", "-jar", "app.jar"],
                Path(temp),
                lambda value: None,
                replace_listeners=lambda port, on_output: True,
                popen=lambda *args, **kwargs: RunningProcess(),
            )

        self.assertEqual(exit_code, 1)
        self.assertEqual(terminated, [27544])

    @unittest.skipUnless(os.name == "nt", "Windows creation flags only apply on Windows")
    def test_backend_restart_uses_visible_windows_console(self):
        restart_backend = getattr(deployer_module, "_restart_backend", lambda *args, **kwargs: 0)
        popen_options = {}

        class RunningProcess:
            pid = 27544

            def poll(self):
                return None

        def fake_popen(*args, **kwargs):
            popen_options.update(kwargs)
            return RunningProcess()

        with (
            TemporaryDirectory() as temp,
            patch.object(deployer_module, "_is_port_open", return_value=True),
            patch.object(deployer_module, "_find_listening_pids", return_value={27544}),
        ):
            exit_code = restart_backend(
                ["java", "-Dserver.port=65530", "-jar", "app.jar"],
                Path(temp),
                lambda value: None,
                replace_listeners=lambda port, on_output: True,
                popen=fake_popen,
            )

        self.assertEqual(exit_code, 0)
        self.assertIn("creationflags", popen_options)
        self.assertTrue(popen_options["creationflags"] & subprocess.CREATE_NEW_CONSOLE)
        self.assertTrue(popen_options["creationflags"] & subprocess.CREATE_NEW_PROCESS_GROUP)

    def test_background_runner_returns_backend_restart_health_result(self):
        class FakeProcess:
            pid = 27544

        with (
            TemporaryDirectory() as temp,
            patch.object(deployer_module, "_restart_backend", return_value=7) as restart_backend,
            patch.object(deployer_module.subprocess, "Popen", return_value=FakeProcess()),
        ):
            exit_code = deployer_module._run_process(
                ["java", "-Dserver.port=8080", "-jar", "app.jar"],
                Path(temp),
                lambda value: None,
                background=True,
            )

        self.assertEqual(exit_code, 7)
        restart_backend.assert_called_once()

    def test_backend_restart_separates_each_run_in_backend_log(self):
        restart_backend = getattr(deployer_module, "_restart_backend", lambda *args, **kwargs: 0)

        class RunningProcess:
            pid = 27544

            def poll(self):
                return None

        with (
            TemporaryDirectory() as temp,
            patch.object(deployer_module, "_is_port_open", return_value=True),
            patch.object(deployer_module, "_find_listening_pids", return_value={27544}),
        ):
            root = Path(temp)
            exit_code = restart_backend(
                ["java", "-Dserver.port=65530", "-jar", "app.jar"],
                root,
                lambda value: None,
                replace_listeners=lambda port, on_output: True,
                popen=lambda *args, **kwargs: RunningProcess(),
            )
            log_text = (root / "backend.log").read_text(encoding="utf-8")

        self.assertEqual(exit_code, 0)
        self.assertIn("后端部署", log_text)

    def test_builds_plan_for_detected_components(self):
        root = Path("C:/project")
        scan = ScanResult(
            root=root,
            backend=ProjectComponent("Spring Boot", root / "server", "Maven project"),
            frontend=ProjectComponent(
                "Vue",
                root / "web",
                "Vue project",
                {"role": "user", "display_name": "用户端", "start_script": "serve"},
            ),
            sql_files=[root / "db" / "schema.sql"],
        )
        config = DeploymentConfig(
            mysql_user="root",
            mysql_password="secret",
            mysql_database="demo",
            backend_port="8080",
            output_dir=root / "deploy",
        )

        plan = build_deployment_plan(scan, config)

        self.assertEqual(
            [step.kind for step in plan],
            [
                "backend_build",
                "frontend_install",
                "sql_import",
                "backend_start",
                "frontend_start",
                "deployment_summary",
            ],
        )
        self.assertIn("mvn", plan[0].command[0])
        self.assertEqual(plan[2].cwd, root)

    def test_builds_and_starts_user_and_admin_frontends_then_logs_all_addresses(self):
        with TemporaryDirectory() as temp:
            root = Path(temp)
            backend = root / "server"
            client = root / "client_code"
            admin = root / "manage_code"
            (backend / "target").mkdir(parents=True)
            (backend / "target" / "app.jar").write_text("jar", encoding="utf-8")

            frontends = []
            for path, role, display_name, port in (
                (client, "user", "用户端", "8082"),
                (admin, "admin", "管理端", "8081"),
            ):
                (path / "node_modules").mkdir(parents=True)
                (path / "dist").mkdir()
                (path / "dist" / "index.html").write_text("<main>app</main>", encoding="utf-8")
                (path / "package.json").write_text(
                    '{"scripts":{"build":"vue-cli-service build","serve":"vue-cli-service serve"}}',
                    encoding="utf-8",
                )
                frontends.append(
                    ProjectComponent(
                        "Vue",
                        path,
                        "Vue project",
                        {"role": role, "display_name": display_name, "port": port, "start_script": "serve"},
                    )
                )

            scan = ScanResult(
                root=root,
                backend=ProjectComponent(
                    "Spring Boot", backend, "Maven project", {"context_path": "/movies"}
                ),
                frontend=frontends[0],
                sql_files=[],
                frontends=frontends,
            )
            plan = build_deployment_plan(scan, DeploymentConfig(output_dir=root / "deploy"))

            self.assertEqual(
                [step.kind for step in plan],
                [
                    "backend_build_skip",
                    "backend_start",
                    "frontend_start",
                    "frontend_start",
                    "deployment_summary",
                ],
            )
            start_steps = [step for step in plan if step.kind == "frontend_start"]
            self.assertEqual([step.port for step in start_steps], [8082, 8081])
            self.assertEqual([step.service_name for step in start_steps], ["用户端", "管理端"])
            self.assertEqual(start_steps[0].output_path, root / "deploy" / "frontend-user.log")
            self.assertEqual(start_steps[1].output_path, root / "deploy" / "frontend-admin.log")

            logs = []
            exit_code = run_steps(plan, logs.append, process_runner=lambda *args: 0)

            self.assertEqual(exit_code, 0)
            combined = "\n".join(logs)
            self.assertIn("后端: http://localhost:8080/movies", combined)
            self.assertIn("用户端: http://localhost:8082/", combined)
            self.assertIn("管理端: http://localhost:8081/", combined)

    def test_plan_uses_executable_steps_without_shell_placeholders(self):
        with TemporaryDirectory() as temp:
            root = Path(temp)
            backend = root / "server"
            frontend = root / "web"
            database = root / "db"
            backend.mkdir()
            frontend.mkdir()
            database.mkdir()
            (backend / "pom.xml").write_text("<project />", encoding="utf-8")
            (frontend / "package.json").write_text(
                '{"scripts":{"build":"vite build","dev":"vite"}}', encoding="utf-8"
            )
            sql_file = database / "schema.sql"
            sql_file.write_text("create table demo(id int);", encoding="utf-8")
            scan = ScanResult(
                root=root,
                backend=ProjectComponent("Spring Boot", backend, "Maven project"),
                frontend=ProjectComponent("Vue", frontend, "Vue project"),
                sql_files=[sql_file],
            )
            config = DeploymentConfig(mysql_database="demo", output_dir=root / "deploy")

            plan = build_deployment_plan(scan, config)

            self.assertNotIn("frontend_build", [step.kind for step in plan])
            self.assertNotIn("frontend_copy", [step.kind for step in plan])
            for step in plan:
                joined = " ".join(step.command)
                self.assertNotIn("<", joined)
                self.assertNotIn(">", joined)
            sql_step = next(step for step in plan if step.kind == "sql_import")
            self.assertEqual(sql_step.input_path, sql_file)
            self.assertNotIn(str(sql_file), sql_step.command)

    def test_run_steps_installs_frontend_imports_sql_and_starts_services(self):
        with TemporaryDirectory() as temp:
            root = Path(temp)
            backend = root / "server"
            frontend = root / "web"
            target = backend / "target"
            database = root / "db"
            target.mkdir(parents=True)
            frontend.mkdir(parents=True)
            database.mkdir()
            (backend / "pom.xml").write_text("<project />", encoding="utf-8")
            (frontend / "package.json").write_text(
                '{"scripts":{"build":"vite build","dev":"vite"}}', encoding="utf-8"
            )
            old_jar = target / "old.jar"
            app_jar = target / "app.jar"
            old_jar.write_text("old", encoding="utf-8")
            app_jar.write_text("new", encoding="utf-8")
            os.utime(old_jar, (1, 1))
            os.utime(app_jar, (2, 2))
            sql_file = database / "schema.sql"
            sql_file.write_text("create table demo(id int);", encoding="utf-8")
            scan = ScanResult(
                root=root,
                backend=ProjectComponent("Spring Boot", backend, "Maven project"),
                frontend=ProjectComponent("Vue", frontend, "Vue project"),
                sql_files=[sql_file],
            )
            config = DeploymentConfig(
                mysql_user="root",
                mysql_password="secret",
                mysql_database="demo",
                backend_port="9090",
                output_dir=root / "deploy",
                tool_search_roots=[root / "missing"],
            )
            plan = build_deployment_plan(scan, config)
            calls = []

            def fake_runner(command, cwd, on_output, input_path=None, background=False):
                calls.append((command, cwd, input_path, background))
                return 0

            exit_code = run_steps(plan, lambda value: None, process_runner=fake_runner)

            self.assertEqual(exit_code, 0)
            self.assertIn((["mysql", "-u", "root", "-psecret", "demo"], root, sql_file, False), calls)
            self.assertIn((["java", "-Dserver.port=9090", "-jar", str(app_jar)], root / "deploy", None, True), calls)

    def test_plan_uses_configured_maven_path_for_backend_build(self):
        with TemporaryDirectory() as temp:
            root = Path(temp)
            backend = root / "server"
            maven = root / "tools" / "apache-maven" / "bin" / "mvn.cmd"
            backend.mkdir()
            maven.parent.mkdir(parents=True)
            (backend / "pom.xml").write_text("<project />", encoding="utf-8")
            maven.write_text("", encoding="utf-8")
            scan = ScanResult(
                root=root,
                backend=ProjectComponent("Spring Boot", backend, "Maven project"),
                frontend=None,
                sql_files=[],
            )
            config = DeploymentConfig(tool_paths={"Maven": maven})

            plan = build_deployment_plan(scan, config)

            self.assertEqual(plan[0].command, [str(maven), "clean", "package", "-DskipTests"])

    def test_plan_uses_discovered_mysql_path_for_sql_import(self):
        with TemporaryDirectory() as temp:
            root = Path(temp)
            mysql = root / "MySQL" / "MySQL Server 8.0" / "bin" / "mysql.exe"
            database = root / "db"
            mysql.parent.mkdir(parents=True)
            database.mkdir()
            mysql.write_text("", encoding="utf-8")
            sql_file = database / "schema.sql"
            sql_file.write_text("create table demo(id int);", encoding="utf-8")
            scan = ScanResult(
                root=root,
                backend=None,
                frontend=None,
                sql_files=[sql_file],
            )
            config = DeploymentConfig(mysql_database="demo", tool_search_roots=[root])

            plan = build_deployment_plan(scan, config)

            self.assertEqual(plan[0].command, [str(mysql), "-u", "root", "demo"])

    def test_sql_dump_with_database_selection_does_not_require_existing_database(self):
        with TemporaryDirectory() as temp:
            root = Path(temp)
            database = root / "db"
            database.mkdir()
            sql_file = database / "movie.sql"
            sql_file.write_text(
                "CREATE DATABASE IF NOT EXISTS `movie`;\nUSE `movie`;\nCREATE TABLE demo(id int);",
                encoding="utf-8",
            )
            scan = ScanResult(
                root=root,
                backend=None,
                frontend=None,
                sql_files=[sql_file],
            )
            config = DeploymentConfig(mysql_database="movie", tool_search_roots=[root / "missing"])

            plan = build_deployment_plan(scan, config)

            self.assertEqual(plan[0].command, ["mysql", "-u", "root"])

    def test_plan_skips_backend_build_when_runnable_jar_already_exists(self):
        with TemporaryDirectory() as temp:
            root = Path(temp)
            backend = root / "server"
            target = backend / "target"
            target.mkdir(parents=True)
            (backend / "pom.xml").write_text("<project />", encoding="utf-8")
            (target / "app.jar").write_text("jar", encoding="utf-8")
            scan = ScanResult(
                root=root,
                backend=ProjectComponent("Spring Boot", backend, "Maven project"),
                frontend=None,
                sql_files=[],
            )

            plan = build_deployment_plan(scan, DeploymentConfig(output_dir=root / "deploy", tool_search_roots=[root / "missing"]))

            self.assertEqual([step.kind for step in plan], ["backend_build_skip", "backend_start", "deployment_summary"])
            self.assertEqual(plan[0].command, [])
            self.assertIn("已有后端 jar", plan[0].description)

    def test_existing_dist_does_not_add_frontend_build_or_copy_steps(self):
        with TemporaryDirectory() as temp:
            root = Path(temp)
            frontend = root / "web"
            dist = frontend / "dist"
            dist.mkdir(parents=True)
            (frontend / "package.json").write_text(
                '{"scripts":{"build":"vite build","dev":"vite"}}', encoding="utf-8"
            )
            (dist / "index.html").write_text("<main>built</main>", encoding="utf-8")
            scan = ScanResult(
                root=root,
                backend=None,
                frontend=ProjectComponent("Vue", frontend, "Vue project"),
                sql_files=[],
            )

            plan = build_deployment_plan(scan, DeploymentConfig(output_dir=root / "deploy", tool_search_roots=[root / "missing"]))

            self.assertEqual(
                [step.kind for step in plan],
                ["frontend_install", "frontend_start", "deployment_summary"],
            )
            self.assertNotIn("frontend_build", [step.kind for step in plan])
            self.assertNotIn("frontend_copy", [step.kind for step in plan])

    def test_run_steps_logs_skip_steps_without_running_a_process(self):
        with TemporaryDirectory() as temp:
            root = Path(temp)
            logs = []
            calls = []
            skip_step = DeploymentStep("跳过构建", [], root, "检测到已有产物，跳过构建", kind="skip")

            def fake_runner(command, cwd, on_output, input_path=None, background=False):
                calls.append(command)
                return 0

            exit_code = run_steps([skip_step], logs.append, process_runner=fake_runner)

            self.assertEqual(exit_code, 0)
            self.assertEqual(calls, [])
            self.assertTrue(any("检测到已有产物" in line for line in logs))

    def test_plan_uses_configured_npm_path_for_frontend_start(self):
        with TemporaryDirectory() as temp:
            root = Path(temp)
            frontend = root / "web"
            npm = root / "nodejs" / "npm.cmd"
            frontend.mkdir()
            (frontend / "node_modules").mkdir()
            npm.parent.mkdir(parents=True)
            (frontend / "package.json").write_text(
                '{"scripts":{"build":"vite build","dev":"vite"}}', encoding="utf-8"
            )
            npm.write_text("", encoding="utf-8")
            scan = ScanResult(
                root=root,
                backend=None,
                frontend=ProjectComponent("Vue", frontend, "Vue project"),
                sql_files=[],
            )
            config = DeploymentConfig(tool_paths={"npm": npm})

            plan = build_deployment_plan(scan, config)

            self.assertEqual(
                plan[0].command,
                [str(npm), "run", "dev", "--", "--host", "0.0.0.0", "--port", "8081"],
            )

    def test_plan_installs_frontend_dependencies_before_start_when_node_modules_missing(self):
        with TemporaryDirectory() as temp:
            root = Path(temp)
            frontend = root / "web"
            frontend.mkdir()
            (frontend / "package.json").write_text(
                '{"scripts":{"build":"vue-cli-service build","serve":"vue-cli-service serve"},'
                '"devDependencies":{"@vue/cli-service":"^5.0.0"}}',
                encoding="utf-8",
            )
            scan = ScanResult(
                root=root,
                backend=None,
                frontend=ProjectComponent("Vue", frontend, "Vue project"),
                sql_files=[],
            )

            plan = build_deployment_plan(scan, DeploymentConfig(output_dir=root / "deploy", tool_search_roots=[root / "missing"]))

            npm = "npm.cmd" if os.name == "nt" else "npm"
            self.assertEqual(
                [step.kind for step in plan],
                ["frontend_install", "frontend_start", "deployment_summary"],
            )
            self.assertEqual(plan[0].command, [npm, "install"])
            self.assertEqual(plan[1].command[1:3], ["run", "serve"])

    def test_resolves_windows_cmd_shims_before_starting_process(self):
        with TemporaryDirectory() as temp:
            root = Path(temp)
            npm = root / "npm.cmd"
            npm.write_text("", encoding="utf-8")

            command = resolve_process_command(["npm", "run", "build"], path=str(root))

            self.assertEqual(os.path.normcase(command[0]), os.path.normcase(str(npm)))
            self.assertEqual(command[1:], ["run", "build"])


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