79 lines
3.3 KiB
Python
79 lines
3.3 KiB
Python
import asyncio
|
|
import importlib
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent.parent))
|
|
|
|
from main import app # noqa: F401 - initializes the project import path used by module tests
|
|
locations_router = importlib.import_module("app.modules.locations.backend.router")
|
|
from app.modules.locations.models.schemas import WallOutletCreate
|
|
|
|
|
|
def test_wall_outlet_requires_supported_location_type(monkeypatch):
|
|
monkeypatch.setattr(
|
|
locations_router,
|
|
"execute_query",
|
|
lambda query, params=None: [{"id": 1, "name": "HQ", "location_type": "kompleks"}],
|
|
)
|
|
|
|
with pytest.raises(HTTPException) as exc:
|
|
asyncio.run(locations_router.create_wall_outlet(WallOutletCreate(location_id=1, outlet_number="A-01")))
|
|
|
|
assert exc.value.status_code == 400
|
|
|
|
|
|
def test_wall_outlet_create_returns_location_context(monkeypatch):
|
|
calls = []
|
|
|
|
def fake_execute_query(query, params=None):
|
|
calls.append((query, params))
|
|
if "SELECT id, name, location_type FROM locations_locations" in query:
|
|
return [{"id": 2, "name": "1. sal", "location_type": "etage"}]
|
|
if "INSERT INTO locations_wall_outlets" in query:
|
|
return [{"id": 33}]
|
|
return [{
|
|
"id": 33, "location_id": 2, "outlet_number": "A-12", "category": "Cat6a",
|
|
"patch_panel": "PP-A", "patch_port": "12", "switch_name": "SW-1",
|
|
"switch_port": "Gi1/0/12", "status": "active", "notes": None,
|
|
"is_active": True, "created_at": "2026-07-17T12:00:00",
|
|
"updated_at": "2026-07-17T12:00:00", "deleted_at": None,
|
|
"location_name": "1. sal", "location_type": "etage", "customer_name": "BMC",
|
|
"hierarchy_path": "HQ > 1. sal",
|
|
}]
|
|
|
|
monkeypatch.setattr(locations_router, "execute_query", fake_execute_query)
|
|
result = asyncio.run(locations_router.create_wall_outlet(
|
|
WallOutletCreate(location_id=2, outlet_number="A-12", category="Cat6a", status="active")
|
|
))
|
|
|
|
assert result.id == 33
|
|
assert result.hierarchy_path == "HQ > 1. sal"
|
|
assert any("INSERT INTO locations_wall_outlets" in query for query, _ in calls)
|
|
|
|
|
|
def test_wall_outlet_allows_customer_site(monkeypatch):
|
|
def fake_execute_query(query, params=None):
|
|
if "SELECT id, name, location_type FROM locations_locations" in query:
|
|
return [{"id": 2, "name": "Kundesite", "location_type": "customer_site"}]
|
|
if "INSERT INTO locations_wall_outlets" in query:
|
|
return [{"id": 34}]
|
|
return [{
|
|
"id": 34, "location_id": 2, "outlet_number": "A-01", "category": None,
|
|
"patch_panel": None, "patch_port": None, "switch_name": None, "switch_port": None,
|
|
"status": "unknown", "notes": None, "is_active": True,
|
|
"created_at": "2026-07-17T12:00:00", "updated_at": "2026-07-17T12:00:00",
|
|
"deleted_at": None, "location_name": "Kundesite", "location_type": "customer_site",
|
|
"customer_name": "BMC", "hierarchy_path": "Kundesite",
|
|
}]
|
|
|
|
monkeypatch.setattr(locations_router, "execute_query", fake_execute_query)
|
|
result = asyncio.run(locations_router.create_wall_outlet(
|
|
WallOutletCreate(location_id=2, outlet_number="A-01")
|
|
))
|
|
|
|
assert result.location_type == "customer_site"
|