admin/codex-app-server
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

test: close unclosed file handles in tests (Path read/write)

Browse Source 
Replace bare open().read()/write() in tests with Path.read_text/
write_text (or json.loads over read_text); the five remaining
open(...).close() empty-file creations become Path.write_text('').
No assertion, predicate, or production code changed; no behaviour
change beyond closing the handle deterministically.

Long-function splits and server.py module split intentionally not
included: the longest functions (_dispatch, _create_task,
_dispatch_project, ~375-395 lines in a 10430-line server.py) are
central routing/creation paths, and splitting them is not low-risk;
see PR body.
This commit is contained in:
openqj authored and host committed 2026-11-04 20:53:46 +00:00
1 parent fa4588a119
commit d779a1b426
9 files changed
+61 -57
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

cloud/tests/test_docker_sandbox.py
+1 -1
View File
@@ -347,7 +347,7 @@ class TestDockerExtraMounts(DockerCommandTestCase):
work = os.path.join(td, "work")
os.makedirs(work)
state = os.path.join(td, "host-state.json")
open(state, "w").write("{}")
Path(state).write_text("{}", encoding="utf-8")
with mock.patch("cloud.sandbox.shutil.which",
return_value="/usr/bin/docker"):
ex = DockerExecutor("/fake/host.py", runner=_ok_runner_factory([]),
cloud/tests/test_imagegen_worker.py
+10 -10
View File
@@ -28,13 +28,13 @@ class WorkerImagegenTestCase(unittest.TestCase):
init_db(self.db_path)
self.template_home = os.path.join(self.tmp.name, "tpl-home")
os.makedirs(self.template_home)
open(os.path.join(self.template_home, "config.toml"),
"w").write('[model_provider.xcmapi]\n'
'base_url="https://xcmapi.org/v1"\n')
Path(os.path.join(self.template_home, "config.toml")).write_text(
'[model_provider.xcmapi]\n'
'base_url="https://xcmapi.org/v1"\n', encoding="utf-8")
self.template_repo = os.path.join(self.tmp.name, "tpl-repo")
os.makedirs(os.path.join(self.template_repo, ".git"))
open(os.path.join(self.template_repo, "sample.py"),
"w").write("def add(a, b):\n return a + b\n")
Path(os.path.join(self.template_repo, "sample.py")).write_text(
"def add(a, b):\n return a + b\n", encoding="utf-8")
def tearDown(self):
self.tmp.cleanup()
@@ -52,8 +52,8 @@ class WorkerImagegenTestCase(unittest.TestCase):
worker = self._worker()
self.assertFalse(worker.imagegen_mcp)
dirs = worker._tenant_dirs("projA")
config = open(os.path.join(dirs["codex_home"],
"config.toml")).read()
config = Path(os.path.join(dirs["codex_home"],
"config.toml")).read_text(encoding="utf-8")
self.assertNotIn("[mcp_servers.imagegen]", config)
self.assertFalse(os.path.exists(
os.path.join(dirs["codex_home"], "imagegen_mcp.py")))
@@ -62,7 +62,7 @@ class WorkerImagegenTestCase(unittest.TestCase):
worker = self._worker(imagegen_mcp=True)
dirs = worker._tenant_dirs("projA")
home = dirs["codex_home"]
config = open(os.path.join(home, "config.toml")).read()
config = Path(os.path.join(home, "config.toml")).read_text(encoding="utf-8")
self.assertIn("[mcp_servers.imagegen]", config)
self.assertIn("/var/lib/codex-app-server/codex-home/"
"imagegen_mcp.py", config)
@@ -71,7 +71,7 @@ class WorkerImagegenTestCase(unittest.TestCase):
os.path.join(home, "imagegen.py")))
self.assertTrue(os.path.exists(
os.path.join(home, "imagegen_mcp.py")))
script = open(os.path.join(home, "imagegen_mcp.py")).read()
script = Path(os.path.join(home, "imagegen_mcp.py")).read_text(encoding="utf-8")
self.assertIn("image_generate", script)
def test_env_helper_writes_imagegen_section_only(self):
@@ -82,7 +82,7 @@ class WorkerImagegenTestCase(unittest.TestCase):
home, {"IMAGE_GEN_API_KEY": "k",
"IMAGE_GEN_MODEL": "gpt-image-2.5-flare",
"NO_PROXY": "localhost,127.0.0.1"}))
config = open(os.path.join(home, "config.toml")).read()
config = Path(os.path.join(home, "config.toml")).read_text(encoding="utf-8")
self.assertIn("[mcp_servers.imagegen.env]", config)
self.assertIn("gpt-image-2.5-flare", config)
# Idempotent: same values again is a no-op.
cloud/tests/test_ops_batch.py
+2 -2
View File
@@ -79,7 +79,7 @@ class RestorePostgresTestCase(unittest.TestCase):
def test_empty_backup_file_rejected(self):
empty = os.path.join(self.tmp.name, "empty.sql")
open(empty, "w").close()
Path(empty).write_text("", encoding="utf-8")
with self.assertRaises(ValueError):
restore_postgres(empty, "restore_test",
runner=lambda c, e: _fake_result())
@@ -126,7 +126,7 @@ class WorkerHeartbeatTestCase(unittest.TestCase):
# 心跳目录不可写时只记日志，不抛异常（写到一个文件路径下当目录）。
tmp = tempfile.TemporaryDirectory()
blocker = os.path.join(tmp.name, "blocker")
open(blocker, "w").close()
Path(blocker).write_text("", encoding="utf-8")
worker = Worker(os.path.join(tmp.name, "db.db"), "/nonexistent/host.py",
env_key="TEST_KEY",
heartbeat_file=os.path.join(blocker, "hb.json"))
cloud/tests/test_skill_seed_worker.py
+16 -15
View File
@@ -35,8 +35,9 @@ class SkillSeedHelperTestCase(unittest.TestCase):
for name in ("live-info", "document-tools"):
d = os.path.join(src, name)
os.makedirs(d, exist_ok=True)
open(os.path.join(d, "SKILL.md"), "w").write(
"---\nname: %s\ndescription: test\n---\nbody\n" % name)
Path(os.path.join(d, "SKILL.md")).write_text(
"---\nname: %s\ndescription: test\n---\nbody\n" % name,
encoding="utf-8")
return src
def test_copy_idempotent_and_missing_source(self):
@@ -58,10 +59,10 @@ class SkillSeedHelperTestCase(unittest.TestCase):
ensure_codex_home_skills(home, src)
dest = os.path.join(home, "skills",
"live-info", "SKILL.md")
open(dest, "w").write("tenant edited version\n")
Path(dest).write_text("tenant edited version\n", encoding="utf-8")
out = ensure_codex_home_skills(home, src)
self.assertEqual(out["skills_skipped_differ"], ["live-info"])
self.assertEqual(open(dest).read(), "tenant edited version\n")
self.assertEqual(Path(dest).read_text(encoding="utf-8"), "tenant edited version\n")
def test_real_pack_lists_all_skills(self):
names = [n for n, _p in list_skills()]
@@ -78,13 +79,13 @@ class WorkerSkillSeedTestCase(unittest.TestCase):
init_db(self.db_path)
self.template_home = os.path.join(self.tmp.name, "tpl-home")
os.makedirs(self.template_home)
open(os.path.join(self.template_home, "config.toml"),
"w").write('[model_provider.xcmapi]\n'
'base_url="https://xcmapi.org/v1"\n')
Path(os.path.join(self.template_home, "config.toml")).write_text(
'[model_provider.xcmapi]\n'
'base_url="https://xcmapi.org/v1"\n', encoding="utf-8")
self.template_repo = os.path.join(self.tmp.name, "tpl-repo")
os.makedirs(os.path.join(self.template_repo, ".git"))
open(os.path.join(self.template_repo, "sample.py"),
"w").write("def add(a, b):\n return a + b\n")
Path(os.path.join(self.template_repo, "sample.py")).write_text(
"def add(a, b):\n return a + b\n", encoding="utf-8")
def tearDown(self):
self.tmp.cleanup()
@@ -115,22 +116,22 @@ class WorkerSkillSeedTestCase(unittest.TestCase):
for name in ("live-info", "document-tools", "image-workflow"):
path = self._skill_path(dirs, name)
self.assertTrue(os.path.exists(path), path)
self.assertIn("name: %s" % name, open(path).read())
self.assertIn("name: %s" % name, Path(path).read_text(encoding="utf-8"))
def test_flag_on_seeds_software_engineering_and_user_edit_survives(self):
worker = self._worker(seed_skills=True)
dirs = worker._tenant_dirs("projA")
path = self._skill_path(dirs, "software-engineering")
self.assertTrue(os.path.exists(path), path)
text = open(path).read()
text = Path(path).read_text(encoding="utf-8")
self.assertIn("name: software-engineering", text)
# v0.2 P0-1: the appendix travels inside SKILL.md itself,
# because skill_seed only copies SKILL.md, not templates/.
self.assertIn("附录角色：00-coordinator", text)
self.assertIn("附录角色：08-shared-core", text)
open(path, "w").write("tenant edited version\n")
Path(path).write_text("tenant edited version\n", encoding="utf-8")
worker._tenant_dirs("projA")
self.assertEqual(open(path).read(), "tenant edited version\n")
self.assertEqual(Path(path).read_text(encoding="utf-8"), "tenant edited version\n")
def test_old_tenant_merges_and_user_edit_survives(self):
worker = self._worker()
@@ -139,9 +140,9 @@ class WorkerSkillSeedTestCase(unittest.TestCase):
worker2._tenant_dirs("projA")
path = self._skill_path(dirs, "live-info")
self.assertTrue(os.path.exists(path))
open(path, "w").write("tenant edited version\n")
Path(path).write_text("tenant edited version\n", encoding="utf-8")
worker2._tenant_dirs("projA")
self.assertEqual(open(path).read(), "tenant edited version\n")
self.assertEqual(Path(path).read_text(encoding="utf-8"), "tenant edited version\n")
if __name__ == "__main__":
cloud/tests/test_tenant_isolation.py
+12 -11
View File
@@ -41,16 +41,17 @@ class TenantDirsTestCase(unittest.TestCase):
# 与假密钥文件；模板 repo：sample.py + .git。
self.template_home = os.path.join(self.tmp.name, "tpl-home")
os.makedirs(self.template_home)
open(os.path.join(self.template_home, "config.toml"),
"w").write('[model_provider.xcmapi]\nbase_url="https://xcmapi.org/v1"\n')
open(os.path.join(self.template_home, "thread_history_1.sqlite"),
"w").write("OTHER TENANT HISTORY")
open(os.path.join(self.template_home, "auth.json"),
"w").write("SECRET")
Path(os.path.join(self.template_home, "config.toml")).write_text(
'[model_provider.xcmapi]\nbase_url="https://xcmapi.org/v1"\n',
encoding="utf-8")
Path(os.path.join(self.template_home, "thread_history_1.sqlite")).write_text(
"OTHER TENANT HISTORY", encoding="utf-8")
Path(os.path.join(self.template_home, "auth.json")).write_text(
"SECRET", encoding="utf-8")
self.template_repo = os.path.join(self.tmp.name, "tpl-repo")
os.makedirs(os.path.join(self.template_repo, ".git"))
open(os.path.join(self.template_repo, "sample.py"),
"w").write("def add(a, b):\n return a + b\n")
Path(os.path.join(self.template_repo, "sample.py")).write_text(
"def add(a, b):\n return a + b\n", encoding="utf-8")
def tearDown(self):
self.tmp.cleanup()
@@ -77,7 +78,7 @@ class TenantDirsTestCase(unittest.TestCase):
os.path.join(dirs["test_repo"], "sample.py")))
# 第二次调用不覆盖租户自己的改动。
marker = os.path.join(dirs["test_repo"], "mine.txt")
open(marker, "w").write("mine")
Path(marker).write_text("mine", encoding="utf-8")
worker._tenant_dirs("projA")
self.assertTrue(os.path.exists(marker))
@@ -95,8 +96,8 @@ class TenantDirsTestCase(unittest.TestCase):
hosts_b = {m.split(":")[0] for m in mounts_b}
self.assertFalse(hosts_a & hosts_b)
# 租户 A 的 work 里写文件，B 的挂载里看不到（宿主侧）。
open(os.path.join(worker._tenant_dirs("projA")["work"],
"secret-a.txt"), "w").write("SECRET-A")
Path(os.path.join(worker._tenant_dirs("projA")["work"],
"secret-a.txt")).write_text("SECRET-A", encoding="utf-8")
self.assertFalse(os.path.exists(os.path.join(
worker._tenant_dirs("projB")["work"], "secret-a.txt")))
cloud/tests/test_websearch_worker.py
+14 -14
View File
@@ -29,13 +29,13 @@ class WorkerWebsearchTestCase(unittest.TestCase):
init_db(self.db_path)
self.template_home = os.path.join(self.tmp.name, "tpl-home")
os.makedirs(self.template_home)
open(os.path.join(self.template_home, "config.toml"),
"w").write('[model_provider.xcmapi]\n'
'base_url="https://xcmapi.org/v1"\n')
Path(os.path.join(self.template_home, "config.toml")).write_text(
'[model_provider.xcmapi]\n'
'base_url="https://xcmapi.org/v1"\n', encoding="utf-8")
self.template_repo = os.path.join(self.tmp.name, "tpl-repo")
os.makedirs(os.path.join(self.template_repo, ".git"))
open(os.path.join(self.template_repo, "sample.py"),
"w").write("def add(a, b):\n return a + b\n")
Path(os.path.join(self.template_repo, "sample.py")).write_text(
"def add(a, b):\n return a + b\n", encoding="utf-8")
def tearDown(self):
self.tmp.cleanup()
@@ -53,8 +53,8 @@ class WorkerWebsearchTestCase(unittest.TestCase):
worker = self._worker()
self.assertFalse(worker.websearch_mcp)
dirs = worker._tenant_dirs("projA")
config = open(os.path.join(dirs["codex_home"],
"config.toml")).read()
config = Path(os.path.join(dirs["codex_home"],
"config.toml")).read_text(encoding="utf-8")
self.assertNotIn("[mcp_servers.websearch]", config)
self.assertFalse(os.path.exists(
os.path.join(dirs["codex_home"], "websearch_mcp.py")))
@@ -63,7 +63,7 @@ class WorkerWebsearchTestCase(unittest.TestCase):
worker = self._worker(websearch_mcp=True)
dirs = worker._tenant_dirs("projA")
home = dirs["codex_home"]
config = open(os.path.join(home, "config.toml")).read()
config = Path(os.path.join(home, "config.toml")).read_text(encoding="utf-8")
self.assertIn("[mcp_servers.websearch]", config)
# Container path, not the host temp path: this config is
# read inside the task container, where the tenant home is
@@ -76,7 +76,7 @@ class WorkerWebsearchTestCase(unittest.TestCase):
self.assertTrue(os.path.exists(
os.path.join(home, "websearch_mcp.py")))
# The copied MCP script is the real one (not a stub).
script = open(os.path.join(home, "websearch_mcp.py")).read()
script = Path(os.path.join(home, "websearch_mcp.py")).read_text(encoding="utf-8")
self.assertIn("web_search", script)
self.assertIn("web_fetch", script)
@@ -85,18 +85,18 @@ class WorkerWebsearchTestCase(unittest.TestCase):
dirs = worker._tenant_dirs("projA")
config_path = os.path.join(dirs["codex_home"], "config.toml")
self.assertNotIn("[mcp_servers.websearch]",
open(config_path).read())
Path(config_path).read_text(encoding="utf-8"))
# Flag flips on later (deploy enables it): next task's
# _tenant_dirs must merge, not require a new project.
worker2 = self._worker(websearch_mcp=True)
worker2._tenant_dirs("projA")
config = open(config_path).read()
config = Path(config_path).read_text(encoding="utf-8")
self.assertIn("[mcp_servers.websearch]", config)
self.assertIn("xcmapi.org", config)
self.assertEqual(config.count("[mcp_servers.websearch]"), 1)
# Third call stays idempotent.
worker2._tenant_dirs("projA")
config = open(config_path).read()
config = Path(config_path).read_text(encoding="utf-8")
self.assertEqual(config.count("[mcp_servers.websearch]"), 1)
@@ -152,8 +152,8 @@ class DockerEnvTestCase(unittest.TestCase):
class EgressFilterTestCase(unittest.TestCase):
def test_search_domains_present_in_basic_form(self):
lines = open(REPO_ROOT / "deploy" / "egress" /
"tinyproxy-filter").read().splitlines()
lines = (REPO_ROOT / "deploy" / "egress" /
"tinyproxy-filter").read_text(encoding="utf-8").splitlines()
entries = [ln for ln in lines
if ln and not ln.startswith("#")]
# No extended-regex syntax in any entry: + ? | and bare