46fba8875b
PUT …/goodwill {rows: [...]} вместо построчного PUT …/goodwill/{head}/{sub}: все строки
проверяются до записи, ошибка любой (NOT_FOUND, INVALID_SHARE, COMMENT_REQUIRED,
MONEY_OUT_OF_RANGE, DUPLICATE_ROW) несёт ключ пары и откатывает пакет; неизменённые
строки не перезаписываются и аудит не пишут (DOC-259025 §5.1 п. 4).
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
231 lines
12 KiB
Python
231 lines
12 KiB
Python
"""Проверка ввода строки до обращения к хранилищу."""
|
||
|
||
from decimal import Decimal
|
||
from http import HTTPStatus
|
||
|
||
import pytest
|
||
from sqlalchemy import func, select
|
||
|
||
from src.api.dependencies.audit import RequestAudit
|
||
from src.common.exceptions import ConflictError, DomainValidationError
|
||
from src.models.audit_ex import AuditEvent
|
||
from src.models.enums import AuditAction
|
||
from src.models.kfo import Goodwill, KfoResult
|
||
from src.models.osv import OsvEntry
|
||
from src.models.statements import FinancialStatementValue
|
||
from src.schemas.goodwill import GoodwillRowIn
|
||
from src.services.goodwill import validate_decision
|
||
from tests.test_api_consolidated_osv import KA_RF
|
||
from tests.test_api_goodwill import batch, goodwill_world
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
"value", ["-1", "100.01", "100.0001", "80.123", "80,123", "12.3456", "1e2", "NaN", "Infinity", "abc", ""]
|
||
)
|
||
def test_invalid_share(value):
|
||
with pytest.raises(DomainValidationError) as error:
|
||
validate_decision(GoodwillRowIn(ownership_share=value, include_in_kfo=False))
|
||
assert error.value.code == "INVALID_SHARE"
|
||
|
||
|
||
@pytest.mark.parametrize("value", ["0", "100", "7.5", "80.12", "80,12"])
|
||
def test_valid_share(value):
|
||
result = validate_decision(GoodwillRowIn(ownership_share=value, include_in_kfo=True, comment=" Обоснование "))
|
||
assert str(result.ownership_share) == value.replace(",", ".")
|
||
assert result.comment == "Обоснование"
|
||
|
||
|
||
def test_required_fields():
|
||
assert validate_decision(GoodwillRowIn(ownership_share=None, include_in_kfo=False)).ownership_share is None
|
||
with pytest.raises(DomainValidationError) as error:
|
||
validate_decision(GoodwillRowIn(ownership_share=None, include_in_kfo=True, comment="Да"))
|
||
assert error.value.code == "INVALID_SHARE"
|
||
with pytest.raises(DomainValidationError) as error:
|
||
validate_decision(GoodwillRowIn(ownership_share="100", include_in_kfo=True, comment=" "))
|
||
assert error.value.code == "COMMENT_REQUIRED"
|
||
|
||
|
||
@pytest.mark.integration
|
||
@pytest.mark.fixtures
|
||
async def test_save_settings_and_disappearing_pair(db_session, auth_client, fixtures_dir):
|
||
|
||
w = await goodwill_world(db_session, fixtures_dir)
|
||
client = await auth_client(KA_RF)
|
||
base = f"/consolidations/versions/{w.version.id}"
|
||
path = base + "/goodwill"
|
||
body = dict(ownership_share="80.12", include_in_kfo=True, comment="Подтверждено")
|
||
response = await client.put(path, json=batch(w.a.id, w.b.id, **body))
|
||
assert response.status_code == HTTPStatus.CONFLICT and response.json()["detail"]["code"] == "LOCK_REQUIRED"
|
||
assert (await client.post(base + "/lock")).status_code == HTTPStatus.CREATED
|
||
other = await auth_client("rka@armka.local")
|
||
rejected = await other.put(path, json=batch(w.a.id, w.b.id, **body))
|
||
assert rejected.status_code == HTTPStatus.CONFLICT
|
||
assert rejected.json()["detail"]["code"] == "LOCKED_BY_OTHER"
|
||
response = await client.put(path, json=batch(w.a.id, w.b.id, **body))
|
||
assert response.status_code == HTTPStatus.OK, response.text
|
||
row = response.json()["rows"][0]
|
||
assert row["ownership_share"] == "80.12"
|
||
assert row["participation_amount"] == "80120000.00"
|
||
assert row["goodwill_amount"] == "189877300.00"
|
||
again = await client.put(path, json=batch(w.a.id, w.b.id, **body))
|
||
assert again.status_code == HTTPStatus.OK
|
||
assert await db_session.scalar(select(func.count()).select_from(Goodwill)) == 1
|
||
on = await client.put(base + "/goodwill-settings", json=dict(goodwill_enabled=True))
|
||
assert on.status_code == HTTPStatus.OK
|
||
assert on.json()["totals"]["goodwill_in_kfo_amount"] == "189877300.00"
|
||
audit_count = await db_session.scalar(
|
||
select(func.count()).select_from(AuditEvent).where(AuditEvent.action == AuditAction.GOODWILL_EDIT)
|
||
)
|
||
assert (
|
||
await client.put(base + "/goodwill-settings", json=dict(goodwill_enabled=True))
|
||
).status_code == HTTPStatus.OK
|
||
assert audit_count == await db_session.scalar(
|
||
select(func.count()).select_from(AuditEvent).where(AuditEvent.action == AuditAction.GOODWILL_EDIT)
|
||
)
|
||
off = await client.put(base + "/goodwill-settings", json=dict(goodwill_enabled=False))
|
||
assert off.json()["totals"]["goodwill_in_kfo_amount"] == "0.00"
|
||
assert off.json()["rows"][0]["ownership_share"] == "80.12"
|
||
body["include_in_kfo"] = False
|
||
assert (await client.put(path, json=batch(w.a.id, w.b.id, **body))).status_code == HTTPStatus.OK
|
||
assert (await client.get(base + "/goodwill")).json()["totals"]["goodwill_amount"] == "0.00"
|
||
entries = list(
|
||
(
|
||
await db_session.scalars(
|
||
select(OsvEntry).where(OsvEntry.account_code == "58.01", OsvEntry.owner_legal_entity_id == w.a.id)
|
||
)
|
||
).all()
|
||
)
|
||
for entry in entries:
|
||
entry.closing_debit = 0
|
||
await db_session.flush()
|
||
row = (await client.get(base + "/goodwill")).json()["rows"][0]
|
||
assert not row["has_investments"] and row["goodwill_amount"] is None
|
||
assert row["comment"] == "Подтверждено"
|
||
assert (await client.put(path, json=batch(w.a.id, w.b.id, **body))).status_code == HTTPStatus.OK
|
||
body["include_in_kfo"] = True
|
||
response = await client.put(path, json=batch(w.a.id, w.b.id, **body))
|
||
assert response.status_code == HTTPStatus.OK
|
||
assert response.json()["rows"][0]["financial_investment_amount"] is None
|
||
assert response.json()["rows"][0]["goodwill_amount"] is None
|
||
assert await db_session.scalar(select(func.count()).select_from(KfoResult)) == 0
|
||
|
||
|
||
@pytest.mark.integration
|
||
@pytest.mark.fixtures
|
||
async def test_mutation_validation_and_rollback(db_session, auth_client, fixtures_dir, monkeypatch):
|
||
|
||
w = await goodwill_world(db_session, fixtures_dir)
|
||
client = await auth_client(KA_RF)
|
||
base = f"/consolidations/versions/{w.version.id}"
|
||
path = base + "/goodwill"
|
||
assert (await client.post(base + "/lock")).status_code == HTTPStatus.CREATED
|
||
body = dict(ownership_share="100", include_in_kfo=True, comment="")
|
||
response = await client.put(path, json=batch(w.a.id, w.b.id, **body))
|
||
assert (
|
||
response.status_code == HTTPStatus.UNPROCESSABLE_ENTITY
|
||
and response.json()["detail"]["code"] == "COMMENT_REQUIRED"
|
||
)
|
||
body.update(ownership_share="100.00001", comment="Да")
|
||
assert (await client.put(path, json=batch(w.a.id, w.b.id, **body))).json()["detail"]["code"] == "INVALID_SHARE"
|
||
assert await db_session.scalar(select(func.count()).select_from(Goodwill)) == 0
|
||
|
||
async def reject_audit(*args, **kwargs):
|
||
raise ConflictError("TEST_AUDIT_FAILURE", "Ошибка тестового аудита")
|
||
|
||
monkeypatch.setattr(RequestAudit, "record", reject_audit)
|
||
body["ownership_share"] = "100"
|
||
response = await client.put(path, json=batch(w.a.id, w.b.id, **body))
|
||
assert response.status_code == HTTPStatus.CONFLICT
|
||
assert await db_session.scalar(select(func.count()).select_from(Goodwill)) == 0
|
||
|
||
|
||
@pytest.mark.integration
|
||
@pytest.mark.fixtures
|
||
async def test_money_overflow_keeps_database_unchanged(db_session, auth_client, fixtures_dir):
|
||
w = await goodwill_world(db_session, fixtures_dir)
|
||
client = await auth_client(KA_RF)
|
||
base = f"/consolidations/versions/{w.version.id}"
|
||
assert (await client.post(base + "/lock")).status_code == HTTPStatus.CREATED
|
||
value = (await db_session.scalars(select(FinancialStatementValue))).one()
|
||
value.amount = Decimal("9999999999999999.99")
|
||
await db_session.flush()
|
||
response = await client.put(
|
||
base + "/goodwill", json=batch(w.a.id, w.b.id, ownership_share="100", include_in_kfo=True, comment="Проверка")
|
||
)
|
||
assert response.status_code == HTTPStatus.UNPROCESSABLE_ENTITY
|
||
assert response.json()["detail"]["code"] == "MONEY_OUT_OF_RANGE"
|
||
assert await db_session.scalar(select(func.count()).select_from(Goodwill)) == 0
|
||
|
||
|
||
async def _locked_world(db_session, auth_client, fixtures_dir):
|
||
w = await goodwill_world(db_session, fixtures_dir)
|
||
client = await auth_client(KA_RF)
|
||
base = f"/consolidations/versions/{w.version.id}"
|
||
assert (await client.post(base + "/lock")).status_code == HTTPStatus.CREATED
|
||
return w, client, base
|
||
|
||
|
||
async def _goodwill_audit_count(db_session) -> int:
|
||
return await db_session.scalar(
|
||
select(func.count()).select_from(AuditEvent).where(AuditEvent.action == AuditAction.GOODWILL_EDIT)
|
||
)
|
||
|
||
|
||
@pytest.mark.integration
|
||
@pytest.mark.fixtures
|
||
async def test_batch_rolls_back_when_one_row_invalid(db_session, auth_client, fixtures_dir):
|
||
w, client, base = await _locked_world(db_session, auth_client, fixtures_dir)
|
||
valid = batch(w.a.id, w.b.id, ownership_share="100", include_in_kfo=True, comment="Да")["rows"][0]
|
||
invalid = batch(w.b.id, w.a.id, ownership_share="50", include_in_kfo=True, comment=" ")["rows"][0]
|
||
# обратная пара Б → А в таблице не существует: сперва проверяем порядок ошибок на существующей паре
|
||
response = await client.put(base + "/goodwill", json={"rows": [valid, {**valid, "comment": " "}]})
|
||
assert response.status_code == HTTPStatus.UNPROCESSABLE_ENTITY
|
||
assert response.json()["detail"]["code"] == "DUPLICATE_ROW"
|
||
response = await client.put(base + "/goodwill", json={"rows": [valid, invalid]})
|
||
assert response.status_code == HTTPStatus.NOT_FOUND
|
||
assert response.json()["detail"]["subsidiary_legal_entity_id"] == str(w.a.id)
|
||
assert await db_session.scalar(select(func.count()).select_from(Goodwill)) == 0
|
||
assert await _goodwill_audit_count(db_session) == 0
|
||
|
||
|
||
@pytest.mark.integration
|
||
@pytest.mark.fixtures
|
||
async def test_batch_reports_invalid_row_key(db_session, auth_client, fixtures_dir):
|
||
w, client, base = await _locked_world(db_session, auth_client, fixtures_dir)
|
||
body = batch(w.a.id, w.b.id, ownership_share="100", include_in_kfo=True, comment=" ")
|
||
response = await client.put(base + "/goodwill", json=body)
|
||
assert response.status_code == HTTPStatus.UNPROCESSABLE_ENTITY
|
||
detail = response.json()["detail"]
|
||
assert detail["code"] == "COMMENT_REQUIRED"
|
||
assert (detail["head_legal_entity_id"], detail["subsidiary_legal_entity_id"]) == (str(w.a.id), str(w.b.id))
|
||
assert await db_session.scalar(select(func.count()).select_from(Goodwill)) == 0
|
||
assert await _goodwill_audit_count(db_session) == 0
|
||
|
||
|
||
@pytest.mark.integration
|
||
@pytest.mark.fixtures
|
||
async def test_unchanged_rows_are_not_rewritten(db_session, auth_client, fixtures_dir):
|
||
w, client, base = await _locked_world(db_session, auth_client, fixtures_dir)
|
||
body = batch(w.a.id, w.b.id, ownership_share="80,5", include_in_kfo=True, comment="Устав")
|
||
first = await client.put(base + "/goodwill", json=body)
|
||
assert first.status_code == HTTPStatus.OK, first.text
|
||
saved_at = first.json()["rows"][0]["updated_at"]
|
||
again = await client.put(
|
||
base + "/goodwill", json=batch(w.a.id, w.b.id, ownership_share="80.50", include_in_kfo=True, comment="Устав")
|
||
)
|
||
assert again.status_code == HTTPStatus.OK
|
||
assert again.json()["rows"][0]["updated_at"] == saved_at
|
||
assert await _goodwill_audit_count(db_session) == 1
|
||
|
||
|
||
@pytest.mark.integration
|
||
@pytest.mark.fixtures
|
||
async def test_new_row_with_defaults_is_not_created(db_session, auth_client, fixtures_dir):
|
||
w, client, base = await _locked_world(db_session, auth_client, fixtures_dir)
|
||
body = batch(w.a.id, w.b.id, ownership_share=None, include_in_kfo=False, comment=None)
|
||
response = await client.put(base + "/goodwill", json=body)
|
||
assert response.status_code == HTTPStatus.OK, response.text
|
||
assert response.json()["rows"][0]["include_in_kfo"] is False
|
||
assert await db_session.scalar(select(func.count()).select_from(Goodwill)) == 0
|
||
assert await _goodwill_audit_count(db_session) == 0
|