mirror of
https://github.com/pewdiepie-archdaemon/odysseus.git
synced 2026-06-17 02:05:22 -04:00
fix(integrations): prevent blank API integrations (#3840)
* fix(integrations): validate unified API form fields * fix(integrations): validate API integration fields server-side
This commit is contained in:
@@ -1,4 +1,8 @@
|
||||
import json
|
||||
import asyncio
|
||||
from types import SimpleNamespace
|
||||
|
||||
import pytest
|
||||
|
||||
from src import integrations
|
||||
|
||||
@@ -9,3 +13,117 @@ def test_load_integrations_skips_non_object_rows(tmp_path, monkeypatch):
|
||||
monkeypatch.setattr(integrations, "DATA_FILE", str(data_file))
|
||||
|
||||
assert integrations.load_integrations() == [{"id": "good", "name": "Good"}]
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def integrations_routes(tmp_path, monkeypatch):
|
||||
fastapi = pytest.importorskip("fastapi")
|
||||
from routes import auth_routes
|
||||
|
||||
monkeypatch.setattr(integrations, "DATA_FILE", str(tmp_path / "integrations.json"))
|
||||
monkeypatch.setattr(auth_routes, "migrate_from_settings", lambda: None)
|
||||
|
||||
class _AuthManager:
|
||||
def get_username_for_token(self, token):
|
||||
return "admin" if token == "session-token" else None
|
||||
|
||||
def is_admin(self, user):
|
||||
return user == "admin"
|
||||
|
||||
router = auth_routes.setup_auth_routes(_AuthManager())
|
||||
|
||||
def endpoint(path, method):
|
||||
for route in router.routes:
|
||||
if getattr(route, "path", "") == path and method in getattr(route, "methods", set()):
|
||||
return route.endpoint
|
||||
raise AssertionError(f"{method} {path} route not registered")
|
||||
|
||||
return endpoint, auth_routes.SESSION_COOKIE, fastapi.HTTPException
|
||||
|
||||
|
||||
class _JsonRequest(SimpleNamespace):
|
||||
def __init__(self, body, session_cookie):
|
||||
super().__init__(
|
||||
cookies={session_cookie: "session-token"},
|
||||
client=SimpleNamespace(host="127.0.0.1"),
|
||||
_body=body,
|
||||
)
|
||||
|
||||
async def json(self):
|
||||
return self._body
|
||||
|
||||
|
||||
@pytest.mark.parametrize("blank_name", ["", " "])
|
||||
def test_create_integration_rejects_blank_name_without_persisting(integrations_routes, blank_name):
|
||||
endpoint, session_cookie, http_exception = integrations_routes
|
||||
create_integration = endpoint("/api/auth/integrations", "POST")
|
||||
|
||||
with pytest.raises(http_exception) as exc:
|
||||
asyncio.run(create_integration(
|
||||
_JsonRequest({"name": blank_name, "base_url": "https://example.test"}, session_cookie)
|
||||
))
|
||||
|
||||
assert exc.value.status_code == 400
|
||||
assert exc.value.detail == "Integration name is required"
|
||||
assert integrations.load_integrations() == []
|
||||
|
||||
|
||||
@pytest.mark.parametrize("blank_base_url", ["", " "])
|
||||
def test_create_integration_rejects_blank_base_url_without_persisting(integrations_routes, blank_base_url):
|
||||
endpoint, session_cookie, http_exception = integrations_routes
|
||||
create_integration = endpoint("/api/auth/integrations", "POST")
|
||||
|
||||
with pytest.raises(http_exception) as exc:
|
||||
asyncio.run(create_integration(
|
||||
_JsonRequest({"name": "Example", "base_url": blank_base_url}, session_cookie)
|
||||
))
|
||||
|
||||
assert exc.value.status_code == 400
|
||||
assert exc.value.detail == "Integration base URL is required"
|
||||
assert integrations.load_integrations() == []
|
||||
|
||||
|
||||
@pytest.mark.parametrize("blank_name", ["", " "])
|
||||
def test_update_integration_rejects_blank_name_without_changing_existing(integrations_routes, blank_name):
|
||||
endpoint, session_cookie, http_exception = integrations_routes
|
||||
update_integration = endpoint("/api/auth/integrations/{integration_id}", "PUT")
|
||||
integrations.save_integrations([
|
||||
{
|
||||
"id": "existing",
|
||||
"name": "Original",
|
||||
"base_url": "https://example.test",
|
||||
}
|
||||
])
|
||||
|
||||
with pytest.raises(http_exception) as exc:
|
||||
asyncio.run(update_integration(
|
||||
integration_id="existing",
|
||||
request=_JsonRequest({"name": blank_name}, session_cookie),
|
||||
))
|
||||
|
||||
assert exc.value.status_code == 400
|
||||
assert exc.value.detail == "Integration name is required"
|
||||
assert integrations.load_integrations()[0]["name"] == "Original"
|
||||
|
||||
|
||||
@pytest.mark.parametrize("blank_base_url", ["", " "])
|
||||
def test_update_integration_rejects_blank_base_url_without_changing_existing(integrations_routes, blank_base_url):
|
||||
endpoint, session_cookie, http_exception = integrations_routes
|
||||
update_integration = endpoint("/api/auth/integrations/{integration_id}", "PUT")
|
||||
integrations.save_integrations([
|
||||
{
|
||||
"id": "existing",
|
||||
"name": "Original",
|
||||
"base_url": "https://example.test",
|
||||
}
|
||||
])
|
||||
|
||||
with pytest.raises(http_exception) as exc:
|
||||
asyncio.run(update_integration(
|
||||
integration_id="existing",
|
||||
request=_JsonRequest({"base_url": blank_base_url}, session_cookie),
|
||||
))
|
||||
|
||||
assert exc.value.status_code == 400
|
||||
assert exc.value.detail == "Integration base URL is required"
|
||||
assert integrations.load_integrations()[0]["base_url"] == "https://example.test"
|
||||
|
||||
Reference in New Issue
Block a user