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

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.

cloud/tests/test_docker_sandbox.py

@@ -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

@@ -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"),
+        Path(os.path.join(self.template_home, "config.toml")).write_text(
-             "w").write('[model_provider.xcmapi]\n'
+            '[model_provider.xcmapi]\n'
-                         'base_url="https://xcmapi.org/v1"\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"),
+        Path(os.path.join(self.template_repo, "sample.py")).write_text(
-             "w").write("def add(a, b):\n    return a + b\n")
+            "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 = Path(os.path.join(dirs["codex_home"],
-                                   "config.toml")).read()
+                                   "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

@@ -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

@@ -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(
+            Path(os.path.join(d, "SKILL.md")).write_text(
-                "---\nname: %s\ndescription: test\n---\nbody\n" % name)
+                "---\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"),
+        Path(os.path.join(self.template_home, "config.toml")).write_text(
-             "w").write('[model_provider.xcmapi]\n'
+            '[model_provider.xcmapi]\n'
-                         'base_url="https://xcmapi.org/v1"\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"),
+        Path(os.path.join(self.template_repo, "sample.py")).write_text(
-             "w").write("def add(a, b):\n    return a + b\n")
+            "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

@@ -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"),
+        Path(os.path.join(self.template_home, "config.toml")).write_text(
-             "w").write('[model_provider.xcmapi]\nbase_url="https://xcmapi.org/v1"\n')
+            '[model_provider.xcmapi]\nbase_url="https://xcmapi.org/v1"\n',
-        open(os.path.join(self.template_home, "thread_history_1.sqlite"),
+            encoding="utf-8")
-             "w").write("OTHER TENANT HISTORY")
+        Path(os.path.join(self.template_home, "thread_history_1.sqlite")).write_text(
-        open(os.path.join(self.template_home, "auth.json"),
+            "OTHER TENANT HISTORY", encoding="utf-8")
-             "w").write("SECRET")
+        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"),
+        Path(os.path.join(self.template_repo, "sample.py")).write_text(
-             "w").write("def add(a, b):\n    return a + b\n")
+            "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"],
+        Path(os.path.join(worker._tenant_dirs("projA")["work"],
-                          "secret-a.txt"), "w").write("SECRET-A")
+                          "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

@@ -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"),
+        Path(os.path.join(self.template_home, "config.toml")).write_text(
-             "w").write('[model_provider.xcmapi]\n'
+            '[model_provider.xcmapi]\n'
-                         'base_url="https://xcmapi.org/v1"\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"),
+        Path(os.path.join(self.template_repo, "sample.py")).write_text(
-             "w").write("def add(a, b):\n    return a + b\n")
+            "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 = Path(os.path.join(dirs["codex_home"],
-                                   "config.toml")).read()
+                                   "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" /
+        lines = (REPO_ROOT / "deploy" / "egress" /
-                     "tinyproxy-filter").read().splitlines()
+                 "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

deploy/tests/test_host_deploy.py

@@ -10,6 +10,7 @@ import subprocess
 import tarfile
 import tempfile
 import unittest
+from pathlib import Path
 
 REPO_ROOT = os.path.dirname(os.path.dirname(os.path.dirname(
     os.path.abspath(__file__))))
@@ -59,7 +60,7 @@ class HostDeploySim(unittest.TestCase):
         self.fake_bin = fake_bin
         self.health_path = os.path.join(base, "health.json")
         self.log_path = os.path.join(base, "systemctl.log")
-        open(self.log_path, "w").close()
+        Path(self.log_path).write_text("", encoding="utf-8")
         self.env = dict(os.environ)
         self.env["PATH"] = fake_bin + os.pathsep + os.environ["PATH"]
         self.env["CODEX_DEPLOY_APP_DIR"] = self.app

deploy/tests/test_p2_ops.py

@@ -13,6 +13,7 @@ import tarfile
 import tempfile
 import threading
 import unittest
+from pathlib import Path
 
 REPO_ROOT = os.path.dirname(os.path.dirname(os.path.dirname(
     os.path.abspath(__file__))))
@@ -61,7 +62,7 @@ class MonitorSim(unittest.TestCase):
                    "exit 0\n")
         real_curl = REAL_CURL or "/usr/bin/curl"
         self.curl_log = os.path.join(base, "curl.log")
-        open(self.curl_log, "w").close()
+        Path(self.curl_log).write_text("", encoding="utf-8")
         _write_exe(os.path.join(fake_bin, "curl"),
                    "#!/bin/sh\n"
                    "echo \"$@\" >> \"$FAKE_CURL_LOG\"\n"
@@ -200,7 +201,7 @@ class RollbackSim(unittest.TestCase):
         fake_bin = os.path.join(base, "fakebin")
         os.makedirs(fake_bin)
         self.log_path = os.path.join(base, "systemctl.log")
-        open(self.log_path, "w").close()
+        Path(self.log_path).write_text("", encoding="utf-8")
         _write_exe(os.path.join(fake_bin, "ssh"),
                    "#!/bin/sh\n"
                    "# 假 ssh：忽略目标参数，直接把 heredoc 脚本在本地跑。\n"

tests/test_host_thread_cli.py

@@ -252,7 +252,7 @@ class ThreadCliTest(unittest.TestCase):
                             code = e.code if isinstance(e.code, int) else 1
             self.assertEqual(code, 0)
             self.assertIn("archived t1", out.getvalue())
-            saved = json.load(open(state_path, encoding="utf-8"))
+            saved = json.loads(Path(state_path).read_text(encoding="utf-8"))
             self.assertEqual(saved["tasks"]["t1"]["status"], "archived")
             calls = self._calls_before_close(fake)
             self.assertEqual(calls, [("archive", "th-1")])