diff --git a/api/src/db/session.py b/api/src/db/session.py index eb04b72..83da0bc 100644 --- a/api/src/db/session.py +++ b/api/src/db/session.py @@ -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() diff --git a/api/src/repository/form_phase_repository.py b/api/src/repository/form_phase_repository.py index ba6558c..046b13c 100644 --- a/api/src/repository/form_phase_repository.py +++ b/api/src/repository/form_phase_repository.py @@ -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) diff --git a/api/src/repository/user_repository.py b/api/src/repository/user_repository.py index af89cdc..733ef56 100644 --- a/api/src/repository/user_repository.py +++ b/api/src/repository/user_repository.py @@ -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.commit() - return True - except Exception: - await self.db.rollback() - return False + for ssp_id in ssp_ids: + 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 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(query) - await self.db.commit() - return True - except Exception: - await self.db.rollback() - return False + for ssp_id in ssp_ids: + 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.commit() + return True - 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) - await self.db.commit() - return True - except Exception: - await self.db.rollback() - return False + 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() + + 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) diff --git a/api/tests/integration/test_audit_events_integration.py b/api/tests/integration/test_audit_events_integration.py new file mode 100644 index 0000000..e806cd4 --- /dev/null +++ b/api/tests/integration/test_audit_events_integration.py @@ -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 diff --git a/api/tests/unit/test_auditlog_repository.py b/api/tests/unit/test_auditlog_repository.py index 1f4c8c8..0954440 100644 --- a/api/tests/unit/test_auditlog_repository.py +++ b/api/tests/unit/test_auditlog_repository.py @@ -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