148 lines
4.0 KiB
Python
148 lines
4.0 KiB
Python
import importlib
|
|
|
|
from core.security import get_auth_context, require_authenticated_user
|
|
from models.proxy import ProxyDocument
|
|
from models.service import ServiceEndpointDocument
|
|
|
|
proxy_router_module = importlib.import_module("api.proxy.router")
|
|
|
|
|
|
def make_proxy_document(
|
|
*,
|
|
proxy_port: int,
|
|
name: str = "svc",
|
|
code: str = "proxy-code",
|
|
address: str = "127.0.0.1",
|
|
port: int = 9000,
|
|
) -> ProxyDocument:
|
|
return ProxyDocument(
|
|
proxy_port=proxy_port,
|
|
name=name,
|
|
code=code,
|
|
service=ServiceEndpointDocument(address=address, port=port),
|
|
)
|
|
|
|
|
|
def test_proxy_requires_auth_without_dependency_overrides(client):
|
|
client.app.dependency_overrides.pop(require_authenticated_user, None)
|
|
client.app.dependency_overrides.pop(get_auth_context, None)
|
|
|
|
response = client.get("/proxy")
|
|
|
|
assert response.status_code == 401
|
|
assert response.json()["detail"] == "Could not validate credentials"
|
|
|
|
|
|
def test_list_proxies_returns_documents(client, monkeypatch: object):
|
|
docs = [
|
|
make_proxy_document(proxy_port=8080, name="api"),
|
|
make_proxy_document(proxy_port=8081, name="worker"),
|
|
]
|
|
|
|
async def fake_list_proxies() -> list[ProxyDocument]:
|
|
return docs
|
|
|
|
monkeypatch.setattr(proxy_router_module, "list_proxy_documents", fake_list_proxies)
|
|
|
|
response = client.get("/proxy")
|
|
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
assert [item["proxy_port"] for item in body] == [8080, 8081]
|
|
|
|
|
|
def test_get_proxy_returns_404_when_not_found(client, monkeypatch: object):
|
|
async def fake_get_proxy_by_port(proxy_port: int) -> None:
|
|
return None
|
|
|
|
monkeypatch.setattr(
|
|
proxy_router_module,
|
|
"get_proxy_by_port",
|
|
fake_get_proxy_by_port,
|
|
)
|
|
|
|
response = client.get("/proxy/9999")
|
|
|
|
assert response.status_code == 404
|
|
assert response.json()["detail"] == "Proxy not found"
|
|
|
|
|
|
def test_create_proxy_returns_409_on_port_conflict(client, monkeypatch: object):
|
|
existing = make_proxy_document(proxy_port=3671)
|
|
|
|
async def fake_get_proxy_by_port(proxy_port: int) -> ProxyDocument | None:
|
|
assert proxy_port == 3671
|
|
return existing
|
|
|
|
monkeypatch.setattr(
|
|
proxy_router_module,
|
|
"get_proxy_by_port",
|
|
fake_get_proxy_by_port,
|
|
)
|
|
|
|
response = client.post(
|
|
"/proxy",
|
|
json={
|
|
"proxy_port": 3671,
|
|
"name": "new",
|
|
"code": "abc",
|
|
"service": {"address": "localhost", "port": 8080},
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 409
|
|
assert response.json()["detail"] == "Proxy already exists on port 3671"
|
|
|
|
|
|
def test_create_proxy_returns_created_document(client, monkeypatch: object):
|
|
captured: dict[str, object] = {}
|
|
|
|
async def fake_get_proxy_by_port(proxy_port: int) -> ProxyDocument | None:
|
|
assert proxy_port == 4040
|
|
return None
|
|
|
|
async def fake_create_proxy(
|
|
*,
|
|
proxy_port: int,
|
|
name: str,
|
|
service: dict[str, object],
|
|
code: str,
|
|
) -> ProxyDocument:
|
|
captured["proxy_port"] = proxy_port
|
|
captured["name"] = name
|
|
captured["service"] = service
|
|
captured["code"] = code
|
|
return make_proxy_document(
|
|
proxy_port=proxy_port,
|
|
name=name,
|
|
code=code,
|
|
address=str(service["address"]),
|
|
port=int(service["port"]),
|
|
)
|
|
|
|
monkeypatch.setattr(
|
|
proxy_router_module,
|
|
"get_proxy_by_port",
|
|
fake_get_proxy_by_port,
|
|
)
|
|
monkeypatch.setattr(proxy_router_module, "create_proxy", fake_create_proxy)
|
|
|
|
response = client.post(
|
|
"/proxy",
|
|
json={
|
|
"proxy_port": 4040,
|
|
"name": "edge",
|
|
"code": "edge-code",
|
|
"service": {"address": "service.local", "port": 8181},
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 201
|
|
assert response.json()["proxy_port"] == 4040
|
|
assert captured == {
|
|
"proxy_port": 4040,
|
|
"name": "edge",
|
|
"service": {"address": "service.local", "port": 8181},
|
|
"code": "edge-code",
|
|
}
|