правки по ревью
This commit is contained in:
PotapovaA
parent
c6a10e4604
commit
abe1d890d0
@ -180,7 +180,7 @@ async def get_unpaired_osv(
|
||||
service: IntercompanyMatrixServiceDep,
|
||||
) -> UnpairedOsvOut:
|
||||
"""Все строки односторонних ячеек шахматки и их число."""
|
||||
return await service.unpaired_osv(version_id, rule_id)
|
||||
return await service.get_unpaired_osv(version_id, rule_id)
|
||||
|
||||
|
||||
@router.get(
|
||||
|
||||
@ -1,56 +1,39 @@
|
||||
"""Чтение версионированных правил элиминации и счетов их сторон."""
|
||||
|
||||
import datetime
|
||||
import uuid
|
||||
from collections.abc import Collection
|
||||
|
||||
from sqlalchemy import func, or_, select
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from src.models.accounting import ChartOfAccount, EliminationRule, EliminationRuleAccount
|
||||
|
||||
|
||||
async def list_rules_with_accounts(
|
||||
session: AsyncSession, version_no: int, *, as_of: datetime.date | None = None
|
||||
session: AsyncSession, version_no: int | None
|
||||
) -> list[tuple[EliminationRule, EliminationRuleAccount | None]]:
|
||||
"""Правила выбранной версии и счета сторон; для текущей версии — только действующие на дату."""
|
||||
"""Правила заданной версии либо все правила без `valid_to`."""
|
||||
condition = (
|
||||
EliminationRule.version_no == version_no
|
||||
if version_no is not None
|
||||
else EliminationRule.valid_to.is_(None)
|
||||
)
|
||||
stmt = (
|
||||
select(EliminationRule, EliminationRuleAccount)
|
||||
.outerjoin(EliminationRuleAccount, EliminationRuleAccount.rule_id == EliminationRule.id)
|
||||
.where(EliminationRule.version_no == version_no)
|
||||
.where(condition)
|
||||
.order_by(
|
||||
EliminationRule.statement_type,
|
||||
EliminationRule.rule_no,
|
||||
EliminationRuleAccount.company_no,
|
||||
)
|
||||
)
|
||||
if as_of is not None:
|
||||
stmt = stmt.where(
|
||||
EliminationRule.valid_from <= as_of,
|
||||
or_(EliminationRule.valid_to.is_(None), EliminationRule.valid_to >= as_of),
|
||||
)
|
||||
return [(rule, account) for rule, account in (await session.execute(stmt)).all()]
|
||||
|
||||
|
||||
async def current_version_no(session: AsyncSession, as_of: datetime.date) -> int | None:
|
||||
"""Старшая версия правил, у которой есть действующие на дату записи."""
|
||||
stmt = select(func.max(EliminationRule.version_no)).where(
|
||||
EliminationRule.valid_from <= as_of,
|
||||
or_(EliminationRule.valid_to.is_(None), EliminationRule.valid_to >= as_of),
|
||||
)
|
||||
return await session.scalar(stmt)
|
||||
|
||||
|
||||
def is_valid_for_version(
|
||||
rule: EliminationRule | None, version_no: int | None, *, as_of: datetime.date | None = None
|
||||
) -> bool:
|
||||
"""Проверить принадлежность правила версии и, если нужно, срок его действия."""
|
||||
return (
|
||||
rule is not None
|
||||
and version_no is not None
|
||||
and rule.version_no == version_no
|
||||
and (as_of is None or (rule.valid_from <= as_of and (rule.valid_to is None or rule.valid_to >= as_of)))
|
||||
)
|
||||
def is_valid_for_version(rule: EliminationRule | None, version_no: int | None) -> bool:
|
||||
"""Проверить номер версии либо отсутствие `valid_to`."""
|
||||
return rule is not None and (rule.version_no == version_no if version_no is not None else rule.valid_to is None)
|
||||
|
||||
|
||||
async def list_rules(session: AsyncSession, version_no: int) -> list[EliminationRule]:
|
||||
|
||||
@ -4,7 +4,6 @@ import uuid
|
||||
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from src.common.clock import moscow_date, utc_now
|
||||
from src.common.exceptions import NotFoundError
|
||||
from src.domain.policy import VisibilityScope
|
||||
from src.repositories import consolidations as consolidations_repo
|
||||
@ -27,17 +26,8 @@ class EliminationRuleService:
|
||||
raise NotFoundError(VERSION_NOT_FOUND)
|
||||
version, _ = found
|
||||
is_version_fixed = version.elimination_rules_version is not None
|
||||
if is_version_fixed:
|
||||
as_of = None
|
||||
version_no = version.elimination_rules_version
|
||||
else:
|
||||
as_of = moscow_date(utc_now())
|
||||
version_no = await rules_repo.current_version_no(self._session, as_of)
|
||||
rows = (
|
||||
await rules_repo.list_rules_with_accounts(self._session, version_no, as_of=as_of)
|
||||
if version_no is not None
|
||||
else []
|
||||
)
|
||||
version_no = version.elimination_rules_version
|
||||
rows = await rules_repo.list_rules_with_accounts(self._session, version_no)
|
||||
by_id: dict[uuid.UUID, EliminationRuleOut] = {}
|
||||
for rule, account in rows:
|
||||
result = by_id.get(rule.id)
|
||||
@ -64,6 +54,6 @@ class EliminationRuleService:
|
||||
return EliminationRulesOut(
|
||||
version_no=version_no,
|
||||
is_version_fixed=is_version_fixed,
|
||||
as_of=as_of,
|
||||
as_of=None,
|
||||
rules=list(by_id.values()),
|
||||
)
|
||||
|
||||
@ -10,7 +10,6 @@ from typing import Literal
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from src.common.clock import moscow_date, utc_now
|
||||
from src.common.exceptions import DomainValidationError, NotFoundError
|
||||
from src.domain.consolidated_osv import ConsolidatedRow, Entity
|
||||
from src.domain.policy import VisibilityScope
|
||||
@ -79,6 +78,18 @@ class MatrixSelection:
|
||||
measure: Measure
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class MatrixBuildResult:
|
||||
"""Матрица и данные свода, использованные при её построении."""
|
||||
|
||||
matrix: IntercompanyMatrixOut
|
||||
version: ConsolidationVersion
|
||||
consolidation: Consolidation
|
||||
accounts: list[ChartOfAccount]
|
||||
selection: MatrixSelection
|
||||
rows: tuple[ConsolidatedRow, ...]
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class MatrixExportInfo:
|
||||
group_name: str
|
||||
@ -97,7 +108,7 @@ class MatrixExportInfo:
|
||||
return f"intercompany-matrix_{self.group_name}_{self.year}-{self.quarter.value}_v{self.version_no}{preset}.xlsx"
|
||||
|
||||
|
||||
def _amount(row: ConsolidatedRow, measure: Measure, side: Side) -> Decimal:
|
||||
def _get_amount(row: ConsolidatedRow, measure: Measure, side: Side) -> Decimal:
|
||||
if measure == "closing":
|
||||
value = row.closing_debit if side == "D" else row.closing_credit
|
||||
else:
|
||||
@ -136,11 +147,11 @@ def calculate_matrix(
|
||||
continue
|
||||
if row.account_id in selection.company_1_account_ids:
|
||||
item = amounts_1[pair]
|
||||
item["amount"] += _amount(row, selection.measure, "D")
|
||||
item["amount"] += _get_amount(row, selection.measure, "D")
|
||||
item["entry_osv_ids"].update(row.entry_ids)
|
||||
if row.account_id in selection.company_2_account_ids:
|
||||
item = amounts_2[pair]
|
||||
item["amount"] += _amount(row, selection.measure, "K")
|
||||
item["amount"] += _get_amount(row, selection.measure, "K")
|
||||
item["entry_osv_ids"].update(row.entry_ids)
|
||||
|
||||
cells: list[IntercompanyMatrixCellOut] = []
|
||||
@ -179,7 +190,7 @@ def calculate_matrix(
|
||||
return IntercompanyMatrixOut(companies=list(company_info), cells=cells)
|
||||
|
||||
|
||||
def unpaired_rows(rows: Iterable[ConsolidatedRow], matrix: IntercompanyMatrixOut) -> list[ConsolidatedRow]:
|
||||
def get_unpaired_rows(rows: Iterable[ConsolidatedRow], matrix: IntercompanyMatrixOut) -> list[ConsolidatedRow]:
|
||||
"""Строки свода, попавшие только на одну сторону пары компаний в шахматке."""
|
||||
entry_ids = {
|
||||
entry_id
|
||||
@ -190,7 +201,7 @@ def unpaired_rows(rows: Iterable[ConsolidatedRow], matrix: IntercompanyMatrixOut
|
||||
return [row for row in rows if entry_ids.intersection(row.entry_ids)]
|
||||
|
||||
|
||||
def _account_tree(
|
||||
def _make_account_tree(
|
||||
accounts: Sequence[ChartOfAccount],
|
||||
) -> tuple[dict[uuid.UUID, ChartOfAccount], dict[uuid.UUID, list[uuid.UUID]]]:
|
||||
by_id = {account.id: account for account in accounts}
|
||||
@ -213,11 +224,11 @@ def _expand(roots: Iterable[uuid.UUID], children: dict[uuid.UUID, list[uuid.UUID
|
||||
return result
|
||||
|
||||
|
||||
def _kinds(side: Side) -> frozenset[AccountKind]:
|
||||
def _get_kinds(side: Side) -> frozenset[AccountKind]:
|
||||
return DEBIT_KINDS if side == "D" else CREDIT_KINDS
|
||||
|
||||
|
||||
def _selected_accounts(
|
||||
def _select_accounts(
|
||||
included: Iterable[uuid.UUID],
|
||||
excluded: Iterable[uuid.UUID],
|
||||
side: Side,
|
||||
@ -229,10 +240,10 @@ def _selected_accounts(
|
||||
if missing:
|
||||
missing_ids = ", ".join(map(str, sorted(missing)))
|
||||
raise DomainValidationError("ACCOUNT_NOT_FOUND", f"Выбранного счёта нет ({missing_ids})")
|
||||
if any(by_id[account_id].kind not in _kinds(side) for account_id in roots):
|
||||
if any(by_id[account_id].kind not in _get_kinds(side) for account_id in roots):
|
||||
raise DomainValidationError("ACCOUNT_KIND_MISMATCH", "Сторона счёта не соответствует направлению расчёта")
|
||||
selected = _expand(roots, children) - _expand(excluded, children)
|
||||
return frozenset(account_id for account_id in selected if by_id[account_id].kind in _kinds(side))
|
||||
return frozenset(account_id for account_id in selected if by_id[account_id].kind in _get_kinds(side))
|
||||
|
||||
|
||||
def calculate_result_table(
|
||||
@ -241,7 +252,7 @@ def calculate_result_table(
|
||||
accounts: Sequence[ChartOfAccount],
|
||||
) -> IntercompanyResultTableOut:
|
||||
"""Фиксированные статьи ВГО: сальдо по всем контрагентам периметра для каждого ЮЛ."""
|
||||
by_id, children = _account_tree(accounts)
|
||||
by_id, children = _make_account_tree(accounts)
|
||||
by_code = {account.code: account.id for account in accounts}
|
||||
required_codes = {code for article in RESULT_ARTICLES for code in (*article.account_codes, *article.excluded_codes)}
|
||||
missing = sorted(required_codes - by_code.keys())
|
||||
@ -249,7 +260,7 @@ def calculate_result_table(
|
||||
raise DomainValidationError("RESULT_ACCOUNTS_MISSING", f"В плане счетов нет счетов: {', '.join(missing)}")
|
||||
|
||||
selected = {
|
||||
article.code: _selected_accounts(
|
||||
article.code: _select_accounts(
|
||||
(by_code[code] for code in article.account_codes),
|
||||
(by_code[code] for code in article.excluded_codes),
|
||||
article.side,
|
||||
@ -273,7 +284,7 @@ def calculate_result_table(
|
||||
continue
|
||||
for article in RESULT_ARTICLES:
|
||||
if row.account_id in selected[article.code]:
|
||||
amounts[article.code][row.legal_entity_id] += _amount(row, "closing", article.side)
|
||||
amounts[article.code][row.legal_entity_id] += _get_amount(row, "closing", article.side)
|
||||
|
||||
articles: list[IntercompanyResultArticleOut] = []
|
||||
for article in RESULT_ARTICLES:
|
||||
@ -332,15 +343,19 @@ class IntercompanyMatrixService:
|
||||
credit_account_ids: Sequence[uuid.UUID] = (),
|
||||
rule_id: uuid.UUID | None = None,
|
||||
) -> IntercompanyMatrixOut:
|
||||
"""Рассчитать матрицу по всем счетам, ручному выбору или правилу-пресету."""
|
||||
matrix, _, _, _, _, _ = await self._build_accounts(
|
||||
"""Построить шахматку по выбранному режиму.
|
||||
|
||||
ALL включает все счета соответствующей стороны; MANUAL — выбранные счета и их потомков;
|
||||
PRESET берёт счета и показатель из правила с идентификатором `rule_id`.
|
||||
"""
|
||||
result = await self._build_accounts(
|
||||
version_id,
|
||||
mode=mode,
|
||||
debit_account_ids=debit_account_ids,
|
||||
credit_account_ids=credit_account_ids,
|
||||
rule_id=rule_id,
|
||||
)
|
||||
return matrix
|
||||
return result.matrix
|
||||
|
||||
async def build_export(
|
||||
self,
|
||||
@ -352,7 +367,7 @@ class IntercompanyMatrixService:
|
||||
rule_id: uuid.UUID | None = None,
|
||||
) -> tuple[IntercompanyMatrixOut, MatrixExportInfo]:
|
||||
"""Матрица и реквизиты книги из одного расчёта свода."""
|
||||
matrix, version, consolidation, accounts, selection, _ = await self._build_accounts(
|
||||
result = await self._build_accounts(
|
||||
version_id,
|
||||
mode=mode,
|
||||
debit_account_ids=debit_account_ids,
|
||||
@ -361,35 +376,35 @@ class IntercompanyMatrixService:
|
||||
)
|
||||
group_name = (
|
||||
await self._session.execute(
|
||||
select(CompanyGroup.name).where(CompanyGroup.id == consolidation.company_group_id)
|
||||
select(CompanyGroup.name).where(CompanyGroup.id == result.consolidation.company_group_id)
|
||||
)
|
||||
).scalar_one()
|
||||
by_id = {account.id: account.code for account in accounts}
|
||||
by_id = {account.id: account.code for account in result.accounts}
|
||||
if mode == "PRESET":
|
||||
debit_codes = tuple(sorted(by_id[account_id] for account_id in selection.company_1_account_ids))
|
||||
credit_codes = tuple(sorted(by_id[account_id] for account_id in selection.company_2_account_ids))
|
||||
debit_codes = tuple(sorted(by_id[account_id] for account_id in result.selection.company_1_account_ids))
|
||||
credit_codes = tuple(sorted(by_id[account_id] for account_id in result.selection.company_2_account_ids))
|
||||
else:
|
||||
debit_codes = tuple(by_id[account_id] for account_id in debit_account_ids)
|
||||
credit_codes = tuple(by_id[account_id] for account_id in credit_account_ids)
|
||||
info = MatrixExportInfo(
|
||||
group_name=group_name,
|
||||
year=consolidation.year,
|
||||
quarter=consolidation.quarter,
|
||||
version_no=version.version_no,
|
||||
year=result.consolidation.year,
|
||||
quarter=result.consolidation.quarter,
|
||||
version_no=result.version.version_no,
|
||||
mode=mode,
|
||||
debit_account_codes=debit_codes,
|
||||
credit_account_codes=credit_codes,
|
||||
measure=selection.measure,
|
||||
measure=result.selection.measure,
|
||||
rule_id=rule_id,
|
||||
)
|
||||
return matrix, info
|
||||
return result.matrix, info
|
||||
|
||||
async def unpaired_osv(self, version_id: uuid.UUID, rule_id: uuid.UUID) -> UnpairedOsvOut:
|
||||
async def get_unpaired_osv(self, version_id: uuid.UUID, rule_id: uuid.UUID) -> UnpairedOsvOut:
|
||||
"""Все строки свода без встречной стороны в шахматке заданного пресета."""
|
||||
matrix, _, _, _, _, rows = await self._build_accounts(
|
||||
result = await self._build_accounts(
|
||||
version_id, mode="PRESET", debit_account_ids=(), credit_account_ids=(), rule_id=rule_id
|
||||
)
|
||||
unmatched = unpaired_rows(rows, matrix)
|
||||
unmatched = get_unpaired_rows(result.rows, result.matrix)
|
||||
return UnpairedOsvOut(items=[row_out(row) for row in unmatched], total=len(unmatched))
|
||||
|
||||
async def _apply_corrections(
|
||||
@ -449,7 +464,44 @@ class IntercompanyMatrixService:
|
||||
)
|
||||
return IntercompanyMatrixOut(companies=matrix.companies, cells=cells)
|
||||
|
||||
async def _build_accounts( # noqa: PLR0912 - три режима выбора счетов и проверка пресета
|
||||
async def _resolve_preset_selection(
|
||||
self,
|
||||
rule_id: uuid.UUID | None,
|
||||
version: ConsolidationVersion,
|
||||
by_id: dict[uuid.UUID, ChartOfAccount],
|
||||
children: dict[uuid.UUID, list[uuid.UUID]],
|
||||
) -> tuple[MatrixSelection, EliminationRule]:
|
||||
"""Выбрать счета и показатель пресета для версии консолидации."""
|
||||
if rule_id is None:
|
||||
raise DomainValidationError("RULE_REQUIRED", "Выберите правило-пресет")
|
||||
rule, rule_accounts = await elimination_rules_repo.rule_with_account(self._session, rule_id)
|
||||
if rule is None or not elimination_rules_repo.is_valid_for_version(rule, version.elimination_rules_version):
|
||||
raise NotFoundError("правило-пресет не найдено для версии консолидации")
|
||||
if rule.side_1 not in ("D", "K") or rule.measure not in ("closing", "turnover"):
|
||||
raise DomainValidationError("INVALID_RULE", "Неверное направление или показатель правила-пресета")
|
||||
|
||||
sides: dict[int, tuple[list[uuid.UUID], list[uuid.UUID]]] = {
|
||||
1: ([], []),
|
||||
2: ([], []),
|
||||
}
|
||||
for rule_account in rule_accounts:
|
||||
if rule_account.company_no not in sides:
|
||||
raise DomainValidationError("INVALID_RULE", "Неверная сторона счёта в правиле-пресете")
|
||||
included, excluded = sides[rule_account.company_no]
|
||||
(excluded if rule_account.is_excluded else included).append(rule_account.account_id)
|
||||
debit_side = 1 if rule.side_1 == "D" else 2
|
||||
credit_side = 2 if debit_side == 1 else 1
|
||||
debit_ids, debit_excluded = sides[debit_side]
|
||||
credit_ids, credit_excluded = sides[credit_side]
|
||||
if not debit_ids or not credit_ids:
|
||||
raise DomainValidationError("ACCOUNTS_REQUIRED", "В правиле-пресете нужны счета обеих сторон")
|
||||
debit = _select_accounts(debit_ids, debit_excluded, "D", by_id, children)
|
||||
credit = _select_accounts(credit_ids, credit_excluded, "K", by_id, children)
|
||||
measure = "closing" if rule.measure == "closing" else "turnover"
|
||||
|
||||
return MatrixSelection(debit, credit, measure), rule
|
||||
|
||||
async def _build_accounts(
|
||||
self,
|
||||
version_id: uuid.UUID,
|
||||
*,
|
||||
@ -457,67 +509,36 @@ class IntercompanyMatrixService:
|
||||
debit_account_ids: Sequence[uuid.UUID],
|
||||
credit_account_ids: Sequence[uuid.UUID],
|
||||
rule_id: uuid.UUID | None = None,
|
||||
) -> tuple[
|
||||
IntercompanyMatrixOut,
|
||||
ConsolidationVersion,
|
||||
Consolidation,
|
||||
list[ChartOfAccount],
|
||||
MatrixSelection,
|
||||
tuple[ConsolidatedRow, ...],
|
||||
]:
|
||||
) -> MatrixBuildResult:
|
||||
consolidated, version, consolidation, members = await self._osv_service.build_with_members(version_id)
|
||||
rows = consolidated.rows
|
||||
accounts = await directory_repo.list_accounts(self._session)
|
||||
by_id, children = _account_tree(accounts)
|
||||
measure: Measure = "closing"
|
||||
by_id, children = _make_account_tree(accounts)
|
||||
rule: EliminationRule | None = None
|
||||
if mode == "ALL":
|
||||
if debit_account_ids or credit_account_ids:
|
||||
raise DomainValidationError("ACCOUNTS_NOT_ALLOWED", "Для режима ALL не нужно указывать счета")
|
||||
debit = frozenset(account.id for account in accounts if account.kind in DEBIT_KINDS)
|
||||
credit = frozenset(account.id for account in accounts if account.kind in CREDIT_KINDS)
|
||||
selection = MatrixSelection(debit, credit, "closing")
|
||||
elif mode == "MANUAL":
|
||||
if not debit_account_ids or not credit_account_ids:
|
||||
raise DomainValidationError("ACCOUNTS_REQUIRED", "Выберите дебетовые и кредитовые счета")
|
||||
debit = _selected_accounts(debit_account_ids, (), "D", by_id, children)
|
||||
credit = _selected_accounts(credit_account_ids, (), "K", by_id, children)
|
||||
debit = _select_accounts(debit_account_ids, (), "D", by_id, children)
|
||||
credit = _select_accounts(credit_account_ids, (), "K", by_id, children)
|
||||
selection = MatrixSelection(debit, credit, "closing")
|
||||
elif mode == "PRESET":
|
||||
if rule_id is None:
|
||||
raise DomainValidationError("RULE_REQUIRED", "Выберите правило-пресет")
|
||||
rule, rule_accounts = await elimination_rules_repo.rule_with_account(self._session, rule_id)
|
||||
as_of = None if version.elimination_rules_version is not None else moscow_date(utc_now())
|
||||
version_no = (
|
||||
version.elimination_rules_version
|
||||
if as_of is None
|
||||
else await elimination_rules_repo.current_version_no(self._session, as_of)
|
||||
)
|
||||
if rule is None or not elimination_rules_repo.is_valid_for_version(rule, version_no, as_of=as_of):
|
||||
raise NotFoundError("правило-пресет не найдено для версии консолидации")
|
||||
if rule.side_1 not in ("D", "K") or rule.measure not in ("closing", "turnover"):
|
||||
raise DomainValidationError("INVALID_RULE", "Неверное направление или показатель правила-пресета")
|
||||
|
||||
sides: dict[int, tuple[list[uuid.UUID], list[uuid.UUID]]] = {
|
||||
1: ([], []),
|
||||
2: ([], []),
|
||||
}
|
||||
for rule_account in rule_accounts:
|
||||
if rule_account.company_no not in sides:
|
||||
raise DomainValidationError("INVALID_RULE", "Неверная сторона счёта в правиле-пресете")
|
||||
included, excluded = sides[rule_account.company_no]
|
||||
(excluded if rule_account.is_excluded else included).append(rule_account.account_id)
|
||||
debit_side = 1 if rule.side_1 == "D" else 2
|
||||
credit_side = 2 if debit_side == 1 else 1
|
||||
debit_ids, debit_excluded = sides[debit_side]
|
||||
credit_ids, credit_excluded = sides[credit_side]
|
||||
if not debit_ids or not credit_ids:
|
||||
raise DomainValidationError("ACCOUNTS_REQUIRED", "В правиле-пресете нужны счета обеих сторон")
|
||||
debit = _selected_accounts(debit_ids, debit_excluded, "D", by_id, children)
|
||||
credit = _selected_accounts(credit_ids, credit_excluded, "K", by_id, children)
|
||||
measure = "closing" if rule.measure == "closing" else "turnover"
|
||||
selection, rule = await self._resolve_preset_selection(rule_id, version, by_id, children)
|
||||
else:
|
||||
raise DomainValidationError("INVALID_MODE", "Неизвестный режим выбора счетов")
|
||||
selection = MatrixSelection(debit, credit, measure)
|
||||
matrix = self._calculate(rows, members, selection)
|
||||
if rule is not None:
|
||||
matrix = await self._apply_corrections(matrix, version_id, rule.id)
|
||||
return matrix, version, consolidation, accounts, selection, rows
|
||||
return MatrixBuildResult(
|
||||
matrix=matrix,
|
||||
version=version,
|
||||
consolidation=consolidation,
|
||||
accounts=accounts,
|
||||
selection=selection,
|
||||
rows=rows,
|
||||
)
|
||||
|
||||
@ -1,6 +1,5 @@
|
||||
"""Выбор версии правил для шахматки без обращения к БД."""
|
||||
|
||||
import datetime
|
||||
import uuid
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import AsyncMock
|
||||
@ -24,39 +23,35 @@ def _rule() -> SimpleNamespace:
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("rule", "version_no", "as_of", "expected"),
|
||||
("rule", "version_no", "expected"),
|
||||
[
|
||||
(None, 3, None, False),
|
||||
(SimpleNamespace(version_no=3), None, None, False),
|
||||
(SimpleNamespace(version_no=3), 2, None, False),
|
||||
(SimpleNamespace(version_no=3), 3, None, True),
|
||||
(
|
||||
SimpleNamespace(version_no=3, valid_from=datetime.date(2024, 1, 1), valid_to=datetime.date(2024, 12, 31)),
|
||||
3,
|
||||
datetime.date(2023, 12, 31),
|
||||
False,
|
||||
),
|
||||
(
|
||||
SimpleNamespace(version_no=3, valid_from=datetime.date(2024, 1, 1), valid_to=datetime.date(2024, 12, 31)),
|
||||
3,
|
||||
datetime.date(2024, 12, 31),
|
||||
True,
|
||||
),
|
||||
(
|
||||
SimpleNamespace(version_no=3, valid_from=datetime.date(2024, 1, 1), valid_to=datetime.date(2024, 12, 31)),
|
||||
3,
|
||||
datetime.date(2025, 1, 1),
|
||||
False,
|
||||
),
|
||||
(None, 3, False),
|
||||
(SimpleNamespace(version_no=3, valid_to=None), None, True),
|
||||
(SimpleNamespace(version_no=3, valid_to="2026-12-31"), None, False),
|
||||
(SimpleNamespace(version_no=3), 2, False),
|
||||
(SimpleNamespace(version_no=3), 3, True),
|
||||
(SimpleNamespace(version_no=3, valid_to="2026-12-31"), 3, True),
|
||||
],
|
||||
)
|
||||
def test_rule_is_valid_for_version(
|
||||
rule: SimpleNamespace | None, version_no: int | None, as_of: datetime.date | None, expected: bool
|
||||
rule: SimpleNamespace | None, version_no: int | None, expected: bool
|
||||
) -> None:
|
||||
assert rules_repo.is_valid_for_version(rule, version_no, as_of=as_of) is expected
|
||||
assert rules_repo.is_valid_for_version(rule, version_no) is expected
|
||||
|
||||
|
||||
async def test_fixed_rules_version_ignores_current_date(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
async def test_rules_filter_by_version_or_open_valid_to() -> None:
|
||||
version_no = 3
|
||||
session = SimpleNamespace(execute=AsyncMock(return_value=SimpleNamespace(all=lambda: [])))
|
||||
|
||||
await rules_repo.list_rules_with_accounts(session, None)
|
||||
assert "valid_to IS NULL" in str(session.execute.await_args.args[0])
|
||||
|
||||
await rules_repo.list_rules_with_accounts(session, version_no)
|
||||
assert "valid_to IS NULL" not in str(session.execute.await_args.args[0])
|
||||
assert "version_no =" in str(session.execute.await_args.args[0])
|
||||
|
||||
|
||||
async def test_fixed_rules_use_version_number(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
session, scope = object(), object()
|
||||
version_id = uuid.uuid4()
|
||||
rule = _rule()
|
||||
@ -67,9 +62,7 @@ async def test_fixed_rules_version_ignores_current_date(monkeypatch: pytest.Monk
|
||||
"get_version",
|
||||
AsyncMock(return_value=(SimpleNamespace(elimination_rules_version=rule.version_no), None)),
|
||||
)
|
||||
current = AsyncMock()
|
||||
listed = AsyncMock(return_value=[(rule, account)])
|
||||
monkeypatch.setattr(service_module.rules_repo, "current_version_no", current)
|
||||
monkeypatch.setattr(service_module.rules_repo, "list_rules_with_accounts", listed)
|
||||
|
||||
result = await service_module.EliminationRuleService(session, scope).list_rules(version_id)
|
||||
@ -79,34 +72,28 @@ async def test_fixed_rules_version_ignores_current_date(monkeypatch: pytest.Monk
|
||||
assert result.as_of is None
|
||||
assert [item.id for item in result.rules] == [rule.id]
|
||||
assert result.rules[0].company_1.accounts == [account_id]
|
||||
current.assert_not_awaited()
|
||||
listed.assert_awaited_once_with(session, rule.version_no, as_of=None)
|
||||
listed.assert_awaited_once_with(session, rule.version_no)
|
||||
|
||||
|
||||
async def test_unfixed_rules_use_latest_active_version_on_moscow_date(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
async def test_unfixed_rules_use_open_valid_to(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
session, scope = object(), object()
|
||||
version_id = uuid.uuid4()
|
||||
rule = _rule()
|
||||
expected_date = datetime.date(2026, 10, 19)
|
||||
monkeypatch.setattr(service_module, "utc_now", lambda: datetime.datetime(2026, 10, 18, 22, tzinfo=datetime.UTC))
|
||||
monkeypatch.setattr(
|
||||
service_module.consolidations_repo,
|
||||
"get_version",
|
||||
AsyncMock(return_value=(SimpleNamespace(elimination_rules_version=None), None)),
|
||||
)
|
||||
current = AsyncMock(return_value=rule.version_no)
|
||||
listed = AsyncMock(return_value=[(rule, None)])
|
||||
monkeypatch.setattr(service_module.rules_repo, "current_version_no", current)
|
||||
monkeypatch.setattr(service_module.rules_repo, "list_rules_with_accounts", listed)
|
||||
|
||||
result = await service_module.EliminationRuleService(session, scope).list_rules(version_id)
|
||||
|
||||
assert result.is_version_fixed is False
|
||||
assert result.version_no == rule.version_no
|
||||
assert result.as_of == expected_date
|
||||
assert result.version_no is None
|
||||
assert result.as_of is None
|
||||
assert [item.id for item in result.rules] == [rule.id]
|
||||
current.assert_awaited_once_with(session, expected_date)
|
||||
listed.assert_awaited_once_with(session, rule.version_no, as_of=expected_date)
|
||||
listed.assert_awaited_once_with(session, None)
|
||||
|
||||
|
||||
async def test_no_current_rules_returns_empty_with_unfixed_flag(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
@ -116,14 +103,13 @@ async def test_no_current_rules_returns_empty_with_unfixed_flag(monkeypatch: pyt
|
||||
"get_version",
|
||||
AsyncMock(return_value=(SimpleNamespace(elimination_rules_version=None), None)),
|
||||
)
|
||||
monkeypatch.setattr(service_module.rules_repo, "current_version_no", AsyncMock(return_value=None))
|
||||
listed = AsyncMock()
|
||||
listed = AsyncMock(return_value=[])
|
||||
monkeypatch.setattr(service_module.rules_repo, "list_rules_with_accounts", listed)
|
||||
|
||||
result = await service_module.EliminationRuleService(session, scope).list_rules(uuid.uuid4())
|
||||
|
||||
assert result.is_version_fixed is False
|
||||
assert result.version_no is None
|
||||
assert result.as_of is not None
|
||||
assert result.as_of is None
|
||||
assert result.rules == []
|
||||
listed.assert_not_awaited()
|
||||
listed.assert_awaited_once_with(session, None)
|
||||
|
||||
@ -58,9 +58,9 @@ def test_routes_are_separate_and_matrix_expands_manual_accounts() -> None:
|
||||
debit_child = _account(AccountKind.A, debit_parent.id)
|
||||
credit_parent = _account(AccountKind.P)
|
||||
credit_child = _account(AccountKind.P, credit_parent.id)
|
||||
by_id, children = matrix._account_tree([debit_parent, debit_child, credit_parent, credit_child])
|
||||
debit = matrix._selected_accounts([debit_parent.id], [], "D", by_id, children)
|
||||
credit = matrix._selected_accounts([credit_parent.id], [], "K", by_id, children)
|
||||
by_id, children = matrix._make_account_tree([debit_parent, debit_child, credit_parent, credit_child])
|
||||
debit = matrix._select_accounts([debit_parent.id], [], "D", by_id, children)
|
||||
credit = matrix._select_accounts([credit_parent.id], [], "K", by_id, children)
|
||||
assert debit == {debit_parent.id, debit_child.id}
|
||||
assert credit == {credit_parent.id, credit_child.id}
|
||||
|
||||
@ -109,7 +109,7 @@ def test_unpaired_rows_uses_source_presence_and_preserves_osv_order() -> None:
|
||||
rows, companies, matrix.MatrixSelection(frozenset({debit_account}), frozenset({credit_account}), "closing")
|
||||
)
|
||||
|
||||
assert matrix.unpaired_rows(rows, result) == [unpaired_credit]
|
||||
assert matrix.get_unpaired_rows(rows, result) == [unpaired_credit]
|
||||
|
||||
|
||||
async def test_preset_applies_exclusions_to_its_own_side_and_rule_measure(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
@ -152,15 +152,12 @@ async def test_preset_applies_exclusions_to_its_own_side_and_rule_measure(monkey
|
||||
assert captured[0].company_2_account_ids == {parent.id, child.id}
|
||||
assert captured[0].measure == "turnover"
|
||||
|
||||
current_version = AsyncMock(return_value=1)
|
||||
monkeypatch.setattr(matrix.elimination_rules_repo, "current_version_no", current_version)
|
||||
version.elimination_rules_version = None
|
||||
_, info = await service.build_export(uuid.uuid4(), mode="PRESET", rule_id=uuid.uuid4())
|
||||
assert info.mode == "PRESET"
|
||||
assert info.debit_account_codes == ("76",)
|
||||
assert info.credit_account_codes == ("76", "76.01")
|
||||
assert info.measure == "turnover"
|
||||
current_version.assert_awaited_once()
|
||||
|
||||
|
||||
async def test_preset_corrections_exclude_amounts_without_changing_osv(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user