import io import urllib.error import core import margins from margins import income_direction, expense_direction, DIRECTIONS, split_lump, merge_direct # ── ftmut: ретраи безопасны только для идемпотентных методов ── # POST (создание операции) не ретраится, иначе потерянный ответ = дубль в живых деньгах. class _Resp: def __init__(self, status=200, body=b'{"id": 7}'): self.status = status self._body = body def read(self, *a): return self._body def _http_error(code): return urllib.error.HTTPError('http://x', code, 'err', {}, io.BytesIO(b'')) def _patch(monkeypatch, fake): # клиент переехал в core.py; margins.ftmut — реэкспорт того же объекта monkeypatch.setattr(core.urllib.request, 'urlopen', fake) monkeypatch.setattr(core.time, 'sleep', lambda *a, **k: None) def test_ftmut_post_success_single_call(monkeypatch): calls = [] def fake(req, timeout=None): calls.append(1); return _Resp(body=b'{"id": 7}') _patch(monkeypatch, fake) assert margins.ftmut('POST', '/v1/transaction', {'value': 1}) == {'id': 7} assert len(calls) == 1 def test_ftmut_post_no_retry_on_5xx(monkeypatch): calls = [] def fake(req, timeout=None): calls.append(1); raise _http_error(503) _patch(monkeypatch, fake) try: margins.ftmut('POST', '/v1/transaction', {'value': 1}) assert False, "POST должен был бросить, не проглотить" except urllib.error.HTTPError: pass assert len(calls) == 1 # ключевое: POST не повторяется def test_ftmut_post_raises_on_4xx(monkeypatch): _patch(monkeypatch, lambda req, timeout=None: (_ for _ in ()).throw(_http_error(400))) try: margins.ftmut('POST', '/x', {}); assert False except urllib.error.HTTPError as e: assert e.code == 400 def test_ftmut_delete_retries_on_5xx(monkeypatch): calls = [] def fake(req, timeout=None): calls.append(1) if len(calls) < 3: raise _http_error(503) return _Resp(status=204) _patch(monkeypatch, fake) assert margins.ftmut('DELETE', '/v1/transaction/1') == 204 assert len(calls) == 3 # DELETE идемпотентен — ретраится def test_ftmut_delete_404_on_retry_is_success(monkeypatch): calls = [] def fake(req, timeout=None): calls.append(1) raise _http_error(503) if len(calls) == 1 else _http_error(404) _patch(monkeypatch, fake) # первый DELETE мог удалить, ответ потерялся; повтор ловит 404 → уже удалено → успех assert margins.ftmut('DELETE', '/v1/transaction/1') == 404 assert len(calls) == 2 def test_directions_order(): assert DIRECTIONS == ["1-1", "Групповые", "Хор", "B2B", "Внутренние мероприятия", "Съёмки", "Аренда"] def test_income_direction_basic(): assert income_direction("Абонемент оффлайн") == "1-1" assert income_direction("Подарочный сертификат") == "1-1" assert income_direction("Парный абонемент") == "1-1" assert income_direction("Фитмост разовое") == "1-1" assert income_direction("Групповой абонемент") == "Групповые" assert income_direction("Хор (курс)") == "Хор" assert income_direction("Квартирник") == "Внутренние мероприятия" assert income_direction("Cover session") == "Съёмки" assert income_direction("Аренда") == "Аренда" def test_income_direction_b2b_fallback(): # b2b-статей в CRM много и они меняются — ловим по подстроке, а не списком assert income_direction("B2B Абонемент") == "B2B" assert income_direction("b2b") == "B2B" def test_income_direction_excluded(): assert income_direction("Мерч") is None assert income_direction("Мероприятия внешние на привлечение") is None def test_expense_direction(): assert expense_direction("Расходы на хоровые курсы") == "Хор" assert expense_direction("Расходы на съёмки") == "Съёмки" assert expense_direction("Расходы на съемки ") == "Съёмки" # без ё и с пробелом assert expense_direction("Расходы на внутренние ивенты") == "Внутренние мероприятия" assert expense_direction("Расходы на b2b-ивенты") == "B2B" assert expense_direction("Зарплаты педагогов: Расходы на занятия 1-1") == "__lump__" assert expense_direction("Аренда и содержание здания (ком.услуги)") == "overhead" assert expense_direction("Зарплаты бэк-офис") == "overhead" assert expense_direction("что угодно неизвестное") == "overhead" def test_split_lump_basic(): # 10 индив (1200) = 12000 вес → 1-1; 1 групповой (2700) = 2700 вес → Групповые counts = {"Индивидуальный": 10, "Групповой": 1} r = split_lump(14700.0, counts) assert round(r["1-1"], 2) == round(14700 * 12000/14700, 2) assert round(r["Групповые"], 2) == round(14700 * 2700/14700, 2) assert round(r["1-1"] + r["Групповые"] + r["B2B"], 2) == 14700.0 def test_split_lump_b2b_bucket(): counts = {"B2B Индивидуальный": 5} # 5*1200 = вес только в B2B r = split_lump(6000.0, counts) assert round(r["B2B"], 2) == 6000.0 assert r["1-1"] == 0 and r["Групповые"] == 0 def test_split_lump_ignores_events(): # Квартирник/Концерт не участвуют в разбивке ЗП counts = {"Индивидуальный": 1, "Квартирник": 99, "Концерт": 99} r = split_lump(1200.0, counts) assert round(r["1-1"], 2) == 1200.0 def test_split_lump_no_lessons(): # нет уроков (сбой CRM / пустой месяц) → НЕ сваливаем всю ЗП на «1-1», отдаём нули; # неразнесённый остаток month_pnl кладёт в overhead assert split_lump(5000.0, {}) == {"1-1": 0.0, "Групповые": 0.0, "B2B": 0.0} def test_merge_direct(): expenses = {"1-1": 0.0, "Групповые": 0.0, "Хор": 200.0, "B2B": 50.0, "Внутренние мероприятия": 100.0, "Съёмки": 30.0, "Аренда": 0.0, "overhead": 999.0, "__lump__": 3000.0} salary = {"1-1": 2000.0, "Групповые": 500.0, "B2B": 500.0} # сумма == лампе direct = merge_direct(expenses, salary) assert direct["1-1"] == 2000.0 assert direct["Групповые"] == 500.0 assert direct["Хор"] == 200.0 assert direct["Съёмки"] == 30.0 # B2B = прямые b2b-ивенты (50) + доля лампы (500) assert direct["B2B"] == 550.0 assert direct["Внутренние мероприятия"] == 100.0 assert direct["Аренда"] == 0.0 assert "overhead" not in direct and "__lump__" not in direct assert set(direct.keys()) == set(DIRECTIONS)