277 lines
9.9 KiB
Python
277 lines
9.9 KiB
Python
#
|
||
# закоментированы для избежания переполнения бд
|
||
#
|
||
|
||
|
||
# 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
|