mirror of
https://github.com/hansjone/netx.git
synced 2026-10-14 22:20:50 +08:00
oliver
312f6f2f96
Add column filters and clearer API error mapping for NE management, tighten port validation, and slim the workbench into reusable modules with cleaner styles. Co-authored-by: Cursor <cursoragent@cursor.com>
205 lines
9.2 KiB
Python
205 lines
9.2 KiB
Python
"""NE management page: column filters, create/edit validation, batch and import fixes."""
|
|
import json
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
from cryptography.fernet import Fernet
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import Session, sessionmaker
|
|
from sqlalchemy.pool import StaticPool
|
|
|
|
from netx_api.config import settings
|
|
from netx_api.db import Base, get_db
|
|
from netx_api.managed_ne_router import router
|
|
from netx_api.models import ManagedNE
|
|
from netx_api.ne_crypto import decrypt_secret, encrypt_secret
|
|
from netx_api.ne_service import import_managed_ne
|
|
|
|
|
|
@pytest.fixture
|
|
def engine():
|
|
orig = settings.credential_secret_key
|
|
settings.credential_secret_key = Fernet.generate_key().decode()
|
|
eng = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool)
|
|
Base.metadata.create_all(eng)
|
|
with Session(eng) as db:
|
|
db.add_all([
|
|
ManagedNE(id="m1", name="PE-1", ip_address="10.0.0.1", vendor="ZTE", device_type="zte_zxros",
|
|
username="u1", tags="core edge", connect_status="pass", source=""),
|
|
ManagedNE(id="m2", name="", ip_address="10.0.0.2", vendor="Huawei", device_type="huawei",
|
|
username="u2", tags="edge", connect_status="fail", source="webcrt"),
|
|
ManagedNE(id="m3", name="SW-3", ip_address="10.0.0.3", vendor="Cisco", device_type="cisco_ios",
|
|
username="u3", tags="", connect_status="unknown", source="lldp"),
|
|
ManagedNE(id="m4", name="Lab_1", ip_address="10.0.0.4", vendor="Cisco", device_type="cisco_ios",
|
|
username="u4", tags="lab%x", connect_status="pass"),
|
|
])
|
|
db.commit()
|
|
yield eng
|
|
eng.dispose()
|
|
settings.credential_secret_key = orig
|
|
|
|
|
|
@pytest.fixture
|
|
def client(engine):
|
|
app = FastAPI()
|
|
app.include_router(router)
|
|
|
|
def database():
|
|
with Session(engine) as db:
|
|
yield db
|
|
|
|
app.dependency_overrides[get_db] = database
|
|
with patch("netx_api.auth_deps.settings.auth_enabled", False), TestClient(app) as test_client:
|
|
yield test_client
|
|
|
|
|
|
def _list(client, filters=None, **params):
|
|
if filters:
|
|
params["column_filters"] = json.dumps(filters)
|
|
r = client.get("/v1/managed-ne", params=params)
|
|
assert r.status_code == 200, r.text
|
|
return r.json()
|
|
|
|
|
|
def _ids(body):
|
|
return sorted(x["id"] for x in body["items"])
|
|
|
|
|
|
def test_column_filter_options_cover_full_dataset(client):
|
|
opts = _list(client, filter_field="vendor", page_size=1)
|
|
assert opts == {"field": "vendor", "options": ["Cisco", "Huawei", "ZTE"]}
|
|
# Blank names fall back to the IP shown in the table; blank source shows as manual.
|
|
assert "10.0.0.2" in _list(client, filter_field="name")["options"]
|
|
assert _list(client, filter_field="source")["options"] == ["lldp", "manual", "webcrt"]
|
|
|
|
|
|
def test_column_filters_apply_before_pagination(client):
|
|
body = _list(client, {"connectStatus": {"mode": "include", "values": ["pass"]}}, page_size=1)
|
|
assert body["total"] == 2
|
|
body = _list(client, {"vendor": {"mode": "exclude", "values": ["Cisco"]}})
|
|
assert _ids(body) == ["m1", "m2"]
|
|
body = _list(client, {"source": {"mode": "include", "values": ["manual"]}})
|
|
assert _ids(body) == ["m1", "m4"]
|
|
|
|
|
|
def test_tags_filter_matches_individual_tags(client):
|
|
assert _list(client, filter_field="tags")["options"] == ["", "core", "edge", "lab%x"]
|
|
assert _ids(_list(client, {"tags": {"mode": "include", "values": ["edge"]}})) == ["m1", "m2"]
|
|
assert _ids(_list(client, {"tags": {"mode": "include", "values": [""]}})) == ["m3"]
|
|
assert _ids(_list(client, {"tags": {"mode": "exclude", "values": ["edge", ""]}})) == ["m4"]
|
|
# LIKE wildcards inside a tag are literal.
|
|
assert _ids(_list(client, {"tags": {"mode": "include", "values": ["lab%x"]}})) == ["m4"]
|
|
# Tag options respect the other columns' filters.
|
|
opts = _list(client, {"vendor": {"mode": "include", "values": ["Huawei"]}}, filter_field="tags")
|
|
assert opts["options"] == ["edge"]
|
|
|
|
|
|
def test_invalid_filter_field_rejected(client):
|
|
r = client.get("/v1/managed-ne", params={"filter_field": "password_enc"})
|
|
assert r.status_code == 400
|
|
|
|
|
|
@pytest.mark.parametrize("ip", ["10.0.0.300", "a b", "http://x", "10.0.0.9:22"])
|
|
def test_create_rejects_bad_address(client, ip):
|
|
r = client.post("/v1/managed-ne", json={"vendor": "ZTE", "device_type": "zte_zxros",
|
|
"ip_address": ip, "username": "u"})
|
|
assert r.status_code == 400 and r.json()["detail"] == "invalid_ip_address"
|
|
|
|
|
|
@pytest.mark.parametrize("ip", ["10.0.0.9", "2001:db8::1", "pe-01.example.net"])
|
|
def test_create_accepts_ip_or_hostname(client, ip):
|
|
r = client.post("/v1/managed-ne", json={"vendor": "ZTE", "device_type": "zte_zxros",
|
|
"ip_address": ip, "username": "u"})
|
|
assert r.status_code == 200, r.text
|
|
|
|
|
|
@pytest.mark.parametrize("device_type", ["linux", "generic"])
|
|
def test_create_accepts_form_device_types(client, device_type):
|
|
r = client.post("/v1/managed-ne", json={"vendor": "Other", "device_type": device_type,
|
|
"ip_address": "10.0.0.9", "username": "u"})
|
|
assert r.status_code == 200, r.text
|
|
|
|
|
|
@pytest.mark.parametrize("port", [0, 65536, -1])
|
|
def test_create_rejects_bad_port(client, port):
|
|
r = client.post("/v1/managed-ne", json={"vendor": "ZTE", "device_type": "zte_zxros",
|
|
"ip_address": "10.0.0.9", "username": "u", "port": port})
|
|
assert r.status_code == 422
|
|
|
|
|
|
def test_update_resets_status_when_login_changes_and_keeps_name(client):
|
|
r = client.patch("/v1/managed-ne/m1", json={"remark": "note"})
|
|
assert r.json()["connect_status"] == "pass"
|
|
r = client.patch("/v1/managed-ne/m1", json={"port": 2222})
|
|
assert r.json()["connect_status"] == "unknown"
|
|
r = client.patch("/v1/managed-ne/m1", json={"name": " "})
|
|
assert r.json()["name"] == "10.0.0.1"
|
|
|
|
|
|
def test_batch_hop_requires_password_for_ne_without_one(client, engine):
|
|
hop = {"hop_vendor": "zte", "hop_host": "1.1.1.1", "hop_username": "jump"}
|
|
r = client.post("/v1/managed-ne/batch-hop", json={"ids": ["m1"], "hop": hop})
|
|
assert r.status_code == 400 and r.json()["detail"].startswith("hop_password_required")
|
|
with Session(engine) as db:
|
|
db.get(ManagedNE, "m1").hop_password_enc = encrypt_secret("old")
|
|
db.commit()
|
|
# Blank password keeps the stored one.
|
|
r = client.post("/v1/managed-ne/batch-hop", json={"ids": ["m1"], "hop": hop})
|
|
assert r.status_code == 200, r.text
|
|
with Session(engine) as db:
|
|
row = db.get(ManagedNE, "m1")
|
|
assert decrypt_secret(row.hop_password_enc) == "old"
|
|
assert row.connect_status == "unknown"
|
|
# A bastion that manages target auth may run without a hop password.
|
|
bastion = {**hop, "hop_vendor": "bastion", "hop_target_auth_mode": "bastion_managed"}
|
|
r = client.post("/v1/managed-ne/batch-hop", json={"ids": ["m2"], "hop": bastion})
|
|
assert r.status_code == 200, r.text
|
|
|
|
|
|
def test_batch_account_resets_status(client, engine):
|
|
r = client.post("/v1/managed-ne/batch-account", json={"ids": ["m1"], "account": {"username": "new"}})
|
|
assert r.status_code == 200
|
|
with Session(engine) as db:
|
|
assert db.get(ManagedNE, "m1").connect_status == "unknown"
|
|
|
|
|
|
def test_import_keeps_password_skips_bad_rows_and_promotes_placeholder(engine):
|
|
db = sessionmaker(bind=engine)()
|
|
try:
|
|
row = db.get(ManagedNE, "m1")
|
|
row.password_enc = encrypt_secret("secret")
|
|
db.commit()
|
|
csv = (
|
|
"device_type,ip,username,password,port,protocol,name,vendor,tags\n"
|
|
"zte_zxros,10.0.0.1,u1,,22,ssh,PE-1,ZTE,core\n" # blank password keeps stored one
|
|
"zte_zxros,10.0.0.3,u3,,22,ssh,SW-3,ZTE,\n" # LLDP placeholder becomes manual
|
|
"zte_zxros,10.9.9.9,u,,70000,ssh,BAD,ZTE,\n" # bad port
|
|
"zte_zxros,999.1.1.1,u,,22,ssh,BAD,ZTE,\n" # bad ip
|
|
"zte_zxros,10.9.9.8,u,,22,ssh,NEW,ZTE,\n"
|
|
).encode("utf-8")
|
|
result = import_managed_ne(db, csv, "devices.csv")
|
|
assert (result.inserted, result.updated) == (1, 2)
|
|
assert [(f.row, f.reason) for f in result.failed] == [(4, "invalid_port"), (5, "invalid_ip_address")]
|
|
db.expire_all()
|
|
assert decrypt_secret(db.get(ManagedNE, "m1").password_enc) == "secret"
|
|
assert db.get(ManagedNE, "m3").source == ""
|
|
assert db.query(ManagedNE).filter(ManagedNE.ip_address == "10.9.9.9").count() == 0
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def test_import_failure_mid_row_leaves_no_partial_ne(engine):
|
|
db = sessionmaker(bind=engine)()
|
|
try:
|
|
csv = (
|
|
"device_type,ip,username,password,port,protocol,name,vendor\n"
|
|
"zte_zxros,10.7.7.7,u,pw,22,ssh,X,ZTE\n"
|
|
).encode("utf-8")
|
|
with patch("netx_api.ne_service_import.encrypt_secret", side_effect=RuntimeError("boom")):
|
|
result = import_managed_ne(db, csv, "devices.csv")
|
|
assert (result.inserted, result.updated, len(result.failed)) == (0, 0, 1)
|
|
assert db.query(ManagedNE).filter(ManagedNE.ip_address == "10.7.7.7").count() == 0
|
|
finally:
|
|
db.close()
|