openqj
7b537c31a6
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.
97 lines
4.0 KiB
Python
97 lines
4.0 KiB
Python
"""Worker wiring for image generation MCP: tenant seeding + env.
|
|
|
|
No Docker, no network: _tenant_dirs runs against temp dirs, and
|
|
the MCP env helper only edits the tenant config.toml.
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
try:
|
|
from cloud.server import init_db
|
|
from cloud.worker import Worker
|
|
from cloud.imagegen import ensure_imagegen_mcp_env
|
|
except ImportError:
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
|
|
from cloud.server import init_db
|
|
from cloud.worker import Worker
|
|
from cloud.imagegen import ensure_imagegen_mcp_env
|
|
|
|
|
|
class WorkerImagegenTestCase(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.db_path = os.path.join(self.tmp.name, "control.db")
|
|
init_db(self.db_path)
|
|
self.template_home = os.path.join(self.tmp.name, "tpl-home")
|
|
os.makedirs(self.template_home)
|
|
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"))
|
|
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()
|
|
|
|
def _worker(self, **kwargs):
|
|
kwargs.setdefault("env", {"XCM_API_KEY": "test-placeholder"})
|
|
kwargs.setdefault("tenants_root",
|
|
os.path.join(self.tmp.name, "tenants"))
|
|
kwargs.setdefault("tenant_template_home", self.template_home)
|
|
kwargs.setdefault("tenant_template_repo", self.template_repo)
|
|
kwargs.setdefault("tenant_isolation", True)
|
|
return Worker(self.db_path, "/fake/host.py", **kwargs)
|
|
|
|
def test_flag_off_is_old_behaviour(self):
|
|
worker = self._worker()
|
|
self.assertFalse(worker.imagegen_mcp)
|
|
dirs = worker._tenant_dirs("projA")
|
|
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")))
|
|
|
|
def test_flag_on_seeds_container_path_and_scripts(self):
|
|
worker = self._worker(imagegen_mcp=True)
|
|
dirs = worker._tenant_dirs("projA")
|
|
home = dirs["codex_home"]
|
|
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)
|
|
self.assertNotIn(self.tmp.name, config)
|
|
self.assertTrue(os.path.exists(
|
|
os.path.join(home, "imagegen.py")))
|
|
self.assertTrue(os.path.exists(
|
|
os.path.join(home, "imagegen_mcp.py")))
|
|
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):
|
|
worker = self._worker(imagegen_mcp=True)
|
|
dirs = worker._tenant_dirs("projA")
|
|
home = dirs["codex_home"]
|
|
self.assertTrue(ensure_imagegen_mcp_env(
|
|
home, {"IMAGE_GEN_API_KEY": "k",
|
|
"IMAGE_GEN_MODEL": "gpt-image-2.5-flare",
|
|
"NO_PROXY": "localhost,127.0.0.1"}))
|
|
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.
|
|
self.assertFalse(ensure_imagegen_mcp_env(
|
|
home, {"IMAGE_GEN_API_KEY": "k",
|
|
"IMAGE_GEN_MODEL": "gpt-image-2.5-flare",
|
|
"NO_PROXY": "localhost,127.0.0.1"}))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|