107 lines
3.8 KiB
Python
107 lines
3.8 KiB
Python
# ruff: noqa: PLR2004
|
|
|
|
import hashlib
|
|
from datetime import UTC, datetime, timedelta
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
import pytest
|
|
|
|
from external.pally import BillStatus
|
|
from schemas.plans import AddonData, PricingPlans
|
|
from services.payments import validate_pally_signature
|
|
from services.subscriptions import calculate_order_total, should_apply_immediately
|
|
|
|
|
|
def pricing():
|
|
return PricingPlans(
|
|
device_price=100,
|
|
addons=[
|
|
AddonData(id="a", name="A", price=25, free_threshold=0, is_enabled=True),
|
|
AddonData(id="b", name="B", price=50, free_threshold=0, is_enabled=True),
|
|
],
|
|
)
|
|
|
|
|
|
def test_price_calculation_handles_fractional_months_and_unknown_addons():
|
|
assert calculate_order_total(pricing(), devices=2, addon_ids=["a"], duration_days=15) == 112.5
|
|
|
|
with pytest.raises(KeyError):
|
|
calculate_order_total(pricing(), devices=1, addon_ids=["missing"], duration_days=30)
|
|
|
|
|
|
def test_should_apply_immediately_distinguishes_upgrade_from_downgrade():
|
|
now = datetime.now(UTC)
|
|
subscription = SimpleNamespace(
|
|
devices=2,
|
|
expires_at=now + timedelta(days=10),
|
|
addons=[SimpleNamespace(addon_id="a")],
|
|
)
|
|
upgrade = SimpleNamespace(
|
|
devices=3, addons=[SimpleNamespace(addon_id="a"), SimpleNamespace(addon_id="b")]
|
|
)
|
|
downgrade = SimpleNamespace(devices=1, addons=[SimpleNamespace(addon_id="a")])
|
|
|
|
assert should_apply_immediately(
|
|
order=upgrade, subscription=subscription, pricing=pricing(), now=now
|
|
)
|
|
assert not should_apply_immediately(
|
|
order=downgrade, subscription=subscription, pricing=pricing(), now=now
|
|
)
|
|
assert should_apply_immediately(order=downgrade, subscription=None, pricing=pricing(), now=now)
|
|
|
|
|
|
def test_pally_signature_is_exact_and_case_sensitive():
|
|
signature = hashlib.md5(b"10:42:test-token").hexdigest().upper()
|
|
with patch("services.payments.cfg.pally_token", "test-token"):
|
|
assert validate_pally_signature("10", "42", signature)
|
|
assert not validate_pally_signature("10", "42", signature.lower())
|
|
assert not validate_pally_signature("11", "42", signature)
|
|
|
|
|
|
def callback_data(**overrides):
|
|
data = {
|
|
"InvId": "42",
|
|
"OutSum": "10",
|
|
"Commission": "0",
|
|
"TrsId": "transaction",
|
|
"Status": BillStatus.SUCCESS,
|
|
"CurrencyIn": "RUB",
|
|
"SignatureValue": "valid",
|
|
}
|
|
data.update(overrides)
|
|
return data
|
|
|
|
|
|
def test_callback_rejects_invalid_signature_before_processing(client):
|
|
with patch("routes.payments.pally.validate_pally_signature", return_value=False):
|
|
response = client.post("/payments/pally/result", data=callback_data())
|
|
|
|
assert response.status_code == 403
|
|
assert response.json()["detail"] == "Invalid signature."
|
|
|
|
|
|
def test_callback_acknowledges_valid_non_success_and_non_numeric_invoice(client):
|
|
with patch("routes.payments.pally.validate_pally_signature", return_value=True):
|
|
failed_response = client.post("/payments/pally/result", data=callback_data(Status="Failed"))
|
|
invalid_id_response = client.post(
|
|
"/payments/pally/result", data=callback_data(InvId="order-42")
|
|
)
|
|
|
|
assert failed_response.status_code == invalid_id_response.status_code == 200
|
|
assert failed_response.json() == invalid_id_response.json() == "OK"
|
|
|
|
|
|
def test_callback_uses_balance_amount_when_present(client):
|
|
process = AsyncMock()
|
|
with (
|
|
patch("routes.payments.pally.validate_pally_signature", return_value=True),
|
|
patch("routes.payments.pally.process_subscription_purchase", process),
|
|
):
|
|
response = client.post("/payments/pally/result", data=callback_data(BalanceAmount="7.9"))
|
|
|
|
assert response.status_code == 200
|
|
process.assert_awaited_once()
|
|
assert process.await_args.kwargs["invoice_id"] == 42
|
|
assert process.await_args.kwargs["amount"] == 7
|