""" 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"]