a9c1c698b0
Test-only refactor continuing #2523. Adds a shared import-state isolation helper with focused coverage and migrates two pilot tests that manually preserved sys.modules and parent package attributes.
113 lines
3.4 KiB
Python
113 lines
3.4 KiB
Python
import sys
|
|
import json
|
|
from datetime import datetime
|
|
|
|
import pytest
|
|
|
|
from tests.helpers.import_state import clear_module, preserve_import_state
|
|
|
|
# conftest.py stubs src.database; drop the stub so webhook_manager imports the
|
|
# real module. preserve_import_state restores both sys.modules and the parent
|
|
# src.database attribute after the block, preventing stub leakage into siblings.
|
|
with preserve_import_state("src.database"):
|
|
clear_module("src.database")
|
|
_core_database = sys.modules.get("core.database")
|
|
_core_database_all = (
|
|
getattr(_core_database, "__all__", None) if _core_database is not None else None
|
|
)
|
|
if _core_database is not None and (
|
|
not getattr(_core_database, "__file__", None)
|
|
or (
|
|
_core_database_all is not None
|
|
and (
|
|
not isinstance(_core_database_all, (list, tuple, set))
|
|
or not all(isinstance(name, str) for name in _core_database_all)
|
|
)
|
|
)
|
|
):
|
|
del sys.modules["core.database"]
|
|
from src.webhook_manager import validate_webhook_url
|
|
|
|
|
|
def test_webhook_url_ssrf_mitigation():
|
|
# SSRF bypasses that must be rejected, including IPv6 unspecified and
|
|
# IPv4-mapped IPv6 (loopback + cloud metadata).
|
|
private_urls = [
|
|
"http://[::]/",
|
|
"http://[::ffff:127.0.0.1]/",
|
|
"http://[::ffff:169.254.169.254]/",
|
|
"http://127.0.0.1/",
|
|
"http://0.0.0.0/",
|
|
]
|
|
for url in private_urls:
|
|
with pytest.raises(ValueError) as exc:
|
|
validate_webhook_url(url)
|
|
assert "private/internal addresses" in str(exc.value)
|
|
|
|
# A clearly public IP literal must still be accepted.
|
|
public_url = "http://93.184.216.34/"
|
|
assert validate_webhook_url(public_url) == public_url
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_webhook_delivery_uses_naive_utc_timestamps(monkeypatch):
|
|
import src.webhook_manager as wm
|
|
|
|
class _Query:
|
|
def __init__(self, updates):
|
|
self.updates = updates
|
|
|
|
def filter(self, *_args, **_kwargs):
|
|
return self
|
|
|
|
def update(self, values):
|
|
self.updates.append(values)
|
|
|
|
class _Db:
|
|
def __init__(self):
|
|
self.updates = []
|
|
self.committed = False
|
|
self.closed = False
|
|
|
|
def query(self, _model):
|
|
return _Query(self.updates)
|
|
|
|
def commit(self):
|
|
self.committed = True
|
|
|
|
def rollback(self):
|
|
pass
|
|
|
|
def close(self):
|
|
self.closed = True
|
|
|
|
class _Response:
|
|
status_code = 204
|
|
|
|
class _Client:
|
|
def __init__(self):
|
|
self.content = ""
|
|
|
|
async def post(self, _url, content, headers):
|
|
self.content = content
|
|
assert headers["X-Odysseus-Event"] == "webhook.test"
|
|
return _Response()
|
|
|
|
db = _Db()
|
|
client = _Client()
|
|
monkeypatch.setattr(wm, "SessionLocal", lambda: db)
|
|
|
|
manager = wm.WebhookManager()
|
|
await manager._client.aclose()
|
|
manager._client = client
|
|
|
|
await manager._deliver("hook-1", "http://93.184.216.34/", None, "webhook.test", {"ok": True})
|
|
|
|
body = json.loads(client.content)
|
|
payload_timestamp = datetime.fromisoformat(body["timestamp"])
|
|
assert payload_timestamp.tzinfo is None
|
|
assert db.updates[0]["last_triggered_at"].tzinfo is None
|
|
assert db.updates[0]["last_status_code"] == 204
|
|
assert db.committed is True
|
|
assert db.closed is True
|