refactor: harden client, clean tests, and fix timeouts

- Security & validation: tighten hash regex, add HTTP timeouts to all
  requests, pass client.timeout through get_fragment_hash and AsyncClient
- Constants: move all constants to types/constants.py; remove re-exports
  from types/__init__.py; add DEFAULT_TIMEOUT, REQUIRED_COOKIE_KEYS
- FragmentClient: add timeout param (default 30 s); async-context-manager
  support; remove WALLET_CLASSES from public API
- Exceptions: remove dead INVALID_USERNAME constant (Fragment validates
  server-side); keep full hierarchy intact
- Tests: add 006_test_methods_mock.py (6 mock tests for all 3 methods);
  DRY-refactor 004_test_balance.py (_patch_wallet context manager);
  clean up 005_test_methods.py (remove fragile network test, rename tests)
- Examples: switch all 4 examples to async-with; align error messages;
  replace %-format with f-strings
- README: rewrite usage section with single comprehensive async-with
  example covering all 3 methods and full exception hierarchy
- CI: add mypy step to lint job; add pytest-mock and mypy to dev deps;
  set FORCE_JAVASCRIPT_ACTIONS_TO_NODE24 on all jobs; fix COOKIES_JSON
  to job-level env var
This commit is contained in:
bohd4nx
2026-03-20 20:16:43 +02:00
parent c5edfad06f
commit 3e14a01c92
22 changed files with 335 additions and 309 deletions
+1 -1
View File
@@ -5,7 +5,7 @@ import re
import pytest
from pyfragment.types import BASE_HEADERS, STARS_PAGE
from pyfragment.types.constants import BASE_HEADERS, STARS_PAGE
from pyfragment.utils import get_fragment_hash
+26 -62
View File
@@ -1,21 +1,21 @@
"""Unit tests for process_transaction() — balance checks before broadcast."""
from contextlib import contextmanager
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from pyfragment.types import WalletError
from pyfragment.types import TransactionError, WalletError
from pyfragment.utils.wallet import process_transaction
VALID_SEED = "abandon " * 23 + "about"
# Minimal transaction payload: 0.5 TON = 500_000_000 nanotons
TRANSACTION_DATA = {
"transaction": {
"messages": [
{
"address": "0:852443f8599fe6a5da34fe43049ac4e0beb3071bb2bfb56635ea9421287c283a",
"amount": "500000000",
"amount": "500000000", # 0.5 TON
"payload": "",
}
]
@@ -23,9 +23,9 @@ TRANSACTION_DATA = {
}
def _make_client(api_key: str = "test_key") -> MagicMock:
def _make_client() -> MagicMock:
client = MagicMock()
client.api_key = api_key
client.api_key = "test_key"
client.seed = VALID_SEED.split()
client.wallet_version = "V5R1"
return client
@@ -39,89 +39,53 @@ def _make_wallet(balance_nanotons: int) -> MagicMock:
return wallet
@pytest.mark.asyncio
async def test_sufficient_balance_broadcasts() -> None:
# 0.5 TON amount + 0.056 TON gas = 0.556 TON required; wallet has 1 TON
client = _make_client()
wallet = _make_wallet(balance_nanotons=1_000_000_000)
@contextmanager
def _patch_wallet(wallet: MagicMock):
with (
patch("pyfragment.utils.wallet.TonapiClient") as mock_tonapi,
patch("pyfragment.utils.wallet.WALLET_CLASSES") as mock_classes,
patch("pyfragment.utils.wallet.clean_decode", return_value="50 Telegram Stars"),
):
mock_tonapi.return_value.__aenter__ = AsyncMock(return_value=MagicMock())
mock_tonapi.return_value.__aexit__ = AsyncMock(return_value=False)
mock_classes["V5R1"].from_mnemonic.return_value = (wallet, MagicMock(), None, None)
yield
result = await process_transaction(client, TRANSACTION_DATA)
@pytest.mark.asyncio
async def test_sufficient_balance_broadcasts() -> None:
wallet = _make_wallet(balance_nanotons=1_000_000_000) # 1 TON, needs 0.556 TON
with _patch_wallet(wallet), patch("pyfragment.utils.wallet.clean_decode", return_value="50 Telegram Stars"):
result = await process_transaction(_make_client(), TRANSACTION_DATA)
assert result == "abc123"
wallet.transfer.assert_called_once()
@pytest.mark.asyncio
async def test_insufficient_balance_raises_wallet_error() -> None:
# wallet has 0.1 TON, needs 0.556 TON
client = _make_client()
wallet = _make_wallet(balance_nanotons=100_000_000)
with (
patch("pyfragment.utils.wallet.TonapiClient") as mock_tonapi,
patch("pyfragment.utils.wallet.WALLET_CLASSES") as mock_classes,
):
mock_tonapi.return_value.__aenter__ = AsyncMock(return_value=MagicMock())
mock_tonapi.return_value.__aexit__ = AsyncMock(return_value=False)
mock_classes["V5R1"].from_mnemonic.return_value = (wallet, MagicMock(), None, None)
async def test_insufficient_balance_raises() -> None:
wallet = _make_wallet(balance_nanotons=100_000_000) # 0.1 TON, needs 0.556 TON
with _patch_wallet(wallet):
with pytest.raises(WalletError, match="required"):
await process_transaction(client, TRANSACTION_DATA)
await process_transaction(_make_client(), TRANSACTION_DATA)
wallet.transfer.assert_not_called()
@pytest.mark.asyncio
async def test_exactly_minimum_balance_broadcasts() -> None:
# exactly amount + gas: 500_000_000 + 56_000_000 = 556_000_000 nanotons
client = _make_client()
wallet = _make_wallet(balance_nanotons=556_000_000)
with (
patch("pyfragment.utils.wallet.TonapiClient") as mock_tonapi,
patch("pyfragment.utils.wallet.WALLET_CLASSES") as mock_classes,
patch("pyfragment.utils.wallet.clean_decode", return_value="50 Telegram Stars"),
):
mock_tonapi.return_value.__aenter__ = AsyncMock(return_value=MagicMock())
mock_tonapi.return_value.__aexit__ = AsyncMock(return_value=False)
mock_classes["V5R1"].from_mnemonic.return_value = (wallet, MagicMock(), None, None)
result = await process_transaction(client, TRANSACTION_DATA)
async def test_exact_minimum_balance_broadcasts() -> None:
wallet = _make_wallet(balance_nanotons=556_000_000) # exactly 0.5 + 0.056 TON
with _patch_wallet(wallet), patch("pyfragment.utils.wallet.clean_decode", return_value="50 Telegram Stars"):
result = await process_transaction(_make_client(), TRANSACTION_DATA)
assert result == "abc123"
@pytest.mark.asyncio
async def test_one_nanoton_below_minimum_raises() -> None:
# 556_000_000 - 1 nanoton: just below threshold
client = _make_client()
wallet = _make_wallet(balance_nanotons=555_999_999)
with (
patch("pyfragment.utils.wallet.TonapiClient") as mock_tonapi,
patch("pyfragment.utils.wallet.WALLET_CLASSES") as mock_classes,
):
mock_tonapi.return_value.__aenter__ = AsyncMock(return_value=MagicMock())
mock_tonapi.return_value.__aexit__ = AsyncMock(return_value=False)
mock_classes["V5R1"].from_mnemonic.return_value = (wallet, MagicMock(), None, None)
wallet = _make_wallet(balance_nanotons=555_999_999) # 1 nanoton below threshold
with _patch_wallet(wallet):
with pytest.raises(WalletError, match="required"):
await process_transaction(client, TRANSACTION_DATA)
await process_transaction(_make_client(), TRANSACTION_DATA)
@pytest.mark.asyncio
async def test_invalid_payload_raises_transaction_error() -> None:
from pyfragment.types import TransactionError
client = _make_client()
async def test_invalid_payload_raises() -> None:
with pytest.raises(TransactionError):
await process_transaction(client, {"transaction": {}})
await process_transaction(_make_client(), {"transaction": {}})
+6 -15
View File
@@ -21,45 +21,36 @@ def client() -> FragmentClient:
@pytest.mark.asyncio
async def test_purchase_premium_invalid_months_raises(client: FragmentClient) -> None:
async def test_purchase_premium_invalid_months(client: FragmentClient) -> None:
with pytest.raises(ConfigurationError):
await client.purchase_premium("@user", months=5)
@pytest.mark.asyncio
async def test_purchase_premium_valid_months(client: FragmentClient) -> None:
"""Validation passes for 3/6/12 — network error expected, not ConfigurationError."""
for months in (3, 6, 12):
with pytest.raises(Exception) as exc_info:
await client.purchase_premium("@user", months=months)
assert not isinstance(exc_info.value, ConfigurationError)
@pytest.mark.asyncio
async def test_purchase_stars_amount_too_low_raises(client: FragmentClient) -> None:
async def test_purchase_stars_amount_too_low(client: FragmentClient) -> None:
with pytest.raises(ConfigurationError):
await client.purchase_stars("@user", amount=49)
@pytest.mark.asyncio
async def test_purchase_stars_amount_too_high_raises(client: FragmentClient) -> None:
async def test_purchase_stars_amount_too_high(client: FragmentClient) -> None:
with pytest.raises(ConfigurationError):
await client.purchase_stars("@user", amount=1_000_001)
@pytest.mark.asyncio
async def test_purchase_stars_float_raises(client: FragmentClient) -> None:
async def test_purchase_stars_float_amount(client: FragmentClient) -> None:
with pytest.raises(ConfigurationError):
await client.purchase_stars("@user", amount=100.5) # type: ignore[arg-type]
@pytest.mark.asyncio
async def test_topup_ton_amount_zero_raises(client: FragmentClient) -> None:
async def test_topup_ton_amount_zero(client: FragmentClient) -> None:
with pytest.raises(ConfigurationError):
await client.topup_ton("@user", amount=0)
@pytest.mark.asyncio
async def test_topup_ton_amount_too_high_raises(client: FragmentClient) -> None:
async def test_topup_ton_amount_too_high(client: FragmentClient) -> None:
with pytest.raises(ConfigurationError):
await client.topup_ton("@user", amount=1_000_000_001)
+146
View File
@@ -0,0 +1,146 @@
"""Tests for purchase methods with all network calls mocked."""
from unittest.mock import AsyncMock, patch
import pytest
from pyfragment import FragmentClient
from pyfragment.types import AdsTopupResult, PremiumResult, StarsResult, UserNotFoundError
VALID_SEED = "abandon " * 23 + "about"
VALID_API_KEY = "A" * 68
VALID_COOKIES = {
"stel_ssid": "x",
"stel_dt": "x",
"stel_token": "x",
"stel_ton_token": "x",
}
FAKE_HASH = "abc123"
FAKE_RECIPIENT = "recipient_token"
FAKE_REQ_ID = "req_42"
FAKE_TX_HASH = "deadbeef" * 8
FAKE_ACCOUNT = {"address": "0:abc", "publicKey": "pub", "chain": "-239", "walletStateInit": "base64=="}
FAKE_TRANSACTION = {"transaction": {"messages": [{"address": "0:abc", "amount": "100000000", "payload": ""}]}}
@pytest.fixture
def client() -> FragmentClient:
return FragmentClient(seed=VALID_SEED, api_key=VALID_API_KEY, cookies=VALID_COOKIES)
@pytest.mark.asyncio
async def test_purchase_stars_success(client: FragmentClient) -> None:
with (
patch("pyfragment.methods.stars.get_fragment_hash", AsyncMock(return_value=FAKE_HASH)),
patch("pyfragment.methods.stars.get_account_info", AsyncMock(return_value=FAKE_ACCOUNT)),
patch(
"pyfragment.methods.stars.fragment_post",
AsyncMock(
side_effect=[
{"found": {"recipient": FAKE_RECIPIENT}}, # searchStarsRecipient
{"req_id": FAKE_REQ_ID}, # initBuyStarsRequest
]
),
),
patch("pyfragment.methods.stars.execute_transaction_request", AsyncMock(return_value=FAKE_TRANSACTION)),
patch("pyfragment.methods.stars.process_transaction", AsyncMock(return_value=FAKE_TX_HASH)),
):
result = await client.purchase_stars("testuser", amount=100)
assert isinstance(result, StarsResult)
assert result.transaction_id == FAKE_TX_HASH
assert result.username == "testuser"
assert result.stars == 100
@pytest.mark.asyncio
async def test_purchase_stars_user_not_found(client: FragmentClient) -> None:
with (
patch("pyfragment.methods.stars.get_fragment_hash", AsyncMock(return_value=FAKE_HASH)),
patch("pyfragment.methods.stars.get_account_info", AsyncMock(return_value=FAKE_ACCOUNT)),
patch("pyfragment.methods.stars.fragment_post", AsyncMock(return_value={"found": {}})),
):
with pytest.raises(UserNotFoundError):
await client.purchase_stars("ghost", amount=100)
@pytest.mark.asyncio
async def test_purchase_premium_success(client: FragmentClient) -> None:
with (
patch("pyfragment.methods.premium.get_fragment_hash", AsyncMock(return_value=FAKE_HASH)),
patch("pyfragment.methods.premium.get_account_info", AsyncMock(return_value=FAKE_ACCOUNT)),
patch(
"pyfragment.methods.premium.fragment_post",
AsyncMock(
side_effect=[
{"found": {"recipient": FAKE_RECIPIENT}}, # searchPremiumGiftRecipient
{}, # updatePremiumState
{"req_id": FAKE_REQ_ID}, # initGiftPremiumRequest
]
),
),
patch("pyfragment.methods.premium.execute_transaction_request", AsyncMock(return_value=FAKE_TRANSACTION)),
patch("pyfragment.methods.premium.process_transaction", AsyncMock(return_value=FAKE_TX_HASH)),
):
result = await client.purchase_premium("testuser", months=6)
assert isinstance(result, PremiumResult)
assert result.transaction_id == FAKE_TX_HASH
assert result.username == "testuser"
assert result.months == 6
@pytest.mark.asyncio
async def test_purchase_premium_user_not_found(client: FragmentClient) -> None:
with (
patch("pyfragment.methods.premium.get_fragment_hash", AsyncMock(return_value=FAKE_HASH)),
patch("pyfragment.methods.premium.get_account_info", AsyncMock(return_value=FAKE_ACCOUNT)),
patch("pyfragment.methods.premium.fragment_post", AsyncMock(return_value={"found": {}})),
):
with pytest.raises(UserNotFoundError):
await client.purchase_premium("ghost", months=3)
@pytest.mark.asyncio
async def test_topup_ton_success(client: FragmentClient) -> None:
with (
patch("pyfragment.methods.ton.get_fragment_hash", AsyncMock(return_value=FAKE_HASH)),
patch("pyfragment.methods.ton.get_account_info", AsyncMock(return_value=FAKE_ACCOUNT)),
patch(
"pyfragment.methods.ton.fragment_post",
AsyncMock(
side_effect=[
{}, # updateAdsTopupState
{"found": {"recipient": FAKE_RECIPIENT}}, # searchAdsTopupRecipient
{"req_id": FAKE_REQ_ID}, # initAdsTopupRequest
]
),
),
patch("pyfragment.methods.ton.execute_transaction_request", AsyncMock(return_value=FAKE_TRANSACTION)),
patch("pyfragment.methods.ton.process_transaction", AsyncMock(return_value=FAKE_TX_HASH)),
):
result = await client.topup_ton("testuser", amount=10)
assert isinstance(result, AdsTopupResult)
assert result.transaction_id == FAKE_TX_HASH
assert result.username == "testuser"
assert result.amount == 10
@pytest.mark.asyncio
async def test_topup_ton_user_not_found(client: FragmentClient) -> None:
with (
patch("pyfragment.methods.ton.get_fragment_hash", AsyncMock(return_value=FAKE_HASH)),
patch("pyfragment.methods.ton.get_account_info", AsyncMock(return_value=FAKE_ACCOUNT)),
patch(
"pyfragment.methods.ton.fragment_post",
AsyncMock(
side_effect=[
{}, # updateAdsTopupState
{"found": {}}, # searchAdsTopupRecipient → not found
]
),
),
):
with pytest.raises(UserNotFoundError):
await client.topup_ton("ghost", amount=10)
+6 -7
View File
@@ -1,17 +1,16 @@
import json
from pathlib import Path
import os
import pytest
@pytest.fixture
def cookies():
"""Load Fragment cookies from cookies.json; skip the test if unavailable."""
cookies_path = Path(__file__).resolve().parents[1] / "cookies.json"
if not cookies_path.exists():
pytest.skip("cookies.json not found")
"""Load Fragment cookies from COOKIES_JSON env var; skip if unavailable."""
raw = os.environ.get("COOKIES_JSON")
if not raw:
pytest.skip("COOKIES_JSON env var not set")
try:
with cookies_path.open("r", encoding="utf-8") as f:
return json.load(f)
return json.loads(raw)
except Exception as exc:
pytest.skip(f"Cookies unavailable — {exc}")