175 lines
6.8 KiB
Python
175 lines
6.8 KiB
Python
import pytest
|
|
from unittest.mock import AsyncMock, patch
|
|
from aiohttp import web
|
|
from aiohttp.test_utils import AioHTTPTestCase
|
|
|
|
from server import create_app
|
|
from usage import BalanceResult
|
|
|
|
|
|
class TestServer(AioHTTPTestCase):
|
|
async def get_application(self):
|
|
return create_app()
|
|
|
|
async def test_health(self):
|
|
resp = await self.client.get("/health")
|
|
assert resp.status == 200
|
|
text = await resp.text()
|
|
assert text == "ok"
|
|
|
|
@patch("server.get_first_token", new_callable=AsyncMock)
|
|
@patch("server.wait_for_token", new_callable=AsyncMock)
|
|
async def test_token_empty_pool_no_accounts(self, mock_wait, mock_first):
|
|
mock_first.return_value = None
|
|
mock_wait.return_value = None
|
|
|
|
resp = await self.client.get("/token")
|
|
assert resp.status == 200
|
|
data = await resp.json()
|
|
assert data["token"] == "public"
|
|
|
|
@patch("server.get_first_token", new_callable=AsyncMock)
|
|
@patch("server.wait_for_token", new_callable=AsyncMock)
|
|
async def test_token_empty_pool_waits_then_gets(self, mock_wait, mock_first):
|
|
mock_first.side_effect = [None, "new_token"]
|
|
mock_wait.return_value = "new_token"
|
|
|
|
resp = await self.client.get("/token")
|
|
assert resp.status == 200
|
|
data = await resp.json()
|
|
assert data["token"] == "new_token"
|
|
|
|
@patch("server.get_first_token", new_callable=AsyncMock)
|
|
@patch("server.pop_token", new_callable=AsyncMock)
|
|
@patch("server.trigger_refill", new_callable=AsyncMock)
|
|
@patch("server.pool_size", new_callable=AsyncMock)
|
|
@patch("server.get_balance", new_callable=AsyncMock)
|
|
async def test_token_valid(
|
|
self, mock_balance, mock_size, mock_refill, mock_pop, mock_first
|
|
):
|
|
mock_first.return_value = "test_token_12345"
|
|
mock_balance.return_value = BalanceResult(
|
|
balance=10.0, is_invalid=False, error=None
|
|
)
|
|
mock_size.return_value = 3
|
|
|
|
resp = await self.client.get("/token")
|
|
assert resp.status == 200
|
|
data = await resp.json()
|
|
assert data["token"] == "test_token_12345"
|
|
mock_pop.assert_not_called()
|
|
|
|
@patch("server.get_first_token", new_callable=AsyncMock)
|
|
@patch("server.pop_token", new_callable=AsyncMock)
|
|
@patch("server.trigger_refill", new_callable=AsyncMock)
|
|
@patch("server.pool_size", new_callable=AsyncMock)
|
|
@patch("server.get_balance", new_callable=AsyncMock)
|
|
async def test_token_zero_balance_removed(
|
|
self, mock_balance, mock_size, mock_refill, mock_pop, mock_first
|
|
):
|
|
mock_first.side_effect = ["bad_token", "good_token"]
|
|
mock_balance.side_effect = [
|
|
BalanceResult(balance=0.0, is_invalid=False, error=None),
|
|
BalanceResult(balance=15.0, is_invalid=False, error=None),
|
|
]
|
|
mock_size.return_value = 2
|
|
|
|
resp = await self.client.get("/token")
|
|
assert resp.status == 200
|
|
data = await resp.json()
|
|
assert data["token"] == "good_token"
|
|
assert mock_pop.call_count == 1
|
|
|
|
@patch("server.get_first_token", new_callable=AsyncMock)
|
|
@patch("server.pop_token", new_callable=AsyncMock)
|
|
@patch("server.trigger_refill", new_callable=AsyncMock)
|
|
@patch("server.pool_size", new_callable=AsyncMock)
|
|
@patch("server.get_balance", new_callable=AsyncMock)
|
|
async def test_token_network_error_kept(
|
|
self, mock_balance, mock_size, mock_refill, mock_pop, mock_first
|
|
):
|
|
mock_first.return_value = "test_token"
|
|
mock_balance.return_value = BalanceResult(
|
|
balance=None, is_invalid=False, error="network_error"
|
|
)
|
|
mock_size.return_value = 2
|
|
|
|
resp = await self.client.get("/token")
|
|
assert resp.status == 200
|
|
data = await resp.json()
|
|
assert data["token"] == "test_token"
|
|
mock_pop.assert_not_called()
|
|
|
|
@patch("server.get_first_token", new_callable=AsyncMock)
|
|
@patch("server.pop_token", new_callable=AsyncMock)
|
|
@patch("server.trigger_refill", new_callable=AsyncMock)
|
|
@patch("server.pool_size", new_callable=AsyncMock)
|
|
@patch("server.get_balance", new_callable=AsyncMock)
|
|
async def test_token_invalid_removed(
|
|
self, mock_balance, mock_size, mock_refill, mock_pop, mock_first
|
|
):
|
|
mock_first.side_effect = ["invalid_token", "good_token"]
|
|
mock_balance.side_effect = [
|
|
BalanceResult(balance=None, is_invalid=True, error="Invalid token (xxx)"),
|
|
BalanceResult(balance=10.0, is_invalid=False, error=None),
|
|
]
|
|
mock_size.return_value = 2
|
|
|
|
resp = await self.client.get("/token")
|
|
assert resp.status == 200
|
|
data = await resp.json()
|
|
assert data["token"] == "good_token"
|
|
assert mock_pop.call_count == 1
|
|
|
|
@patch("server.get_first_token", new_callable=AsyncMock)
|
|
@patch("server.pop_token", new_callable=AsyncMock)
|
|
@patch("server.trigger_refill", new_callable=AsyncMock)
|
|
@patch("server.pool_size", new_callable=AsyncMock)
|
|
@patch("server.get_balance", new_callable=AsyncMock)
|
|
async def test_token_balance_with_remaining_key(
|
|
self, mock_balance, mock_size, mock_refill, mock_pop, mock_first
|
|
):
|
|
mock_first.return_value = "test_token"
|
|
mock_balance.return_value = BalanceResult(
|
|
balance=20.0, is_invalid=False, error=None
|
|
)
|
|
mock_size.return_value = 1
|
|
|
|
resp = await self.client.get("/token")
|
|
assert resp.status == 200
|
|
data = await resp.json()
|
|
assert data["token"] == "test_token"
|
|
|
|
@patch("server.has_accounts", new_callable=AsyncMock)
|
|
@patch("server.pool_size", new_callable=AsyncMock)
|
|
async def test_status(self, mock_size, mock_has):
|
|
mock_size.return_value = 5
|
|
mock_has.return_value = True
|
|
|
|
resp = await self.client.get("/status")
|
|
assert resp.status == 200
|
|
data = await resp.json()
|
|
assert data["pool_size"] == 5
|
|
assert data["has_accounts"] is True
|
|
|
|
@patch("server.get_first_token", new_callable=AsyncMock)
|
|
@patch("server.pop_token", new_callable=AsyncMock)
|
|
@patch("server.trigger_refill", new_callable=AsyncMock)
|
|
@patch("server.wait_for_token", new_callable=AsyncMock)
|
|
@patch("server.pool_size", new_callable=AsyncMock)
|
|
@patch("server.get_balance", new_callable=AsyncMock)
|
|
async def test_all_tokens_exhausted_then_public(
|
|
self, mock_balance, mock_size, mock_wait, mock_refill, mock_pop, mock_first
|
|
):
|
|
mock_first.side_effect = ["token1", "token2", None]
|
|
mock_balance.side_effect = [
|
|
BalanceResult(balance=0.0, is_invalid=False, error=None),
|
|
BalanceResult(balance=-5.0, is_invalid=False, error=None),
|
|
]
|
|
mock_wait.return_value = None
|
|
mock_size.return_value = 0
|
|
|
|
resp = await self.client.get("/token")
|
|
assert resp.status == 200
|
|
data = await resp.json()
|
|
assert data["token"] == "public"
|