davideandClaude Sonnet 5
a21e058cdd
Minimal Electrum protocol client (client.py) plus a scripthash subscription listener (listener.py) that watches user deposit addresses for confirmed UTXOs, with the address->scripthash conversion helper and a manual smoke-test script against the dev bootstrap server. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
78 lines
2.3 KiB
Python
78 lines
2.3 KiB
Python
import asyncio
|
|
import json
|
|
|
|
from app.electrum.client import ElectrumClient, ElectrumError
|
|
|
|
|
|
class FakeWriter:
|
|
def __init__(self):
|
|
self.written = b""
|
|
|
|
def write(self, data: bytes) -> None:
|
|
self.written += data
|
|
|
|
async def drain(self) -> None:
|
|
pass
|
|
|
|
def close(self) -> None:
|
|
pass
|
|
|
|
async def wait_closed(self) -> None:
|
|
pass
|
|
|
|
|
|
async def _client_with_fake_transport() -> tuple[ElectrumClient, asyncio.StreamReader, FakeWriter]:
|
|
client = ElectrumClient("localhost", 1234)
|
|
reader = asyncio.StreamReader()
|
|
writer = FakeWriter()
|
|
client._reader = reader
|
|
client._writer = writer
|
|
client._read_task = asyncio.create_task(client._read_loop())
|
|
return client, reader, writer
|
|
|
|
|
|
async def test_request_resolves_on_matching_response():
|
|
client, reader, writer = await _client_with_fake_transport()
|
|
|
|
task = asyncio.create_task(client.request("blockchain.headers.subscribe"))
|
|
await asyncio.sleep(0) # let request() write the payload
|
|
sent = json.loads(writer.written.decode())
|
|
assert sent["method"] == "blockchain.headers.subscribe"
|
|
|
|
reader.feed_data((json.dumps({"id": sent["id"], "result": {"height": 100}}) + "\n").encode())
|
|
result = await task
|
|
assert result == {"height": 100}
|
|
|
|
client._read_task.cancel()
|
|
|
|
|
|
async def test_error_response_raises_electrum_error():
|
|
client, reader, writer = await _client_with_fake_transport()
|
|
|
|
task = asyncio.create_task(client.request("blockchain.transaction.broadcast", ["deadbeef"]))
|
|
await asyncio.sleep(0)
|
|
sent = json.loads(writer.written.decode())
|
|
|
|
reader.feed_data((json.dumps({"id": sent["id"], "error": "bad tx"}) + "\n").encode())
|
|
try:
|
|
await task
|
|
assert False, "expected ElectrumError"
|
|
except ElectrumError:
|
|
pass
|
|
|
|
client._read_task.cancel()
|
|
|
|
|
|
async def test_notification_delivered_to_subscription_queue():
|
|
client, reader, writer = await _client_with_fake_transport()
|
|
queue = client.notifications("blockchain.scripthash.subscribe")
|
|
|
|
push = {"method": "blockchain.scripthash.subscribe", "params": ["abcd", "newstatus"]}
|
|
reader.feed_data((json.dumps(push) + "\n").encode())
|
|
await asyncio.sleep(0)
|
|
|
|
params = await asyncio.wait_for(queue.get(), timeout=1)
|
|
assert params == ["abcd", "newstatus"]
|
|
|
|
client._read_task.cancel()
|