mirror of
https://github.com/BEDOLAGA-DEV/remnawave-bedolaga-telegram-bot.git
synced 2026-02-23 21:01:17 +00:00
9a2aea038a
- Add pyproject.toml with uv and ruff configuration - Pin Python version to 3.13 via .python-version - Add Makefile commands: lint, format, fix - Apply ruff formatting to entire codebase - Remove unused imports (base64 in yookassa/simple_subscription) - Update .gitignore for new config files
99 lines
2.7 KiB
Python
99 lines
2.7 KiB
Python
"""Unit tests for the WATA webhook handler."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from cryptography.hazmat.primitives import hashes, serialization
|
|
from cryptography.hazmat.primitives.asymmetric import padding, rsa
|
|
|
|
|
|
ROOT_DIR = Path(__file__).resolve().parents[2]
|
|
if str(ROOT_DIR) not in sys.path:
|
|
sys.path.insert(0, str(ROOT_DIR))
|
|
|
|
from app.external.wata_webhook import WataWebhookHandler
|
|
|
|
|
|
class DummyPaymentService:
|
|
async def process_wata_webhook(self, *args, **kwargs): # pragma: no cover - not used in tests
|
|
return True
|
|
|
|
|
|
class StubPublicKeyProvider:
|
|
def __init__(self, public_key_pem: str | None) -> None:
|
|
self.public_key_pem = public_key_pem
|
|
|
|
async def get_public_key(self) -> str | None:
|
|
return self.public_key_pem
|
|
|
|
|
|
@pytest.fixture
|
|
def anyio_backend() -> str:
|
|
return 'asyncio'
|
|
|
|
|
|
@pytest.mark.anyio('asyncio')
|
|
async def test_verify_signature_success() -> None:
|
|
private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
|
|
public_key = (
|
|
private_key.public_key()
|
|
.public_bytes(
|
|
encoding=serialization.Encoding.PEM,
|
|
format=serialization.PublicFormat.SubjectPublicKeyInfo,
|
|
)
|
|
.decode('utf-8')
|
|
)
|
|
|
|
payload = '{"status": "Paid"}'
|
|
signature = base64.b64encode(
|
|
private_key.sign(
|
|
payload.encode('utf-8'),
|
|
padding.PKCS1v15(),
|
|
hashes.SHA512(),
|
|
)
|
|
).decode('utf-8')
|
|
|
|
handler = WataWebhookHandler(
|
|
DummyPaymentService(),
|
|
public_key_provider=StubPublicKeyProvider(public_key),
|
|
)
|
|
|
|
assert await handler._verify_signature(payload, signature) is True
|
|
|
|
|
|
@pytest.mark.anyio('asyncio')
|
|
async def test_verify_signature_fails_with_invalid_signature() -> None:
|
|
private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
|
|
public_key = (
|
|
private_key.public_key()
|
|
.public_bytes(
|
|
encoding=serialization.Encoding.PEM,
|
|
format=serialization.PublicFormat.SubjectPublicKeyInfo,
|
|
)
|
|
.decode('utf-8')
|
|
)
|
|
|
|
payload = '{"status": "Paid"}'
|
|
bad_signature = base64.b64encode(b'not-a-signature').decode('utf-8')
|
|
|
|
handler = WataWebhookHandler(
|
|
DummyPaymentService(),
|
|
public_key_provider=StubPublicKeyProvider(public_key),
|
|
)
|
|
|
|
assert await handler._verify_signature(payload, bad_signature) is False
|
|
|
|
|
|
@pytest.mark.anyio('asyncio')
|
|
async def test_verify_signature_fails_without_public_key() -> None:
|
|
handler = WataWebhookHandler(
|
|
DummyPaymentService(),
|
|
public_key_provider=StubPublicKeyProvider(None),
|
|
)
|
|
|
|
assert await handler._verify_signature('{}', 'signature') is False
|