document-management/backend/tests/dms/test_ip_entries.py

"""1.10 部门IP登录限制测试。

覆盖：
- normalize_ip_value 规范化（精确IP/CIDR/IPv6压缩）与非法输入
- IP条目 CRUD、重复409、host bits 400、多余字段400
- 部门分配整存（乱序输入升序返回、无效条目400、空数组清空、rowVersion递增）
- 删除守卫（被分配引用409，取消分配后可删）
- 登录IP校验（网段命中/精确命中/未命中拒绝/ADMIN豁免/清空后放行/失败审计）
- 非 ADMIN 访问 403
- 审计记录（CREATE/EDIT/DELETE_IP_ENTRY、ASSIGN_ORG_IP）
"""

from __future__ import annotations

import pytest

from dms.common.errors import InvalidArgumentError
from dms.services.ip_entry_service import normalize_ip_value


def _auth(token: str) -> dict[str, str]:
    return {"Authorization": f"Bearer {token}"}


# ─── 单元：规范化 ───────────────────────────────────────────────────────────


def test_normalize_accepts_exact_ips():
    assert normalize_ip_value(" 192.168.1.5 ") == "192.168.1.5"
    assert normalize_ip_value("2001:0DB8::0001") == "2001:db8::1"


def test_normalize_accepts_strict_cidr():
    assert normalize_ip_value("192.168.1.0/24") == "192.168.1.0/24"
    assert normalize_ip_value("10.0.0.0/8") == "10.0.0.0/8"


def test_normalize_rejects_host_bits_with_hint():
    with pytest.raises(InvalidArgumentError) as exc_info:
        normalize_ip_value("192.168.1.5/24")
    assert "主机位" in str(exc_info.value)


@pytest.mark.parametrize(
    "bad_value",
    ["", "   ", "abc", "999.1.1.1", "192.168.1.0/33", 123, None],
)
def test_normalize_rejects_invalid_input(bad_value):
    with pytest.raises(InvalidArgumentError):
        normalize_ip_value(bad_value)


# ─── HTTP 集成 ──────────────────────────────────────────────────────────────


@pytest.fixture()
def ip_cleanup(b2_app):
    """前后置物理清空IP两表：前置隔离测试数据，后置避免分配表组织FK
    RESTRICT 阻止 conftest 对 sys_organization 的物理清理。"""
    from sqlalchemy import delete

    from dms.extensions import db
    from dms.models import IpEntry, OrgIpAssignment

    def purge():
        with b2_app.app_context():
            db.session.execute(delete(OrgIpAssignment))
            db.session.execute(delete(IpEntry))
            db.session.commit()

    purge()
    yield
    purge()


def _create_entry(client, token, ip_value: str, remark=None):
    response = client.post(
        "/api/v1/ip-entries",
        json={"ipValue": ip_value, "remark": remark},
        headers=_auth(token),
    )
    assert response.status_code == 201, response.get_json()
    return response.get_json()["data"]


def _org_id(app, code: str) -> int:
    from dms.extensions import db
    from dms.models import Organization

    with app.app_context():
        org = db.session.scalar(
            db.select(Organization).where(Organization.org_code == code)
        )
        assert org is not None, f"organization {code} not found"
        return org.id


def _save_assignments(client, token, org_id: int, entry_ids: list[str], row_version: int):
    return client.put(
        f"/api/v1/organizations/{org_id}/ip-assignments",
        json={"ipEntryIds": entry_ids, "rowVersion": row_version},
        headers=_auth(token),
    )


