311 lines
11 KiB
Python
311 lines
11 KiB
Python
import uuid
|
||
|
||
import pytest
|
||
|
||
|
||
def _auth_headers(tokens: dict) -> dict:
|
||
return {"Authorization": f"Bearer {tokens['access_token']}"}
|
||
|
||
|
||
def _pick_form_and_sheet(client, admin_tokens) -> tuple[int, str, str | None]:
|
||
response = client.get("/api/v1/form/", headers=_auth_headers(admin_tokens))
|
||
assert response.status_code == 200
|
||
forms = response.json().get("result") or []
|
||
|
||
for form in forms:
|
||
form_id = form.get("id")
|
||
form_type = form.get("form_type_code")
|
||
if not form_id or not form_type:
|
||
continue
|
||
|
||
sheets_response = client.get(
|
||
f"/api/v1/form/{form_id}/sheets",
|
||
headers=_auth_headers(admin_tokens),
|
||
)
|
||
if sheets_response.status_code != 200:
|
||
continue
|
||
|
||
sheets = (sheets_response.json().get("result") or {}).get("sheets") or []
|
||
if "AHR" not in sheets:
|
||
continue
|
||
|
||
direction = "Support" if form_type == "FORM_1" else None
|
||
return int(form_id), "AHR", direction
|
||
|
||
pytest.skip("В тестовой БД не найдена форма с листом AHR")
|
||
|
||
|
||
def _pick_form_with_sheet(
|
||
client, admin_tokens, target_sheet: str, form_type_code: str | None = None
|
||
) -> tuple[int, str]:
|
||
response = client.get("/api/v1/form/", headers=_auth_headers(admin_tokens))
|
||
assert response.status_code == 200
|
||
forms = response.json().get("result") or []
|
||
for form in forms:
|
||
form_id = form.get("id")
|
||
if not form_id:
|
||
continue
|
||
if form_type_code and form.get("form_type_code") != form_type_code:
|
||
continue
|
||
sheets_response = client.get(
|
||
f"/api/v1/form/{form_id}/sheets",
|
||
headers=_auth_headers(admin_tokens),
|
||
)
|
||
if sheets_response.status_code != 200:
|
||
continue
|
||
sheets = (sheets_response.json().get("result") or {}).get("sheets") or []
|
||
if target_sheet in sheets:
|
||
return int(form_id), target_sheet
|
||
pytest.skip(f"Не найдена форма с листом {target_sheet}")
|
||
|
||
|
||
def _pick_input_line_id(client, admin_tokens, form_id: int, sheet: str, direction: str | None) -> int:
|
||
url = f"/api/v1/form/{form_id}/sheet/{sheet}"
|
||
params = []
|
||
if direction:
|
||
params.append(f"direction={direction}")
|
||
if params:
|
||
url = f"{url}?{'&'.join(params)}"
|
||
|
||
response = client.get(url, headers=_auth_headers(admin_tokens))
|
||
assert response.status_code == 200
|
||
rows = response.json().get("result") or []
|
||
for row in rows:
|
||
data = row.get("data") or {}
|
||
if row.get("row_type") == "INPUT" and data.get("line_id"):
|
||
return int(data["line_id"])
|
||
|
||
pytest.skip("Не найден INPUT line_id для проверки upd_form_cell")
|
||
|
||
|
||
def test_backend_calls_v_form_view_via_sheet_endpoint(client, admin_tokens):
|
||
form_id, sheet, direction = _pick_form_and_sheet(client, admin_tokens)
|
||
url = f"/api/v1/form/{form_id}/sheet/{sheet}"
|
||
if direction:
|
||
url = f"{url}?direction={direction}"
|
||
|
||
response = client.get(url, headers=_auth_headers(admin_tokens))
|
||
assert response.status_code == 200
|
||
payload = response.json()
|
||
assert isinstance(payload.get("result"), list)
|
||
if payload["result"]:
|
||
first_row = payload["result"][0]
|
||
assert "row_type" in first_row
|
||
assert "data" in first_row
|
||
|
||
|
||
def test_backend_calls_upd_form_cell_and_maps_sql_validation_error(client, admin_tokens):
|
||
form_id, sheet, direction = _pick_form_and_sheet(client, admin_tokens)
|
||
line_id = _pick_input_line_id(client, admin_tokens, form_id, sheet, direction)
|
||
|
||
url = f"/api/v1/form/{form_id}/sheet/{sheet}/cell"
|
||
if direction:
|
||
url = f"{url}?direction={direction}"
|
||
|
||
response = client.patch(
|
||
url,
|
||
json={"line_id": line_id, "column": "bad.scope", "value": 1},
|
||
headers=_auth_headers(admin_tokens),
|
||
)
|
||
|
||
# SQL-функция отдает prefixed validation error.
|
||
assert response.status_code in (400, 422)
|
||
payload = response.json()
|
||
assert isinstance(payload, dict)
|
||
assert "message" in payload
|
||
assert isinstance(payload["message"], str)
|
||
|
||
|
||
def test_backend_calls_form3_add_project_and_add_line_functions(client, admin_tokens):
|
||
create = client.post(
|
||
"/api/v1/projects",
|
||
json={"name": f"ITEST_DB_FUNC_{uuid.uuid4().hex[:6]}", "year": 2026, "branch_id": 1},
|
||
headers=_auth_headers(admin_tokens),
|
||
)
|
||
assert create.status_code == 200
|
||
project_payload = create.json()
|
||
assert "project_id" in project_payload
|
||
assert "limit_report_id" in project_payload
|
||
assert "current_expenses_report_id" in project_payload
|
||
|
||
project_id = int(project_payload["project_id"])
|
||
add_line = client.post(
|
||
f"/api/v1/projects/{project_id}/report/2026/LIMIT/line",
|
||
json={"expense_item_id": 1},
|
||
headers=_auth_headers(admin_tokens),
|
||
)
|
||
assert add_line.status_code == 200
|
||
rows = add_line.json()
|
||
assert isinstance(rows, list)
|
||
|
||
|
||
def test_forms_validation_invalid_sections_returns_400(client, admin_tokens):
|
||
form_id, sheet = _pick_form_with_sheet(client, admin_tokens, "AHR")
|
||
response = client.get(
|
||
f"/api/v1/form/{form_id}/sheet/{sheet}?sections=bad_section",
|
||
headers=_auth_headers(admin_tokens),
|
||
)
|
||
assert response.status_code == 400
|
||
payload = response.json()
|
||
assert isinstance(payload, dict)
|
||
assert "message" in payload
|
||
|
||
|
||
def test_forms_validation_empty_sections_csv_is_ignored(client, admin_tokens):
|
||
form_id, sheet, direction = _pick_form_and_sheet(client, admin_tokens)
|
||
url = f"/api/v1/form/{form_id}/sheet/{sheet}?sections= , , "
|
||
if direction:
|
||
url += f"&direction={direction}"
|
||
response = client.get(url, headers=_auth_headers(admin_tokens))
|
||
assert response.status_code == 200
|
||
payload = response.json()
|
||
assert isinstance(payload.get("result"), list)
|
||
|
||
|
||
def test_forms_validation_direction_required_for_form1_ahr(client, admin_tokens):
|
||
form_id, sheet = _pick_form_with_sheet(client, admin_tokens, "AHR", form_type_code="FORM_1")
|
||
response = client.get(
|
||
f"/api/v1/form/{form_id}/sheet/{sheet}",
|
||
headers=_auth_headers(admin_tokens),
|
||
)
|
||
assert response.status_code == 400
|
||
payload = response.json()
|
||
assert isinstance(payload, dict)
|
||
assert "message" in payload
|
||
assert "direction" in payload["message"].lower()
|
||
|
||
|
||
def test_forms_validation_direction_case_sensitive(client, admin_tokens):
|
||
form_id, sheet = _pick_form_with_sheet(client, admin_tokens, "AHR", form_type_code="FORM_1")
|
||
response = client.get(
|
||
f"/api/v1/form/{form_id}/sheet/{sheet}?direction=support",
|
||
headers=_auth_headers(admin_tokens),
|
||
)
|
||
assert response.status_code == 422
|
||
|
||
|
||
def test_forms_validation_direction_forbidden_on_non_form1(client, admin_tokens):
|
||
form_id, sheet = _pick_form_with_sheet(client, admin_tokens, "AHR", form_type_code="FORM_2")
|
||
response = client.get(
|
||
f"/api/v1/form/{form_id}/sheet/{sheet}?direction=Support",
|
||
headers=_auth_headers(admin_tokens),
|
||
)
|
||
assert response.status_code == 400
|
||
payload = response.json()
|
||
assert isinstance(payload, dict)
|
||
assert "message" in payload
|
||
assert "direction" in payload["message"].lower()
|
||
|
||
|
||
def test_form3_validation_invalid_sections_returns_400(client, admin_tokens):
|
||
create = client.post(
|
||
"/api/v1/projects",
|
||
json={"name": f"ITEST_F3_SECT_{uuid.uuid4().hex[:6]}", "year": 2026, "branch_id": 1},
|
||
headers=_auth_headers(admin_tokens),
|
||
)
|
||
assert create.status_code == 200
|
||
project_id = int(create.json()["project_id"])
|
||
response = client.get(
|
||
f"/api/v1/projects/{project_id}/report/2026/LIMIT?sections=q1,q5",
|
||
headers=_auth_headers(admin_tokens),
|
||
)
|
||
assert response.status_code == 400
|
||
payload = response.json()
|
||
assert isinstance(payload, dict)
|
||
assert "message" in payload
|
||
|
||
|
||
def test_form3_validation_empty_sections_csv_is_ignored(client, admin_tokens):
|
||
create = client.post(
|
||
"/api/v1/projects",
|
||
json={"name": f"ITEST_F3_EMPTY_{uuid.uuid4().hex[:6]}", "year": 2026, "branch_id": 1},
|
||
headers=_auth_headers(admin_tokens),
|
||
)
|
||
assert create.status_code == 200
|
||
project_id = int(create.json()["project_id"])
|
||
response = client.get(
|
||
f"/api/v1/projects/{project_id}/report/2026/LIMIT?sections= , , ",
|
||
headers=_auth_headers(admin_tokens),
|
||
)
|
||
assert response.status_code == 200
|
||
assert isinstance(response.json().get("result"), list)
|
||
|
||
|
||
def test_form3_validation_duplicate_sections_is_allowed(client, admin_tokens):
|
||
create = client.post(
|
||
"/api/v1/projects",
|
||
json={"name": f"ITEST_F3_DUP_{uuid.uuid4().hex[:6]}", "year": 2026, "branch_id": 1},
|
||
headers=_auth_headers(admin_tokens),
|
||
)
|
||
assert create.status_code == 200
|
||
project_id = int(create.json()["project_id"])
|
||
response = client.get(
|
||
f"/api/v1/projects/{project_id}/report/2026/LIMIT?sections=q1,q1",
|
||
headers=_auth_headers(admin_tokens),
|
||
)
|
||
assert response.status_code == 200
|
||
assert isinstance(response.json().get("result"), list)
|
||
|
||
|
||
def test_form3_add_project_invalid_name_returns_422(client, admin_tokens):
|
||
# Пробел запрещён regex-правилом AddProjectBody.
|
||
response = client.post(
|
||
"/api/v1/projects",
|
||
json={"name": "INVALID NAME", "year": 2026, "branch_id": 1},
|
||
headers=_auth_headers(admin_tokens),
|
||
)
|
||
assert response.status_code == 422
|
||
|
||
|
||
def test_health_and_ready_endpoints(client):
|
||
healthz = client.get("/healthz")
|
||
assert healthz.status_code == 200
|
||
health_payload = healthz.json()
|
||
assert health_payload.get("status") == "ok"
|
||
|
||
readyz = client.get("/readyz")
|
||
assert readyz.status_code in (200, 503)
|
||
ready_payload = readyz.json()
|
||
assert "status" in ready_payload
|
||
assert "db_status" in ready_payload
|
||
assert "mv_expense_item_tree" in ready_payload
|
||
|
||
|
||
def test_admin_refresh_tree_requires_admin_role(client, admin_tokens):
|
||
admin_resp = client.post(
|
||
"/api/v1/admin/refresh-tree",
|
||
headers=_auth_headers(admin_tokens),
|
||
)
|
||
assert admin_resp.status_code in (200, 503)
|
||
|
||
uname = f"itest_non_admin_{uuid.uuid4().hex[:6]}"
|
||
create_user = client.put(
|
||
"/api/v1/users/",
|
||
json={
|
||
"email": f"{uname}@example.com",
|
||
"username": uname,
|
||
"password": "pass123",
|
||
"full_name": "Integration User",
|
||
"role_id": 2,
|
||
},
|
||
headers=_auth_headers(admin_tokens),
|
||
)
|
||
if create_user.status_code != 200:
|
||
pytest.skip("Не удалось создать non-admin пользователя в текущей БД")
|
||
|
||
login_resp = client.post("/api/v1/auth/login", json={"username": uname, "password": "pass123"})
|
||
if login_resp.status_code != 200:
|
||
pytest.skip("Не удалось залогинить non-admin пользователя в текущей БД")
|
||
user_tokens = login_resp.json()
|
||
|
||
non_admin_resp = client.post(
|
||
"/api/v1/admin/refresh-tree",
|
||
headers=_auth_headers(user_tokens),
|
||
)
|
||
assert non_admin_resp.status_code == 403
|
||
|
||
|
||
def test_admin_refresh_tree_requires_auth(client):
|
||
response = client.post("/api/v1/admin/refresh-tree")
|
||
assert response.status_code in (401, 403)
|