Simone
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>
118 lines
3.9 KiB
Python
118 lines
3.9 KiB
Python
"""Tests for ci_orchestrator.ip_pool — IpSlotPool acquire/release."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from ci_orchestrator.ip_pool import IpSlotPool
|
|
|
|
|
|
def make_pool(tmp_path: Path, addresses: list[str] | None = None) -> IpSlotPool:
|
|
return IpSlotPool(
|
|
addresses=addresses or ["10.0.0.1", "10.0.0.2"],
|
|
netmask="255.255.255.0",
|
|
gateway="10.0.0.254",
|
|
lease_file=tmp_path / "ip-pool.json",
|
|
)
|
|
|
|
|
|
# ── construction ────────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_empty_addresses_raises(tmp_path: Path) -> None:
|
|
with pytest.raises(ValueError, match="at least one"):
|
|
IpSlotPool([], "255.255.255.0", "", tmp_path / "pool.json")
|
|
|
|
|
|
# ── acquire ─────────────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_acquire_returns_first_free_ip(tmp_path: Path) -> None:
|
|
pool = make_pool(tmp_path)
|
|
ip = pool.acquire("owner-1")
|
|
assert ip == "10.0.0.1"
|
|
|
|
|
|
def test_acquire_second_slot(tmp_path: Path) -> None:
|
|
pool = make_pool(tmp_path)
|
|
pool.acquire("owner-1")
|
|
ip2 = pool.acquire("owner-2")
|
|
assert ip2 == "10.0.0.2"
|
|
|
|
|
|
def test_acquire_writes_lease_file(tmp_path: Path) -> None:
|
|
pool = make_pool(tmp_path)
|
|
pool.acquire("owner-1")
|
|
leases = json.loads((tmp_path / "ip-pool.json").read_text())
|
|
assert leases["10.0.0.1"] == "owner-1"
|
|
assert leases["10.0.0.2"] is None
|
|
|
|
|
|
def test_acquire_exhausted_raises(tmp_path: Path) -> None:
|
|
pool = make_pool(tmp_path, addresses=["10.0.0.1"])
|
|
pool.acquire("owner-1")
|
|
with pytest.raises(RuntimeError, match="exhausted"):
|
|
pool.acquire("owner-2", timeout=0.1)
|
|
|
|
|
|
# ── release ──────────────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_release_frees_slot(tmp_path: Path) -> None:
|
|
pool = make_pool(tmp_path)
|
|
ip = pool.acquire("owner-1")
|
|
pool.release(ip)
|
|
leases = json.loads((tmp_path / "ip-pool.json").read_text())
|
|
assert leases[ip] is None
|
|
|
|
|
|
def test_release_unknown_ip_is_noop(tmp_path: Path) -> None:
|
|
pool = make_pool(tmp_path)
|
|
pool.release("99.99.99.99") # should not raise
|
|
|
|
|
|
def test_release_and_reacquire(tmp_path: Path) -> None:
|
|
pool = make_pool(tmp_path, addresses=["10.0.0.1"])
|
|
ip = pool.acquire("owner-1")
|
|
pool.release(ip)
|
|
ip2 = pool.acquire("owner-2")
|
|
assert ip2 == "10.0.0.1"
|
|
|
|
|
|
# ── persistence ──────────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_second_pool_instance_sees_leases(tmp_path: Path) -> None:
|
|
lease = tmp_path / "ip-pool.json"
|
|
p1 = IpSlotPool(["10.0.0.1", "10.0.0.2"], "255.255.255.0", "", lease)
|
|
p1.acquire("owner-1")
|
|
|
|
p2 = IpSlotPool(["10.0.0.1", "10.0.0.2"], "255.255.255.0", "", lease)
|
|
ip = p2.acquire("owner-2")
|
|
assert ip == "10.0.0.2"
|
|
|
|
|
|
def test_corrupt_lease_file_treated_as_empty(tmp_path: Path) -> None:
|
|
lease = tmp_path / "ip-pool.json"
|
|
lease.write_text("NOT JSON", encoding="utf-8")
|
|
pool = make_pool(tmp_path)
|
|
ip = pool.acquire("owner-1")
|
|
assert ip == "10.0.0.1"
|
|
|
|
|
|
def test_missing_lease_file_treated_as_empty(tmp_path: Path) -> None:
|
|
pool = make_pool(tmp_path)
|
|
ip = pool.acquire("owner-1")
|
|
assert ip == "10.0.0.1"
|
|
|
|
|
|
def test_non_dict_lease_json_treated_as_empty(tmp_path: Path) -> None:
|
|
lease = tmp_path / "ip-pool.json"
|
|
lease.write_text("[1, 2, 3]", encoding="utf-8") # valid JSON but not a dict
|
|
pool = make_pool(tmp_path)
|
|
ip = pool.acquire("owner-1")
|
|
assert ip == "10.0.0.1"
|