ipam_control/tests/test_api.py

import ipaddress
import uuid

import httpx

from tests.conftest import BASE, ENV


def _prefix(client, org, cidr, **kw):
    return client.post("/prefixes", json={"organization_id": org["id"], "vrf_id": org["vrf_id"], "prefix": cidr, **kw})


def test_auth_required_and_login():
    anon = httpx.Client(base_url=BASE)
    assert anon.get("/prefixes").status_code == 401
    assert anon.post("/auth/login", json={"username": ENV["ADMIN_USERNAME"], "password": "wrong"}).status_code == 401
    ok = anon.post("/auth/login", json={"username": ENV["ADMIN_USERNAME"], "password": ENV["ADMIN_PASSWORD"]})
    assert ok.status_code == 200 and ok.json()["access_token"]


def test_prefix_and_address_validation(client, org):
    assert _prefix(client, org, "10.201.0.0/24").status_code == 201
    assert _prefix(client, org, "10.201.0.0/24").status_code == 409  # дубль в VRF
    assert _prefix(client, org, "10.201.1.5/24").status_code == 422  # биты хоста
    pid = client.get("/prefixes", params={"organization_id": org["id"]}).json()["items"][0]["id"]
    assert client.post(f"/prefixes/{pid}/addresses", json={"address": "10.202.0.1"}).status_code == 422  # вне префикса


def test_tree_utilization_and_next_free(client, org):
    parent = _prefix(client, org, "10.203.0.0/16").json()
    leaf = _prefix(client, org, "10.203.1.0/29", is_pool=True).json()
    assert leaf["parent_id"] == parent["id"]
    a = client.post(f"/prefixes/{leaf['id']}/addresses/next").json()
    assert a["address"] == "10.203.1.1"
    client.post(f"/prefixes/{leaf['id']}/addresses", json={"address": "10.203.1.2", "status": "reserved"})
    page = client.get(f"/prefixes/{leaf['id']}/addresses").json()
    assert page["summary"] == {"assigned": 1, "reserved": 1, "deprecated": 0, "free": 4, "capacity": 6}

    # изменение 042: подряд идущие свободные адреса (.3–.6) сворачиваются в одну строку free_range;
    # раскрытие — тот же GET .../addresses?status=free&offset=<free_offset>&limit=<range_count>
    grouped = client.get(f"/prefixes/{leaf['id']}/addresses", params={"group_free": "true"}).json()
    rng = next(x for x in grouped["items"] if x["status"] == "free_range")
    assert rng["address"] == "10.203.1.3" and rng["range_end"] == "10.203.1.6" and rng["range_count"] == 4 and rng["free_offset"] == 0
    expanded = client.get(f"/prefixes/{leaf['id']}/addresses", params={"status": "free", "offset": rng["free_offset"], "limit": rng["range_count"]}).json()
    assert [x["address"] for x in expanded["items"]] == ["10.203.1.3", "10.203.1.4", "10.203.1.5", "10.203.1.6"]
    assert client.get(f"/prefixes/{leaf['id']}/addresses").json() == page  # без group_free ответ прежний

    assert client.post(f"/prefixes/{leaf['id']}/addresses/next").json()["address"] == "10.203.1.3"
    parent = client.get(f"/prefixes/{parent['id']}").json()
    assert parent["capacity"] == 65534 and parent["used"] == 2  # ёмкость родителя — размер его подсети, used — по поддереву (изменение 025)


def test_in_use_objects_cannot_be_deleted(client, org):
    _prefix(client, org, "10.204.0.0/24")
    assert client.delete(f"/vrfs/{org['vrf_id']}").status_code == 409
    t = client.get("/device-types").json()["items"][0]
    client.post("/devices", json={"name": "t-1.internal", "device_type_id": t["id"], "organization_id": org["id"]})
    assert client.delete(f"/device-types/{t['id']}").status_code == 409

    # изменение 033, находка №4: отказ в удалении VRF попадает в журнал с organization_id и виден админу этой организации
    name, uid, admin = f"qa-{uuid.uuid4().hex[:8]}", None, None
    try:
        created = client.post("/users", json={"username": name, "password": "start-pass-123", "role": "admin", "organization_id": org["id"]})
        assert created.status_code == 201, created.text
        uid = created.json()["id"]
        admin = httpx.Client(base_url=BASE, timeout=30)
        r = admin.post("/auth/login", json={"username": name, "password": "start-pass-123"})
        admin.headers["Authorization"] = "Bearer " + r.json()["access_token"]
        assert admin.get("/audit", params={"event_type": "vrf.delete_blocked"}).json()["total"] >= 1
    finally:
        if admin is not None:
            admin.close()
        if uid is not None:
            client.delete(f"/users/{uid}")


