Merge pull request 'Логи с уходом от орм' (#16) from future/AURORA-1012 into test
Reviewed-on: #16 Reviewed-by: tsygankoviva <tsygankov.itis@gmail.com>
This commit is contained in:
commit
e37eb09e0a
@ -15,6 +15,9 @@ async def get_db() -> AsyncGenerator:
|
||||
try:
|
||||
yield db
|
||||
await db.commit()
|
||||
except Exception:
|
||||
await db.rollback()
|
||||
raise
|
||||
finally:
|
||||
await db.close()
|
||||
|
||||
|
||||
@ -1,6 +1,6 @@
|
||||
from datetime import datetime
|
||||
|
||||
from sqlalchemy import delete, select, update
|
||||
from sqlalchemy import select, text
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from src.db.models.form_phase import FormPhase
|
||||
@ -45,40 +45,79 @@ class FormPhaseRepository:
|
||||
opens_at: datetime,
|
||||
closes_at: datetime,
|
||||
) -> FormPhase:
|
||||
form_phase = FormPhase(
|
||||
budget_form_id=budget_form_id,
|
||||
sheet=sheet,
|
||||
phase_code=phase_code,
|
||||
role=role,
|
||||
column_keys=column_keys,
|
||||
opens_at=opens_at,
|
||||
closes_at=closes_at,
|
||||
result = await self.db.execute(
|
||||
text(
|
||||
"""
|
||||
SELECT
|
||||
(v3.add_form_phase(
|
||||
:budget_form_id,
|
||||
:sheet,
|
||||
:phase_code,
|
||||
:role,
|
||||
:column_keys,
|
||||
:opens_at,
|
||||
:closes_at
|
||||
)).budget_form_id
|
||||
"""
|
||||
),
|
||||
{
|
||||
"budget_form_id": budget_form_id,
|
||||
"sheet": sheet,
|
||||
"phase_code": phase_code,
|
||||
"role": role,
|
||||
"column_keys": column_keys,
|
||||
"opens_at": opens_at,
|
||||
"closes_at": closes_at,
|
||||
},
|
||||
)
|
||||
self.db.add(form_phase)
|
||||
await self.db.flush()
|
||||
return form_phase
|
||||
if result.scalar_one_or_none() is None:
|
||||
return None
|
||||
return await self.get(budget_form_id, sheet, phase_code)
|
||||
|
||||
async def update(
|
||||
self, budget_form_id: int, sheet: str, phase_code: str, data: dict
|
||||
) -> FormPhase | None:
|
||||
query = (
|
||||
update(FormPhase)
|
||||
.where(FormPhase.budget_form_id == budget_form_id)
|
||||
.where(FormPhase.sheet == sheet)
|
||||
.where(FormPhase.phase_code == phase_code)
|
||||
.values(**data)
|
||||
.returning(FormPhase)
|
||||
result = await self.db.execute(
|
||||
text(
|
||||
"""
|
||||
SELECT
|
||||
(v3.upd_form_phase(
|
||||
:budget_form_id,
|
||||
:sheet,
|
||||
:phase_code,
|
||||
:role,
|
||||
:column_keys,
|
||||
:opens_at,
|
||||
:closes_at
|
||||
)).budget_form_id
|
||||
"""
|
||||
),
|
||||
{
|
||||
"budget_form_id": budget_form_id,
|
||||
"sheet": sheet,
|
||||
"phase_code": phase_code,
|
||||
"role": data.get("role"),
|
||||
"column_keys": data.get("column_keys"),
|
||||
"opens_at": data.get("opens_at"),
|
||||
"closes_at": data.get("closes_at"),
|
||||
},
|
||||
)
|
||||
return (await self.db.execute(query)).scalar_one_or_none()
|
||||
if result.scalar_one_or_none() is None:
|
||||
return None
|
||||
return await self.get(budget_form_id, sheet, phase_code)
|
||||
|
||||
async def delete(
|
||||
self, budget_form_id: int, sheet: str, phase_code: str
|
||||
) -> bool:
|
||||
query = (
|
||||
delete(FormPhase)
|
||||
.where(FormPhase.budget_form_id == budget_form_id)
|
||||
.where(FormPhase.sheet == sheet)
|
||||
.where(FormPhase.phase_code == phase_code)
|
||||
result = await self.db.execute(
|
||||
text(
|
||||
"SELECT v3.del_form_phase(:budget_form_id, :sheet, :phase_code)"
|
||||
),
|
||||
{
|
||||
"budget_form_id": budget_form_id,
|
||||
"sheet": sheet,
|
||||
"phase_code": phase_code,
|
||||
},
|
||||
)
|
||||
result = await self.db.execute(query)
|
||||
return result.rowcount > 0
|
||||
deleted = result.scalar_one_or_none()
|
||||
return bool(deleted)
|
||||
|
||||
@ -1,6 +1,6 @@
|
||||
from typing import Optional
|
||||
|
||||
from sqlalchemy import delete, select, text, update
|
||||
from sqlalchemy import select, text, update
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
from sqlalchemy.orm import selectinload
|
||||
|
||||
@ -75,36 +75,6 @@ class UserRepository:
|
||||
)
|
||||
return (await self.db.execute(query)).scalars().all()
|
||||
|
||||
async def create(self, user_data: dict) -> AppUser:
|
||||
hashed_password = (
|
||||
get_password_hash(user_data["password"]) if user_data.get("password") else None
|
||||
)
|
||||
db_user = AppUser(
|
||||
email=user_data["email"],
|
||||
username=user_data["username"],
|
||||
hashed_password=hashed_password,
|
||||
full_name=user_data.get("full_name"),
|
||||
role_id=user_data.get("role_id"),
|
||||
)
|
||||
self.db.add(db_user)
|
||||
await self.db.commit()
|
||||
await self.db.refresh(db_user)
|
||||
return db_user
|
||||
|
||||
async def update(self, user_id: int, user_data: dict) -> Optional[AppUser]:
|
||||
user = await self.get(user_id)
|
||||
if not user:
|
||||
return None
|
||||
|
||||
for field, value in user_data.items():
|
||||
if field == "password" and value:
|
||||
value = get_password_hash(value)
|
||||
setattr(user, field, value)
|
||||
|
||||
await self.db.commit()
|
||||
await self.db.refresh(user)
|
||||
return user
|
||||
|
||||
async def logical_delete(self, user_id: int) -> bool:
|
||||
query = update(AppUser).where(AppUser.id == user_id).values(is_active=False)
|
||||
result = await self.db.execute(query)
|
||||
@ -126,45 +96,108 @@ class UserRepository:
|
||||
return (await self.db.execute(select(Role).order_by(Role.id))).scalars().all()
|
||||
|
||||
async def get_many_ssp(self, user_id: int) -> list[OrgUnit]:
|
||||
query = select(OrgUnit).join(UserOrg, UserOrg.ssp_id == OrgUnit.id).where(
|
||||
query = select(OrgUnit).join(UserOrg, UserOrg.org_unit_id == OrgUnit.id).where(
|
||||
UserOrg.user_id == user_id
|
||||
)
|
||||
return (await self.db.execute(query)).scalars().all()
|
||||
|
||||
async def get_many_ssp_ids(self, user_id: int) -> list[int]:
|
||||
query = select(UserOrg.ssp_id).where(UserOrg.user_id == user_id)
|
||||
query = select(UserOrg.org_unit_id).where(UserOrg.user_id == user_id)
|
||||
return (await self.db.execute(query)).scalars().all()
|
||||
|
||||
async def set_many_ssp(self, user_id: int, ssp_ids: list[int]) -> bool:
|
||||
try:
|
||||
for ssp_id in ssp_ids:
|
||||
link = UserOrg(user_id=user_id, ssp_id=ssp_id)
|
||||
self.db.add(link)
|
||||
await self.db.execute(
|
||||
text(
|
||||
"SELECT v3.grant_user_org_access(:user_id, :org_unit_id)"
|
||||
),
|
||||
{"user_id": user_id, "org_unit_id": ssp_id},
|
||||
)
|
||||
await self.db.commit()
|
||||
return True
|
||||
except Exception:
|
||||
await self.db.rollback()
|
||||
return False
|
||||
|
||||
async def unset_many_ssp(self, user_id: int, ssp_ids: list[int]) -> bool:
|
||||
try:
|
||||
for ssp_id in ssp_ids:
|
||||
query = delete(UserOrg).where(
|
||||
(UserOrg.user_id == user_id) & (UserOrg.ssp_id == ssp_id)
|
||||
await self.db.execute(
|
||||
text(
|
||||
"SELECT v3.revoke_user_org_access(:user_id, :org_unit_id)"
|
||||
),
|
||||
{"user_id": user_id, "org_unit_id": ssp_id},
|
||||
)
|
||||
await self.db.execute(query)
|
||||
await self.db.commit()
|
||||
return True
|
||||
except Exception:
|
||||
await self.db.rollback()
|
||||
return False
|
||||
|
||||
async def clear_many_ssp(self, user_id: int) -> bool:
|
||||
try:
|
||||
query = delete(UserOrg).where(UserOrg.user_id == user_id)
|
||||
await self.db.execute(query)
|
||||
async def clear_many_ssp(self, user_id: int) -> None:
|
||||
org_unit_ids = await self.get_many_ssp_ids(user_id)
|
||||
await self.db.execute(
|
||||
text(
|
||||
"SELECT v3.revoke_many_user_org_access(:user_id, :org_unit_ids)"
|
||||
),
|
||||
{"user_id": user_id, "org_unit_ids": org_unit_ids},
|
||||
)
|
||||
await self.db.commit()
|
||||
return True
|
||||
except Exception:
|
||||
await self.db.rollback()
|
||||
return False
|
||||
|
||||
async def create(self, user_data: dict) -> AppUser:
|
||||
hashed_password = (
|
||||
get_password_hash(user_data["password"]) if user_data.get("password") else None
|
||||
)
|
||||
user_id = (
|
||||
await self.db.execute(
|
||||
text(
|
||||
"""
|
||||
SELECT (v3.add_user(
|
||||
:email,
|
||||
:username,
|
||||
:hashed_password,
|
||||
:full_name,
|
||||
:role_id
|
||||
)).id
|
||||
"""
|
||||
),
|
||||
{
|
||||
"email": user_data["email"],
|
||||
"username": user_data["username"],
|
||||
"hashed_password": hashed_password,
|
||||
"full_name": user_data.get("full_name"),
|
||||
"role_id": user_data.get("role_id"),
|
||||
},
|
||||
)
|
||||
).scalar_one()
|
||||
await self.db.commit()
|
||||
return await self.get(user_id)
|
||||
|
||||
async def update(self, user_id: int, user_data: dict) -> Optional[AppUser]:
|
||||
user = await self.get(user_id)
|
||||
if not user:
|
||||
return None
|
||||
|
||||
hashed_password = None
|
||||
if "password" in user_data and user_data["password"]:
|
||||
hashed_password = get_password_hash(user_data["password"])
|
||||
|
||||
await self.db.execute(
|
||||
text(
|
||||
"""
|
||||
SELECT (v3.upd_user(
|
||||
:user_id,
|
||||
:email,
|
||||
:username,
|
||||
:hashed_password,
|
||||
:full_name,
|
||||
:role_id,
|
||||
:is_active
|
||||
)).id
|
||||
"""
|
||||
),
|
||||
{
|
||||
"user_id": user_id,
|
||||
"email": user_data.get("email"),
|
||||
"username": user_data.get("username"),
|
||||
"hashed_password": hashed_password,
|
||||
"full_name": user_data.get("full_name"),
|
||||
"role_id": user_data.get("role_id"),
|
||||
"is_active": user_data.get("is_active"),
|
||||
},
|
||||
)
|
||||
await self.db.commit()
|
||||
return await self.get(user_id)
|
||||
|
||||
276
api/tests/integration/test_audit_events_integration.py
Normal file
276
api/tests/integration/test_audit_events_integration.py
Normal file
@ -0,0 +1,276 @@
|
||||
#
|
||||
# закоментированы для избежания переполнения бд
|
||||
#
|
||||
|
||||
|
||||
# from datetime import datetime, timedelta
|
||||
# import uuid
|
||||
|
||||
# import pytest
|
||||
|
||||
|
||||
# def _find_event(
|
||||
# client,
|
||||
# admin_tokens: dict,
|
||||
# auth_headers,
|
||||
# event_type: str,
|
||||
# predicate,
|
||||
# limit: int = 100,
|
||||
# ):
|
||||
# response = client.get(
|
||||
# f"/api/v1/audit-logs?event_type={event_type}&limit={limit}",
|
||||
# headers=auth_headers(admin_tokens),
|
||||
# )
|
||||
# assert response.status_code == 200
|
||||
# result = response.json().get("result", [])
|
||||
# for item in result:
|
||||
# payload = item.get("payload_json") or {}
|
||||
# if predicate(payload):
|
||||
# return item
|
||||
# return None
|
||||
|
||||
|
||||
# def _event_count(client, admin_tokens: dict, auth_headers, event_type: str) -> int:
|
||||
# response = client.get(
|
||||
# f"/api/v1/audit-logs?event_type={event_type}&limit=1",
|
||||
# headers=auth_headers(admin_tokens),
|
||||
# )
|
||||
# assert response.status_code == 200
|
||||
# return int(response.json().get("count") or 0)
|
||||
|
||||
|
||||
# def _pick_target_user_id(client, admin_tokens: dict, auth_headers) -> int:
|
||||
# me_resp = client.get(
|
||||
# "/api/v1/users/me",
|
||||
# headers=auth_headers(admin_tokens),
|
||||
# )
|
||||
# assert me_resp.status_code == 200
|
||||
# me = me_resp.json() or {}
|
||||
# assert me.get("id") is not None
|
||||
# return int(me["id"])
|
||||
|
||||
|
||||
# def _pick_org_unit_id(client, admin_tokens: dict, auth_headers, user_id: int) -> int:
|
||||
# forms_resp = client.get(
|
||||
# "/api/v1/form/?limit=100",
|
||||
# headers=auth_headers(admin_tokens),
|
||||
# )
|
||||
# assert forms_resp.status_code == 200
|
||||
# forms = forms_resp.json().get("result") or []
|
||||
# candidate_ids = [f.get("org_unit_id") for f in forms if f.get("org_unit_id") is not None]
|
||||
# assert candidate_ids
|
||||
|
||||
# assigned_resp = client.get(
|
||||
# f"/api/v1/users/dfip-many-ssp/?user_id={user_id}",
|
||||
# headers=auth_headers(admin_tokens),
|
||||
# )
|
||||
# assert assigned_resp.status_code == 200
|
||||
# assigned = {
|
||||
# int(item["id"])
|
||||
# for item in (assigned_resp.json().get("result") or {}).get("ssp", [])
|
||||
# if item.get("id") is not None
|
||||
# }
|
||||
# for org_unit_id in candidate_ids:
|
||||
# if int(org_unit_id) not in assigned:
|
||||
# return int(org_unit_id)
|
||||
# raise AssertionError("Не найден org_unit_id, который можно прикрепить пользователю")
|
||||
|
||||
|
||||
# def _create_temp_user(client, admin_tokens: dict, auth_headers, role_id: int = 2) -> int:
|
||||
# suffix = uuid.uuid4().hex[:8]
|
||||
# username = f"audit_u_{suffix}"
|
||||
# response = client.put(
|
||||
# "/api/v1/users/",
|
||||
# json={
|
||||
# "email": f"{username}@example.com",
|
||||
# "username": username,
|
||||
# "password": "pass123",
|
||||
# "full_name": "Audit User",
|
||||
# "role_id": role_id,
|
||||
# },
|
||||
# headers=auth_headers(admin_tokens),
|
||||
# )
|
||||
# assert response.status_code == 200, response.text
|
||||
# result = response.json().get("result") or {}
|
||||
# return int(result["id"])
|
||||
|
||||
|
||||
# def _create_project(client, admin_tokens: dict, auth_headers) -> int:
|
||||
# response = client.post(
|
||||
# "/api/v1/projects",
|
||||
# json={"name": f"PT_{uuid.uuid4().hex[:10]}", "year": 2026, "branch_id": 1},
|
||||
# headers=auth_headers(admin_tokens),
|
||||
# )
|
||||
# assert response.status_code == 200, response.text
|
||||
# return int(response.json()["project_id"])
|
||||
|
||||
|
||||
# def _add_project_line(client, admin_tokens: dict, auth_headers, project_id: int) -> int:
|
||||
# for expense_item_id in range(1, 80):
|
||||
# response = client.post(
|
||||
# f"/api/v1/projects/{project_id}/report/2026/LIMIT/line",
|
||||
# json={"expense_item_id": expense_item_id},
|
||||
# headers=auth_headers(admin_tokens),
|
||||
# )
|
||||
# if response.status_code != 200:
|
||||
# continue
|
||||
# rows = response.json()
|
||||
# input_row = next(
|
||||
# (
|
||||
# row
|
||||
# for row in rows
|
||||
# if row.get("row_type") == "INPUT" and row.get("data", {}).get("line_id")
|
||||
# ),
|
||||
# None,
|
||||
# )
|
||||
# if input_row:
|
||||
# return int(input_row["data"]["line_id"])
|
||||
# raise AssertionError("Не удалось добавить строку в проектный отчёт для ROW_* тестов")
|
||||
|
||||
|
||||
# def _create_stage(client, admin_tokens, auth_headers, form_id: int, sheet: str) -> dict:
|
||||
# phase_code = f"autotest_{uuid.uuid4().hex[:8]}"
|
||||
# opens_at = datetime.utcnow().replace(microsecond=0)
|
||||
# closes_at = opens_at + timedelta(days=2)
|
||||
# response = client.post(
|
||||
# f"/api/v1/stages/form/{form_id}",
|
||||
# json={
|
||||
# "sheet": sheet,
|
||||
# "phase_code": phase_code,
|
||||
# "role": "DFIP",
|
||||
# "column_keys": ["plan.q1"],
|
||||
# "opens_at": opens_at.isoformat() + "Z",
|
||||
# "closes_at": closes_at.isoformat() + "Z",
|
||||
# },
|
||||
# headers=auth_headers(admin_tokens),
|
||||
# )
|
||||
# assert response.status_code == 200, response.text
|
||||
# result = response.json().get("result") or {}
|
||||
# assert result.get("phase_code") == phase_code
|
||||
# return result
|
||||
|
||||
|
||||
# def test_audit_user_access_granted_event(client, admin_tokens, auth_headers):
|
||||
# user_id = _pick_target_user_id(client, admin_tokens, auth_headers)
|
||||
# org_unit_id = _pick_org_unit_id(client, admin_tokens, auth_headers, user_id)
|
||||
|
||||
# response = client.put(
|
||||
# f"/api/v1/users/dfip-many-ssp/?user_id={user_id}",
|
||||
# json={"ssp_ids": [org_unit_id]},
|
||||
# headers=auth_headers(admin_tokens),
|
||||
# )
|
||||
# assert response.status_code == 200
|
||||
# assert response.json().get("success") is True
|
||||
|
||||
# event = _find_event(
|
||||
# client,
|
||||
# admin_tokens,
|
||||
# auth_headers,
|
||||
# "USER_ACCESS_GRANTED",
|
||||
# lambda payload: int(payload.get("user_id", -1)) == user_id
|
||||
# and int(payload.get("org_unit_id", -1)) == org_unit_id,
|
||||
# )
|
||||
# assert event is not None
|
||||
|
||||
|
||||
# def test_audit_user_access_revoked_event(client, admin_tokens, auth_headers):
|
||||
# user_id = _pick_target_user_id(client, admin_tokens, auth_headers)
|
||||
# assigned_resp = client.get(
|
||||
# f"/api/v1/users/dfip-many-ssp/?user_id={user_id}",
|
||||
# headers=auth_headers(admin_tokens),
|
||||
# )
|
||||
# assert assigned_resp.status_code == 200
|
||||
# assigned = (assigned_resp.json().get("result") or {}).get("ssp", [])
|
||||
|
||||
# if assigned:
|
||||
# org_unit_id = int(assigned[0]["id"])
|
||||
# else:
|
||||
# org_unit_id = _pick_org_unit_id(client, admin_tokens, auth_headers, user_id)
|
||||
# attach_resp = client.put(
|
||||
# f"/api/v1/users/dfip-many-ssp/?user_id={user_id}",
|
||||
# json={"ssp_ids": [org_unit_id]},
|
||||
# headers=auth_headers(admin_tokens),
|
||||
# )
|
||||
# assert attach_resp.status_code == 200
|
||||
# assert attach_resp.json().get("success") is True
|
||||
|
||||
# response = client.request(
|
||||
# "DELETE",
|
||||
# f"/api/v1/users/dfip-many-ssp/?user_id={user_id}",
|
||||
# json={"ssp_ids": [org_unit_id]},
|
||||
# headers=auth_headers(admin_tokens),
|
||||
# )
|
||||
# assert response.status_code == 200
|
||||
# assert response.json().get("success") is True
|
||||
|
||||
# event = _find_event(
|
||||
# client,
|
||||
# admin_tokens,
|
||||
# auth_headers,
|
||||
# "USER_ACCESS_REVOKED",
|
||||
# lambda payload: int(payload.get("user_id", -1)) == user_id
|
||||
# and int(payload.get("org_unit_id", -1)) == org_unit_id,
|
||||
# )
|
||||
# assert event is not None
|
||||
|
||||
|
||||
# def test_audit_user_create_event(client, admin_tokens, auth_headers):
|
||||
# before = _event_count(client, admin_tokens, auth_headers, "USER_CREATE")
|
||||
# user_id = _create_temp_user(client, admin_tokens, auth_headers, role_id=2)
|
||||
# after = _event_count(client, admin_tokens, auth_headers, "USER_CREATE")
|
||||
# assert after > before
|
||||
|
||||
# delete_resp = client.delete(
|
||||
# f"/api/v1/users/{user_id}",
|
||||
# headers=auth_headers(admin_tokens),
|
||||
# )
|
||||
# assert delete_resp.status_code == 200
|
||||
|
||||
|
||||
# def test_audit_user_role_change_event(client, admin_tokens, auth_headers):
|
||||
# user_id = _create_temp_user(client, admin_tokens, auth_headers, role_id=2)
|
||||
# before = _event_count(client, admin_tokens, auth_headers, "USER_ROLE_CHANGE")
|
||||
|
||||
# update_resp = client.patch(
|
||||
# f"/api/v1/users/{user_id}",
|
||||
# json={"role_id": 3},
|
||||
# headers=auth_headers(admin_tokens),
|
||||
# )
|
||||
# assert update_resp.status_code == 200, update_resp.text
|
||||
# after = _event_count(client, admin_tokens, auth_headers, "USER_ROLE_CHANGE")
|
||||
# assert after > before
|
||||
|
||||
# delete_resp = client.delete(
|
||||
# f"/api/v1/users/{user_id}",
|
||||
# headers=auth_headers(admin_tokens),
|
||||
# )
|
||||
# assert delete_resp.status_code == 200
|
||||
|
||||
|
||||
# def test_audit_project_create_event(client, admin_tokens, auth_headers):
|
||||
# before = _event_count(client, admin_tokens, auth_headers, "PROJECT_CREATE")
|
||||
# _create_project(client, admin_tokens, auth_headers)
|
||||
# after = _event_count(client, admin_tokens, auth_headers, "PROJECT_CREATE")
|
||||
# assert after > before
|
||||
|
||||
|
||||
# def test_audit_row_create_event(client, admin_tokens, auth_headers):
|
||||
# project_id = _create_project(client, admin_tokens, auth_headers)
|
||||
# before = _event_count(client, admin_tokens, auth_headers, "ROW_CREATE")
|
||||
# _add_project_line(client, admin_tokens, auth_headers, project_id)
|
||||
# after = _event_count(client, admin_tokens, auth_headers, "ROW_CREATE")
|
||||
# assert after > before
|
||||
|
||||
|
||||
# def test_audit_row_delete_event(client, admin_tokens, auth_headers):
|
||||
# project_id = _create_project(client, admin_tokens, auth_headers)
|
||||
# line_id = _add_project_line(client, admin_tokens, auth_headers, project_id)
|
||||
# before = _event_count(client, admin_tokens, auth_headers, "ROW_DELETE")
|
||||
|
||||
# delete_resp = client.delete(
|
||||
# f"/api/v1/projects/{project_id}/report/2026/LIMIT/line/{line_id}",
|
||||
# headers=auth_headers(admin_tokens),
|
||||
# )
|
||||
# assert delete_resp.status_code == 200, delete_resp.text
|
||||
# after = _event_count(client, admin_tokens, auth_headers, "ROW_DELETE")
|
||||
# assert after > before
|
||||
@ -16,6 +16,12 @@ def _result_with_scalars(items):
|
||||
return result
|
||||
|
||||
|
||||
def _result_with_scalar_one(value):
|
||||
result = MagicMock()
|
||||
result.scalar_one.return_value = value
|
||||
return result
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_db():
|
||||
return AsyncMock(spec=AsyncSession)
|
||||
@ -49,9 +55,12 @@ async def test_get_returns_none_for_missing_id(repository, mock_db):
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_all_applies_filters(repository, mock_db):
|
||||
row = MagicMock(id=11, user_id=2, org_unit_id=7, event_type="UPDATE")
|
||||
mock_db.execute.return_value = _result_with_scalars([row])
|
||||
mock_db.execute.side_effect = [
|
||||
_result_with_scalars([row]),
|
||||
_result_with_scalar_one(1),
|
||||
]
|
||||
|
||||
result = await repository.get_all(
|
||||
result, count = await repository.get_all(
|
||||
limit=10,
|
||||
offset=5,
|
||||
user_id=2,
|
||||
@ -64,5 +73,6 @@ async def test_get_all_applies_filters(repository, mock_db):
|
||||
)
|
||||
|
||||
assert result == [row]
|
||||
mock_db.execute.assert_awaited_once()
|
||||
assert count == 1
|
||||
assert mock_db.execute.await_count == 2
|
||||
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user