Andrey Epifancev 39ba9763a3 Merge origin/test в feat/330-aurora-1883-closed-vsp
Конфликты после #323/#331/#334: импорты в Cell.jsx и тесты в конце
test_vsp.py — оставлены обе стороны. Дерево совпадает с пробным
слиянием всех PR (backend 1041 passed / 10 известных, фронт 263/263).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-11-12 18:04:24 +04:00

576 lines
20 KiB
Python

import uuid
import pytest
def _unique_reg_number() -> str:
return f"TEST_{uuid.uuid4().hex[:10]}"
def test_vsp_list_smoke(client, admin_tokens, auth_headers):
response = client.get("/api/v1/dict/info", headers=auth_headers(admin_tokens))
assert response.status_code == 200
payload = response.json()
assert "result" in payload
assert "count" in payload
assert isinstance(payload["result"], list)
def test_vsp_list_with_filters(client, admin_tokens, auth_headers):
response = client.get(
"/api/v1/dict/info?ssp_id=1",
headers=auth_headers(admin_tokens),
)
assert response.status_code == 200
payload = response.json()
assert isinstance(payload["result"], list)
def test_vsp_dropdown_smoke(client, admin_tokens, auth_headers):
response = client.get(
"/api/v1/dict/info/dropdown", headers=auth_headers(admin_tokens)
)
assert response.status_code == 200
payload = response.json()
assert "result" in payload
assert "count" in payload
assert isinstance(payload["result"], list)
@pytest.mark.parametrize(
"params, count",
[
([("ssp_id", 2)], 2),
([("form_id", 1)], 0),
([("form_id", 2)], 2),
([("ssp_id", 2), ("form_id", 1)], 2),
]
)
def test_vsp_dropdown_with_filter(client, admin_tokens, auth_headers, params, count):
filters = "&".join([f"{el[0]}={el[1]}" for el in params])
response = client.get(
f"/api/v1/dict/info/dropdown?{filters}",
headers=auth_headers(admin_tokens),
)
assert response.status_code == 200
payload = response.json()
assert isinstance(payload["result"], list)
assert len(payload["result"]) == count
def test_vsp_get_by_id(client, isp_tokens, auth_headers):
response = client.get("/api/v1/dict/info/1", headers=auth_headers(isp_tokens))
assert response.status_code == 200
payload = response.json()
assert "result" in payload
assert isinstance(payload["result"], dict)
assert payload["result"]["id"] == 1
def test_vsp_get_not_found(client, admin_tokens, auth_headers):
response = client.get(
"/api/v1/dict/info/99999", headers=auth_headers(admin_tokens)
)
assert response.status_code == 404
payload = response.json()
assert isinstance(payload, dict)
assert "message" in payload
def test_vsp_create_smoke(client, admin_tokens, auth_headers):
reg_number = _unique_reg_number()
response = client.post(
"/api/v1/dict/info",
json={
"registration_number": reg_number,
"address": "Test Address",
"ssp_id": 1,
},
headers=auth_headers(admin_tokens),
)
assert response.status_code == 201
payload = response.json()
assert payload["success"] is True
assert payload["result"]["registration_number"] == reg_number
assert payload["result"]["address"] == "Test Address"
assert payload["result"]["ssp_id"] == 1
def test_vsp_create_with_all_fields(client, admin_tokens, auth_headers):
reg_number = _unique_reg_number()
response = client.post(
"/api/v1/dict/info",
json={
"registration_number": reg_number,
"address": "Full Address",
"ssp_id": 1,
"system_code": "SYS001",
"vsp_type": "Тип 1",
"open_date": "2024-01-15",
"close_date": "2025-12-31",
"approved_format": "Стандарт",
"notes": "Тестовое примечание",
"placement_type": "Аренда",
"staff_count": 10,
"area": 150.5,
"rent_contract_num": "Д-123/24",
"rent_end_date": "2025-12-31",
},
headers=auth_headers(admin_tokens),
)
assert response.status_code == 201
payload = response.json()
assert payload["success"] is True
assert payload["result"]["registration_number"] == reg_number
assert payload["result"]["system_code"] == "SYS001"
assert payload["result"]["vsp_type"] == "Тип 1"
assert payload["result"]["open_date"] == "2024-01-15"
assert payload["result"]["close_date"] == "2025-12-31"
assert payload["result"]["approved_format"] == "Стандарт"
assert payload["result"]["placement_type"] == "Аренда"
assert payload["result"]["staff_count"] == 10
assert payload["result"]["area"] == 150.5
assert payload["result"]["rent_contract_num"] == "Д-123/24"
assert payload["result"]["rent_end_date"] == "2025-12-31"
def test_vsp_create_forbidden_for_executor(client, isp_tokens, auth_headers):
response = client.post(
"/api/v1/dict/info",
json={
"registration_number": _unique_reg_number(),
"address": "Should Fail",
"ssp_id": 1,
},
headers=auth_headers(isp_tokens),
)
assert response.status_code == 403
payload = response.json()
assert isinstance(payload, dict)
assert "message" in payload
def test_vsp_create_with_inactive_ssp_fails(client, admin_tokens, auth_headers):
created_org = client.post(
"/api/v1/org-unit/",
json={"title": f"Inactive SSP {uuid.uuid4().hex[:6]}", "is_ssp": True},
headers=auth_headers(admin_tokens),
)
assert created_org.status_code == 200
org_unit_id = created_org.json()["result"]["id"]
deactivate = client.patch(
f"/api/v1/org-unit/{org_unit_id}",
json={"is_active": False},
headers=auth_headers(admin_tokens),
)
assert deactivate.status_code == 200
response = client.post(
"/api/v1/dict/info",
json={
"registration_number": _unique_reg_number(),
"address": "Should Fail",
"ssp_id": org_unit_id,
},
headers=auth_headers(admin_tokens),
)
assert response.status_code == 400
payload = response.json()
assert "message" in payload
def test_vsp_update_smoke(client, admin_tokens, auth_headers):
reg_number = _unique_reg_number()
created = client.post(
"/api/v1/dict/info",
json={
"registration_number": reg_number,
"address": "Original Address",
"ssp_id": 1,
},
headers=auth_headers(admin_tokens),
)
assert created.status_code == 201
vsp_id = created.json()["result"]["id"]
new_address = f"Updated Address {uuid.uuid4().hex[:4]}"
response = client.put(
f"/api/v1/dict/info/{vsp_id}",
json={"address": new_address},
headers=auth_headers(admin_tokens),
)
assert response.status_code == 200
payload = response.json()
assert payload["success"] is True
assert payload["result"]["address"] == new_address
def test_vsp_update_not_found(client, admin_tokens, auth_headers):
response = client.put(
"/api/v1/dict/info/99999",
json={"address": "No Matter"},
headers=auth_headers(admin_tokens),
)
assert response.status_code == 404
payload = response.json()
assert isinstance(payload, dict)
assert "message" in payload
def test_vsp_delete_smoke(client, admin_tokens, auth_headers):
reg_number = _unique_reg_number()
created = client.post(
"/api/v1/dict/info",
json={
"registration_number": reg_number,
"address": "To Be Deleted",
"ssp_id": 2,
},
headers=auth_headers(admin_tokens),
)
assert created.status_code == 201
vsp_id = created.json()["result"]["id"]
response = client.delete(
f"/api/v1/dict/info/{vsp_id}",
headers=auth_headers(admin_tokens),
)
assert response.status_code == 200
payload = response.json()
assert payload["success"] is True
assert payload["message"] == "Запись INFO удалена"
def test_vsp_delete_with_close_date_smoke(client, admin_tokens, auth_headers):
reg_number = _unique_reg_number()
created = client.post(
"/api/v1/dict/info",
json={
"registration_number": reg_number,
"address": "To Be Deleted with close_date",
"ssp_id": 2,
"open_date": "2026-01-01",
"close_date": "2030-01-01",
},
headers=auth_headers(admin_tokens),
)
assert created.status_code == 201
vsp_id = created.json()["result"]["id"]
response = client.delete(
f"/api/v1/dict/info/{vsp_id}",
headers=auth_headers(admin_tokens),
)
assert response.status_code == 200
payload = response.json()
assert payload["success"] is True
assert payload["message"] == "Запись INFO удалена"
def test_vsp_delete_not_found(client, admin_tokens, auth_headers):
response = client.delete(
"/api/v1/dict/info/99999",
headers=auth_headers(admin_tokens),
)
assert response.status_code == 404
payload = response.json()
assert isinstance(payload, dict)
assert "message" in payload
def test_vsp_delete_forbidden_for_executor(client, isp_tokens, auth_headers):
response = client.delete(
"/api/v1/dict/info/1",
headers=auth_headers(isp_tokens),
)
assert response.status_code == 403
payload = response.json()
assert isinstance(payload, dict)
assert "message" in payload
def test_vsp_requires_auth(client):
response = client.get("/api/v1/dict/info")
assert response.status_code == 403
def _create_vsp(client, admin_tokens, auth_headers, **extra) -> int:
created = client.post(
"/api/v1/dict/info",
json={
"registration_number": _unique_reg_number(),
"address": "AURORA-1885",
"ssp_id": 1,
**extra,
},
headers=auth_headers(admin_tokens),
)
assert created.status_code == 201
return created.json()["result"]["id"]
def test_vsp_get_by_id_admin_active(client, admin_tokens, auth_headers):
vsp_id = _create_vsp(client, admin_tokens, auth_headers)
response = client.get(f"/api/v1/dict/info/{vsp_id}", headers=auth_headers(admin_tokens))
assert response.status_code == 200
assert response.json()["result"]["id"] == vsp_id
assert response.json()["result"]["is_deleted"] is False
def test_vsp_get_by_id_admin_sees_deleted(client, admin_tokens, auth_headers):
vsp_id = _create_vsp(client, admin_tokens, auth_headers)
deleted = client.delete(f"/api/v1/dict/info/{vsp_id}", headers=auth_headers(admin_tokens))
assert deleted.status_code == 200
response = client.get(f"/api/v1/dict/info/{vsp_id}", headers=auth_headers(admin_tokens))
assert response.status_code == 200
assert response.json()["result"]["is_deleted"] is True
def test_vsp_get_by_id_executor_deleted_not_found(client, admin_tokens, isp_tokens, auth_headers):
vsp_id = _create_vsp(client, admin_tokens, auth_headers)
client.delete(f"/api/v1/dict/info/{vsp_id}", headers=auth_headers(admin_tokens))
response = client.get(f"/api/v1/dict/info/{vsp_id}", headers=auth_headers(isp_tokens))
assert response.status_code == 404
def test_vsp_update_past_close_date_deactivates(client, admin_tokens, auth_headers):
from datetime import date, timedelta
vsp_id = _create_vsp(client, admin_tokens, auth_headers, open_date="2020-01-01")
yesterday = (date.today() - timedelta(days=1)).isoformat()
response = client.put(
f"/api/v1/dict/info/{vsp_id}",
json={"close_date": yesterday},
headers=auth_headers(admin_tokens),
)
assert response.status_code == 200
assert response.json()["result"]["is_active"] is False
listing = client.get("/api/v1/dict/info", headers=auth_headers(admin_tokens))
assert listing.status_code == 200
assert vsp_id not in [item["id"] for item in listing.json()["result"]]
def _put_vsp(client, admin_tokens, auth_headers, vsp_id, **body):
response = client.put(
f"/api/v1/dict/info/{vsp_id}", json=body, headers=auth_headers(admin_tokens),
)
assert response.status_code == 200, response.text
return response.json()["result"]
def _closed_vsp(client, admin_tokens, auth_headers) -> int:
from datetime import date, timedelta
vsp_id = _create_vsp(client, admin_tokens, auth_headers, open_date="2020-01-01")
result = _put_vsp(
client, admin_tokens, auth_headers, vsp_id,
close_date=(date.today() - timedelta(days=1)).isoformat(),
)
assert result["is_active"] is False
return vsp_id
def test_vsp_update_clear_close_date_reopens(client, admin_tokens, auth_headers):
vsp_id = _closed_vsp(client, admin_tokens, auth_headers)
result = _put_vsp(client, admin_tokens, auth_headers, vsp_id, close_date=None)
assert result["close_date"] is None
assert result["is_active"] is True
listing = client.get("/api/v1/dict/info", headers=auth_headers(admin_tokens))
assert vsp_id in [item["id"] for item in listing.json()["result"]]
def test_vsp_update_future_close_date_reopens(client, admin_tokens, auth_headers):
from datetime import date, timedelta
vsp_id = _closed_vsp(client, admin_tokens, auth_headers)
future = (date.today() + timedelta(days=30)).isoformat()
result = _put_vsp(client, admin_tokens, auth_headers, vsp_id, close_date=future)
assert result["close_date"] == future
assert result["is_active"] is True
def test_vsp_update_without_close_date_keeps_it(client, admin_tokens, auth_headers):
from datetime import date, timedelta
vsp_id = _closed_vsp(client, admin_tokens, auth_headers)
yesterday = (date.today() - timedelta(days=1)).isoformat()
result = _put_vsp(client, admin_tokens, auth_headers, vsp_id, address="AURORA-1886 new")
assert result["close_date"] == yesterday
assert result["is_active"] is False
def test_vsp_clear_close_date_is_audited(client, admin_tokens, auth_headers, db_session):
from sqlalchemy import text
vsp_id = _closed_vsp(client, admin_tokens, auth_headers)
_put_vsp(client, admin_tokens, auth_headers, vsp_id, close_date=None)
async def _last_changes():
row = (
await db_session.execute(
text(
"SELECT event_data->'changes' FROM v3.audit_log "
"WHERE event = 'VSP_UPDATE' AND (event_data->>'vsp_id')::int = :id "
"ORDER BY id DESC LIMIT 1"
),
{"id": vsp_id},
)
).first()
return row[0]
changes = client.portal.call(_last_changes)
assert changes["closed_at"]["before"] is not None
assert changes["closed_at"]["after"] is None
# --- AURORA-1883: статус и период закрытия в справочнике ---
def _create_named_vsp(client, admin_tokens, auth_headers, **extra) -> str:
reg_number = _unique_reg_number()
created = client.post(
"/api/v1/dict/info",
json={
"registration_number": reg_number,
"address": "AURORA-1883",
"ssp_id": 1,
"open_date": "2020-01-01",
**extra,
},
headers=auth_headers(admin_tokens),
)
assert created.status_code == 201, created.text
return reg_number
def _list_reg_numbers(client, tokens, auth_headers, query: str = "") -> list[str]:
response = client.get(f"/api/v1/dict/info{query}", headers=auth_headers(tokens))
assert response.status_code == 200, response.text
return [item["registration_number"] for item in response.json()["result"]]
def test_vsp_list_status_default_active(client, admin_tokens, auth_headers):
active = _create_named_vsp(client, admin_tokens, auth_headers)
closed = _create_named_vsp(client, admin_tokens, auth_headers, close_date="2021-03-15")
regs = _list_reg_numbers(client, admin_tokens, auth_headers)
assert active in regs
assert closed not in regs
def test_vsp_list_status_closed(client, admin_tokens, auth_headers):
active = _create_named_vsp(client, admin_tokens, auth_headers)
closed = _create_named_vsp(client, admin_tokens, auth_headers, close_date="2021-03-15")
response = client.get("/api/v1/dict/info?status=closed", headers=auth_headers(admin_tokens))
assert response.status_code == 200
by_reg = {item["registration_number"]: item for item in response.json()["result"]}
assert closed in by_reg
assert by_reg[closed]["is_active"] is False
assert active not in by_reg
def test_vsp_list_status_all(client, admin_tokens, auth_headers):
active = _create_named_vsp(client, admin_tokens, auth_headers)
closed = _create_named_vsp(client, admin_tokens, auth_headers, close_date="2021-03-15")
regs = _list_reg_numbers(client, admin_tokens, auth_headers, "?status=all")
assert active in regs
assert closed in regs
def test_vsp_list_status_invalid(client, admin_tokens, auth_headers):
response = client.get("/api/v1/dict/info?status=bogus", headers=auth_headers(admin_tokens))
assert response.status_code == 422
def test_vsp_list_deleted_never_returned(client, admin_tokens, auth_headers):
reg = _create_named_vsp(client, admin_tokens, auth_headers, close_date="2021-03-15")
listing = client.get("/api/v1/dict/info?status=all", headers=auth_headers(admin_tokens))
vsp_id = next(i["id"] for i in listing.json()["result"] if i["registration_number"] == reg)
deleted = client.delete(f"/api/v1/dict/info/{vsp_id}", headers=auth_headers(admin_tokens))
assert deleted.status_code == 200
for status in ("active", "closed", "all"):
assert reg not in _list_reg_numbers(client, admin_tokens, auth_headers, f"?status={status}")
def test_vsp_list_close_period(client, admin_tokens, auth_headers):
closed = _create_named_vsp(client, admin_tokens, auth_headers, close_date="2021-03-15")
inside = _list_reg_numbers(
client, admin_tokens, auth_headers,
"?status=closed&close_date_start=2021-01-01&close_date_end=2021-12-31",
)
assert closed in inside
outside = _list_reg_numbers(client, admin_tokens, auth_headers, "?status=closed&close_date_start=2022-01-01")
assert closed not in outside
def test_vsp_list_closed_order(client, admin_tokens, auth_headers):
older = _create_named_vsp(client, admin_tokens, auth_headers, close_date="2021-03-15")
newer = _create_named_vsp(client, admin_tokens, auth_headers, close_date="2024-06-01")
regs = _list_reg_numbers(client, admin_tokens, auth_headers, "?status=closed")
assert regs.index(newer) < regs.index(older)
def test_vsp_list_closed_executor_scope(client, admin_tokens, isp_tokens, auth_headers):
# isp1 привязан к оргединице 2 (fixture.sql, user_org), оргединица 1 ему чужая
own = _create_named_vsp(client, admin_tokens, auth_headers, ssp_id=2, close_date="2021-03-15")
foreign = _create_named_vsp(client, admin_tokens, auth_headers, ssp_id=1, close_date="2021-03-15")
for status in ("closed", "all"):
regs = _list_reg_numbers(client, isp_tokens, auth_headers, f"?status={status}")
assert own in regs
assert foreign not in regs
def test_vsp_dropdown_include_inactive(client, admin_tokens, auth_headers):
closed = _create_named_vsp(client, admin_tokens, auth_headers, ssp_id=2, close_date="2021-03-15")
with_closed = client.get(
"/api/v1/dict/info/dropdown?form_id=2&include_inactive=true", headers=auth_headers(admin_tokens)
)
assert with_closed.status_code == 200
by_reg = {item["registration_number"]: item for item in with_closed.json()["result"]}
assert closed in by_reg
assert by_reg[closed]["is_active"] is False
default = client.get("/api/v1/dict/info/dropdown?form_id=2", headers=auth_headers(admin_tokens))
assert closed not in [item["registration_number"] for item in default.json()["result"]]
def _export_values(client, tokens, auth_headers, body: dict) -> set:
import io
from openpyxl import load_workbook
response = client.post("/api/v1/dict/info/export", json=body, headers=auth_headers(tokens))
assert response.status_code == 200, response.text
sheet = load_workbook(io.BytesIO(response.content)).active
return {cell for row in sheet.iter_rows(values_only=True) for cell in row}
def test_vsp_export_status_closed(client, admin_tokens, auth_headers):
active = _create_named_vsp(client, admin_tokens, auth_headers)
closed = _create_named_vsp(client, admin_tokens, auth_headers, close_date="2021-03-15")
closed_values = _export_values(client, admin_tokens, auth_headers, {"ssp_ids": [1], "status": "closed"})
assert closed in closed_values
assert active not in closed_values
default_values = _export_values(client, admin_tokens, auth_headers, {"ssp_ids": [1]})
assert active in default_values
assert closed not in default_values