forked from bitcoinafterlife/bal-electrum-plugin
152 lines
5.9 KiB
Python
152 lines
5.9 KiB
Python
"""
|
|
Live will-executor / welist query tests.
|
|
|
|
These tests contact the real servers documented in the project spec:
|
|
|
|
* the default will-executor ``https://we.bitcoin-after.life`` (``/bitcoin``)
|
|
* the welist ``https://welist.bitcoin-after.life`` (``/data/bitcoin``)
|
|
|
|
They are automatically **skipped** when the machine is offline or the servers
|
|
are unreachable, so the suite never fails spuriously in a sandbox / CI without
|
|
internet.
|
|
|
|
Two things are verified for each endpoint:
|
|
|
|
1. The live response has the documented JSON shape.
|
|
2. The plugin's own parsing logic (``Willexecutors.get_info_task`` /
|
|
``Willexecutors.download_list``) consumes that live response correctly.
|
|
To test the parsing without needing a running Electrum daemon, the live
|
|
JSON is fetched once with ``urllib`` and then fed back through the plugin
|
|
by mocking only the transport layer.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import sys
|
|
import urllib.error
|
|
import urllib.request
|
|
|
|
import pytest
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
sys.path.insert(0, os.path.join(os.path.dirname(__file__), os.pardir))
|
|
|
|
from bal.core.willexecutors import Willexecutors
|
|
|
|
WE_URL = "https://we.bitcoin-after.life"
|
|
WE_INFO_URL = "https://we.bitcoin-after.life/bitcoin/info"
|
|
WELIST_URL = "https://welist.bitcoin-after.life"
|
|
WELIST_DATA_URL = "https://welist.bitcoin-after.life/data/bitcoin?page=0&limit=100"
|
|
|
|
HTTP_TIMEOUT = 8
|
|
|
|
|
|
def _fetch_json(url):
|
|
"""Fetch and JSON-decode ``url``; skip the test on any network problem."""
|
|
req = urllib.request.Request(
|
|
url, headers={"User-Agent": "BalPluginTest"}
|
|
)
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=HTTP_TIMEOUT) as resp:
|
|
raw = resp.read().decode("utf-8")
|
|
except (urllib.error.URLError, OSError, TimeoutError) as e:
|
|
pytest.skip(f"network unavailable for {url}: {e}")
|
|
try:
|
|
return json.loads(raw)
|
|
except json.JSONDecodeError as e:
|
|
pytest.fail(f"{url} returned non-JSON: {e}: {raw[:200]!r}")
|
|
|
|
|
|
# ------------------------------------------------------------------ #
|
|
# Will-executor /info endpoint
|
|
# ------------------------------------------------------------------ #
|
|
|
|
def test_live_willexecutor_info_shape():
|
|
"""The live /bitcoin/info reply has the documented fields."""
|
|
data = _fetch_json(WE_INFO_URL)
|
|
assert isinstance(data, dict)
|
|
for key in ("address", "base_fee", "info"):
|
|
assert key in data, f"missing '{key}' in info reply: {data}"
|
|
assert isinstance(data["base_fee"], int)
|
|
assert data["base_fee"] > 0
|
|
assert isinstance(data["address"], str) and data["address"]
|
|
|
|
|
|
def test_live_willexecutor_info_parsed_by_plugin():
|
|
"""Feed the live info reply through ``get_info_task`` and check it parses."""
|
|
data = _fetch_json(WE_INFO_URL)
|
|
we = {"url": WE_URL, "selected": True, "status": "New"}
|
|
# The plugin reaches the server via send_request -> Network transport. We
|
|
# make the transport return exactly what the live server returned, so we
|
|
# exercise the real parsing path without a running Electrum daemon.
|
|
with patch("bal.core.willexecutors.Network.get_instance",
|
|
return_value=MagicMock()), \
|
|
patch("bal.core.willexecutors.Network.send_http_on_proxy",
|
|
return_value=data):
|
|
out = Willexecutors.get_info_task(WE_URL, we, max_retries=0)
|
|
assert out["status"] == 200
|
|
assert out["base_fee"] == data["base_fee"]
|
|
assert out["address"] == data["address"]
|
|
assert out["info"] == data["info"]
|
|
|
|
|
|
# ------------------------------------------------------------------ #
|
|
# Welist /data/bitcoin endpoint
|
|
# ------------------------------------------------------------------ #
|
|
|
|
def test_live_welist_shape():
|
|
"""The live welist reply maps URLs to will-executor descriptors."""
|
|
data = _fetch_json(WELIST_DATA_URL)
|
|
assert isinstance(data, dict)
|
|
# Drop the optional bookkeeping keys before inspecting executors.
|
|
executors = {k: v for k, v in data.items() if k not in ("status", "url")}
|
|
assert executors, "welist returned no will-executors"
|
|
for url, we in executors.items():
|
|
assert url.startswith("http"), f"bad executor url: {url}"
|
|
assert isinstance(we, dict)
|
|
assert "base_fee" in we, f"executor {url} missing base_fee"
|
|
assert "url" in we
|
|
|
|
|
|
def test_live_welist_contains_default_executor():
|
|
"""The default will-executor should be listed in the welist."""
|
|
data = _fetch_json(WELIST_DATA_URL)
|
|
assert WE_URL in data, f"{WE_URL} not present in welist: {list(data)[:5]}"
|
|
|
|
|
|
def test_live_welist_parsed_by_plugin():
|
|
"""Feed the live welist reply through ``download_list`` and check parsing.
|
|
|
|
``download_list`` normalises each executor entry (sets url / status /
|
|
selected / address) and must not crash on the real payload.
|
|
"""
|
|
data = _fetch_json(WELIST_DATA_URL)
|
|
with patch("bal.core.willexecutors.Network.get_instance",
|
|
return_value=MagicMock()), \
|
|
patch("bal.core.willexecutors.Network.send_http_on_proxy",
|
|
return_value=data):
|
|
out = Willexecutors.download_list({}, WELIST_URL)
|
|
assert isinstance(out, dict)
|
|
executors = {k: v for k, v in out.items() if k not in ("status", "url")}
|
|
assert executors, "download_list produced no executors"
|
|
for url, we in executors.items():
|
|
# initialize_willexecutor must have stamped the normalised fields.
|
|
assert we["url"] == url
|
|
assert "selected" in we
|
|
assert "address" in we
|
|
assert "status" in we
|
|
|
|
|
|
# ------------------------------------------------------------------ #
|
|
|
|
if __name__ == "__main__":
|
|
import inspect
|
|
for name, obj in sorted(globals().items()):
|
|
if name.startswith("test_") and inspect.isfunction(obj):
|
|
try:
|
|
obj()
|
|
print(f" [OK] {name}")
|
|
except Exception as e: # pragma: no cover
|
|
print(f" [FAIL/SKIP] {name}: {e}")
|
|
print("[done] live willexecutor tests")
|