126 lines
2.8 KiB
Python
126 lines
2.8 KiB
Python
import pytest
|
|
|
|
from pool import (
|
|
append_token,
|
|
pop_token,
|
|
pool_size,
|
|
get_first_token,
|
|
trigger_refill,
|
|
wait_for_token,
|
|
)
|
|
import pool as p
|
|
|
|
|
|
@pytest.fixture
|
|
def fresh_pool(tmp_path, monkeypatch):
|
|
tokens_file = tmp_path / "tokens.txt"
|
|
tokens_file.write_text("")
|
|
monkeypatch.setattr(p, "TOKENS_FILE", tokens_file)
|
|
monkeypatch.setattr(p, "DATA_DIR", tmp_path)
|
|
return tokens_file
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_empty_pool(fresh_pool):
|
|
assert await get_first_token() is None
|
|
assert await pop_token() is None
|
|
assert await pool_size() == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_append_token(fresh_pool):
|
|
await append_token("token1")
|
|
assert await pool_size() == 1
|
|
assert await get_first_token() == "token1"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_append_multiple_tokens(fresh_pool):
|
|
await append_token("token1")
|
|
await append_token("token2")
|
|
await append_token("token3")
|
|
|
|
assert await pool_size() == 3
|
|
assert await get_first_token() == "token1"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_pop_token(fresh_pool):
|
|
await append_token("token1")
|
|
await append_token("token2")
|
|
|
|
t = await pop_token()
|
|
assert t == "token1"
|
|
assert await pool_size() == 1
|
|
assert await get_first_token() == "token2"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_pop_until_empty(fresh_pool):
|
|
await append_token("only_one")
|
|
|
|
t = await pop_token()
|
|
assert t == "only_one"
|
|
|
|
assert await pop_token() is None
|
|
assert await pool_size() == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_first_token_peek(fresh_pool):
|
|
await append_token("token1")
|
|
|
|
t = await get_first_token()
|
|
assert t == "token1"
|
|
|
|
t2 = await get_first_token()
|
|
assert t2 == "token1"
|
|
|
|
assert await pool_size() == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_token_with_whitespace(fresh_pool):
|
|
await append_token(" token_with_spaces ")
|
|
t = await get_first_token()
|
|
assert t == "token_with_spaces"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_persist_to_file(fresh_pool):
|
|
await append_token("token1")
|
|
await append_token("token2")
|
|
|
|
content = fresh_pool.read_text()
|
|
assert "token1" in content
|
|
assert "token2" in content
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_wait_for_token_with_existing(fresh_pool):
|
|
await append_token("existing_token")
|
|
|
|
token = await wait_for_token()
|
|
assert token == "existing_token"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_wait_for_token_empty_no_accounts(fresh_pool, tmp_path, monkeypatch):
|
|
import emails
|
|
|
|
emails_file = tmp_path / "emails.txt"
|
|
emails_file.write_text("")
|
|
monkeypatch.setattr(emails, "EMAILS_FILE", emails_file)
|
|
|
|
token = await wait_for_token()
|
|
assert token is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_trigger_refill_does_not_block(fresh_pool):
|
|
import asyncio
|
|
|
|
await trigger_refill()
|
|
await asyncio.sleep(0.01)
|
|
|
|
assert True
|