feat: AuditService.record — единственная точка записи audit_event, тесты и статический запрет UPDATE/DELETE
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
Andrey Epifancev
parent
d7f6c280bb
commit
a58f5d9df1
119
api/src/services/audit.py
Normal file
119
api/src/services/audit.py
Normal file
@ -0,0 +1,119 @@
|
||||
"""Сервис аудита: единственная точка записи `audit_event` (spec audit-trail, ТЗ 8.2).
|
||||
|
||||
Журнал append-only: у сервиса один публичный метод — `record`; методов изменения и удаления нет и не будет
|
||||
(проверяется тестом `tests/test_audit_service.py`, он же сканирует `src/` на UPDATE/DELETE по `AuditEvent`).
|
||||
|
||||
Запись выполняется в ТЕКУЩЕЙ транзакции сессии: сервис делает `add` + `flush` и не вызывает `commit`/`rollback` —
|
||||
фиксация и откат остаются за вызывающим кодом, поэтому откат изменения данных откатывает и строку аудита.
|
||||
"""
|
||||
|
||||
import datetime
|
||||
import decimal
|
||||
import enum
|
||||
import math
|
||||
import uuid
|
||||
from collections.abc import Mapping
|
||||
from typing import Any
|
||||
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from src.models.audit_ex import AuditEvent
|
||||
from src.models.enums import AuditAction, AuditSection
|
||||
|
||||
|
||||
def _scalar_to_jsonable(value: Any) -> Any:
|
||||
if value is None or isinstance(value, bool | int | str):
|
||||
return value
|
||||
if isinstance(value, enum.Enum):
|
||||
return to_jsonable(value.value)
|
||||
if isinstance(value, float):
|
||||
if not math.isfinite(value):
|
||||
raise ValueError("Нечисловое значение float (NaN/Infinity) нельзя записать в jsonb")
|
||||
return value
|
||||
if isinstance(value, decimal.Decimal):
|
||||
return format(value, "f")
|
||||
if isinstance(value, uuid.UUID):
|
||||
return str(value)
|
||||
if isinstance(value, datetime.datetime | datetime.date | datetime.time):
|
||||
return value.isoformat()
|
||||
raise TypeError(f"Значение типа {type(value).__name__} не сериализуется в jsonb для audit_event")
|
||||
|
||||
|
||||
def to_jsonable(value: Any) -> Any:
|
||||
"""Приводит значение к JSON-совместимому виду для jsonb.
|
||||
|
||||
UUID -> str; Decimal -> строка без экспоненты (деньги не превращаются в float); datetime/date/time -> isoformat;
|
||||
Enum -> `.value`; Mapping/list/tuple/set -> рекурсивно. Неизвестный тип — `TypeError`, а не молчаливое `str()`.
|
||||
"""
|
||||
if isinstance(value, Mapping):
|
||||
return {str(key): to_jsonable(item) for key, item in value.items()}
|
||||
if isinstance(value, list | tuple | set | frozenset):
|
||||
return [to_jsonable(item) for item in value]
|
||||
return _scalar_to_jsonable(value)
|
||||
|
||||
|
||||
def _normalize_payload(name: str, payload: Mapping[str, Any] | None) -> Any:
|
||||
if payload is None:
|
||||
return None
|
||||
if not isinstance(payload, Mapping):
|
||||
raise TypeError(f"{name} должен быть словарём (Mapping) или None, получено {type(payload).__name__}")
|
||||
return to_jsonable(payload)
|
||||
|
||||
|
||||
class AuditService:
|
||||
"""Запись событий журнала аудита в сессию вызывающего (ручной DI: сессия приходит в конструктор)."""
|
||||
|
||||
def __init__(self, session: AsyncSession) -> None:
|
||||
self._session = session
|
||||
|
||||
async def record( # noqa: PLR0913 - сигнатура зафиксирована спекой audit-trail (все поля audit_event)
|
||||
self,
|
||||
*,
|
||||
action: AuditAction,
|
||||
section: AuditSection,
|
||||
object_type: str,
|
||||
object_id: uuid.UUID | None = None,
|
||||
user_id: uuid.UUID | None = None,
|
||||
consolidation_version_id: uuid.UUID | None = None,
|
||||
description: str | None = None,
|
||||
before_value: Mapping[str, Any] | None = None,
|
||||
after_value: Mapping[str, Any] | None = None,
|
||||
comment: str | None = None,
|
||||
result: str | None = None,
|
||||
changed_field: str | None = None,
|
||||
ip_address: str | None = None,
|
||||
user_agent: str | None = None,
|
||||
) -> AuditEvent:
|
||||
"""Добавляет событие в текущую транзакцию и делает `flush` (id, ранние нарушения ограничений); без `commit`.
|
||||
|
||||
`user_id=None` — системное действие. `occurred_at` ставит сервис (UTC), параметром он не принимается.
|
||||
"""
|
||||
if not isinstance(action, AuditAction):
|
||||
raise TypeError(f"action должен быть членом AuditAction, получено {type(action).__name__}: {action!r}")
|
||||
if not isinstance(section, AuditSection):
|
||||
raise TypeError(f"section должен быть членом AuditSection, получено {type(section).__name__}: {section!r}")
|
||||
if not isinstance(object_type, str):
|
||||
raise TypeError(f"object_type должен быть строкой, получено {type(object_type).__name__}")
|
||||
if not object_type.strip():
|
||||
raise ValueError("object_type не может быть пустым")
|
||||
|
||||
event = AuditEvent(
|
||||
occurred_at=datetime.datetime.now(datetime.UTC),
|
||||
user_id=user_id,
|
||||
action=action,
|
||||
section=section,
|
||||
object_type=object_type,
|
||||
object_id=object_id,
|
||||
consolidation_version_id=consolidation_version_id,
|
||||
description=description,
|
||||
before_value=_normalize_payload("before_value", before_value),
|
||||
after_value=_normalize_payload("after_value", after_value),
|
||||
comment=comment,
|
||||
result=result,
|
||||
changed_field=changed_field,
|
||||
ip_address=ip_address,
|
||||
user_agent=user_agent,
|
||||
)
|
||||
self._session.add(event)
|
||||
await self._session.flush()
|
||||
return event
|
||||
381
api/tests/test_audit_service.py
Normal file
381
api/tests/test_audit_service.py
Normal file
@ -0,0 +1,381 @@
|
||||
"""Сервис аудита (spec audit-trail, задача 21): системное событие, откат, append-only, статический запрет UPDATE/DELETE.
|
||||
|
||||
Быстрые тесты (валидация, сканер, статический обход `src/`) идут без БД; интеграционные — на `db_session`.
|
||||
"""
|
||||
|
||||
import ast
|
||||
import datetime
|
||||
import inspect
|
||||
import re
|
||||
import uuid
|
||||
from decimal import Decimal
|
||||
from pathlib import Path
|
||||
from unittest.mock import AsyncMock
|
||||
|
||||
import pytest
|
||||
from sqlalchemy import func, select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from src.models.audit_ex import AuditEvent
|
||||
from src.models.directories import RegionalOffice
|
||||
from src.models.enums import AuditAction, AuditSection
|
||||
from src.services.audit import AuditService, to_jsonable
|
||||
|
||||
SRC_DIR = Path(__file__).resolve().parent.parent / "src"
|
||||
|
||||
# ---------------------------------------------------------------------------------------------------------------------
|
||||
# Статический сканер: обращения, изменяющие или удаляющие audit_event
|
||||
# ---------------------------------------------------------------------------------------------------------------------
|
||||
|
||||
MUTATING_CALLS = {"update", "delete"}
|
||||
AUDIT_FIELDS = {
|
||||
"occurred_at",
|
||||
"user_id",
|
||||
"action",
|
||||
"section",
|
||||
"object_type",
|
||||
"object_id",
|
||||
"consolidation_version_id",
|
||||
"description",
|
||||
"before_value",
|
||||
"after_value",
|
||||
"comment",
|
||||
"result",
|
||||
"changed_field",
|
||||
"ip_address",
|
||||
"user_agent",
|
||||
"id",
|
||||
}
|
||||
RAW_SQL_MUTATION = re.compile(
|
||||
r"\b(update|delete\s+from|truncate(\s+table)?)\s+(only\s+)?(\"?dkb\"?\.)?\"?audit_event\b", re.I
|
||||
)
|
||||
|
||||
|
||||
def _call_name(node: ast.Call) -> str | None:
|
||||
func = node.func
|
||||
if isinstance(func, ast.Name):
|
||||
return func.id
|
||||
if isinstance(func, ast.Attribute):
|
||||
return func.attr
|
||||
return None
|
||||
|
||||
|
||||
def _mentions_audit_event(node: ast.AST) -> bool:
|
||||
return any(
|
||||
(isinstance(sub, ast.Name) and sub.id == "AuditEvent")
|
||||
or (isinstance(sub, ast.Attribute) and sub.attr == "AuditEvent")
|
||||
for sub in ast.walk(node)
|
||||
)
|
||||
|
||||
|
||||
def _is_audit_event_expr(node: ast.AST | None) -> bool:
|
||||
"""Выражение, дающее экземпляр AuditEvent: `AuditEvent(...)` / `mod.AuditEvent(...)`."""
|
||||
return isinstance(node, ast.Call) and _mentions_audit_event(node.func)
|
||||
|
||||
|
||||
def _tainted_names(tree: ast.AST) -> set[str]:
|
||||
"""Имена, связанные с экземплярами AuditEvent: присваивание `AuditEvent(...)`, аннотация `: AuditEvent`."""
|
||||
names: set[str] = set()
|
||||
for node in ast.walk(tree):
|
||||
if isinstance(node, ast.Assign) and _is_audit_event_expr(node.value):
|
||||
names.update(t.id for t in node.targets if isinstance(t, ast.Name))
|
||||
elif isinstance(node, ast.AnnAssign) and isinstance(node.target, ast.Name):
|
||||
if _mentions_audit_event(node.annotation) or _is_audit_event_expr(node.value):
|
||||
names.add(node.target.id)
|
||||
elif isinstance(node, ast.arg) and node.annotation is not None and _mentions_audit_event(node.annotation):
|
||||
names.add(node.arg)
|
||||
elif isinstance(node, ast.For | ast.comprehension) and _is_audit_event_expr(getattr(node, "iter", None)):
|
||||
pass
|
||||
return names
|
||||
|
||||
|
||||
def find_audit_event_mutations(source: str, filename: str = "<src>") -> list[str]:
|
||||
"""Находки вида `file:line: описание` — любое изменение/удаление audit_event в исходнике.
|
||||
|
||||
Ловит: `update(AuditEvent...)` / `delete(AuditEvent...)` (в т.ч. `sa.update`, `AuditEvent.__table__`);
|
||||
`.delete(x)`/`delete(x)`, где x — экземпляр AuditEvent (внутримодульное отслеживание имён); присваивание/удаление
|
||||
атрибутов таких экземпляров и `AuditEvent.<поле> = ...`; сырой SQL UPDATE/DELETE/TRUNCATE по audit_event.
|
||||
"""
|
||||
tree = ast.parse(source, filename=filename)
|
||||
tainted = _tainted_names(tree)
|
||||
findings: list[str] = []
|
||||
|
||||
def add(node: ast.AST, message: str) -> None:
|
||||
findings.append(f"{filename}:{getattr(node, 'lineno', 0)}: {message}")
|
||||
|
||||
for node in ast.walk(tree):
|
||||
if isinstance(node, ast.Call):
|
||||
name = _call_name(node)
|
||||
args: list[ast.AST] = [*node.args, *(kw.value for kw in node.keywords)]
|
||||
if name in MUTATING_CALLS:
|
||||
if any(_mentions_audit_event(arg) for arg in args):
|
||||
add(node, f"{name}() по AuditEvent")
|
||||
elif any(isinstance(arg, ast.Name) and arg.id in tainted for arg in args):
|
||||
add(node, f"{name}() экземпляра AuditEvent")
|
||||
elif isinstance(node, ast.Assign | ast.AugAssign | ast.AnnAssign | ast.Delete):
|
||||
targets = node.targets if isinstance(node, ast.Assign | ast.Delete) else [node.target]
|
||||
for target in targets:
|
||||
if isinstance(target, ast.Attribute) and target.attr in AUDIT_FIELDS:
|
||||
base = target.value
|
||||
if (
|
||||
isinstance(base, ast.Name) and (base.id in tainted or base.id == "AuditEvent")
|
||||
) or _is_audit_event_expr(base):
|
||||
add(node, f"изменение поля {target.attr} у AuditEvent")
|
||||
elif isinstance(node, ast.Constant) and isinstance(node.value, str) and RAW_SQL_MUTATION.search(node.value):
|
||||
add(node, "сырой SQL, изменяющий audit_event")
|
||||
return findings
|
||||
|
||||
|
||||
def scan_directory(root: Path) -> list[str]:
|
||||
findings: list[str] = []
|
||||
for path in sorted(root.rglob("*.py")):
|
||||
findings.extend(find_audit_event_mutations(path.read_text(encoding="utf-8"), str(path.relative_to(root))))
|
||||
return findings
|
||||
|
||||
|
||||
def test_src_has_no_audit_event_update_or_delete() -> None:
|
||||
assert SRC_DIR.is_dir()
|
||||
assert list(SRC_DIR.rglob("audit.py")), "сканер должен видеть src/services/audit.py"
|
||||
assert scan_directory(SRC_DIR) == []
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"source",
|
||||
[
|
||||
"from sqlalchemy import update\nstmt = update(AuditEvent).values(comment='x')",
|
||||
"import sqlalchemy as sa\nstmt = sa.delete(AuditEvent).where(AuditEvent.id == 1)",
|
||||
"stmt = update(AuditEvent.__table__).values(result='x')",
|
||||
"stmt = delete(models.audit_ex.AuditEvent)",
|
||||
"async def f(session):\n event = AuditEvent(object_type='x')\n await session.delete(event)",
|
||||
"def f(event: AuditEvent):\n event.result = 'changed'",
|
||||
"def f(s):\n e = AuditEvent(object_type='x')\n e.comment = 'x'",
|
||||
"def f(s):\n e = AuditEvent(object_type='x')\n del e.comment",
|
||||
"AuditEvent.comment = 'x'",
|
||||
"stmt = text('UPDATE dkb.audit_event SET comment = 1')",
|
||||
"stmt = text('delete from audit_event')",
|
||||
"stmt = text('TRUNCATE TABLE dkb.audit_event')",
|
||||
],
|
||||
)
|
||||
def test_scanner_detects_violation(source: str) -> None:
|
||||
assert find_audit_event_mutations(source), f"сканер пропустил нарушение: {source!r}"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"source",
|
||||
[
|
||||
"stmt = select(AuditEvent).where(AuditEvent.id == 1)",
|
||||
"event = AuditEvent(object_type='x')\nsession.add(event)",
|
||||
"stmt = update(RegionalOffice).values(name='x')",
|
||||
"async def f(session, office):\n await session.delete(office)",
|
||||
"def f(office):\n office.comment = 'x'",
|
||||
"stmt = text('select * from audit_event')",
|
||||
],
|
||||
)
|
||||
def test_scanner_passes_allowed_code(source: str) -> None:
|
||||
assert find_audit_event_mutations(source) == []
|
||||
|
||||
|
||||
def test_scan_directory_fails_on_synthetic_violation(tmp_path: Path) -> None:
|
||||
(tmp_path / "ok.py").write_text("x = 1\n", encoding="utf-8")
|
||||
sub = tmp_path / "services"
|
||||
sub.mkdir()
|
||||
(sub / "bad.py").write_text("from sqlalchemy import update\nstmt = update(AuditEvent)\n", encoding="utf-8")
|
||||
findings = scan_directory(tmp_path)
|
||||
assert len(findings) == 1
|
||||
assert findings[0].startswith("services/bad.py:2:")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------------------------------------------------
|
||||
# Append-only и валидация (без БД)
|
||||
# ---------------------------------------------------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_service_exposes_only_record() -> None:
|
||||
public = {name for name in dir(AuditService) if not name.startswith("_")}
|
||||
assert public == {"record"}
|
||||
|
||||
|
||||
async def test_record_rejects_empty_object_type() -> None:
|
||||
service = AuditService(AsyncMock(spec=AsyncSession))
|
||||
for bad in ("", " "):
|
||||
with pytest.raises(ValueError, match="object_type"):
|
||||
await service.record(action=AuditAction.LOGIN, section=AuditSection.SESSION, object_type=bad)
|
||||
|
||||
|
||||
async def test_record_rejects_string_enums() -> None:
|
||||
session = AsyncMock(spec=AsyncSession)
|
||||
service = AuditService(session)
|
||||
with pytest.raises(TypeError, match="action"):
|
||||
await service.record(action="LOGIN", section=AuditSection.SESSION, object_type="app_user") # type: ignore[arg-type]
|
||||
with pytest.raises(TypeError, match="section"):
|
||||
await service.record(action=AuditAction.LOGIN, section="SESSION", object_type="app_user") # type: ignore[arg-type]
|
||||
session.add.assert_not_called()
|
||||
|
||||
|
||||
async def test_record_rejects_non_mapping_payload_and_unknown_types() -> None:
|
||||
service = AuditService(AsyncMock(spec=AsyncSession))
|
||||
with pytest.raises(TypeError, match="before_value"):
|
||||
await service.record(
|
||||
action=AuditAction.LOGIN,
|
||||
section=AuditSection.SESSION,
|
||||
object_type="app_user",
|
||||
before_value=["x"], # type: ignore[arg-type]
|
||||
)
|
||||
with pytest.raises(TypeError, match="object"):
|
||||
await service.record(
|
||||
action=AuditAction.LOGIN,
|
||||
section=AuditSection.SESSION,
|
||||
object_type="app_user",
|
||||
after_value={"x": object()},
|
||||
)
|
||||
|
||||
|
||||
def test_record_does_not_accept_occurred_at() -> None:
|
||||
assert "occurred_at" not in inspect.signature(AuditService.record).parameters
|
||||
|
||||
|
||||
def test_to_jsonable_types() -> None:
|
||||
uid = uuid.UUID("12345678-1234-5678-1234-567812345678")
|
||||
moment = datetime.datetime(2026, 10, 3, 12, 0, tzinfo=datetime.UTC)
|
||||
result = to_jsonable(
|
||||
{
|
||||
"sum": Decimal("106311116.46"),
|
||||
"tiny": Decimal("1E+2"),
|
||||
"id": uid,
|
||||
"at": moment,
|
||||
"day": datetime.date(2026, 10, 3),
|
||||
"action": AuditAction.LOGIN,
|
||||
"nested": [{"k": Decimal("0.10")}, (1, 2)],
|
||||
3: None,
|
||||
}
|
||||
)
|
||||
assert result == {
|
||||
"sum": "106311116.46",
|
||||
"tiny": "100",
|
||||
"id": str(uid),
|
||||
"at": moment.isoformat(),
|
||||
"day": "2026-10-03",
|
||||
"action": "LOGIN",
|
||||
"nested": [{"k": "0.10"}, [1, 2]],
|
||||
"3": None,
|
||||
}
|
||||
with pytest.raises(ValueError, match="NaN"):
|
||||
to_jsonable({"x": float("nan")})
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------------------------------------------------
|
||||
# Интеграция: запись в текущую транзакцию
|
||||
# ---------------------------------------------------------------------------------------------------------------------
|
||||
|
||||
|
||||
async def _count_events(session: AsyncSession, object_id: uuid.UUID) -> int:
|
||||
return (
|
||||
await session.scalar(select(func.count()).select_from(AuditEvent).where(AuditEvent.object_id == object_id)) or 0
|
||||
)
|
||||
|
||||
|
||||
async def _count_offices(session: AsyncSession, name: str) -> int:
|
||||
return (
|
||||
await session.scalar(select(func.count()).select_from(RegionalOffice).where(RegionalOffice.name == name)) or 0
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_system_event_without_user(db_session: AsyncSession) -> None:
|
||||
before = datetime.datetime.now(datetime.UTC)
|
||||
event = await AuditService(db_session).record(
|
||||
action=AuditAction.RECOGNIZED_DATA_EDIT,
|
||||
section=AuditSection.RECOGNITION,
|
||||
object_type="recognition_job",
|
||||
user_id=None,
|
||||
before_value={"status": "queued"},
|
||||
after_value={"status": "published"},
|
||||
)
|
||||
assert event.id is not None
|
||||
await db_session.refresh(event) # перечитать строку из БД (jsonb round-trip)
|
||||
|
||||
row = (await db_session.execute(select(AuditEvent).where(AuditEvent.id == event.id))).scalar_one()
|
||||
assert row.user_id is None
|
||||
assert row.action is AuditAction.RECOGNIZED_DATA_EDIT
|
||||
assert row.section is AuditSection.RECOGNITION
|
||||
assert row.before_value == {"status": "queued"}
|
||||
assert row.after_value == {"status": "published"}
|
||||
assert row.occurred_at.tzinfo is not None
|
||||
assert row.occurred_at.utcoffset() == datetime.timedelta(0)
|
||||
assert (
|
||||
before - datetime.timedelta(seconds=5)
|
||||
<= row.occurred_at
|
||||
<= datetime.datetime.now(datetime.UTC) + datetime.timedelta(seconds=5)
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_payload_decimal_uuid_datetime_stored_as_strings(db_session: AsyncSession) -> None:
|
||||
uid = uuid.uuid4()
|
||||
moment = datetime.datetime(2026, 10, 3, 9, 30, 15, tzinfo=datetime.UTC)
|
||||
event = await AuditService(db_session).record(
|
||||
action=AuditAction.CORRECTION_CREATE,
|
||||
section=AuditSection.CONSOLIDATION,
|
||||
object_type="correction",
|
||||
object_id=uid,
|
||||
after_value={"amount": Decimal("106311116.46"), "ref": uid, "at": moment, "kind": AuditSection.KFO},
|
||||
changed_field="amount",
|
||||
comment="ручная корректировка",
|
||||
result="ok",
|
||||
ip_address="10.0.0.1",
|
||||
user_agent="pytest",
|
||||
description="тест",
|
||||
)
|
||||
await db_session.refresh(event)
|
||||
assert event.after_value == {
|
||||
"amount": "106311116.46",
|
||||
"ref": str(uid),
|
||||
"at": moment.isoformat(),
|
||||
"kind": "KFO",
|
||||
}
|
||||
assert isinstance(event.after_value["amount"], str)
|
||||
assert (event.object_id, event.changed_field, event.comment, event.result) == (
|
||||
uid,
|
||||
"amount",
|
||||
"ручная корректировка",
|
||||
"ok",
|
||||
)
|
||||
assert (event.ip_address, event.user_agent, event.description) == ("10.0.0.1", "pytest", "тест")
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_rollback_removes_audit_with_data(db_session: AsyncSession) -> None:
|
||||
office_name = f"Филиал-аудит-{uuid.uuid4().hex[:8]}"
|
||||
object_id = uuid.uuid4()
|
||||
|
||||
with pytest.raises(RuntimeError, match="отмена"):
|
||||
async with db_session.begin_nested(): # SAVEPOINT: откат при исключении, как откат транзакции запроса
|
||||
db_session.add(RegionalOffice(name=office_name))
|
||||
event = await AuditService(db_session).record(
|
||||
action=AuditAction.DIRECTORY_IMPORT,
|
||||
section=AuditSection.DIRECTORIES,
|
||||
object_type="regional_office",
|
||||
object_id=object_id,
|
||||
)
|
||||
# внутри транзакции запись и данные видны
|
||||
assert event.occurred_at is not None
|
||||
seen = await _count_events(db_session, object_id)
|
||||
assert seen == 1
|
||||
assert await _count_offices(db_session, office_name) == 1
|
||||
raise RuntimeError("отмена")
|
||||
|
||||
assert await _count_events(db_session, object_id) == 0
|
||||
assert await _count_offices(db_session, office_name) == 0
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
async def test_record_does_not_commit_and_external_rollback_discards(db_session: AsyncSession) -> None:
|
||||
object_id = uuid.uuid4()
|
||||
await AuditService(db_session).record(
|
||||
action=AuditAction.LOGIN, section=AuditSection.SESSION, object_type="app_user", object_id=object_id
|
||||
)
|
||||
assert db_session.in_transaction() # record не закрыл транзакцию (нет commit)
|
||||
assert await _count_events(db_session, object_id) == 1
|
||||
|
||||
await db_session.rollback() # внешний откат (в фикстуре — к SAVEPOINT) уничтожает запись
|
||||
assert await _count_events(db_session, object_id) == 0
|
||||
Loading…
x
Reference in New Issue
Block a user