From d8e260cc5baac619be31e3a83c80f95818fe907c Mon Sep 17 00:00:00 2001 From: Raykov-MS Date: Thu, 21 May 2026 17:24:45 +0300 Subject: [PATCH] add tests --- .../test_audit_events_integration.py | 276 ++++++++++++++++++ 1 file changed, 276 insertions(+) create mode 100644 api/tests/integration/test_audit_events_integration.py 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