diff --git a/tests/_port_alloc.py b/tests/_port_alloc.py new file mode 100644 index 00000000..b7786fbf --- /dev/null +++ b/tests/_port_alloc.py @@ -0,0 +1,13 @@ +"""Shared free-port helper for svc tests (not a test module).""" +import socket + + +def free_test_port(): + """Return a port the kernel reports free on 127.0.0.1 right now. + + Each call opens a fresh reservation and closes it before returning, so the + caller (e.g. `svc deploy`) can bind the port itself. Ports are not pooled. + """ + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as reservation: + reservation.bind(("127.0.0.1", 0)) + return reservation.getsockname()[1] diff --git a/tests/test_port_alloc.py b/tests/test_port_alloc.py new file mode 100644 index 00000000..2d6303a2 --- /dev/null +++ b/tests/test_port_alloc.py @@ -0,0 +1,24 @@ +"""Regression for tests/_port_alloc.free_test_port.""" +import os +import socket +import sys +import unittest + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +from _port_alloc import free_test_port # noqa: E402 + + +class FreeTestPortTests(unittest.TestCase): + def test_two_calls_return_distinct_bindable_ports(self): + first = free_test_port() + second = free_test_port() + self.assertNotEqual(first, second) + for port in (first, second): + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as listener: + listener.bind(("127.0.0.1", port)) + listener.listen(1) + self.assertEqual(listener.getsockname()[1], port) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_svc.py b/tests/test_svc.py index 816cb8f7..e67d20e3 100644 --- a/tests/test_svc.py +++ b/tests/test_svc.py @@ -10,12 +10,16 @@ import os import shutil import subprocess import socket +import sys import tempfile import unittest import urllib.error import urllib.request from pathlib import Path +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +from _port_alloc import free_test_port # noqa: E402 + SOURCE = Path(__file__).resolve().parents[1] SVC_SCRIPT = SOURCE / "scripts" / "svc.sh" @@ -666,7 +670,7 @@ class SvcDbRecoveryTests(unittest.TestCase): 旧进程占口 → 报「非本服务」。env_config 须按实例清空 SVC_PIDCHECK 等。 """ old_sha, new_sha = self.make_src_repo() - ports = {"a": 18841, "b": 18842} + ports = {"a": free_test_port(), "b": free_test_port()} old_pids = {} for name, port in ports.items(): d = self.runtime(name) @@ -897,7 +901,8 @@ class SvcDbRecoveryTests(unittest.TestCase): def test_rdm_source_deploy_end_to_end(self): remote, work = self.make_rdm_repo() env = self.rdm_env(remote) - result = self.svc("rdm1", "deploy", "--from", "rdm", "--port", "18843", env=env) + result = self.svc("rdm1", "deploy", "--from", "rdm", "--port", str(free_test_port()), + env=env) d = self.runtime("rdm1") # U2：组合产物入不可变 release，经 current 指针解析；插件空集、工作台随 release。 active = d / "current" @@ -923,7 +928,7 @@ class SvcDbRecoveryTests(unittest.TestCase): def test_rdm_companion_missing_web_fails_deploy(self): remote, work = self.make_rdm_repo(with_web=False) env = self.rdm_env(remote) - result = self.svc("rdm2", "deploy", "--from", "rdm", "--port", "18843", + result = self.svc("rdm2", "deploy", "--from", "rdm", "--port", str(free_test_port()), env=env, check=False) self.assertNotEqual(result.returncode, 0) self.assertIn("ui-missing", result.stdout) @@ -934,18 +939,19 @@ class SvcDbRecoveryTests(unittest.TestCase): def test_source_switch_requires_flag(self): remote, work = self.make_rdm_repo() env = self.rdm_env(remote) + port = str(free_test_port()) # 先以 local 源部署（内嵌 UI 桩一律 200）。 - self.svc("sw1", "deploy", "--from", "local", "--port", "18844") + self.svc("sw1", "deploy", "--from", "local", "--port", port) old = self.kv("sw1", "source") self.assertTrue(old.startswith("local@")) # 无旗标换到 rdm → 拒绝，source 不变。 - denied = self.svc("sw1", "deploy", "--from", "rdm", "--port", "18844", + denied = self.svc("sw1", "deploy", "--from", "rdm", "--port", port, env=env, check=False) self.assertNotEqual(denied.returncode, 0) self.assertIn("allow-source-switch", denied.stdout) self.assertEqual(self.kv("sw1", "source"), old) # 带旗标允许换源，伴生就位。 - ok = self.svc("sw1", "deploy", "--from", "rdm", "--port", "18844", + ok = self.svc("sw1", "deploy", "--from", "rdm", "--port", port, "--allow-source-switch", env=env) self.assertTrue(self.kv("sw1", "source").startswith("rdm@")) self.assertIn("UI 就绪", ok.stdout) @@ -961,7 +967,7 @@ class SvcDbRecoveryTests(unittest.TestCase): self.assertIn("RDM 仓提交", result.stdout) self.assertIn("rdm v1", result.stdout) # 部署 v1 → 远端推进 v2 → update 检测可更新并真更新（source 保持 rdm@ 轨道）。 - self.svc("rdm1", "deploy", "--from", "rdm", "--port", "18842", env=env) + self.svc("rdm1", "deploy", "--from", "rdm", "--port", str(free_test_port()), env=env) old = self.kv("rdm1", "source") (work / "marker").write_text("v2\n") subprocess.run(["git", "add", "."], cwd=work, env=self.env, timeout=60, check=True) diff --git a/tests/test_svc_realcombo.py b/tests/test_svc_realcombo.py index 2eaf13c4..9e01c195 100644 --- a/tests/test_svc_realcombo.py +++ b/tests/test_svc_realcombo.py @@ -21,6 +21,7 @@ import shutil import socket import sqlite3 import subprocess +import sys import tempfile import time import unittest @@ -28,6 +29,9 @@ import urllib.error import urllib.request from pathlib import Path +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +from _port_alloc import free_test_port # noqa: E402 + SOURCE = Path(__file__).resolve().parents[1] REAL = os.environ.get("SVC_TEST_REAL") == "1" @@ -165,7 +169,8 @@ class RealComboTests(tsr.SvcAutoRecoverTests): def test_U3_real_agent_identity_lifecycle_sqlite(self): """真实组合二进制：领养 ensure → verify → 两次裸重启身份/锚点稳定； capabilities v1 探测契约。""" - self.seed_real("rc1", 9471, self.real_bin) + port = free_test_port() + self.seed_real("rc1", port, self.real_bin) result = self.svc("rc1", "start") self.assertIn("已启动", result.stdout) m = self.manifest("rc1") @@ -178,7 +183,7 @@ class RealComboTests(tsr.SvcAutoRecoverTests): self.assertTrue(con.execute( "SELECT count(*) FROM schema_migrations").fetchone()[0] >= 1) con.close() - self.assertTrue(self.http_ok(9471)) + self.assertTrue(self.http_ok(port)) # capabilities v1 契约（探测无副作用形态） cap = subprocess.run([str(self.runtime("rc1") / "bin" / "agent-system"), @@ -203,7 +208,7 @@ class RealComboTests(tsr.SvcAutoRecoverTests): self.assertEqual(self.identity_row("rc1"), [(iid, "sqlite")], "裸重启不得换身份") self.assertEqual(self.manifest("rc1")["instance_id"], iid) - self.assertTrue(self.http_ok(9471)) + self.assertTrue(self.http_ok(port)) self.svc("rc1", "stop") @unittest.skipUnless(REAL_RDM_BIN, "真实 RDM 专项 NOT_RUN：需 SVC_REAL_RDM_BIN") @@ -211,13 +216,14 @@ class RealComboTests(tsr.SvcAutoRecoverTests): """真实 RDM 组合二进制：identity 接线（Backend 公共 CLI）→ serve 门禁 → 领域层可初始化。""" rdm_bin = Path(REAL_RDM_BIN).resolve() - self.seed_real("rcr", 9473, rdm_bin) + port = free_test_port() + self.seed_real("rcr", port, rdm_bin) result = self.svc("rcr", "start") self.assertIn("已启动", result.stdout) m = self.manifest("rcr") iid = m["instance_id"] self.assertEqual(self.identity_row("rcr"), [(iid, "sqlite")]) - self.assertTrue(self.http_ok(9473)) + self.assertTrue(self.http_ok(port)) # RDM 未知子命令明确退出（不落服务路径） bogus = subprocess.run([str(self.runtime("rcr") / "bin" / "agent-system"), "totally-bogus"], @@ -236,7 +242,8 @@ class RealComboTests(tsr.SvcAutoRecoverTests): env.update(ci_git_env()) env.update({"SVC_RDM_SRC_DIR": str(self.root / "rdmcache"), "SVC_RDM_ORIGIN": RDM_ORIGIN}) - result = self.svc("rcf", "deploy", "--from", "rdm", "--port", "9474", + agent_port = free_test_port() + result = self.svc("rcf", "deploy", "--from", "rdm", "--port", str(agent_port), env=env, check=False) if result.returncode != 0: log = self.runtime("rcf") / "svc.log" @@ -259,7 +266,7 @@ class RealComboTests(tsr.SvcAutoRecoverTests): else "rdm", "工作台内容") self.svc("rcf", "restart") self.assertEqual(self.identity_row("rcf"), [(iid, "sqlite")]) - self.assertTrue(self.http_ok(9474)) + self.assertTrue(self.http_ok(agent_port)) self.svc("rcf", "stop") self.svc("rcfweb", "stop")