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)
|
|
|
|
|
|
# ── 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"
|