107 lines
3.8 KiB
Python
107 lines
3.8 KiB
Python
"""Tests for cloud scout CRUD routes."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
import pytest
|
|
from httpx import ASGITransport, AsyncClient
|
|
|
|
from app.db import get_session
|
|
from app.main import app
|
|
from app.models import CloudScoutConfig
|
|
from tests.conftest import _TestSessionLocal, make_jwt
|
|
|
|
|
|
def _auth_headers(tier: str = "power") -> dict:
|
|
return {"Authorization": f"Bearer {make_jwt(tier)}"}
|
|
|
|
|
|
async def _test_get_session():
|
|
async with _TestSessionLocal() as session:
|
|
yield session
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _override_session():
|
|
# FastAPI resolves Depends() by the original function object, so patching the
|
|
# module-level name does not take effect — use dependency_overrides instead.
|
|
app.dependency_overrides[get_session] = _test_get_session
|
|
yield
|
|
app.dependency_overrides.pop(get_session, None)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_cloud_scout_defaults_schedule():
|
|
payload = {
|
|
"name": "Inbox",
|
|
"provider": "gmail",
|
|
"data_types": [],
|
|
"prompt_template": "client requests",
|
|
"auto_trash_spam": True,
|
|
# schedule_cron omitted → server default
|
|
}
|
|
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
|
|
resp = await client.post("/api/v1/scouts/cloud", json=payload, headers=_auth_headers())
|
|
assert resp.status_code == 201, resp.text
|
|
body = resp.json()
|
|
assert body["name"] == "Inbox"
|
|
assert body["provider"] == "gmail"
|
|
assert body["auto_trash_spam"] is True
|
|
assert body["prompt_template"] == "client requests"
|
|
assert body["schedule_cron"] # non-empty default applied
|
|
assert body["oauth_connected"] is False
|
|
assert body["gmail_address"] is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_cloud_scouts_returns_only_own():
|
|
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
|
|
await client.post(
|
|
"/api/v1/scouts/cloud",
|
|
json={"name": "A", "provider": "gmail"},
|
|
headers=_auth_headers(),
|
|
)
|
|
resp = await client.get("/api/v1/scouts/cloud", headers=_auth_headers())
|
|
assert resp.status_code == 200
|
|
rows = resp.json()
|
|
assert all(r["provider"] == "gmail" for r in rows)
|
|
assert any(r["name"] == "A" for r in rows)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_cloud_scout_applies_filter_and_autotrash():
|
|
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
|
|
created = (await client.post(
|
|
"/api/v1/scouts/cloud",
|
|
json={"name": "B", "provider": "gmail"},
|
|
headers=_auth_headers(),
|
|
)).json()
|
|
sid = created["id"]
|
|
resp = await client.put(
|
|
f"/api/v1/scouts/cloud/{sid}",
|
|
json={"filter_config": {"labels": ["INBOX"], "senders": ["@client.co"]}, "auto_trash_spam": True, "prompt_template": "invoices"},
|
|
headers=_auth_headers(),
|
|
)
|
|
assert resp.status_code == 200, resp.text
|
|
body = resp.json()
|
|
assert body["filter_config"] == {"labels": ["INBOX"], "senders": ["@client.co"]}
|
|
assert body["auto_trash_spam"] is True
|
|
assert body["prompt_template"] == "invoices"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_delete_cloud_scout():
|
|
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
|
|
created = (await client.post(
|
|
"/api/v1/scouts/cloud",
|
|
json={"name": "C", "provider": "gmail"},
|
|
headers=_auth_headers(),
|
|
)).json()
|
|
sid = created["id"]
|
|
resp = await client.delete(f"/api/v1/scouts/cloud/{sid}", headers=_auth_headers())
|
|
assert resp.status_code == 200
|
|
listing = (await client.get("/api/v1/scouts/cloud", headers=_auth_headers())).json()
|
|
assert all(r["id"] != sid for r in listing)
|