def test_device_status_on_address(client, org):
    """Изменение 039: статус устройства отдаётся в списке адресов префикса и меняется через PATCH /devices/{id}."""
    p = _prefix(client, org, "10.205.0.0/24").json()
    t = client.get("/device-types").json()["items"][0]
    dev = client.post("/devices", json={"name": "svc-1.internal", "device_type_id": t["id"], "organization_id": org["id"], "status": "maintenance"}).json()
    assert dev["status"] == "maintenance"
    client.post(f"/prefixes/{p['id']}/addresses", json={"address": "10.205.0.1", "device_id": dev["id"]})
    page = client.get(f"/prefixes/{p['id']}/addresses").json()
    a = next(x for x in page["items"] if x["address"] == "10.205.0.1")
    assert a["device_name"] == "svc-1.internal" and a["device_type_name"] == t["name"] and a["device_status"] == "maintenance"

    upd = client.patch(f"/devices/{dev['id']}", json={"status": "off"})
    assert upd.status_code == 200 and upd.json()["status"] == "off"


def test_delete_organization(client, org):
    _prefix(client, org, "10.206.0.0/24")
    busy = client.delete(f"/organizations/{org['id']}")  # с префиксом — нельзя
    assert busy.status_code == 409 and "Нельзя удалить" in busy.json()["message"]
    logged = client.get("/audit", params={"event_type": "organization.delete_blocked", "q": org["name"]}).json()["items"]  # отказ попадает в журнал
    assert logged and logged[0]["entity_id"] == org["id"] and "удаление отклонено" in logged[0]["message"]
    assert logged[0]["diff"]["blocked_by"]["prefixes"] == {"total": 1, "items": ["10.206.0.0/24 (default)"]} and "devices" not in logged[0]["diff"]["blocked_by"]
    empty = client.post("/organizations", json={"name": f"empty-{org['name']}", "inn": "".join(reversed(org["inn"]))}).json()
    assert client.delete(f"/organizations/{empty['id']}").status_code == 204  # служебный VRF default удаляется вместе с организацией
    assert client.get(f"/organizations/{empty['id']}").status_code == 404
    assert client.get("/vrfs", params={"organization_id": empty["id"]}).json()["items"] == []


def test_allocate_next_subnet(client, org):
    parent = _prefix(client, org, "10.207.0.0/24").json()
    assert _prefix(client, org, "10.207.0.0/30").json()["parent_id"] == parent["id"]
    url = f"/prefixes/{parent['id']}/subnets/next"
    assert client.get(url, params={"length": 30}).json()["prefix"] == "10.207.0.4/30"  # предпросмотр
    a = client.post(url, json={"length": 30, "description": "auto"})
    assert a.status_code == 201 and a.json()["prefix"] == "10.207.0.4/30" and a.json()["parent_id"] == parent["id"] and a.json()["vrf_id"] == org["vrf_id"]
    assert client.post(url, json={"length": 30}).json()["prefix"] == "10.207.0.8/30"
    assert client.post(url, json={"length": 29}).json()["prefix"] == "10.207.0.16/29"  # выравнивание по размеру блока
    assert client.post(url, json={"length": 24}).status_code == 422 and client.post(url, json={"length": 33}).status_code == 422
    assert client.post(url, json={"length": 25}).json()["prefix"] == "10.207.0.128/25"
    assert client.post(url, json={"length": 25}).status_code == 409  # свободной половины больше нет
    assert client.get(url, params={"length": 25}).json()["prefix"] is None
    from app.services import next_free_subnet
    v6 = int(ipaddress.ip_address("fd00::"))
    assert next_free_subnet("fd00::/64", 66, [(v6, v6 + 5)]) == "fd00::4000:0:0:0/66"  # IPv6: первый блок занят, берётся второй


