80 lines
3.2 KiB
Python
80 lines
3.2 KiB
Python
from types import SimpleNamespace
|
||
from unittest.mock import AsyncMock
|
||
|
||
import pytest
|
||
|
||
from src.services.summary_service import EMPTY_SVOD2_FACETS, SummaryService
|
||
|
||
ROW = ("INPUT", 3, 1, {"line_id": 1})
|
||
|
||
|
||
def make_service(monkeypatch, allowed):
|
||
service = SummaryService.__new__(SummaryService)
|
||
service.summary_repo = AsyncMock()
|
||
service.user_repo = AsyncMock()
|
||
monkeypatch.setattr(service, "_allowed_org_ids", AsyncMock(return_value=allowed))
|
||
return service
|
||
|
||
|
||
USER = SimpleNamespace(id=1)
|
||
|
||
|
||
async def test_page_passes_params_and_compact_flag(monkeypatch):
|
||
service = make_service(monkeypatch, [1, 2])
|
||
service.summary_repo.get_svod2_page.return_value = ([ROW], 7)
|
||
result = await service.get_svod2_page(
|
||
2026, "GO", USER, org_ids=[1, 2], offset=0, limit=50,
|
||
sort_by="name", sort_direction="desc", search="x",
|
||
smeta_type="КВ", smeta_direction="Support", org_name="Орг", view="compact",
|
||
)
|
||
assert result == ([ROW], 7)
|
||
service.summary_repo.get_svod2_page.assert_awaited_once_with(
|
||
year=2026, sheet="GO", org_ids=[1, 2], offset=0, limit=50,
|
||
sort_by="name", sort_direction="desc", search="x",
|
||
smeta_type="КВ", smeta_direction="Support", org_name="Орг", compact=True,
|
||
)
|
||
|
||
|
||
async def test_page_empty_intersection_skips_sql(monkeypatch):
|
||
service = make_service(monkeypatch, [])
|
||
assert await service.get_svod2_page(2026, "GO", USER, offset=0, limit=10) == ([], 0)
|
||
service.summary_repo.get_svod2_page.assert_not_awaited()
|
||
|
||
|
||
async def test_page_past_end_second_call_for_total(monkeypatch):
|
||
service = make_service(monkeypatch, None)
|
||
service.summary_repo.get_svod2_page.side_effect = [([], 0), ([ROW], 42)]
|
||
result = await service.get_svod2_page(2026, "GO", USER, offset=600, limit=50)
|
||
assert result == ([], 42)
|
||
second = service.summary_repo.get_svod2_page.await_args_list[1].kwargs
|
||
assert second["offset"] == 0 and second["limit"] == 1
|
||
|
||
|
||
async def test_page_empty_first_page_no_second_call(monkeypatch):
|
||
service = make_service(monkeypatch, None)
|
||
service.summary_repo.get_svod2_page.return_value = ([], 0)
|
||
assert await service.get_svod2_page(2026, "GO", USER, offset=0, limit=50) == ([], 0)
|
||
assert service.summary_repo.get_svod2_page.await_count == 1
|
||
|
||
|
||
async def test_facets_empty_intersection_zeros_without_sql(monkeypatch):
|
||
service = make_service(monkeypatch, [])
|
||
result = await service.get_svod2_facets(2026, "GO", USER, org_ids=[5])
|
||
assert result == EMPTY_SVOD2_FACETS
|
||
assert result is not EMPTY_SVOD2_FACETS
|
||
service.summary_repo.get_svod2_facets.assert_not_awaited()
|
||
|
||
|
||
async def test_facets_passes_allowed_org_ids_and_filters(monkeypatch):
|
||
service = make_service(monkeypatch, [1])
|
||
service.summary_repo.get_svod2_facets.return_value = {"total": 3}
|
||
result = await service.get_svod2_facets(
|
||
2026, "RF", USER, org_ids=[1, 9], search="x", smeta_type="КВ",
|
||
smeta_direction="Support", org_name="Орг",
|
||
)
|
||
assert result == {"total": 3}
|
||
service.summary_repo.get_svod2_facets.assert_awaited_once_with(
|
||
year=2026, sheet="RF", org_ids=[1], search="x", smeta_type="КВ",
|
||
smeta_direction="Support", org_name="Орг",
|
||
)
|