@pytest.mark.integration
def test_ip_entry_crud_roundtrip(b2_client, login_token, ip_cleanup):
    created = b2_client.post(
        "/api/v1/ip-entries",
        json={"ipValue": " 192.168.1.5 ", "remark": " 网关 "},
        headers=_auth(login_token),
    )
    assert created.status_code == 201, created.get_json()
    entry = created.get_json()["data"]
    assert entry["ipValue"] == "192.168.1.5"
    assert entry["remark"] == "网关"
    assert entry["rowVersion"] == 0

    listed = b2_client.get(
        "/api/v1/ip-entries?keyword=网关", headers=_auth(login_token)
    ).get_json()["data"]
    assert listed["total"] == 1
    assert [item["id"] for item in listed["items"]] == [entry["id"]]

    updated = b2_client.put(
        f"/api/v1/ip-entries/{entry['id']}",
        json={"ipValue": "10.10.10.10", "remark": None, "rowVersion": 0},
        headers=_auth(login_token),
    )
    assert updated.status_code == 200, updated.get_json()
    assert updated.get_json()["data"]["ipValue"] == "10.10.10.10"
    assert updated.get_json()["data"]["remark"] == ""
    assert updated.get_json()["data"]["rowVersion"] == 1

    deleted = b2_client.delete(
        f"/api/v1/ip-entries/{entry['id']}?rowVersion=1",
        headers=_auth(login_token),
    )
    assert deleted.status_code == 200
    assert deleted.get_json()["data"]["deleted"] is True

    remaining = b2_client.get(
        "/api/v1/ip-entries", headers=_auth(login_token)
    ).get_json()["data"]
    assert remaining["total"] == 0


@pytest.mark.integration
def test_create_duplicate_ip_conflicts(b2_client, login_token, ip_cleanup):
    first = _create_entry(b2_client, login_token, "10.0.0.5")
    second = b2_client.post(
        "/api/v1/ip-entries",
        json={"ipValue": "10.0.0.5", "remark": None},
        headers=_auth(login_token),
    )
    assert second.status_code == 409
    assert second.get_json()["code"] == "DATA_VERSION_CONFLICT"
    assert first["ipValue"] == "10.0.0.5"


@pytest.mark.integration
def test_create_rejects_host_bits_and_extra_fields(
    b2_client, login_token, ip_cleanup
):
    host_bits = b2_client.post(
        "/api/v1/ip-entries",
        json={"ipValue": "192.168.1.5/24", "remark": None},
        headers=_auth(login_token),
    )
    assert host_bits.status_code == 400
    assert host_bits.get_json()["code"] == "INVALID_ARGUMENT"

    extra = b2_client.post(
        "/api/v1/ip-entries",
        json={"ipValue": "1.2.3.4", "remark": None, "surplus": True},
        headers=_auth(login_token),
    )
    assert extra.status_code == 400
    assert extra.get_json()["code"] == "INVALID_ARGUMENT"


@pytest.mark.integration
def test_org_assignments_full_replace(b2_app, b2_client, login_token, ip_cleanup):
    ops_id = _org_id(b2_app, "ORG_OPS")
    entry_one = _create_entry(b2_client, login_token, "192.168.10.0/24")
    entry_two = _create_entry(b2_client, login_token, "10.10.10.10")

    saved = _save_assignments(
        b2_client,
        login_token,
        ops_id,
        [entry_two["id"], entry_one["id"]],
        0,
    )
    assert saved.status_code == 200, saved.get_json()
    assert saved.get_json()["data"]["ipEntryIds"] == [
        entry_one["id"],
        entry_two["id"],
    ]
    assert saved.get_json()["data"]["rowVersion"] == 1

    listed = b2_client.get(
        f"/api/v1/organizations/{ops_id}/ip-assignments",
        headers=_auth(login_token),
    )
    assert listed.status_code == 200
    assert listed.get_json()["data"]["ipEntryIds"] == [
        entry_one["id"],
        entry_two["id"],
    ]

    invalid = _save_assignments(
        b2_client, login_token, ops_id, ["99999"], 1
    )
    assert invalid.status_code == 400
    assert invalid.get_json()["code"] == "INVALID_ARGUMENT"

    cleared = _save_assignments(b2_client, login_token, ops_id, [], 1)
    assert cleared.status_code == 200, cleared.get_json()
    assert cleared.get_json()["data"]["ipEntryIds"] == []
    assert cleared.get_json()["data"]["rowVersion"] == 2


