sakshamandClaude Sonnet 5
18305b545c
Reports: the LLM reliably used company_enrichment for prose fields but inconsistently populated the parallel Finding-list/string-list fields from the same evidence, even with progressively more explicit prompting. Add a code-level backfill (products, recent developments, financial signals, strategic initiatives, regulatory signals, risks/opportunities mirrored from SWOT, unknowns, monitoring recommendations) that only ever fills in what the model left empty, never overwrites what it produced. Enrichment tab: reorder sections (Products/Recent updates before Customers/Competitors) and add a per-section "Refresh" button that re-fetches just one of NinjaPear's six independent per-company endpoints when it came back empty - confirmed live that a data-coverage gap (e.g. Amazon returning no products) is real provider behavior, not a bug. Auth: the first account registered on a deployment with zero existing admins is now auto-promoted to admin, closing the chicken-and-egg gap where the only path to admin access was direct DB access. Self-heals if the last admin ever deletes their account. Also bumps nginx's proxy_read_timeout for api.ciagent.org to cover the enrichment refresh's synchronous funding-endpoint call (up to 5 minutes per NinjaPear's docs). Co-Authored-By: Claude Sonnet 5 <[email protected]>
289 lines
11 KiB
Python
289 lines
11 KiB
Python
"""Auth flow tests. Runs under AUTH_MODE=jwt (the suite default) except where
|
|
`local_mode_client` explicitly exercises the local-dev-user path."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
from datetime import UTC, datetime, timedelta
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.core.config import get_settings
|
|
from app.main import app
|
|
from app.models.enums import SecurityEventType
|
|
from app.models.user import LOCAL_DEV_USER_ID
|
|
from app.repositories.user_security_event_repository import UserSecurityEventRepository
|
|
|
|
|
|
def _unique_email() -> str:
|
|
return f"user-{uuid.uuid4().hex[:12]}@example.com"
|
|
|
|
|
|
def test_register_then_login(client):
|
|
email = _unique_email()
|
|
register_resp = client.post(
|
|
"/api/v1/auth/register",
|
|
json={
|
|
"email": email,
|
|
"password": "correct-horse-1",
|
|
"display_name": "Test User",
|
|
},
|
|
)
|
|
assert register_resp.status_code == 201
|
|
body = register_resp.json()
|
|
assert body["email"] == email
|
|
assert "password" not in body
|
|
assert "password_hash" not in body
|
|
|
|
login_resp = client.post(
|
|
"/api/v1/auth/login", json={"email": email, "password": "correct-horse-1"}
|
|
)
|
|
assert login_resp.status_code == 200
|
|
tokens = login_resp.json()
|
|
assert tokens["access_token"]
|
|
assert tokens["refresh_token"]
|
|
assert tokens["token_type"] == "bearer"
|
|
|
|
|
|
def test_register_duplicate_email_conflicts(client):
|
|
email = _unique_email()
|
|
payload = {"email": email, "password": "correct-horse-1", "display_name": "Test User"}
|
|
first = client.post("/api/v1/auth/register", json=payload)
|
|
assert first.status_code == 201
|
|
|
|
second = client.post("/api/v1/auth/register", json=payload)
|
|
assert second.status_code == 409
|
|
|
|
|
|
def test_register_rejects_weak_password(client):
|
|
resp = client.post(
|
|
"/api/v1/auth/register",
|
|
json={"email": _unique_email(), "password": "allletters", "display_name": "Test User"},
|
|
)
|
|
assert resp.status_code == 422
|
|
|
|
|
|
async def test_first_registration_on_an_admin_less_deployment_becomes_admin(db_session, settings):
|
|
"""Bootstraps admin access on a fresh deployment - without this, the
|
|
only way to ever get an admin account is direct DB access. Patches
|
|
count_admins() to simulate a genuinely admin-less deployment rather
|
|
than manipulating the shared session-wide test DB's real admin count
|
|
(conftest.py seeds one admin precisely so ordinary registrations in
|
|
other tests never accidentally trip this path)."""
|
|
from unittest.mock import patch
|
|
|
|
from app.repositories.user_repository import UserRepository
|
|
from app.schemas.auth import RegisterRequest
|
|
from app.services.auth_service import register
|
|
|
|
payload = RegisterRequest(
|
|
email=_unique_email(), password="correct-horse-1", display_name="First User"
|
|
)
|
|
with patch.object(UserRepository, "count_admins", return_value=0):
|
|
user = await register(db_session, settings, "127.0.0.1", payload)
|
|
|
|
assert user.is_admin is True
|
|
|
|
|
|
async def test_registration_after_an_admin_already_exists_is_not_promoted(db_session, settings):
|
|
from app.schemas.auth import RegisterRequest
|
|
from app.services.auth_service import register
|
|
|
|
payload = RegisterRequest(
|
|
email=_unique_email(), password="correct-horse-1", display_name="Second User"
|
|
)
|
|
user = await register(db_session, settings, "127.0.0.1", payload)
|
|
|
|
assert user.is_admin is False
|
|
|
|
|
|
def test_login_wrong_password_rejected(client):
|
|
email = _unique_email()
|
|
client.post(
|
|
"/api/v1/auth/register",
|
|
json={"email": email, "password": "correct-horse-1", "display_name": "Test User"},
|
|
)
|
|
resp = client.post("/api/v1/auth/login", json={"email": email, "password": "wrong-password-1"})
|
|
assert resp.status_code == 401
|
|
|
|
|
|
def test_me_requires_bearer_token(client):
|
|
resp = client.get("/api/v1/auth/me")
|
|
assert resp.status_code == 401
|
|
|
|
|
|
def test_me_returns_current_user(client):
|
|
email = _unique_email()
|
|
client.post(
|
|
"/api/v1/auth/register",
|
|
json={"email": email, "password": "correct-horse-1", "display_name": "Test User"},
|
|
)
|
|
login = client.post(
|
|
"/api/v1/auth/login", json={"email": email, "password": "correct-horse-1"}
|
|
).json()
|
|
|
|
resp = client.get(
|
|
"/api/v1/auth/me", headers={"Authorization": f"Bearer {login['access_token']}"}
|
|
)
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
assert body["email"] == email
|
|
assert body["auth_mode"] == "jwt"
|
|
|
|
|
|
def test_refresh_rotates_and_invalidates_old_token(client):
|
|
email = _unique_email()
|
|
client.post(
|
|
"/api/v1/auth/register",
|
|
json={"email": email, "password": "correct-horse-1", "display_name": "Test User"},
|
|
)
|
|
login = client.post(
|
|
"/api/v1/auth/login", json={"email": email, "password": "correct-horse-1"}
|
|
).json()
|
|
|
|
refresh_resp = client.post(
|
|
"/api/v1/auth/refresh", json={"refresh_token": login["refresh_token"]}
|
|
)
|
|
assert refresh_resp.status_code == 200
|
|
new_tokens = refresh_resp.json()
|
|
assert new_tokens["refresh_token"] != login["refresh_token"]
|
|
|
|
# The old refresh token was rotated out and must not be reusable.
|
|
reuse_resp = client.post("/api/v1/auth/refresh", json={"refresh_token": login["refresh_token"]})
|
|
assert reuse_resp.status_code == 401
|
|
|
|
|
|
def test_logout_revokes_refresh_token(client):
|
|
email = _unique_email()
|
|
client.post(
|
|
"/api/v1/auth/register",
|
|
json={"email": email, "password": "correct-horse-1", "display_name": "Test User"},
|
|
)
|
|
login = client.post(
|
|
"/api/v1/auth/login", json={"email": email, "password": "correct-horse-1"}
|
|
).json()
|
|
|
|
logout_resp = client.post("/api/v1/auth/logout", json={"refresh_token": login["refresh_token"]})
|
|
assert logout_resp.status_code == 204
|
|
|
|
reuse_resp = client.post("/api/v1/auth/refresh", json={"refresh_token": login["refresh_token"]})
|
|
assert reuse_resp.status_code == 401
|
|
|
|
|
|
def test_local_mode_me_returns_fixed_dev_user_without_token(local_mode_client):
|
|
resp = local_mode_client.get("/api/v1/auth/me")
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
assert body["email"] == "[email protected]"
|
|
assert body["auth_mode"] == "local"
|
|
|
|
|
|
def test_register_login_available_even_in_local_mode(local_mode_client):
|
|
"""Registering/logging in a real account must always be possible,
|
|
regardless of AUTH_MODE - the loopback convenience only affects whether
|
|
a request can skip auth entirely, not whether real accounts exist."""
|
|
email = _unique_email()
|
|
register_resp = local_mode_client.post(
|
|
"/api/v1/auth/register",
|
|
json={"email": email, "password": "correct-horse-1", "display_name": "X"},
|
|
)
|
|
assert register_resp.status_code == 201
|
|
|
|
login_resp = local_mode_client.post(
|
|
"/api/v1/auth/login", json={"email": email, "password": "correct-horse-1"}
|
|
)
|
|
assert login_resp.status_code == 200
|
|
assert login_resp.json()["access_token"]
|
|
|
|
|
|
def test_security_events_requires_auth(client):
|
|
resp = client.get("/api/v1/auth/security-events")
|
|
assert resp.status_code == 401
|
|
|
|
|
|
def _login_success_count(events: list[dict]) -> int:
|
|
return sum(1 for e in events if e["event_type"] == "login_success")
|
|
|
|
|
|
def test_local_dev_sign_in_is_logged_to_account_activity(local_mode_client):
|
|
"""The local-dev bypass has no real login step - hitting any
|
|
authenticated endpoint (here /auth/me, the same check the frontend
|
|
performs on app load) must still show up as a sign-in."""
|
|
local_mode_client.get("/api/v1/auth/me")
|
|
events = local_mode_client.get("/api/v1/auth/security-events").json()
|
|
assert any(e["event_type"] == "login_success" for e in events)
|
|
|
|
|
|
def test_local_dev_sign_in_is_not_logged_twice_within_cooldown(local_mode_client):
|
|
"""Every authenticated request runs get_or_create_local_user - without
|
|
a cooldown, browsing the app at all would flood Account activity with
|
|
one login_success per request."""
|
|
local_mode_client.get("/api/v1/auth/me")
|
|
before = _login_success_count(local_mode_client.get("/api/v1/auth/security-events").json())
|
|
|
|
local_mode_client.get("/api/v1/auth/me")
|
|
after = _login_success_count(local_mode_client.get("/api/v1/auth/security-events").json())
|
|
|
|
assert after == before
|
|
|
|
|
|
async def test_local_dev_sign_in_logs_again_after_cooldown_expires(local_mode_client, db_session):
|
|
local_mode_client.get("/api/v1/auth/me")
|
|
before = _login_success_count(local_mode_client.get("/api/v1/auth/security-events").json())
|
|
|
|
event_repo = UserSecurityEventRepository(db_session)
|
|
last_login = await event_repo.most_recent_of_type(
|
|
LOCAL_DEV_USER_ID, SecurityEventType.LOGIN_SUCCESS
|
|
)
|
|
assert last_login is not None
|
|
last_login.created_at = datetime.now(UTC) - timedelta(minutes=31)
|
|
await db_session.commit()
|
|
|
|
local_mode_client.get("/api/v1/auth/me")
|
|
after = _login_success_count(local_mode_client.get("/api/v1/auth/security-events").json())
|
|
|
|
assert after == before + 1
|
|
|
|
|
|
def test_security_events_returns_own_login_events_only(client):
|
|
email_a = _unique_email()
|
|
client.post(
|
|
"/api/v1/auth/register",
|
|
json={"email": email_a, "password": "correct-horse-1", "display_name": "A"},
|
|
)
|
|
login_a = client.post(
|
|
"/api/v1/auth/login", json={"email": email_a, "password": "correct-horse-1"}
|
|
).json()
|
|
|
|
email_b = _unique_email()
|
|
client.post(
|
|
"/api/v1/auth/register",
|
|
json={"email": email_b, "password": "correct-horse-1", "display_name": "B"},
|
|
)
|
|
client.post("/api/v1/auth/login", json={"email": email_b, "password": "correct-horse-1"})
|
|
|
|
resp = client.get(
|
|
"/api/v1/auth/security-events",
|
|
headers={"Authorization": f"Bearer {login_a['access_token']}"},
|
|
)
|
|
assert resp.status_code == 200
|
|
events = resp.json()
|
|
event_types = {e["event_type"] for e in events}
|
|
# registration writes email_verification_sent, login writes login_success -
|
|
# both belong to user A only, never user B's events.
|
|
assert event_types == {"email_verification_sent", "login_success"}
|
|
|
|
|
|
def test_local_mode_setting_alone_does_not_bypass_auth_for_non_loopback_callers():
|
|
"""AUTH_MODE=local is not a blanket switch - a request that isn't
|
|
actually from loopback (e.g. a LAN/WAN caller, or here Starlette's
|
|
TestClient default fake peer) still needs a real bearer token."""
|
|
settings = get_settings().model_copy(update={"auth_mode": "local"})
|
|
app.dependency_overrides[get_settings] = lambda: settings
|
|
try:
|
|
with TestClient(app) as non_loopback_client: # default peer: ("testclient", 50000)
|
|
resp = non_loopback_client.get("/api/v1/auth/me")
|
|
assert resp.status_code == 401
|
|
finally:
|
|
app.dependency_overrides.pop(get_settings, None)
|