927d6927fd
Eliminates VMware-Tools IP-detect polling (observed: 39–177 s variance on
Linux host B7 benchmark). When [ip_pool] is configured the orchestrator:
1. Allocates a slot IP from a JSON-backed pool (fcntl-locked on Linux)
2. Injects guestinfo.ip-assignment / netmask / gateway into the cloned VMX
before vmrun start
3. Skips _wait_for_ip entirely — uses the pre-assigned IP immediately
4. Releases the slot in _destroy_clone (all paths: success/failure/interrupt)
New modules / changes:
src/ci_orchestrator/ip_pool.py IpSlotPool + _file_lock context manager
src/ci_orchestrator/config.py IpPoolConfig dataclass; [ip_pool] TOML
src/ci_orchestrator/commands/job.py _inject_guestinfo_ip helper; pool wiring
tests/python/test_ip_pool.py 12 unit tests (acquire/release/persist)
tests/python/test_config.py ip_pool config parsing tests
tests/python/test_commands_job.py pool path + _inject_guestinfo_ip tests
config.example.toml [ip_pool] section (commented example)
Coverage 90.02%; mypy --strict clean.
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
148 lines
4.6 KiB
Python
148 lines
4.6 KiB
Python
"""Tests for ci_orchestrator.config."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from ci_orchestrator.config import load_config
|
|
|
|
|
|
def test_defaults_are_os_aware() -> None:
|
|
cfg = load_config(env={})
|
|
if os.name == "nt":
|
|
assert str(cfg.paths.root) == r"F:\CI"
|
|
else:
|
|
assert str(cfg.paths.root) == "/var/lib/ci"
|
|
|
|
|
|
def test_env_overrides_paths(tmp_path: Path) -> None:
|
|
cfg = load_config(
|
|
env={
|
|
"CI_ROOT": str(tmp_path / "root"),
|
|
"CI_TEMPLATES": str(tmp_path / "tpl"),
|
|
"CI_BUILD_VMS": str(tmp_path / "bvm"),
|
|
"CI_ARTIFACTS": str(tmp_path / "art"),
|
|
"CI_KEYS": str(tmp_path / "keys"),
|
|
}
|
|
)
|
|
assert cfg.paths.root == tmp_path / "root"
|
|
assert cfg.paths.templates == tmp_path / "tpl"
|
|
assert cfg.paths.build_vms == tmp_path / "bvm"
|
|
assert cfg.paths.artifacts == tmp_path / "art"
|
|
assert cfg.paths.keys == tmp_path / "keys"
|
|
|
|
|
|
def test_toml_overrides_defaults(tmp_path: Path) -> None:
|
|
toml = tmp_path / "config.toml"
|
|
toml.write_text(
|
|
'[paths]\n'
|
|
f'root = "{(tmp_path / "altroot").as_posix()}"\n'
|
|
'[backend]\n'
|
|
'type = "workstation"\n'
|
|
'extra = "value"\n',
|
|
encoding="utf-8",
|
|
)
|
|
cfg = load_config(config_path=toml, env={})
|
|
assert cfg.paths.root == tmp_path / "altroot"
|
|
assert cfg.backend.type == "workstation"
|
|
assert cfg.backend.options == {"extra": "value"}
|
|
|
|
|
|
def test_env_wins_over_toml(tmp_path: Path) -> None:
|
|
toml = tmp_path / "config.toml"
|
|
toml.write_text(
|
|
f'[paths]\nroot = "{(tmp_path / "fromtoml").as_posix()}"\n',
|
|
encoding="utf-8",
|
|
)
|
|
cfg = load_config(
|
|
config_path=toml,
|
|
env={"CI_ROOT": str(tmp_path / "fromenv")},
|
|
)
|
|
assert cfg.paths.root == tmp_path / "fromenv"
|
|
|
|
|
|
def test_ci_config_env_loads_toml(tmp_path: Path) -> None:
|
|
toml = tmp_path / "alt.toml"
|
|
toml.write_text(
|
|
f'[paths]\nroot = "{(tmp_path / "rootenv").as_posix()}"\n', encoding="utf-8"
|
|
)
|
|
cfg = load_config(env={"CI_CONFIG": str(toml)})
|
|
assert cfg.paths.root == tmp_path / "rootenv"
|
|
|
|
|
|
def test_default_config_toml_in_root_is_loaded(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""When no config_path/env is given, ``<root>/config.toml`` is autoloaded."""
|
|
import ci_orchestrator.config as cfg_mod
|
|
from ci_orchestrator.config import Paths
|
|
|
|
fake = Paths(
|
|
root=tmp_path,
|
|
templates=tmp_path / "t",
|
|
build_vms=tmp_path / "b",
|
|
artifacts=tmp_path / "a",
|
|
keys=tmp_path / "k",
|
|
)
|
|
monkeypatch.setattr(cfg_mod, "_platform_defaults", lambda: fake)
|
|
(tmp_path / "config.toml").write_text(
|
|
'[backend]\ntype = "workstation"\noption = "auto"\n', encoding="utf-8"
|
|
)
|
|
cfg = load_config(env={})
|
|
assert cfg.backend.options == {"option": "auto"}
|
|
|
|
|
|
def test_toml_provides_ssh_key_path_and_guest_cred_target(tmp_path: Path) -> None:
|
|
toml = tmp_path / "config.toml"
|
|
key = tmp_path / "id_ed25519"
|
|
key.write_text("k")
|
|
toml.write_text(
|
|
f'ssh_key_path = "{key.as_posix()}"\n'
|
|
'guest_cred_target = "MyTarget"\n',
|
|
encoding="utf-8",
|
|
)
|
|
cfg = load_config(config_path=toml, env={})
|
|
assert cfg.ssh_key_path == key
|
|
assert cfg.guest_cred_target == "MyTarget"
|
|
|
|
|
|
def test_toml_ip_pool_parsed(tmp_path: Path) -> None:
|
|
lease = tmp_path / "ip-pool.json"
|
|
toml = tmp_path / "config.toml"
|
|
toml.write_text(
|
|
"[ip_pool]\n"
|
|
'addresses = ["192.168.79.201", "192.168.79.202"]\n'
|
|
'netmask = "255.255.255.0"\n'
|
|
'gateway = "192.168.79.2"\n'
|
|
f'lease_file = "{lease.as_posix()}"\n',
|
|
encoding="utf-8",
|
|
)
|
|
cfg = load_config(config_path=toml, env={})
|
|
assert cfg.ip_pool is not None
|
|
assert cfg.ip_pool.addresses == ["192.168.79.201", "192.168.79.202"]
|
|
assert cfg.ip_pool.netmask == "255.255.255.0"
|
|
assert cfg.ip_pool.gateway == "192.168.79.2"
|
|
assert cfg.ip_pool.lease_file == lease
|
|
|
|
|
|
def test_toml_ip_pool_default_lease_file(tmp_path: Path) -> None:
|
|
toml = tmp_path / "config.toml"
|
|
toml.write_text(
|
|
"[ip_pool]\n"
|
|
'addresses = ["192.168.79.201"]\n',
|
|
encoding="utf-8",
|
|
)
|
|
cfg = load_config(config_path=toml, env={})
|
|
assert cfg.ip_pool is not None
|
|
assert cfg.ip_pool.lease_file.name == "ip-pool.json"
|
|
|
|
|
|
def test_ip_pool_absent_when_not_configured(tmp_path: Path) -> None:
|
|
toml = tmp_path / "config.toml"
|
|
toml.write_text("[backend]\ntype = \"workstation\"\n", encoding="utf-8")
|
|
cfg = load_config(config_path=toml, env={})
|
|
assert cfg.ip_pool is None
|