d1922c9354
Миграция 0050_budget_line_delete_owner_guard -> 0056 (down 0055, стек поверх 0055 AURORA-1938). Тело v3.del_budget_line = актуальное тело из 0053_form2_ahr_source_sheet (AURORA-1907: запрет удаления АХР-строк с source_sheet IS NOT NULL, ручные удаляются) + только owner-guard; раньше копия опиралась на 0040 и откатывала 1907. Тесты: доступ к форме id=1 даёт отдельная пара исполнителей isp2/isp3 (isp1 больше не добавляется в org 1 — это ломало тесты «исполнитель из чужой организации»); окно фазы DFIP в фикстуре продлено (истекло 2026-06-06, add_line возвращал 403); маркер internal_order не приходит в INPUT-строке add_line — берём единственную INPUT-строку; добавлен тест 1907 (строка из сателлита не удаляется напрямую). Тест «владелец удаляет свою строку» помечен xfail(strict): add_budget_line не заполняет budget_line.created_by (O-1). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
113 lines
3.4 KiB
Python
113 lines
3.4 KiB
Python
import asyncio
|
||
from contextlib import contextmanager
|
||
import os
|
||
|
||
import pytest
|
||
from fastapi.testclient import TestClient
|
||
|
||
|
||
from src.db.base import SessionLocal
|
||
from src.db.session import get_db
|
||
from src.main import app
|
||
|
||
|
||
|
||
@pytest.fixture
|
||
def db_session():
|
||
"""Сессия БД теста: её же использует client, всё откатывается в конце теста."""
|
||
db_ref = SessionLocal()
|
||
yield db_ref
|
||
|
||
|
||
@pytest.fixture
|
||
def client(db_session):
|
||
db_ref = db_session
|
||
loop_ref = None
|
||
|
||
async def _get_db():
|
||
nonlocal loop_ref
|
||
loop_ref = asyncio.get_running_loop()
|
||
try:
|
||
yield db_ref
|
||
await db_ref.flush()
|
||
except:
|
||
await db_ref.rollback()
|
||
raise
|
||
|
||
app.dependency_overrides[get_db] = _get_db
|
||
with TestClient(app) as test_client:
|
||
try:
|
||
yield test_client
|
||
finally:
|
||
if db_ref is not None and loop_ref:
|
||
async def cleanup():
|
||
await db_ref.rollback()
|
||
await db_ref.close()
|
||
|
||
future = asyncio.run_coroutine_threadsafe(cleanup(), loop_ref)
|
||
future.result()
|
||
app.dependency_overrides.clear()
|
||
|
||
|
||
@pytest.fixture
|
||
def admin_password() -> str:
|
||
# Пароль админа берём из окружения, чтобы не хардкодить локальные отличия.
|
||
return os.getenv("OPENBAO__TEST_ADMIN_PASSWORD", "admin123")
|
||
|
||
|
||
@pytest.fixture
|
||
def isp_password() -> str:
|
||
# Пароль админа берём из окружения, чтобы не хардкодить локальные отличия.
|
||
return os.getenv("OPENBAO__TEST_EXECUTOR_PASSWORD", "admin123")
|
||
|
||
|
||
@pytest.fixture
|
||
def admin_tokens(client, admin_password: str):
|
||
response = client.post(
|
||
"/api/v1/auth/login",
|
||
json={"username": "admin", "password": admin_password},
|
||
)
|
||
assert response.status_code == 200
|
||
return response.json()
|
||
|
||
|
||
@pytest.fixture
|
||
def auth_headers():
|
||
def _build(tokens: dict) -> dict:
|
||
return {"Authorization": f"Bearer {tokens['access_token']}"}
|
||
|
||
return _build
|
||
|
||
@pytest.fixture
|
||
def isp_tokens(client, isp_password: str):
|
||
response = client.post(
|
||
"/api/v1/auth/login",
|
||
json={"username": "isp1", "password": isp_password},
|
||
)
|
||
assert response.status_code == 200
|
||
return response.json()
|
||
|
||
|
||
@pytest.fixture
|
||
def isp2_tokens(client, isp_password: str):
|
||
# Исполнитель в org_unit формы id=1 (isp1 в нём намеренно не состоит — на этом
|
||
# держатся тесты «исполнитель из чужой организации»). Нужен вместе с isp3_tokens,
|
||
# чтобы проверять, что исполнитель не может удалить чужую строку.
|
||
response = client.post(
|
||
"/api/v1/auth/login",
|
||
json={"username": "isp2", "password": isp_password},
|
||
)
|
||
assert response.status_code == 200
|
||
return response.json()
|
||
|
||
|
||
@pytest.fixture
|
||
def isp3_tokens(client, isp_password: str):
|
||
# Второй исполнитель в том же org_unit, что и isp2 (форма id=1).
|
||
response = client.post(
|
||
"/api/v1/auth/login",
|
||
json={"username": "isp3", "password": isp_password},
|
||
)
|
||
assert response.status_code == 200
|
||
return response.json()
|