675 lines
27 KiB
Python
675 lines
27 KiB
Python
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()
|