122 lines
3.7 KiB
Python
122 lines
3.7 KiB
Python
from unittest.mock import AsyncMock
|
|
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
from fastapi.testclient import TestClient
|
|
from pydantic import ValidationError
|
|
|
|
from app.webhook.models import NextcloudEvent, EventType
|
|
from app.webhook.auth import verify_secret
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"raw,expected",
|
|
[
|
|
("created", EventType.CREATED),
|
|
("updated", EventType.UPDATED),
|
|
("deleted", EventType.DELETED),
|
|
],
|
|
)
|
|
def test_event_parses_valid_types(raw, expected):
|
|
evt = NextcloudEvent(event_type=raw, file_path="a/b.pdf", file_name="b.pdf")
|
|
assert evt.event_type == expected
|
|
|
|
|
|
def test_event_invalid_type_raises():
|
|
with pytest.raises(ValidationError):
|
|
NextcloudEvent(event_type="exploded", file_path="a", file_name="a")
|
|
|
|
|
|
def test_verify_secret_pass():
|
|
verify_secret(provided="abc", expected="abc") # no exception
|
|
|
|
|
|
def test_verify_secret_fail():
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
verify_secret(provided="wrong", expected="abc")
|
|
assert exc_info.value.status_code == 401
|
|
|
|
|
|
def test_verify_secret_missing_fail():
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
verify_secret(provided=None, expected="abc")
|
|
assert exc_info.value.status_code == 401
|
|
|
|
|
|
def _make_app(monkeypatch):
|
|
"""Build the FastAPI app with all external clients stubbed."""
|
|
monkeypatch.setenv("NEXTCLOUD_WEBDAV_URL", "http://nc")
|
|
monkeypatch.setenv("NEXTCLOUD_USER", "u")
|
|
monkeypatch.setenv("NEXTCLOUD_APP_PASSWORD", "p")
|
|
monkeypatch.setenv("OLLAMA_URL", "http://ollama")
|
|
monkeypatch.setenv("OLLAMA_EMBED_MODEL", "m")
|
|
monkeypatch.setenv("QDRANT_URL", "http://qdrant")
|
|
monkeypatch.setenv("QDRANT_COLLECTION", "rag_test")
|
|
monkeypatch.setenv("WEBHOOK_SECRET", "abc")
|
|
|
|
# Reset cached settings/clients
|
|
from app.config import get_settings
|
|
get_settings.cache_clear()
|
|
import app.ingest.pipeline as pipe
|
|
pipe._qdrant_client.cache_clear()
|
|
|
|
# Stub the lifespan startup so it doesn't try to talk to real services
|
|
monkeypatch.setattr("app.main._startup_ensure_collection", AsyncMock())
|
|
|
|
from app.main import app
|
|
return app
|
|
|
|
|
|
def test_health_endpoint_no_auth(monkeypatch):
|
|
app = _make_app(monkeypatch)
|
|
with TestClient(app) as client:
|
|
r = client.get("/health")
|
|
assert r.status_code == 200
|
|
assert r.json() == {"status": "ok"}
|
|
|
|
|
|
def test_webhook_rejects_missing_secret(monkeypatch):
|
|
app = _make_app(monkeypatch)
|
|
with TestClient(app) as client:
|
|
r = client.post("/webhook", json={
|
|
"event_type": "created",
|
|
"file_path": "a/b.pdf",
|
|
"file_name": "b.pdf",
|
|
})
|
|
assert r.status_code == 401
|
|
|
|
|
|
def test_webhook_rejects_wrong_secret(monkeypatch):
|
|
app = _make_app(monkeypatch)
|
|
with TestClient(app) as client:
|
|
r = client.post(
|
|
"/webhook",
|
|
json={"event_type": "created", "file_path": "a/b.pdf", "file_name": "b.pdf"},
|
|
headers={"X-Webhook-Secret": "wrong"},
|
|
)
|
|
assert r.status_code == 401
|
|
|
|
|
|
def test_webhook_dispatches_background_task(monkeypatch):
|
|
app = _make_app(monkeypatch)
|
|
|
|
process_mock = AsyncMock()
|
|
monkeypatch.setattr("app.webhook.handler.process_file", process_mock)
|
|
|
|
with TestClient(app) as client:
|
|
r = client.post(
|
|
"/webhook",
|
|
json={
|
|
"event_type": "created",
|
|
"file_path": "Documents/THB/Studium/2.Semester/Databases/x.pdf",
|
|
"file_name": "x.pdf",
|
|
},
|
|
headers={"X-Webhook-Secret": "abc"},
|
|
)
|
|
|
|
assert r.status_code == 202
|
|
process_mock.assert_awaited_once_with(
|
|
"Documents/THB/Studium/2.Semester/Databases/x.pdf",
|
|
EventType.CREATED,
|
|
)
|