Cheng Zhou
63 lines
1.6 KiB
Python
63 lines
1.6 KiB
Python
import uuid
|
|
|
|
import pytest
|
|
|
|
from app.core.request_context import (
|
|
RequestAuditContext,
|
|
reset_request_audit_context,
|
|
set_request_audit_context,
|
|
)
|
|
from app.crud.audit import log_action
|
|
|
|
|
|
class FakeAsyncSession:
|
|
def __init__(self) -> None:
|
|
self.added = None
|
|
self.flushed = False
|
|
|
|
def add(self, value) -> None:
|
|
self.added = value
|
|
|
|
async def flush(self) -> None:
|
|
self.flushed = True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_audit_log_action_captures_request_access_context():
|
|
context_token = set_request_audit_context(
|
|
RequestAuditContext(
|
|
client_ip="203.0.113.10",
|
|
user_agent="curl/8.1.2",
|
|
client_type="desktop",
|
|
client_version="1.2.3",
|
|
client_platform="macos",
|
|
build_channel="release",
|
|
build_commit="abcdef1",
|
|
)
|
|
)
|
|
db = FakeAsyncSession()
|
|
try:
|
|
log = await log_action(
|
|
db,
|
|
study_id=uuid.uuid4(),
|
|
entity_type="subject",
|
|
entity_id=uuid.uuid4(),
|
|
action="UPDATE_SUBJECT",
|
|
detail=None,
|
|
operator_id=uuid.uuid4(),
|
|
operator_role="PM",
|
|
auto_commit=False,
|
|
)
|
|
finally:
|
|
reset_request_audit_context(context_token)
|
|
|
|
assert db.added is log
|
|
assert db.flushed is True
|
|
assert log.client_ip == "203.0.113.10"
|
|
assert log.user_agent == "curl/8.1.2"
|
|
assert log.client_type == "desktop"
|
|
assert log.client_version == "1.2.3"
|
|
assert log.client_platform == "macos"
|
|
assert log.build_channel == "release"
|
|
assert log.build_commit == "abcdef1"
|