435 lines
14 KiB
Python
435 lines
14 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
|