From 88369eec5047e733d26d2450a74abd0d600b2e1b Mon Sep 17 00:00:00 2001 From: Fringg Date: Thu, 12 Mar 2026 22:20:48 +0300 Subject: [PATCH] feat: add _calculate_servers_price (fixed fallback) and _calculate_traffic_price _calculate_servers_price ALWAYS uses real server.price_kopeks even when is_available=False or is_full=True, fixing the silent zero-price bug. _calculate_traffic_price separates base from purchased GB to prevent purchased top-ups from inflating the tier lookup. --- app/services/pricing_engine.py | 81 +++++++++++++++++++++++++++++ tests/test_pricing_engine.py | 93 ++++++++++++++++++++++++++++++++++ 2 files changed, 174 insertions(+) diff --git a/app/services/pricing_engine.py b/app/services/pricing_engine.py index a8b1bb1a..e00343e8 100644 --- a/app/services/pricing_engine.py +++ b/app/services/pricing_engine.py @@ -3,6 +3,9 @@ from __future__ import annotations import structlog from dataclasses import dataclass, field +from app.config import settings +from app.database.crud.server_squad import get_server_squad_by_uuid + logger = structlog.get_logger() @@ -46,3 +49,81 @@ class PricingEngine: after_offer = PricingEngine.apply_discount(after_group, offer_percent) offer_discount_value = after_group - after_offer return after_offer, group_discount_value, offer_discount_value + + async def _calculate_servers_price( + self, + country_uuids: list[str], + db, # AsyncSession + *, + promo_group_id: int | None = None, + ) -> tuple[int, list[dict]]: + """Calculate total server price from connected squad UUIDs. + + Unlike the old implementation, ALWAYS uses real price_kopeks + even when server is unavailable or full. Only orphaned UUIDs + (not found in DB) get price=0. + """ + total_price = 0 + details: list[dict] = [] + + for uuid in country_uuids: + try: + server = await get_server_squad_by_uuid(db, uuid) + except Exception as e: + logger.error("Ошибка загрузки сервера", squad_uuid=uuid, error=str(e)) + details.append({"uuid": uuid, "price": 0, "status": "error"}) + continue + + if server is None: + logger.error("Сервер не найден в БД", squad_uuid=uuid) + details.append({"uuid": uuid, "price": 0, "status": "not_found"}) + continue + + price = server.price_kopeks or 0 + status = "available" + + if not server.is_available: + status = "unavailable" + logger.warning( + "Сервер недоступен, используем реальную цену", + squad_uuid=uuid, + price_kopeks=price, + ) + elif server.is_full: + status = "full" + logger.warning( + "Сервер переполнен, используем реальную цену", + squad_uuid=uuid, + price_kopeks=price, + ) + elif promo_group_id is not None: + allowed_ids = [pg.id for pg in (server.allowed_promo_groups or [])] + if allowed_ids and promo_group_id not in allowed_ids: + status = "not_allowed" + logger.warning( + "Сервер недоступен для промогруппы, используем реальную цену", + squad_uuid=uuid, + promo_group_id=promo_group_id, + price_kopeks=price, + ) + + total_price += price + details.append({"uuid": uuid, "price": price, "status": status}) + + return total_price, details + + def _calculate_traffic_price( + self, + traffic_limit_gb: int, + purchased_traffic_gb: int, + ) -> int: + """Calculate traffic price, separating base from purchased GB. + Prevents purchased top-ups from inflating the tier lookup.""" + total_gb = traffic_limit_gb or 0 + purchased_gb = purchased_traffic_gb or 0 + base_gb = max(0, total_gb - purchased_gb) + + base_price = settings.get_traffic_price(base_gb) if base_gb > 0 else 0 + purchased_price = settings.get_traffic_price(purchased_gb) if purchased_gb > 0 else 0 + + return base_price + purchased_price diff --git a/tests/test_pricing_engine.py b/tests/test_pricing_engine.py index 59d8fd5b..fda62c3e 100644 --- a/tests/test_pricing_engine.py +++ b/tests/test_pricing_engine.py @@ -57,3 +57,96 @@ class TestStackedDiscounts: assert final == 8500 assert g_val == 0 assert o_val == 1500 + + +from unittest.mock import AsyncMock, MagicMock, patch + + +def _make_server(price_kopeks=5000, is_available=True, is_full=False, allowed_promo_groups=None): + server = MagicMock() + server.price_kopeks = price_kopeks + server.is_available = is_available + server.is_full = is_full + server.allowed_promo_groups = allowed_promo_groups or [] + return server + + +class TestCalculateServersPrice: + @pytest.mark.asyncio + async def test_available_server(self): + engine = PricingEngine() + db = AsyncMock() + server = _make_server(price_kopeks=5000) + with patch("app.services.pricing_engine.get_server_squad_by_uuid", return_value=server): + total, details = await engine._calculate_servers_price(["uuid-1"], db, promo_group_id=None) + assert total == 5000 + assert len(details) == 1 + assert details[0]["price"] == 5000 + + @pytest.mark.asyncio + async def test_unavailable_server_uses_real_price(self): + engine = PricingEngine() + db = AsyncMock() + server = _make_server(price_kopeks=7000, is_available=False) + with patch("app.services.pricing_engine.get_server_squad_by_uuid", return_value=server): + total, details = await engine._calculate_servers_price(["uuid-1"], db, promo_group_id=None) + assert total == 7000 # NOT 0! + assert details[0]["status"] == "unavailable" + + @pytest.mark.asyncio + async def test_full_server_uses_real_price(self): + engine = PricingEngine() + db = AsyncMock() + server = _make_server(price_kopeks=3000, is_full=True) + with patch("app.services.pricing_engine.get_server_squad_by_uuid", return_value=server): + total, details = await engine._calculate_servers_price(["uuid-1"], db, promo_group_id=None) + assert total == 3000 # NOT 0! + + @pytest.mark.asyncio + async def test_server_not_found_zero_price(self): + engine = PricingEngine() + db = AsyncMock() + with patch("app.services.pricing_engine.get_server_squad_by_uuid", return_value=None): + total, details = await engine._calculate_servers_price(["uuid-orphan"], db, promo_group_id=None) + assert total == 0 + assert details[0]["status"] == "not_found" + + @pytest.mark.asyncio + async def test_multiple_servers(self): + engine = PricingEngine() + db = AsyncMock() + s1 = _make_server(price_kopeks=5000) + s2 = _make_server(price_kopeks=3000, is_available=False) + with patch("app.services.pricing_engine.get_server_squad_by_uuid", side_effect=[s1, s2]): + total, details = await engine._calculate_servers_price(["uuid-1", "uuid-2"], db, promo_group_id=None) + assert total == 8000 + + +class TestCalculateTrafficPrice: + def test_base_only(self): + engine = PricingEngine() + with patch("app.services.pricing_engine.settings") as ms: + ms.get_traffic_price.side_effect = lambda gb: {25: 3000, 50: 5000}.get(gb, 0) + price = engine._calculate_traffic_price(traffic_limit_gb=25, purchased_traffic_gb=0) + assert price == 3000 + + def test_purchased_separated(self): + engine = PricingEngine() + with patch("app.services.pricing_engine.settings") as ms: + ms.get_traffic_price.side_effect = lambda gb: {25: 3000, 100: 8000, 125: 12000}.get(gb, 0) + price = engine._calculate_traffic_price(traffic_limit_gb=125, purchased_traffic_gb=100) + assert price == 11000 # NOT 12000 + + def test_zero_traffic(self): + engine = PricingEngine() + with patch("app.services.pricing_engine.settings") as ms: + ms.get_traffic_price.return_value = 0 + price = engine._calculate_traffic_price(traffic_limit_gb=0, purchased_traffic_gb=0) + assert price == 0 + + def test_purchased_exceeds_total(self): + engine = PricingEngine() + with patch("app.services.pricing_engine.settings") as ms: + ms.get_traffic_price.side_effect = lambda gb: {0: 0, 100: 8000}.get(gb, 0) + price = engine._calculate_traffic_price(traffic_limit_gb=80, purchased_traffic_gb=100) + assert price == 8000 # base_gb clamped to 0