174 lines
6.8 KiB
Python
174 lines
6.8 KiB
Python
"""
|
|
tests/test_api_audit.py
|
|
-----------------------
|
|
Tests for audit logging and the recent activity endpoint.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
from unittest.mock import patch
|
|
|
|
import httpx
|
|
import pytest
|
|
import respx
|
|
from fastapi.testclient import TestClient
|
|
|
|
from web.app import app
|
|
from web.session import _COOKIE_NAME, create_test_session
|
|
import web.routers.migrate as migrate_module
|
|
|
|
client = TestClient(app, raise_server_exceptions=True)
|
|
|
|
ADOBE_BASE = "https://api.eu2.adobesign.com/api/rest/v6"
|
|
DS_BASE = "https://demo.docusign.net/restapi"
|
|
DS_ACCOUNT = "test-account-id"
|
|
ADOBE_ID = "adobe-123"
|
|
DS_NEW_ID = "ds-new-456"
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def temp_audit_env(tmp_path, monkeypatch):
|
|
import web.config as cfg
|
|
|
|
monkeypatch.setattr(cfg.settings, "session_store_dir", str(tmp_path / ".session-store"))
|
|
monkeypatch.setattr(cfg.settings, "audit_log_file", str(tmp_path / ".audit-log.jsonl"))
|
|
monkeypatch.setattr(cfg.settings, "docusign_account_id", DS_ACCOUNT)
|
|
monkeypatch.setattr(cfg.settings, "docusign_base_url", DS_BASE)
|
|
monkeypatch.setattr(cfg.settings, "adobe_sign_base_url", ADOBE_BASE)
|
|
monkeypatch.setattr(migrate_module, "_HISTORY_FILE", str(tmp_path / ".history.json"))
|
|
client.cookies.clear()
|
|
yield
|
|
client.cookies.clear()
|
|
|
|
|
|
def test_adobe_connect_writes_audit_event(monkeypatch):
|
|
monkeypatch.setenv("ADOBE_ACCESS_TOKEN", "existing-token")
|
|
monkeypatch.setenv("ADOBE_REFRESH_TOKEN", "existing-refresh")
|
|
|
|
with patch("adobe_api._refresh_access_token", return_value="refreshed-token"), \
|
|
patch("web.routers.auth._fetch_adobe_profile", return_value={
|
|
"adobe_user_name": "Paul Adobe",
|
|
"adobe_user_email": "paul@example.com",
|
|
"adobe_account_name": "Paul Sandbox",
|
|
"adobe_account_id": "adobe-account-123",
|
|
}):
|
|
resp = client.get("/api/auth/adobe/connect")
|
|
|
|
assert resp.status_code == 200
|
|
|
|
session_cookie = resp.cookies.get("migrator_session")
|
|
activity = client.get("/api/audit/recent", cookies={_COOKIE_NAME: session_cookie})
|
|
assert activity.status_code == 200
|
|
events = activity.json()["events"]
|
|
assert events
|
|
assert events[0]["action"] == "adobe_connected"
|
|
assert events[0]["adobe_account_name"] == "Paul Sandbox"
|
|
assert events[0]["details"]["auth_mode"] == "shared_env"
|
|
|
|
|
|
def _mock_compose(template_dir: str, output_path: str):
|
|
with open(output_path, "w", encoding="utf-8") as f:
|
|
json.dump({"name": "Test NDA", "description": "mocked"}, f)
|
|
|
|
|
|
def _mock_download(template_id, access_token, output_dir):
|
|
os.makedirs(output_dir, exist_ok=True)
|
|
with open(os.path.join(output_dir, "metadata.json"), "w", encoding="utf-8") as f:
|
|
json.dump({"name": "Test NDA", "id": template_id}, f)
|
|
with open(os.path.join(output_dir, "form_fields.json"), "w", encoding="utf-8") as f:
|
|
json.dump({"fields": []}, f)
|
|
with open(os.path.join(output_dir, "documents.json"), "w", encoding="utf-8") as f:
|
|
json.dump({"documents": []}, f)
|
|
return True
|
|
|
|
|
|
async def _async_wrap(fn, *args, **kwargs):
|
|
return fn(*args, **kwargs)
|
|
|
|
|
|
@respx.mock
|
|
def test_migration_writes_audit_event():
|
|
respx.get(f"{DS_BASE}/v2.1/accounts/{DS_ACCOUNT}/templates").mock(
|
|
return_value=httpx.Response(200, json={"envelopeTemplates": []})
|
|
)
|
|
respx.post(f"{DS_BASE}/v2.1/accounts/{DS_ACCOUNT}/templates").mock(
|
|
return_value=httpx.Response(201, json={"templateId": DS_NEW_ID})
|
|
)
|
|
|
|
session_cookie = create_test_session({
|
|
"adobe_access_token": "adobe-tok",
|
|
"docusign_access_token": "ds-tok",
|
|
"adobe_account_name": "Adobe QA",
|
|
"docusign_user_name": "Paul Example",
|
|
})
|
|
|
|
with (
|
|
patch.object(migrate_module, "_download_adobe_template", new=lambda *args, **kwargs: _async_wrap(_mock_download, *args, **kwargs)),
|
|
patch.object(migrate_module, "_load_compose", return_value=_mock_compose),
|
|
):
|
|
resp = client.post(
|
|
"/api/migrate",
|
|
json={"adobe_template_ids": [ADOBE_ID]},
|
|
cookies={_COOKIE_NAME: session_cookie},
|
|
)
|
|
|
|
assert resp.status_code == 200
|
|
|
|
activity = client.get("/api/audit/recent", cookies={_COOKIE_NAME: session_cookie})
|
|
assert activity.status_code == 200
|
|
events = activity.json()["events"]
|
|
migrate_event = next(event for event in events if event["action"] == "migration_run")
|
|
assert migrate_event["adobe_account_name"] == "Adobe QA"
|
|
assert migrate_event["details"]["template_count"] == 1
|
|
assert migrate_event["details"]["success_count"] == 1
|
|
|
|
|
|
def test_audit_recent_is_session_scoped_by_default():
|
|
session_a = create_test_session({
|
|
"docusign_user_email": "a@example.com",
|
|
"docusign_user_name": "Tester A",
|
|
}, session_id="session-a")
|
|
session_b = create_test_session({
|
|
"docusign_user_email": "b@example.com",
|
|
"docusign_user_name": "Tester B",
|
|
}, session_id="session-b")
|
|
|
|
import web.config as cfg
|
|
with open(cfg.settings.audit_log_file, "w", encoding="utf-8") as f:
|
|
f.write(json.dumps({"timestamp": "2026-04-22T15:00:00Z", "action": "docusign_connected", "session_id": "session-a"}) + "\n")
|
|
f.write(json.dumps({"timestamp": "2026-04-22T15:01:00Z", "action": "migration_run", "session_id": "session-b"}) + "\n")
|
|
|
|
resp_a = client.get("/api/audit/recent", cookies={_COOKIE_NAME: session_a})
|
|
resp_b = client.get("/api/audit/recent", cookies={_COOKIE_NAME: session_b})
|
|
|
|
assert resp_a.status_code == 200
|
|
assert resp_b.status_code == 200
|
|
assert [event["session_id"] for event in resp_a.json()["events"]] == ["session-a"]
|
|
assert [event["session_id"] for event in resp_b.json()["events"]] == ["session-b"]
|
|
|
|
|
|
def test_admin_can_request_all_audit_events(monkeypatch):
|
|
import web.config as cfg
|
|
|
|
monkeypatch.setattr(cfg.settings, "admin_emails_raw", "admin@example.com")
|
|
admin_cookie = create_test_session({
|
|
"docusign_user_email": "admin@example.com",
|
|
}, session_id="admin-session")
|
|
user_cookie = create_test_session({
|
|
"docusign_user_email": "user@example.com",
|
|
}, session_id="user-session")
|
|
|
|
with open(cfg.settings.audit_log_file, "w", encoding="utf-8") as f:
|
|
f.write(json.dumps({"timestamp": "2026-04-22T15:00:00Z", "action": "docusign_connected", "session_id": "admin-session"}) + "\n")
|
|
f.write(json.dumps({"timestamp": "2026-04-22T15:01:00Z", "action": "migration_run", "session_id": "user-session"}) + "\n")
|
|
|
|
admin_resp = client.get("/api/audit/recent?all=true", cookies={_COOKIE_NAME: admin_cookie})
|
|
user_resp = client.get("/api/audit/recent?all=true", cookies={_COOKIE_NAME: user_cookie})
|
|
|
|
assert admin_resp.status_code == 200
|
|
assert user_resp.status_code == 200
|
|
assert admin_resp.json()["scope"] == "all"
|
|
assert len(admin_resp.json()["events"]) == 2
|
|
assert user_resp.json()["scope"] == "session"
|
|
assert [event["session_id"] for event in user_resp.json()["events"]] == ["user-session"]
|