56 lines
1.5 KiB
Python
56 lines
1.5 KiB
Python
from __future__ import annotations
|
|
|
|
from starlette.requests import Request
|
|
|
|
from src.main import _resolve_request_project_context
|
|
|
|
|
|
def _request(
|
|
path: str,
|
|
*,
|
|
query_string: bytes = b"",
|
|
headers: list[tuple[bytes, bytes]] | None = None,
|
|
) -> Request:
|
|
return Request(
|
|
{
|
|
"type": "http",
|
|
"method": "POST",
|
|
"path": path,
|
|
"scheme": "http",
|
|
"server": ("testserver", 80),
|
|
"client": ("127.0.0.1", 50000),
|
|
"query_string": query_string,
|
|
"headers": headers or [],
|
|
}
|
|
)
|
|
|
|
|
|
def test_alertmanager_webhook_without_header_uses_awoooi_project_context():
|
|
project_id, source = _resolve_request_project_context(
|
|
_request("/api/v1/webhooks/alertmanager")
|
|
)
|
|
|
|
assert project_id == "awoooi"
|
|
assert source == "request.alertmanager.default_project"
|
|
|
|
|
|
def test_alertmanager_webhook_keeps_explicit_project_context_priority():
|
|
project_id, source = _resolve_request_project_context(
|
|
_request(
|
|
"/api/v1/webhooks/alertmanager",
|
|
headers=[(b"x-project-id", b"tenant-a")],
|
|
)
|
|
)
|
|
|
|
assert project_id == "tenant-a"
|
|
assert source == "request.header_or_query"
|
|
|
|
|
|
def test_non_alertmanager_route_without_header_still_fails_closed():
|
|
project_id, source = _resolve_request_project_context(
|
|
_request("/api/v1/security/db-context-guard")
|
|
)
|
|
|
|
assert project_id is None
|
|
assert source == "request.project_id.missing"
|