def test_prefix_isolation_between_organizations(client, org):
    """Изменение 033, находка №9: чужой организации не должен быть виден чужой префикс ни через GET, ни через предпросмотр подсети.
    Находка №11: admin чужой организации получает тот же 404 при попытке создать префикс в VRF организации A."""
    other = client.post("/organizations", json={"name": f"other-{org['name']}", "inn": "".join(reversed(org["inn"]))}).json()
    pid = _prefix(client, org, "10.208.0.0/24").json()["id"]
    name, uid, admin_b = f"qa-{uuid.uuid4().hex[:8]}", None, None
    try:
        created = client.post("/users", json={"username": name, "password": "start-pass-123", "role": "admin", "organization_id": other["id"]})
        assert created.status_code == 201, created.text
        uid = created.json()["id"]

        admin_b = httpx.Client(base_url=BASE, timeout=30)
        r = admin_b.post("/auth/login", json={"username": name, "password": "start-pass-123"})
        assert r.status_code == 200
        admin_b.headers["Authorization"] = "Bearer " + r.json()["access_token"]

        assert admin_b.get(f"/prefixes/{pid}").status_code == 404
        assert admin_b.get(f"/prefixes/{pid}/subnets/next", params={"length": 24}).status_code == 404
        # изменение 033, находка №11: чужой VRF в create_prefix — 404 (раньше было 422)
        cross = admin_b.post("/prefixes", json={"organization_id": other["id"], "vrf_id": org["vrf_id"], "prefix": "10.209.0.0/24"})
        assert cross.status_code == 404
        # изменение 033, находка №11: чужой parent_id в create_prefix — тоже 404 (раньше было 422)
        other_vrf_id = client.get("/vrfs", params={"organization_id": other["id"]}).json()["items"][0]["id"]
        cross_parent = admin_b.post("/prefixes", json={"organization_id": other["id"], "vrf_id": other_vrf_id, "prefix": "10.208.0.0/25", "parent_id": pid})
        assert cross_parent.status_code == 404
    finally:
        if admin_b is not None:
            admin_b.close()
        if uid is not None:
            client.delete(f"/users/{uid}")
        for v in client.get("/vrfs", params={"organization_id": other["id"]}).json()["items"]:
            client.delete(f"/vrfs/{v['id']}")
        client.delete(f"/organizations/{other['id']}")


def test_vrf_name_unique_per_organization(client, org):
    other = client.post("/organizations", json={"name": f"other-{org['name']}", "inn": "".join(reversed(org["inn"]))}).json()
    try:
        assert client.post("/vrfs", json={"organization_id": org["id"], "name": "Lab"}).status_code == 201
        assert client.post("/vrfs", json={"organization_id": org["id"], "name": "lab"}).status_code == 409  # регистр не важен
        assert client.post("/vrfs", json={"organization_id": other["id"], "name": "Lab"}).status_code == 201  # в другой организации можно
    finally:
        for v in client.get("/vrfs", params={"organization_id": other["id"]}).json()["items"]:
            client.delete(f"/vrfs/{v['id']}")
        client.delete(f"/organizations/{other['id']}")


def test_move_prefix_subtree_to_another_vrf(client, org):
    lab = client.post("/vrfs", json={"organization_id": org["id"], "name": "lab"}).json()
    parent = _prefix(client, org, "10.205.0.0/16").json()
    child = _prefix(client, org, "10.205.1.0/24").json()
    assert child["parent_id"] == parent["id"]
    # конфликт в целевом VRF -> 409, ничего не переехало
    client.post("/prefixes", json={"organization_id": org["id"], "vrf_id": lab["id"], "prefix": "10.205.1.0/24"})
    assert client.patch(f"/prefixes/{parent['id']}", json={"vrf_id": lab["id"]}).status_code == 409
    assert client.get(f"/prefixes/{child['id']}").json()["vrf_id"] == org["vrf_id"]
    # VRF другой организации -> 422
    foreign = client.post("/organizations", json={"name": f"f-{org['name']}", "inn": "".join(reversed(org["inn"]))}).json()
    foreign_vrf = client.get("/vrfs", params={"organization_id": foreign["id"]}).json()["items"][0]
    try:
        assert client.patch(f"/prefixes/{parent['id']}", json={"vrf_id": foreign_vrf["id"]}).status_code == 422
    finally:
        client.delete(f"/vrfs/{foreign_vrf['id']}")
        client.delete(f"/organizations/{foreign['id']}")
    # без конфликта: переезжает поддерево, родитель пересчитан
    for p in client.get("/prefixes", params={"organization_id": org["id"], "vrf_id": lab["id"]}).json()["items"]:
        client.delete(f"/prefixes/{p['id']}")
    assert client.patch(f"/prefixes/{parent['id']}", json={"vrf_id": lab["id"]}).status_code == 200
    moved = client.get(f"/prefixes/{child['id']}").json()
    assert moved["vrf_id"] == lab["id"] and moved["parent_id"] == parent["id"]