@pytest.mark.integration
def test_delete_blocked_while_assigned(
    b2_app, b2_client, login_token, ip_cleanup
):
    ops_id = _org_id(b2_app, "ORG_OPS")
    entry = _create_entry(b2_client, login_token, "172.16.0.1")
    assert _save_assignments(
        b2_client, login_token, ops_id, [entry["id"]], 0
    ).status_code == 200

    blocked = b2_client.delete(
        f"/api/v1/ip-entries/{entry['id']}?rowVersion=0",
        headers=_auth(login_token),
    )
    assert blocked.status_code == 409
    assert blocked.get_json()["code"] == "DATA_VERSION_CONFLICT"

    assert (
        _save_assignments(b2_client, login_token, ops_id, [], 1).status_code
        == 200
    )
    deleted = b2_client.delete(
        f"/api/v1/ip-entries/{entry['id']}?rowVersion=0",
        headers=_auth(login_token),
    )
    assert deleted.status_code == 200


@pytest.mark.integration
def test_login_ip_restriction_enforced(
    ip_cleanup, b2_app, b2_client, login_token, b2_password
):
    ops_id = _org_id(b2_app, "ORG_OPS")
    subnet = _create_entry(b2_client, login_token, "192.168.10.0/24")
    exact = _create_entry(b2_client, login_token, "10.10.10.10")
    assert (
        _save_assignments(
            b2_client,
            login_token,
            ops_id,
            [subnet["id"], exact["id"]],
            0,
        ).status_code
        == 200
    )

    def login(username: str, client_ip: str):
        return b2_client.post(
            "/api/v1/auth/login",
            json={
                "username": username,
                "password": b2_password,
                "keepSignedIn": False,
            },
            headers={"X-Forwarded-For": client_ip},
        )

    hit_subnet = login("user", "192.168.10.55")
    assert hit_subnet.status_code == 200

    hit_exact = login("user", "10.10.10.10")
    assert hit_exact.status_code == 200

    miss = login("user", "172.16.0.1")
    assert miss.status_code == 403
    assert miss.get_json()["code"] == "LOGIN_IP_FORBIDDEN"

    admin_exempt = login("admin", "172.16.0.1")
    assert admin_exempt.status_code == 200

    with b2_app.app_context():
        from dms.extensions import db
        from dms.models import AuditLog

        reasons = set(
            db.session.scalars(
                db.select(AuditLog.failure_reason).where(
                    AuditLog.action_type == "LOGIN",
                    AuditLog.operation_result == "FAILURE",
                    AuditLog.username == "user",
                )
            ).all()
        )
    assert "IP_NOT_ALLOWED" in reasons

    assert (
        _save_assignments(b2_client, login_token, ops_id, [], 1).status_code
        == 200
    )
    unrestricted = login("user", "172.16.0.1")
    assert unrestricted.status_code == 200


@pytest.mark.integration
def test_ip_routes_require_admin(ip_cleanup, b2_client, token_for):
    user_token = token_for("user")
    response = b2_client.get("/api/v1/ip-entries", headers=_auth(user_token))
    assert response.status_code == 403


@pytest.mark.integration
def test_ip_audit_trail_written(b2_app, b2_client, login_token, ip_cleanup):
    from dms.extensions import db
    from dms.models import AuditLog

    ops_id = _org_id(b2_app, "ORG_OPS")
    entry = _create_entry(b2_client, login_token, "10.20.30.40", "审计条目")
    b2_client.put(
        f"/api/v1/ip-entries/{entry['id']}",
        json={
            "ipValue": "10.20.30.41",
            "remark": "审计条目",
            "rowVersion": 0,
        },
        headers=_auth(login_token),
    )
    assert (
        _save_assignments(
            b2_client, login_token, ops_id, [entry["id"]], 0
        ).status_code
        == 200
    )
    assert (
        _save_assignments(b2_client, login_token, ops_id, [], 1).status_code
        == 200
    )
    b2_client.delete(
        f"/api/v1/ip-entries/{entry['id']}?rowVersion=1",
        headers=_auth(login_token),
    )

    with b2_app.app_context():
        actions = set(
            db.session.scalars(
                db.select(AuditLog.action_type).where(
                    AuditLog.action_type.in_(
                        [
                            "CREATE_IP_ENTRY",
                            "EDIT_IP_ENTRY",
                            "DELETE_IP_ENTRY",
                            "ASSIGN_ORG_IP",
                        ]
                    )
                )
            ).all()
        )
    assert actions == {
        "CREATE_IP_ENTRY",
        "EDIT_IP_ENTRY",
        "DELETE_IP_ENTRY",
        "ASSIGN_ORG_IP",
    }