fix: all lint errors, test + lint pre-commit passing
Frontend: - Split nodeTypes/edgeTypes into separate .ts files (react-refresh) - Remove setState-in-effect in NodeModal (key prop reset) - Fix handleSave accessed before declaration (useRef pattern) - Exclude src/components/ui/** from eslint (shadcn generated) - Use defineConfig from vitest/config for test type support Backend: - ruff --fix: sort imports, datetime.UTC alias - Raise line-length to 120, ignore E501 in tests - Break long update_node/update_edge signatures - pyproject.toml: per-file-ignores for tests
This commit is contained in:
@@ -1,4 +1,4 @@
|
||||
from datetime import datetime, timezone
|
||||
from datetime import UTC, datetime
|
||||
|
||||
from fastapi import APIRouter, Depends
|
||||
from sqlalchemy import select
|
||||
@@ -34,7 +34,7 @@ async def save_canvas(body: CanvasSaveRequest, db: AsyncSession = Depends(get_db
|
||||
state = await db.get(CanvasState, 1)
|
||||
if state:
|
||||
state.viewport = body.viewport
|
||||
state.saved_at = datetime.now(timezone.utc)
|
||||
state.saved_at = datetime.now(UTC)
|
||||
else:
|
||||
db.add(CanvasState(id=1, viewport=body.viewport))
|
||||
|
||||
|
||||
@@ -35,7 +35,9 @@ async def delete_edge(edge_id: str, db: AsyncSession = Depends(get_db), _: str =
|
||||
|
||||
|
||||
@router.patch("/{edge_id}", response_model=EdgeResponse)
|
||||
async def update_edge(edge_id: str, body: EdgeUpdate, db: AsyncSession = Depends(get_db), _: str = Depends(get_current_user)):
|
||||
async def update_edge(
|
||||
edge_id: str, body: EdgeUpdate, db: AsyncSession = Depends(get_db), _: str = Depends(get_current_user)
|
||||
):
|
||||
edge = await db.get(Edge, edge_id)
|
||||
if not edge:
|
||||
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Edge not found")
|
||||
|
||||
@@ -34,7 +34,9 @@ async def get_node(node_id: str, db: AsyncSession = Depends(get_db), _: str = De
|
||||
|
||||
|
||||
@router.patch("/{node_id}", response_model=NodeResponse)
|
||||
async def update_node(node_id: str, body: NodeUpdate, db: AsyncSession = Depends(get_db), _: str = Depends(get_current_user)):
|
||||
async def update_node(
|
||||
node_id: str, body: NodeUpdate, db: AsyncSession = Depends(get_db), _: str = Depends(get_current_user)
|
||||
):
|
||||
node = await db.get(Node, node_id)
|
||||
if not node:
|
||||
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Node not found")
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from datetime import UTC, datetime, timedelta
|
||||
|
||||
from jose import JWTError, jwt
|
||||
from passlib.context import CryptContext
|
||||
@@ -17,7 +17,7 @@ def hash_password(password: str) -> str:
|
||||
|
||||
|
||||
def create_access_token(subject: str) -> str:
|
||||
expire = datetime.now(timezone.utc) + timedelta(minutes=settings.access_token_expire_minutes)
|
||||
expire = datetime.now(UTC) + timedelta(minutes=settings.access_token_expire_minutes)
|
||||
payload = {"sub": subject, "exp": expire}
|
||||
return jwt.encode(payload, settings.secret_key, algorithm=settings.algorithm)
|
||||
|
||||
|
||||
@@ -1,14 +1,14 @@
|
||||
import uuid
|
||||
from datetime import datetime, timezone
|
||||
from datetime import UTC, datetime
|
||||
|
||||
from sqlalchemy import JSON, Boolean, DateTime, Float, ForeignKey, Integer, String, Text
|
||||
from sqlalchemy import JSON, DateTime, Float, ForeignKey, Integer, String, Text
|
||||
from sqlalchemy.orm import Mapped, mapped_column, relationship
|
||||
|
||||
from app.db.database import Base
|
||||
|
||||
|
||||
def _now() -> datetime:
|
||||
return datetime.now(timezone.utc)
|
||||
return datetime.now(UTC)
|
||||
|
||||
|
||||
def _uuid() -> str:
|
||||
|
||||
+1
-1
@@ -3,9 +3,9 @@ from contextlib import asynccontextmanager
|
||||
from fastapi import FastAPI
|
||||
from fastapi.middleware.cors import CORSMiddleware
|
||||
|
||||
from app.api.routes import auth, canvas, edges, nodes, scan, status
|
||||
from app.core.config import settings
|
||||
from app.db.database import init_db
|
||||
from app.api.routes import auth, nodes, edges, canvas, scan, status
|
||||
|
||||
|
||||
@asynccontextmanager
|
||||
|
||||
+1
-1
@@ -2,7 +2,7 @@ auth:
|
||||
username: admin
|
||||
# Default password: "admin" — change this before deploying!
|
||||
# Generate a new hash: python scripts/hash_password.py yourpassword
|
||||
password_hash: "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW"
|
||||
password_hash: "$2b$12$o/LWyvmBc978CNpSsHxcveXN0WqjAGW/gBR0.U.HURWbaYD3GCDqS"
|
||||
|
||||
scanner:
|
||||
ranges:
|
||||
|
||||
@@ -0,0 +1,27 @@
|
||||
[tool.ruff]
|
||||
target-version = "py313"
|
||||
line-length = 120
|
||||
exclude = ["migrations", ".venv"]
|
||||
|
||||
[tool.ruff.lint]
|
||||
select = ["E", "F", "I", "UP", "B", "SIM"]
|
||||
ignore = ["B008"]
|
||||
|
||||
[tool.ruff.lint.per-file-ignores]
|
||||
"tests/**" = ["E501"]
|
||||
|
||||
[tool.mypy]
|
||||
python_version = "3.13"
|
||||
strict = true
|
||||
ignore_missing_imports = true
|
||||
exclude = ["migrations", ".venv"]
|
||||
|
||||
[tool.pytest.ini_options]
|
||||
asyncio_mode = "auto"
|
||||
asyncio_default_fixture_loop_scope = "function"
|
||||
testpaths = ["tests"]
|
||||
addopts = "--tb=short -q"
|
||||
|
||||
[tool.coverage.run]
|
||||
source = ["app"]
|
||||
omit = ["*/migrations/*", "*/tests/*"]
|
||||
@@ -8,6 +8,7 @@ pydantic==2.9.2
|
||||
pydantic-settings==2.5.2
|
||||
python-jose[cryptography]==3.3.0
|
||||
passlib[bcrypt]==1.7.4
|
||||
bcrypt==4.0.1
|
||||
python-multipart==0.0.12
|
||||
apscheduler==3.10.4
|
||||
python-nmap==0.7.1
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
"""Generate a bcrypt password hash for config.yml."""
|
||||
import sys
|
||||
|
||||
from passlib.context import CryptContext
|
||||
|
||||
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
|
||||
|
||||
@@ -0,0 +1,39 @@
|
||||
import pytest
|
||||
from httpx import ASGITransport, AsyncClient
|
||||
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
|
||||
|
||||
from app.db.database import Base, get_db
|
||||
from app.main import app
|
||||
|
||||
TEST_DB_URL = "sqlite+aiosqlite:///:memory:"
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def db_session():
|
||||
engine = create_async_engine(TEST_DB_URL)
|
||||
async with engine.begin() as conn:
|
||||
await conn.run_sync(Base.metadata.create_all)
|
||||
session_factory = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
|
||||
async with session_factory() as session:
|
||||
yield session
|
||||
async with engine.begin() as conn:
|
||||
await conn.run_sync(Base.metadata.drop_all)
|
||||
await engine.dispose()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def client(db_session: AsyncSession):
|
||||
app.dependency_overrides[get_db] = lambda: db_session
|
||||
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
|
||||
yield c
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def auth_headers(client):
|
||||
"""Returns a coroutine that logs in and returns auth headers."""
|
||||
async def _get():
|
||||
res = await client.post("/api/v1/auth/login", json={"username": "admin", "password": "admin"})
|
||||
token = res.json()["access_token"]
|
||||
return {"Authorization": f"Bearer {token}"}
|
||||
return _get
|
||||
@@ -0,0 +1,39 @@
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
from httpx import AsyncClient
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_credentials():
|
||||
with patch("app.api.routes.auth._load_credentials", return_value=("admin", "$2b$12$o/LWyvmBc978CNpSsHxcveXN0WqjAGW/gBR0.U.HURWbaYD3GCDqS")):
|
||||
yield
|
||||
|
||||
|
||||
async def test_login_success(client: AsyncClient, mock_credentials):
|
||||
res = await client.post("/api/v1/auth/login", json={"username": "admin", "password": "admin"})
|
||||
assert res.status_code == 200
|
||||
data = res.json()
|
||||
assert "access_token" in data
|
||||
assert data["token_type"] == "bearer"
|
||||
|
||||
|
||||
async def test_login_wrong_password(client: AsyncClient, mock_credentials):
|
||||
res = await client.post("/api/v1/auth/login", json={"username": "admin", "password": "wrong"})
|
||||
assert res.status_code == 401
|
||||
|
||||
|
||||
async def test_login_wrong_username(client: AsyncClient, mock_credentials):
|
||||
res = await client.post("/api/v1/auth/login", json={"username": "notadmin", "password": "admin"})
|
||||
assert res.status_code == 401
|
||||
|
||||
|
||||
async def test_protected_route_requires_auth(client: AsyncClient):
|
||||
res = await client.get("/api/v1/nodes/")
|
||||
assert res.status_code == 403
|
||||
|
||||
|
||||
async def test_health_is_public(client: AsyncClient):
|
||||
res = await client.get("/api/v1/health")
|
||||
assert res.status_code == 200
|
||||
assert res.json() == {"status": "ok"}
|
||||
@@ -0,0 +1,54 @@
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
from httpx import AsyncClient
|
||||
|
||||
TOKEN_HASH = "$2b$12$o/LWyvmBc978CNpSsHxcveXN0WqjAGW/gBR0.U.HURWbaYD3GCDqS"
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def headers(client: AsyncClient):
|
||||
with patch("app.api.routes.auth._load_credentials", return_value=("admin", TOKEN_HASH)):
|
||||
res = await client.post("/api/v1/auth/login", json={"username": "admin", "password": "admin"})
|
||||
token = res.json()["access_token"]
|
||||
return {"Authorization": f"Bearer {token}"}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def two_nodes(client: AsyncClient, headers: dict):
|
||||
n1 = (await client.post("/api/v1/nodes/", json={"type": "router", "label": "R1", "status": "online"}, headers=headers)).json()
|
||||
n2 = (await client.post("/api/v1/nodes/", json={"type": "switch", "label": "SW1", "status": "online"}, headers=headers)).json()
|
||||
return n1["id"], n2["id"]
|
||||
|
||||
|
||||
async def test_create_edge(client: AsyncClient, headers: dict, two_nodes):
|
||||
src, tgt = two_nodes
|
||||
res = await client.post("/api/v1/edges/", json={"source": src, "target": tgt, "type": "ethernet"}, headers=headers)
|
||||
assert res.status_code == 201
|
||||
data = res.json()
|
||||
assert data["source"] == src
|
||||
assert data["target"] == tgt
|
||||
assert data["type"] == "ethernet"
|
||||
|
||||
|
||||
async def test_create_vlan_edge(client: AsyncClient, headers: dict, two_nodes):
|
||||
src, tgt = two_nodes
|
||||
res = await client.post("/api/v1/edges/", json={"source": src, "target": tgt, "type": "vlan", "vlan_id": 20, "label": "VLAN 20"}, headers=headers)
|
||||
assert res.status_code == 201
|
||||
assert res.json()["vlan_id"] == 20
|
||||
|
||||
|
||||
async def test_list_edges(client: AsyncClient, headers: dict, two_nodes):
|
||||
src, tgt = two_nodes
|
||||
await client.post("/api/v1/edges/", json={"source": src, "target": tgt, "type": "ethernet"}, headers=headers)
|
||||
res = await client.get("/api/v1/edges/", headers=headers)
|
||||
assert res.status_code == 200
|
||||
assert len(res.json()) == 1
|
||||
|
||||
|
||||
async def test_delete_edge(client: AsyncClient, headers: dict, two_nodes):
|
||||
src, tgt = two_nodes
|
||||
edge_id = (await client.post("/api/v1/edges/", json={"source": src, "target": tgt, "type": "wifi"}, headers=headers)).json()["id"]
|
||||
res = await client.delete(f"/api/v1/edges/{edge_id}", headers=headers)
|
||||
assert res.status_code == 204
|
||||
assert len((await client.get("/api/v1/edges/", headers=headers)).json()) == 0
|
||||
@@ -0,0 +1,67 @@
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
from httpx import AsyncClient
|
||||
|
||||
TOKEN_HASH = "$2b$12$o/LWyvmBc978CNpSsHxcveXN0WqjAGW/gBR0.U.HURWbaYD3GCDqS"
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def headers(client: AsyncClient):
|
||||
with patch("app.api.routes.auth._load_credentials", return_value=("admin", TOKEN_HASH)):
|
||||
res = await client.post("/api/v1/auth/login", json={"username": "admin", "password": "admin"})
|
||||
token = res.json()["access_token"]
|
||||
return {"Authorization": f"Bearer {token}"}
|
||||
|
||||
|
||||
async def test_list_nodes_empty(client: AsyncClient, headers: dict):
|
||||
res = await client.get("/api/v1/nodes/", headers=headers)
|
||||
assert res.status_code == 200
|
||||
assert res.json() == []
|
||||
|
||||
|
||||
async def test_create_node(client: AsyncClient, headers: dict):
|
||||
payload = {"type": "server", "label": "My Server", "ip": "192.168.1.10", "status": "unknown"}
|
||||
res = await client.post("/api/v1/nodes/", json=payload, headers=headers)
|
||||
assert res.status_code == 201
|
||||
data = res.json()
|
||||
assert data["label"] == "My Server"
|
||||
assert data["ip"] == "192.168.1.10"
|
||||
assert "id" in data
|
||||
|
||||
|
||||
async def test_get_node(client: AsyncClient, headers: dict):
|
||||
create = await client.post("/api/v1/nodes/", json={"type": "router", "label": "Router", "status": "online"}, headers=headers)
|
||||
node_id = create.json()["id"]
|
||||
res = await client.get(f"/api/v1/nodes/{node_id}", headers=headers)
|
||||
assert res.status_code == 200
|
||||
assert res.json()["id"] == node_id
|
||||
|
||||
|
||||
async def test_get_node_not_found(client: AsyncClient, headers: dict):
|
||||
res = await client.get("/api/v1/nodes/nonexistent-id", headers=headers)
|
||||
assert res.status_code == 404
|
||||
|
||||
|
||||
async def test_update_node(client: AsyncClient, headers: dict):
|
||||
create = await client.post("/api/v1/nodes/", json={"type": "server", "label": "Old", "status": "unknown"}, headers=headers)
|
||||
node_id = create.json()["id"]
|
||||
res = await client.patch(f"/api/v1/nodes/{node_id}", json={"label": "New", "ip": "10.0.0.1"}, headers=headers)
|
||||
assert res.status_code == 200
|
||||
assert res.json()["label"] == "New"
|
||||
assert res.json()["ip"] == "10.0.0.1"
|
||||
|
||||
|
||||
async def test_delete_node(client: AsyncClient, headers: dict):
|
||||
create = await client.post("/api/v1/nodes/", json={"type": "switch", "label": "Switch", "status": "unknown"}, headers=headers)
|
||||
node_id = create.json()["id"]
|
||||
res = await client.delete(f"/api/v1/nodes/{node_id}", headers=headers)
|
||||
assert res.status_code == 204
|
||||
assert (await client.get(f"/api/v1/nodes/{node_id}", headers=headers)).status_code == 404
|
||||
|
||||
|
||||
async def test_list_nodes_returns_all(client: AsyncClient, headers: dict):
|
||||
for i in range(3):
|
||||
await client.post("/api/v1/nodes/", json={"type": "generic", "label": f"Node {i}", "status": "unknown"}, headers=headers)
|
||||
res = await client.get("/api/v1/nodes/", headers=headers)
|
||||
assert len(res.json()) == 3
|
||||
Reference in New Issue
Block a user