File size: 1,904 Bytes
b641d3d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 | import json
from pathlib import Path
from server.constants import DEFAULT_CONFIGS
from server.env import DistributedDebugEnv
def _write_json(path: Path, payload: dict) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8")
def test_restore_defaults_adds_job_generator_config(tmp_path: Path) -> None:
mesh_root = tmp_path / "mesh"
env = DistributedDebugEnv(project_root=tmp_path, mesh_root=mesh_root)
env._restore_defaults()
payload = json.loads(
(mesh_root / "worker" / "job_generator_config.json").read_text()
)
assert payload == DEFAULT_CONFIGS["job_generator"]
def test_registry_auth_matches_default_detects_corruption(tmp_path: Path) -> None:
mesh_root = tmp_path / "mesh"
env = DistributedDebugEnv(project_root=tmp_path, mesh_root=mesh_root)
env._restore_defaults()
assert env._is_registry_auth_default() is True
_write_json(
mesh_root / "registry.json",
{
"services": {
"auth": {"host": "invalid-host", "port": 3001, "protocol": "http"},
"redis": {"host": "localhost", "port": 6379, "protocol": "tcp"},
"worker": {"host": "localhost", "port": None, "protocol": "internal"},
}
},
)
assert env._is_registry_auth_default() is False
def test_job_generator_rate_resolved_uses_config(tmp_path: Path) -> None:
mesh_root = tmp_path / "mesh"
env = DistributedDebugEnv(project_root=tmp_path, mesh_root=mesh_root)
env._restore_defaults()
assert env._job_generator_interval_ms() == 333
assert env._is_job_generator_rate_resolved() is True
_write_json(mesh_root / "worker" / "job_generator_config.json", {"interval_ms": 10})
assert env._job_generator_interval_ms() == 10
assert env._is_job_generator_rate_resolved() is False
|