Mirror/brk
1
0
Fork 0
mirror of https://github.com/bitcoinresearchkit/brk.git synced 2026-05-19 22:34:46 -07:00
Code Issues Packages Projects Releases Wiki Activity

global: fixes

Browse Source
This commit is contained in:
nym21
2026-04-29 16:51:01 +02:00
parent a7e41df1c6
commit 43f3be4924
101 changed files with 3074 additions and 2869 deletions
Download Patch File Download Diff File Expand all files Collapse all files

48
packages/brk_client/tests/mempool_compat/addresses/test_address_info.py Normal file
View File

@@ -0,0 +1,48 @@
"""GET /api/address/{address}"""
import pytest
from _lib import assert_same_structure, show
@pytest.fixture(params=[
"12cbQLTFMXRnSzktFkuoG3eHoMeFtpTu3S", # P2PKH — early block reward
"3D2oetdNuZUqQHPJmcMDDHYoqkyNVsFk9r", # P2SH
], ids=["p2pkh", "p2sh"])
def static_addr(request):
"""Well-known addresses that always exist."""
return request.param
def test_address_info_static(brk, mempool, static_addr):
"""Address stats structure must match for well-known addresses."""
path = f"/api/address/{static_addr}"
b = brk.get_json(path)
m = mempool.get_json(path)
show("GET", path, b, m)
assert_same_structure(b, m)
assert b["address"] == m["address"]
def test_address_info_discovered(brk, mempool, live_addrs):
"""Address stats structure must match for each discovered type."""
for atype, addr in live_addrs:
path = f"/api/address/{addr}"
b = brk.get_json(path)
m = mempool.get_json(path)
show("GET", f"{path} [{atype}]", b, m)
assert_same_structure(b, m)
assert b["address"] == m["address"]
def test_address_chain_stats_close(brk, mempool, live_addrs):
"""Chain stats values must be close for each discovered address."""
for atype, addr in live_addrs:
path = f"/api/address/{addr}"
b = brk.get_json(path)["chain_stats"]
m = mempool.get_json(path)["chain_stats"]
show("GET", f"{path} [chain_stats, {atype}]", b, m)
assert_same_structure(b, m)
assert abs(b["tx_count"] - m["tx_count"]) <= 5, (
f"{atype} tx_count: brk={b['tx_count']} vs mempool={m['tx_count']}"
)

49
packages/brk_client/tests/mempool_compat/addresses/test_address_txs.py Normal file
View File

@@ -0,0 +1,49 @@
"""GET /api/address/{address}/txs"""
import pytest
from _lib import assert_same_structure, show
@pytest.fixture(params=[
"12cbQLTFMXRnSzktFkuoG3eHoMeFtpTu3S",
"3D2oetdNuZUqQHPJmcMDDHYoqkyNVsFk9r",
], ids=["p2pkh", "p2sh"])
def static_addr(request):
return request.param
def test_address_txs_static(brk, mempool, static_addr):
"""Confirmed+mempool tx list structure must match for well-known addresses."""
path = f"/api/address/{static_addr}/txs"
b = brk.get_json(path)
m = mempool.get_json(path)
show("GET", path, f"({len(b)} txs)", f"({len(m)} txs)")
assert isinstance(b, list) and isinstance(m, list)
if b and m:
assert_same_structure(b[0], m[0])
def test_address_txs_discovered(brk, mempool, live_addrs):
"""Confirmed+mempool tx list structure must match for each discovered type."""
for atype, addr in live_addrs:
path = f"/api/address/{addr}/txs"
b = brk.get_json(path)
m = mempool.get_json(path)
show("GET", f"{path} [{atype}]", f"({len(b)} txs)", f"({len(m)} txs)")
assert isinstance(b, list) and isinstance(m, list)
if b and m:
assert_same_structure(b[0], m[0])
def test_address_txs_fields(brk, mempool, live):
"""Every tx in the list must carry the core mempool.space fields."""
path = f"/api/address/{live.sample_address}/txs"
b = brk.get_json(path)
show("GET", path, f"({len(b)} txs)", "—")
if not b:
pytest.skip("address has no txs in brk")
required = {"txid", "version", "locktime", "vin", "vout", "size", "weight", "fee", "status"}
for tx in b[:5]:
missing = required - set(tx.keys())
assert not missing, f"tx {tx.get('txid', '?')} missing fields: {missing}"

49
packages/brk_client/tests/mempool_compat/addresses/test_address_txs_chain.py Normal file
View File

@@ -0,0 +1,49 @@
"""GET /api/address/{address}/txs/chain"""
import pytest
from _lib import assert_same_structure, show
@pytest.fixture(params=[
"12cbQLTFMXRnSzktFkuoG3eHoMeFtpTu3S",
"3D2oetdNuZUqQHPJmcMDDHYoqkyNVsFk9r",
], ids=["p2pkh", "p2sh"])
def static_addr(request):
return request.param
def test_address_txs_chain_static(brk, mempool, static_addr):
"""Confirmed-only tx list structure must match for well-known addresses."""
path = f"/api/address/{static_addr}/txs/chain"
b = brk.get_json(path)
m = mempool.get_json(path)
show("GET", path, f"({len(b)} txs)", f"({len(m)} txs)")
assert isinstance(b, list) and isinstance(m, list)
if b and m:
assert_same_structure(b[0], m[0])
def test_address_txs_chain_discovered(brk, mempool, live_addrs):
"""Confirmed-only tx list structure must match for each discovered type."""
for atype, addr in live_addrs:
path = f"/api/address/{addr}/txs/chain"
b = brk.get_json(path)
m = mempool.get_json(path)
show("GET", f"{path} [{atype}]", f"({len(b)} txs)", f"({len(m)} txs)")
assert isinstance(b, list) and isinstance(m, list)
if b and m:
assert_same_structure(b[0], m[0])
def test_address_txs_chain_all_confirmed(brk, live):
"""Every tx returned by /txs/chain must have confirmed=True in its status."""
path = f"/api/address/{live.sample_address}/txs/chain"
b = brk.get_json(path)
show("GET", path, f"({len(b)} txs)", "—")
if not b:
pytest.skip("address has no confirmed txs in brk")
unconfirmed = [t for t in b if not t.get("status", {}).get("confirmed", False)]
assert not unconfirmed, (
f"{len(unconfirmed)} unconfirmed tx(s) returned by /txs/chain"
)

33
packages/brk_client/tests/mempool_compat/addresses/test_address_txs_mempool.py Normal file
View File

@@ -0,0 +1,33 @@
"""GET /api/address/{address}/txs/mempool"""
from _lib import show
def test_address_txs_mempool_sample(brk, mempool, live):
"""Mempool tx list must be an array (contents are volatile)."""
path = f"/api/address/{live.sample_address}/txs/mempool"
b = brk.get_json(path)
m = mempool.get_json(path)
show("GET", path, f"({len(b)} txs)", f"({len(m)} txs)")
assert isinstance(b, list) and isinstance(m, list)
def test_address_txs_mempool_discovered(brk, mempool, live_addrs):
"""Mempool tx list must be a (possibly empty) array for each discovered type."""
for atype, addr in live_addrs:
path = f"/api/address/{addr}/txs/mempool"
b = brk.get_json(path)
m = mempool.get_json(path)
show("GET", f"{path} [{atype}]", f"({len(b)} txs)", f"({len(m)} txs)")
assert isinstance(b, list) and isinstance(m, list)
def test_address_txs_mempool_all_unconfirmed(brk, live):
"""Every tx returned by /txs/mempool must have confirmed=False (if any)."""
path = f"/api/address/{live.sample_address}/txs/mempool"
b = brk.get_json(path)
show("GET", path, f"({len(b)} txs)", "—")
confirmed = [t for t in b if t.get("status", {}).get("confirmed", False)]
assert not confirmed, (
f"{len(confirmed)} confirmed tx(s) returned by /txs/mempool"
)

52
packages/brk_client/tests/mempool_compat/addresses/test_address_utxo.py Normal file
View File

@@ -0,0 +1,52 @@
"""GET /api/address/{address}/utxo"""
import pytest
from _lib import assert_same_values, show
@pytest.fixture(params=[
"12cbQLTFMXRnSzktFkuoG3eHoMeFtpTu3S",
"3D2oetdNuZUqQHPJmcMDDHYoqkyNVsFk9r",
], ids=["p2pkh", "p2sh"])
def static_addr(request):
return request.param
def test_address_utxo_static(brk, mempool, static_addr):
"""UTXO list must match — same txids, values, and statuses."""
path = f"/api/address/{static_addr}/utxo"
b = brk.get_json(path)
m = mempool.get_json(path)
show("GET", path, f"({len(b)} utxos)", f"({len(m)} utxos)")
assert isinstance(b, list) and isinstance(m, list)
key = lambda u: (u.get("txid", ""), u.get("vout", 0))
b_sorted = sorted(b, key=key)
m_sorted = sorted(m, key=key)
assert_same_values(b_sorted, m_sorted)
def test_address_utxo_discovered(brk, mempool, live_addrs):
"""UTXO list must match for each discovered address type — same txids, values, and statuses."""
for atype, addr in live_addrs:
path = f"/api/address/{addr}/utxo"
b = brk.get_json(path)
m = mempool.get_json(path)
show("GET", f"{path} [{atype}]", f"({len(b)} utxos)", f"({len(m)} utxos)")
assert isinstance(b, list) and isinstance(m, list)
key = lambda u: (u.get("txid", ""), u.get("vout", 0))
assert_same_values(sorted(b, key=key), sorted(m, key=key))
def test_address_utxo_fields(brk, live):
"""Every utxo must carry the core mempool.space fields."""
path = f"/api/address/{live.sample_address}/utxo"
b = brk.get_json(path)
show("GET", path, f"({len(b)} utxos)", "—")
if not b:
pytest.skip("address has no utxos in brk")
required = {"txid", "vout", "value", "status"}
for u in b[:5]:
missing = required - set(u.keys())
assert not missing, f"utxo {u.get('txid', '?')}:{u.get('vout', '?')} missing fields: {missing}"
assert isinstance(u["value"], int) and u["value"] > 0

53
packages/brk_client/tests/mempool_compat/addresses/test_validate_address.py Normal file
View File

@@ -0,0 +1,53 @@
"""GET /api/v1/validate-address/{address}"""
import pytest
from _lib import assert_same_structure, assert_same_values, show
def test_validate_address_discovered(brk, mempool, live_addrs):
"""Validation of each discovered address type must match exactly."""
for atype, addr in live_addrs:
path = f"/api/v1/validate-address/{addr}"
b = brk.get_json(path)
m = mempool.get_json(path)
show("GET", f"{path} [{atype}]", b, m)
assert_same_values(b, m)
assert b["isvalid"] is True
@pytest.mark.parametrize("addr,kind", [
("1A1zP1eP5QGefi2DMPTfTL5SLmv7DivfNa", "p2pkh-genesis"),
("3J98t1WpEZ73CNmQviecrnyiWrnqRhWNLy", "p2sh"),
("bc1qw508d6qejxtdg4y5r3zarvary0c5xw7kv8f3t4", "p2wpkh"),
("bc1p0xlxvlhemja6c4dqv22uapctqupfhlxm9h8z3k2e72q4k9hcz7vqzk5jj0", "p2tr"),
])
def test_validate_address_static_valid(brk, mempool, addr, kind):
"""Well-known addresses across all script types must validate identically."""
path = f"/api/v1/validate-address/{addr}"
b = brk.get_json(path)
m = mempool.get_json(path)
show("GET", f"{path} [{kind}]", b, m)
assert_same_values(b, m)
assert b["isvalid"] is True
@pytest.mark.parametrize("addr,kind", [
("notanaddress123", "garbage"),
("", "empty"),
("1A1zP1eP5QGefi2DMPTfTL5SLmv7DivfNb", "bad-checksum-p2pkh"),
("bc1qw508d6qejxtdg4y5r3zarvary0c5xw7kv8f3t5", "bad-checksum-p2wpkh"),
("3J98t1WpEZ73CNmQviecrnyiWrnqRhWNLz", "bad-checksum-p2sh"),
])
def test_validate_address_invalid(brk, mempool, addr, kind):
"""Invalid addresses must produce the same rejection structure."""
path = f"/api/v1/validate-address/{addr}"
if kind == "empty":
# An empty path segment routes to a different endpoint — skip.
pytest.skip("empty address routes to a different endpoint")
b = brk.get_json(path)
m = mempool.get_json(path)
show("GET", f"{path} [{kind}]", b, m)
assert b["isvalid"] is False
assert m["isvalid"] is False
assert_same_structure(b, m)

12
packages/brk_client/tests/mempool_compat/blocks/test_block.py Normal file
View File

@@ -0,0 +1,12 @@
"""GET /api/block/{hash}"""
from _lib import assert_same_values, show
def test_block_by_hash(brk, mempool, block):
"""Confirmed block info must be identical."""
path = f"/api/block/{block.hash}"
b = brk.get_json(path)
m = mempool.get_json(path)
show("GET", path, b, m)
assert_same_values(b, m)

13
packages/brk_client/tests/mempool_compat/blocks/test_block_header.py Normal file
View File

@@ -0,0 +1,13 @@
"""GET /api/block/{hash}/header"""
from _lib import show
def test_block_header(brk, mempool, block):
"""80-byte hex block header must be identical."""
path = f"/api/block/{block.hash}/header"
b = brk.get_text(path)
m = mempool.get_text(path)
show("GET", path, b, m)
assert len(b) == 160, f"Expected 160 hex chars (80 bytes), got {len(b)}"
assert b == m

13
packages/brk_client/tests/mempool_compat/blocks/test_block_height.py Normal file
View File

@@ -0,0 +1,13 @@
"""GET /api/block-height/{height}"""
from _lib import show
def test_block_height_to_hash(brk, mempool, block):
"""Block hash at a given height must match."""
path = f"/api/block-height/{block.height}"
b = brk.get_text(path)
m = mempool.get_text(path)
show("GET", path, b, m)
assert b == m
assert b == block.hash

13
packages/brk_client/tests/mempool_compat/blocks/test_block_raw.py Normal file
View File

@@ -0,0 +1,13 @@
"""GET /api/block/{hash}/raw"""
from _lib import show
def test_block_raw(brk, mempool, block):
"""Raw block bytes must be identical and start with the 80-byte header."""
path = f"/api/block/{block.hash}/raw"
b = brk.get_bytes(path)
m = mempool.get_bytes(path)
show("GET", path, f"<{len(b)} bytes>", f"<{len(m)} bytes>")
assert b == m
assert len(b) >= 80

12
packages/brk_client/tests/mempool_compat/blocks/test_block_status.py Normal file
View File

@@ -0,0 +1,12 @@
"""GET /api/block/{hash}/status"""
from _lib import assert_same_values, show
def test_block_status(brk, mempool, block):
"""Block status must be identical for a confirmed block."""
path = f"/api/block/{block.hash}/status"
b = brk.get_json(path)
m = mempool.get_json(path)
show("GET", path, b, m)
assert_same_values(b, m)

37
packages/brk_client/tests/mempool_compat/blocks/test_block_txid_index.py Normal file
View File

@@ -0,0 +1,37 @@
"""GET /api/block/{hash}/txid/{index}"""
import pytest
from _lib import show
def test_block_txid_at_index_0(brk, mempool, block):
"""Txid at position 0 (coinbase) must match."""
path = f"/api/block/{block.hash}/txid/0"
b = brk.get_text(path)
m = mempool.get_text(path)
show("GET", path, b, m)
assert b == m
def test_block_txid_at_index_1(brk, mempool, block):
"""Txid at position 1 (first non-coinbase) must match."""
txids = mempool.get_json(f"/api/block/{block.hash}/txids")
if len(txids) <= 1:
pytest.skip("block has only coinbase")
path = f"/api/block/{block.hash}/txid/1"
b = brk.get_text(path)
m = mempool.get_text(path)
show("GET", path, b, m)
assert b == m
def test_block_txid_at_last_index(brk, mempool, block):
"""Txid at last position must match."""
txids = mempool.get_json(f"/api/block/{block.hash}/txids")
last = len(txids) - 1
path = f"/api/block/{block.hash}/txid/{last}"
b = brk.get_text(path)
m = mempool.get_text(path)
show("GET", path, b, m)
assert b == m

12
packages/brk_client/tests/mempool_compat/blocks/test_block_txids.py Normal file
View File

@@ -0,0 +1,12 @@
"""GET /api/block/{hash}/txids"""
from _lib import show
def test_block_txids(brk, mempool, block):
"""Ordered txid list must be identical."""
path = f"/api/block/{block.hash}/txids"
b = brk.get_json(path)
m = mempool.get_json(path)
show("GET", path, b[:3], m[:3])
assert b == m

14
packages/brk_client/tests/mempool_compat/blocks/test_block_txs.py Normal file
View File

@@ -0,0 +1,14 @@
"""GET /api/block/{hash}/txs"""
from _lib import assert_same_values, show
def test_block_txs_page0(brk, mempool, block):
"""First page of block transactions must match."""
path = f"/api/block/{block.hash}/txs"
b = brk.get_json(path)
m = mempool.get_json(path)
show("GET", path, f"({len(b)} txs)", f"({len(m)} txs)")
assert len(b) == len(m), f"Page size: brk={len(b)} vs mempool={len(m)}"
if b and m:
assert_same_values(b[0], m[0], exclude={"sigops"})

68
packages/brk_client/tests/mempool_compat/blocks/test_block_txs_start.py Normal file
View File

@@ -0,0 +1,68 @@
"""GET /api/block/{hash}/txs/{start_index}"""
import pytest
from _lib import assert_same_structure, show
def test_block_txs_start_index_25(brk, mempool, block):
"""Paginated txs from index 25 must match (skip small blocks)."""
txids = mempool.get_json(f"/api/block/{block.hash}/txids")
if len(txids) <= 25:
pytest.skip(f"block has only {len(txids)} txs")
path = f"/api/block/{block.hash}/txs/25"
b = brk.get_json(path)
m = mempool.get_json(path)
show("GET", path, f"({len(b)} txs)", f"({len(m)} txs)")
assert len(b) == len(m)
if b and m:
assert_same_structure(b[0], m[0])
def test_block_txs_start_index_zero(brk, mempool, block):
"""`/txs/0` must mirror `/txs` (the default page) in length and structure."""
path0 = f"/api/block/{block.hash}/txs/0"
pathx = f"/api/block/{block.hash}/txs"
b0 = brk.get_json(path0)
bx = brk.get_json(pathx)
show("GET", path0, f"({len(b0)} txs)", f"vs /txs ({len(bx)} txs)")
assert len(b0) == len(bx)
if b0 and bx:
assert b0[0]["txid"] == bx[0]["txid"]
def test_block_txs_start_aligned_pagination(brk, mempool, block):
"""Pages at 0, 25, 50 must each be aligned slices of the full txid list."""
txids = mempool.get_json(f"/api/block/{block.hash}/txids")
if len(txids) <= 50:
pytest.skip(f"block has only {len(txids)} txs")
# mempool.space orders txids tip-first inside the block payload, but
# /txids returns them in block order (coinbase-first). Paged /txs follows
# the same coinbase-first order — so page N starts at offset N.
page0 = brk.get_json(f"/api/block/{block.hash}/txs/0")
page25 = brk.get_json(f"/api/block/{block.hash}/txs/25")
page50 = brk.get_json(f"/api/block/{block.hash}/txs/50")
show("GET", f"/api/block/{block.hash}/txs/{{0,25,50}}",
f"page0={len(page0)} page25={len(page25)} page50={len(page50)}", "—")
# The paging origin is what mempool.space does; verify against the live
# /txids list rather than re-deriving the order ourselves.
assert page0 and page0[0]["txid"] == txids[0]
assert page25 and page25[0]["txid"] == txids[25]
assert page50 and page50[0]["txid"] == txids[50]
def test_block_txs_start_past_end(brk, mempool, block):
"""A start index past the last tx must produce the same response on both servers."""
txids = mempool.get_json(f"/api/block/{block.hash}/txids")
past = len(txids) + 1000
path = f"/api/block/{block.hash}/txs/{past}"
b_resp = brk.get_raw(path)
m_resp = mempool.get_raw(path)
show("GET", path, f"brk={b_resp.status_code}", f"mempool={m_resp.status_code}")
assert b_resp.status_code == m_resp.status_code, (
f"past-end status differs: brk={b_resp.status_code} vs mempool={m_resp.status_code}"
)
if b_resp.status_code == 200:
assert b_resp.json() == m_resp.json(), (
f"past-end body differs: brk={b_resp.json()} vs mempool={m_resp.json()}"
)

22
packages/brk_client/tests/mempool_compat/blocks/test_block_v1.py Normal file
View File

@@ -0,0 +1,22 @@
"""GET /api/v1/block/{hash}"""
from _lib import assert_same_structure, assert_same_values, show
def test_block_v1_extras_all_values(brk, mempool, block):
"""Every shared extras field must match — exposes computation differences."""
path = f"/api/v1/block/{block.hash}"
b = brk.get_json(path)["extras"]
m = mempool.get_json(path)["extras"]
show("GET", f"{path} [extras]", b, m, max_lines=50)
assert_same_structure(b, m)
assert_same_values(b, m)
def test_block_v1_extras_pool(brk, mempool, block):
"""Pool identification structure must match."""
path = f"/api/v1/block/{block.hash}"
bp = brk.get_json(path)["extras"]["pool"]
mp = mempool.get_json(path)["extras"]["pool"]
show("GET", f"{path} [extras.pool]", bp, mp)
assert_same_structure(bp, mp)

14
packages/brk_client/tests/mempool_compat/blocks/test_blocks_height.py Normal file
View File

@@ -0,0 +1,14 @@
"""GET /api/blocks/{height}"""
from _lib import assert_same_values, show
def test_blocks_from_height(brk, mempool, block):
"""Confirmed blocks from a fixed height must match exactly."""
path = f"/api/blocks/{block.height}"
b = brk.get_json(path)
m = mempool.get_json(path)
show("GET", path, f"({len(b)} blocks)", f"({len(m)} blocks)")
assert len(b) == len(m)
if b and m:
assert_same_values(b[0], m[0])

35
packages/brk_client/tests/mempool_compat/blocks/test_blocks_recent.py Normal file
View File

@@ -0,0 +1,35 @@
"""GET /api/blocks (most recent confirmed blocks, no height)"""
from _lib import assert_same_structure, show
def test_blocks_recent_structure(brk, mempool):
"""Recent blocks list must have the same element structure."""
path = "/api/blocks"
b = brk.get_json(path)
m = mempool.get_json(path)
show(
"GET", path,
f"({len(b)} blocks, {b[-1]['height']}-{b[0]['height']})" if b else "[]",
f"({len(m)} blocks, {m[-1]['height']}-{m[0]['height']})" if m else "[]",
)
assert len(b) > 0
assert_same_structure(b, m)
def test_blocks_recent_ordering(brk):
"""Returned blocks must be ordered tip-first by strictly decreasing height."""
b = brk.get_json("/api/blocks")
heights = [blk["height"] for blk in b]
show("GET", "/api/blocks", f"heights={heights[:5]}...", "—")
assert heights == sorted(heights, reverse=True), (
f"blocks are not strictly tip-first: {heights}"
)
assert len(set(heights)) == len(heights), "duplicate heights in /api/blocks"
def test_blocks_recent_count(brk):
"""mempool.space returns up to 15 blocks; brk should match that contract."""
b = brk.get_json("/api/blocks")
show("GET", "/api/blocks", f"({len(b)} blocks)", "—")
assert 1 <= len(b) <= 15, f"unexpected block count: {len(b)}"

37
packages/brk_client/tests/mempool_compat/blocks/test_blocks_tip_hash.py Normal file
View File

@@ -0,0 +1,37 @@
"""GET /api/blocks/tip/hash"""
from _lib import show
def test_blocks_tip_hash_format(brk, mempool):
"""Tip hash must be a valid 64-char hex string on both servers."""
path = "/api/blocks/tip/hash"
b = brk.get_text(path)
m = mempool.get_text(path)
show("GET", path, b, m)
assert len(b) == 64 and all(c in "0123456789abcdef" for c in b.lower())
assert len(m) == 64 and all(c in "0123456789abcdef" for c in m.lower())
def test_blocks_tip_hash_matches_height(brk):
"""`tip/hash` must equal `block-height/{tip_height}`."""
h = int(brk.get_text("/api/blocks/tip/height"))
by_height = brk.get_text(f"/api/block-height/{h}")
tip_hash = brk.get_text("/api/blocks/tip/hash")
show("GET", "/api/blocks/tip/hash", tip_hash, by_height)
# Allow a one-block race if a new block landed between the two fetches.
if tip_hash != by_height:
h2 = int(brk.get_text("/api/blocks/tip/height"))
assert h2 != h or tip_hash == by_height, (
f"tip/hash={tip_hash} but block-height/{h}={by_height}"
)
def test_blocks_tip_hash_matches_recent(brk):
"""`tip/hash` must equal the first hash in `/api/blocks`."""
tip_hash = brk.get_text("/api/blocks/tip/hash")
blocks = brk.get_json("/api/blocks")
show("GET", "/api/blocks/tip/hash", tip_hash, blocks[0]["id"])
assert blocks and blocks[0]["id"] == tip_hash, (
f"tip/hash={tip_hash} but /api/blocks[0].id={blocks[0].get('id')}"
)

32
packages/brk_client/tests/mempool_compat/blocks/test_blocks_tip_height.py Normal file
View File

@@ -0,0 +1,32 @@
"""GET /api/blocks/tip/height"""
from _lib import show
def test_blocks_tip_height_close(brk, mempool):
"""Tip heights must be within a few blocks of each other."""
path = "/api/blocks/tip/height"
b = int(brk.get_text(path))
m = int(mempool.get_text(path))
show("GET", path, b, m)
assert abs(b - m) <= 3, f"Tip heights differ by {abs(b - m)}: brk={b}, mempool={m}"
def test_blocks_tip_height_resolves_to_hash(brk):
"""`tip/height` must resolve to a valid hash via `block-height/{tip}`."""
h = int(brk.get_text("/api/blocks/tip/height"))
bh = brk.get_text(f"/api/block-height/{h}")
show("GET", "/api/blocks/tip/height", h, bh)
assert len(bh) == 64 and all(c in "0123456789abcdef" for c in bh.lower()), (
f"block-height/{h} returned non-hash: {bh!r}"
)
def test_blocks_tip_height_matches_recent(brk):
"""`tip/height` must equal the first element's height in `/api/blocks`."""
h = int(brk.get_text("/api/blocks/tip/height"))
blocks = brk.get_json("/api/blocks")
show("GET", "/api/blocks/tip/height", h, blocks[0]["height"])
assert blocks and blocks[0]["height"] == h, (
f"tip/height={h} but /api/blocks[0].height={blocks[0]['height']}"
)

14
packages/brk_client/tests/mempool_compat/blocks/test_blocks_v1_height.py Normal file
View File

@@ -0,0 +1,14 @@
"""GET /api/v1/blocks/{height}"""
from _lib import assert_same_values, show
def test_blocks_v1_from_height(brk, mempool, block):
"""v1 blocks from a confirmed height — all values must match."""
path = f"/api/v1/blocks/{block.height}"
b = brk.get_json(path)
m = mempool.get_json(path)
show("GET", path, f"({len(b)} blocks)", f"({len(m)} blocks)")
assert len(b) == len(m)
if b and m:
assert_same_values(b[0], m[0])

31
packages/brk_client/tests/mempool_compat/blocks/test_blocks_v1_recent.py Normal file
View File

@@ -0,0 +1,31 @@
"""GET /api/v1/blocks (with extras, no height)"""
from _lib import assert_same_structure, show
def test_blocks_v1_recent_structure(brk, mempool):
"""Recent v1 blocks (with extras) must have the same structure."""
path = "/api/v1/blocks"
b = brk.get_json(path)
m = mempool.get_json(path)
show("GET", path, f"({len(b)} blocks)", f"({len(m)} blocks)")
assert len(b) > 0
assert_same_structure(b, m)
def test_blocks_v1_recent_ordering(brk):
"""v1 blocks must also be tip-first."""
b = brk.get_json("/api/v1/blocks")
heights = [blk["height"] for blk in b]
show("GET", "/api/v1/blocks", f"heights={heights[:5]}...", "—")
assert heights == sorted(heights, reverse=True), (
f"v1 blocks are not strictly tip-first: {heights}"
)
def test_blocks_v1_recent_has_extras(brk):
"""Each v1 block must carry the extras envelope (v1 distinguishes itself from /api/blocks)."""
b = brk.get_json("/api/v1/blocks")
show("GET", "/api/v1/blocks", f"({len(b)} blocks)", "—")
assert b
assert "extras" in b[0], f"v1 blocks element missing 'extras': {list(b[0].keys())}"

158
packages/brk_client/tests/mempool_compat/check_endpoints.py Normal file
View File

@@ -0,0 +1,158 @@
"""
Global registry checks for the mempool.space compatibility suite.
These tests don't poke individual endpoints — they verify the *set* of
endpoints brk exposes matches the registry in `_endpoints.py`. If
mempool.space adds a new endpoint, classify it as covered or skipped here so
this file fails loudly on the next CI run.
Checks:
1. Every `covered` endpoint actually appears in brk's live `/openapi.json`.
2. Every `covered` endpoint has a test file at its declared `test_file` path.
3. Every `skipped` endpoint is NOT exposed by brk (proves the skip is real).
4. Every brk path that *looks* like a mempool path is classified — no
orphan routes that we silently added without registering.
5. Brk extensions listed in `BRK_EXTENSIONS` actually exist on brk.
"""
from pathlib import Path
import pytest
from _endpoints import (
BRK_EXTENSIONS,
MEMPOOL_ENDPOINTS,
Endpoint,
covered_endpoints,
skipped_endpoints,
)
HERE = Path(__file__).parent
# ---- Brk-side discovery -------------------------------------------------
_HTTP_METHODS = {"get", "post", "put", "delete", "patch", "head", "options"}
@pytest.fixture(scope="module")
def brk_routes(brk) -> set[tuple[str, str]]:
"""Every `(METHOD, /api/...)` pair brk reports in its OpenAPI spec."""
spec = brk.get_json("/openapi.json")
return {
(method.upper(), path)
for path, ops in spec["paths"].items()
if path.startswith("/api")
for method in ops.keys()
if method.lower() in _HTTP_METHODS
}
@pytest.fixture(scope="module")
def brk_paths(brk_routes) -> set[str]:
"""Just the path strings (collapsed across methods)."""
return {path for _, path in brk_routes}
@pytest.fixture(scope="module")
def brk_compat_paths(brk_paths) -> set[str]:
"""Brk paths that are part of the mempool.space compat surface.
Strips out brk-only namespaces (series, metrics, urpd, vecs, server, etc.)
so we're left with paths that belong in the registry.
"""
brk_only_prefixes = (
"/api/series",
"/api/metric",
"/api/metrics",
"/api/urpd",
"/api/vecs",
"/api/server",
"/api.json",
)
return {p for p in brk_paths if not p.startswith(brk_only_prefixes)}
# ---- Checks -------------------------------------------------------------
@pytest.mark.parametrize("endpoint", covered_endpoints(), ids=lambda e: e.path)
def test_covered_endpoint_exposed_by_brk(brk_routes, endpoint: Endpoint):
"""Every covered endpoint must appear in brk's OpenAPI under the same method."""
assert (endpoint.method, endpoint.path) in brk_routes, (
f"{endpoint.method} {endpoint.path} is marked covered in _endpoints.py "
f"but brk's /openapi.json doesn't expose it"
)
@pytest.mark.parametrize("endpoint", covered_endpoints(), ids=lambda e: e.path)
def test_covered_endpoint_has_test_file(endpoint: Endpoint):
"""Every covered endpoint must have a test file at its declared path."""
assert endpoint.test_file is not None, (
f"{endpoint.path} is covered but has no test_file"
)
file = HERE / endpoint.test_file
assert file.is_file(), (
f"{endpoint.path} declares test_file={endpoint.test_file!r}, "
f"but {file} doesn't exist"
)
@pytest.mark.parametrize("endpoint", skipped_endpoints(), ids=lambda e: e.path)
def test_skipped_endpoint_not_exposed_by_brk(brk_routes, endpoint: Endpoint):
"""Every skipped endpoint must be absent from brk's OpenAPI for that method."""
assert (endpoint.method, endpoint.path) not in brk_routes, (
f"{endpoint.method} {endpoint.path} is marked skipped "
f"({endpoint.skip_reason!r}) but brk now exposes it — please update "
f"_endpoints.py to mark it covered and add a test"
)
def test_no_orphan_brk_routes(brk_compat_paths):
"""Every brk compat path must be classified in the registry.
If this fails, brk has a route that looks like a mempool.space endpoint
but isn't tracked. Either add it to MEMPOOL_ENDPOINTS (covered + a test)
or to BRK_EXTENSIONS (brk-only with a one-line justification in source).
"""
registry_paths = {e.path for e in MEMPOOL_ENDPOINTS}
extension_paths = set(BRK_EXTENSIONS)
known = registry_paths | extension_paths
orphans = brk_compat_paths - known
assert not orphans, (
f"Brk exposes {len(orphans)} unclassified mempool-style routes:\n "
+ "\n ".join(sorted(orphans))
+ "\nClassify each in mempool_compat/_endpoints.py."
)
@pytest.mark.parametrize("path", BRK_EXTENSIONS, ids=lambda p: p)
def test_brk_extension_actually_exists(brk_paths, path: str):
"""Each path in BRK_EXTENSIONS must exist in brk's OpenAPI.
Stale entries get caught here so the list stays accurate.
"""
assert path in brk_paths, (
f"{path} is listed in BRK_EXTENSIONS but brk's /openapi.json doesn't "
f"expose it — remove it from _endpoints.py"
)
def test_registry_has_no_duplicates():
"""Each (method, path) pair appears at most once in MEMPOOL_ENDPOINTS."""
seen: set[tuple[str, str]] = set()
dups: list[tuple[str, str]] = []
for e in MEMPOOL_ENDPOINTS:
key = (e.method, e.path)
if key in seen:
dups.append(key)
seen.add(key)
assert not dups, f"Duplicate registry entries: {dups}"
def test_skipped_endpoints_have_reason():
"""Every skipped endpoint must include a skip_reason."""
bad = [e for e in skipped_endpoints() if not e.skip_reason]
assert not bad, f"Skipped endpoints missing skip_reason: {[e.path for e in bad]}"

216
packages/brk_client/tests/mempool_compat/conftest.py Normal file
View File

@@ -0,0 +1,216 @@
"""
Shared fixtures for mempool.space compatibility tests.
Helper functions live in `_lib.py`; this file holds only fixtures so pytest
can discover them throughout the subtree. Each subtree test imports helpers
with `from _lib import ...` — the conftest puts this directory on sys.path.
Usage:
cd packages/brk_client
uv run pytest tests/mempool_compat -sv # all
uv run pytest tests/mempool_compat/blocks -sv # one category
uv run pytest tests/mempool_compat/blocks/test_block.py -sv # one endpoint
BRK_URL=http://host:port uv run pytest tests/mempool_compat -sv # custom server
Environment variables:
BRK_URL brk server base URL (default: http://localhost:3110)
MEMPOOL_URL mempool.space base URL (default: https://mempool.space)
RATE_LIMIT seconds between mempool.space requests (default: 0.5)
"""
import os
import sys
import time
from dataclasses import dataclass
from pathlib import Path
from typing import Any
import pytest
import requests
# Make `_lib` and `_endpoints` importable from any nested test file.
sys.path.insert(0, str(Path(__file__).parent))
BRK_BASE = os.environ.get("BRK_URL", "http://localhost:3110")
MEMPOOL_BASE = os.environ.get("MEMPOOL_URL", "https://mempool.space")
RATE_LIMIT = float(os.environ.get("RATE_LIMIT", "0.5"))
class ApiClient:
"""HTTP client for a single API server with optional rate limiting."""
def __init__(self, base_url: str, name: str, rate_limit: float = 0.0):
self.base_url = base_url.rstrip("/")
self.name = name
self.rate_limit = rate_limit
self._last_request = 0.0
self.session = requests.Session()
self.session.headers["User-Agent"] = "brk-compat-test/1.0"
def _wait(self):
if self.rate_limit > 0:
elapsed = time.monotonic() - self._last_request
if elapsed < self.rate_limit:
time.sleep(self.rate_limit - elapsed)
self._last_request = time.monotonic()
def get(self, path: str, params=None, timeout: int = 30) -> requests.Response:
self._wait()
url = f"{self.base_url}{path}"
for _ in range(3):
resp = self.session.get(url, params=params, timeout=timeout)
if resp.status_code == 429:
wait = int(resp.headers.get("Retry-After", 5))
time.sleep(wait)
continue
resp.raise_for_status()
return resp
resp.raise_for_status()
return resp
def get_raw(self, path: str, params=None, timeout: int = 30) -> requests.Response:
"""Like `get` but does not raise on non-2xx — returns the raw response."""
self._wait()
url = f"{self.base_url}{path}"
return self.session.get(url, params=params, timeout=timeout)
def get_json(self, path: str, params=None, timeout: int = 30) -> Any:
return self.get(path, params=params, timeout=timeout).json()
def get_text(self, path: str, params=None, timeout: int = 30) -> str:
return self.get(path, params=params, timeout=timeout).text
def get_bytes(self, path: str, params=None, timeout: int = 30) -> bytes:
return self.get(path, params=params, timeout=timeout).content
# Absolute heights for well-known eras + relative depths for recent blocks.
# Covers: genesis-era, early, mid, post-halving, taproot-era, recent, near-tip.
FIXED_HEIGHTS = [100, 100_000, 400_000, 630_000, 800_000]
RELATIVE_DEPTHS = [1000, 100, 10]
@dataclass
class BlockData:
"""A discovered block with associated txids."""
height: int
hash: str
txid: str
coinbase_txid: str
@dataclass
class LiveData:
"""Live blockchain data discovered at session start."""
tip_height: int
tip_hash: str
blocks: list # list[BlockData] — multiple depths for parametrized tests
addresses: dict # dict[str, str] — keyed by scriptpubkey_type
stable_height: int
stable_hash: str
stable_block: dict
sample_txid: str
coinbase_txid: str
sample_address: str
@pytest.fixture(scope="session")
def brk():
return ApiClient(BRK_BASE, "brk")
@pytest.fixture(scope="session")
def mempool():
return ApiClient(MEMPOOL_BASE, "mempool.space", rate_limit=RATE_LIMIT)
@pytest.fixture(scope="session", autouse=True)
def check_servers(brk, mempool):
"""Fail fast if either server is unreachable."""
try:
brk.get("/api/blocks/tip/height")
except Exception as e:
pytest.exit(f"brk server not reachable at {brk.base_url}: {e}")
try:
mempool.get("/api/blocks/tip/height")
except Exception as e:
pytest.exit(f"mempool.space not reachable at {mempool.base_url}: {e}")
@pytest.fixture(scope="session")
def live(mempool) -> LiveData:
"""Discover live blockchain data once per session.
Picks blocks at multiple depths and extracts addresses of different
scriptpubkey types so parametrized tests cover varied real data.
"""
tip_height = int(mempool.get_text("/api/blocks/tip/height"))
tip_hash = mempool.get_text("/api/blocks/tip/hash")
heights = FIXED_HEIGHTS + [tip_height - d for d in RELATIVE_DEPTHS]
heights.sort()
blocks: list[BlockData] = []
addresses: dict[str, str] = {}
for h in heights:
bh = mempool.get_text(f"/api/block-height/{h}")
txids = mempool.get_json(f"/api/block/{bh}/txids")
coinbase = txids[0]
sample = txids[min(1, len(txids) - 1)]
blocks.append(BlockData(height=h, hash=bh, txid=sample, coinbase_txid=coinbase))
if len(addresses) < 8:
tx = mempool.get_json(f"/api/tx/{sample}")
for vout in tx.get("vout", []):
atype = vout.get("scriptpubkey_type")
addr = vout.get("scriptpubkey_address")
if addr and atype and atype not in addresses:
addresses[atype] = addr
stable = blocks[0]
stable_block = mempool.get_json(f"/api/block/{stable.hash}")
sample_address = next(iter(addresses.values()), "1A1zP1eP5QGefi2DMPTfTL5SLmv7DivfNa")
data = LiveData(
tip_height=tip_height,
tip_hash=tip_hash,
blocks=blocks,
addresses=addresses,
stable_height=stable.height,
stable_hash=stable.hash,
stable_block=stable_block,
sample_txid=stable.txid,
coinbase_txid=stable.coinbase_txid,
sample_address=sample_address,
)
print(f"\n{'=' * 70}")
print(f" LIVE TEST DATA (from {MEMPOOL_BASE})")
print(f"{'=' * 70}")
print(f" tip {data.tip_height} {data.tip_hash[:20]}...")
for i, b in enumerate(blocks):
print(f" block[{i}] {b.height} {b.hash[:20]}... tx={b.txid[:16]}...")
for atype, addr in addresses.items():
print(f" addr {atype:12s} {addr}")
print(f"{'=' * 70}\n")
return data
@pytest.fixture(params=range(8), ids=[
"h100", "h100k", "h400k", "h630k", "h800k", "recent1k", "recent100", "recent10",
])
def block(request, live):
"""One BlockData per id — skip if not discovered for this session."""
i = request.param
if i >= len(live.blocks):
pytest.skip("block not discovered")
return live.blocks[i]
@pytest.fixture()
def live_addrs(live):
"""All dynamically discovered addresses, keyed by scriptpubkey_type."""
return list(live.addresses.items())

27
packages/brk_client/tests/mempool_compat/fees/test_mempool_blocks.py Normal file
View File

@@ -0,0 +1,27 @@
"""GET /api/v1/fees/mempool-blocks"""
from _lib import assert_same_structure, show
def test_fees_mempool_blocks(brk, mempool):
"""Projected mempool blocks must have the same element structure."""
path = "/api/v1/fees/mempool-blocks"
b = brk.get_json(path)
m = mempool.get_json(path)
show("GET", path, f"({len(b)} blocks)", f"({len(m)} blocks)")
assert isinstance(b, list) and isinstance(m, list)
assert len(b) > 0
if b and m:
assert_same_structure(b[0], m[0])
def test_fees_mempool_blocks_fee_range(brk, mempool):
"""Each projected block must have a 7-element feeRange."""
path = "/api/v1/fees/mempool-blocks"
for label, client in [("brk", brk), ("mempool", mempool)]:
blocks = client.get_json(path)
for i, block in enumerate(blocks[:3]):
assert "feeRange" in block, f"{label} block {i} missing feeRange"
assert len(block["feeRange"]) == 7, (
f"{label} block {i} feeRange has {len(block['feeRange'])} items, expected 7"
)

42
packages/brk_client/tests/mempool_compat/fees/test_precise.py Normal file
View File

@@ -0,0 +1,42 @@
"""GET /api/v1/fees/precise"""
from _lib import assert_same_structure, show
EXPECTED_FEE_KEYS = [
"fastestFee", "halfHourFee", "hourFee", "economyFee", "minimumFee",
]
def test_fees_precise_structure(brk, mempool):
"""Precise fees must have the same structure as recommended."""
path = "/api/v1/fees/precise"
b = brk.get_json(path)
m = mempool.get_json(path)
show("GET", path, b, m)
assert_same_structure(b, m)
for key in EXPECTED_FEE_KEYS:
assert key in b
def test_fees_precise_ordering(brk, mempool):
"""Precise fee tiers must be ordered: fastest >= halfHour >= hour >= economy >= minimum."""
path = "/api/v1/fees/precise"
for label, client in [("brk", brk), ("mempool", mempool)]:
d = client.get_json(path)
assert d["fastestFee"] >= d["halfHourFee"] >= d["hourFee"], (
f"{label}: precise fee ordering violated {d}"
)
assert d["hourFee"] >= d["economyFee"] >= d["minimumFee"], (
f"{label}: precise fee ordering violated {d}"
)
def test_fees_precise_numeric(brk):
"""Each tier in /precise must be a non-negative number."""
d = brk.get_json("/api/v1/fees/precise")
show("GET", "/api/v1/fees/precise", d, "—")
for key in EXPECTED_FEE_KEYS:
v = d[key]
assert isinstance(v, (int, float)), f"{key} not numeric: {type(v).__name__}"
assert v >= 0, f"{key} is negative: {v}"

33
packages/brk_client/tests/mempool_compat/fees/test_recommended.py Normal file
View File

@@ -0,0 +1,33 @@
"""GET /api/v1/fees/recommended"""
from _lib import assert_same_structure, show
EXPECTED_FEE_KEYS = [
"fastestFee", "halfHourFee", "hourFee", "economyFee", "minimumFee",
]
def test_fees_recommended(brk, mempool):
"""Recommended fees must have the same keys and numeric types."""
path = "/api/v1/fees/recommended"
b = brk.get_json(path)
m = mempool.get_json(path)
show("GET", path, b, m)
assert_same_structure(b, m)
for key in EXPECTED_FEE_KEYS:
assert key in b, f"brk missing '{key}'"
assert isinstance(b[key], (int, float)), f"'{key}' is not numeric: {type(b[key])}"
def test_fees_recommended_ordering(brk, mempool):
"""Fee tiers must be ordered: fastest >= halfHour >= hour >= economy >= minimum."""
path = "/api/v1/fees/recommended"
for label, client in [("brk", brk), ("mempool", mempool)]:
d = client.get_json(path)
assert d["fastestFee"] >= d["halfHourFee"] >= d["hourFee"], (
f"{label}: fee ordering violated {d}"
)
assert d["hourFee"] >= d["economyFee"] >= d["minimumFee"], (
f"{label}: fee ordering violated {d}"
)

35
packages/brk_client/tests/mempool_compat/general/test_difficulty_adjustment.py Normal file
View File

@@ -0,0 +1,35 @@
"""GET /api/v1/difficulty-adjustment"""
from _lib import assert_same_structure, show
DIFFICULTY_KEYS = [
"progressPercent", "difficultyChange", "estimatedRetargetDate",
"remainingBlocks", "remainingTime", "previousRetarget",
"previousTime", "nextRetargetHeight", "timeAvg",
"adjustedTimeAvg", "timeOffset", "expectedBlocks",
]
def test_difficulty_adjustment(brk, mempool):
"""Difficulty adjustment must have the same structure."""
path = "/api/v1/difficulty-adjustment"
b = brk.get_json(path)
m = mempool.get_json(path)
show("GET", path, b, m)
assert_same_structure(b, m)
for key in DIFFICULTY_KEYS:
assert key in b, f"brk missing '{key}'"
def test_difficulty_adjustment_values_sane(brk, mempool):
"""Progress must be 0-100 %, remaining blocks must be 0-2016."""
path = "/api/v1/difficulty-adjustment"
for label, client in [("brk", brk), ("mempool", mempool)]:
d = client.get_json(path)
assert 0 <= d["progressPercent"] <= 100, (
f"{label} progressPercent out of range: {d['progressPercent']}"
)
assert 0 <= d["remainingBlocks"] <= 2016, (
f"{label} remainingBlocks out of range: {d['remainingBlocks']}"
)

54
packages/brk_client/tests/mempool_compat/general/test_historical_price.py Normal file
View File

@@ -0,0 +1,54 @@
"""GET /api/v1/historical-price (with and without timestamp)"""
import pytest
from _lib import assert_same_structure, show
# Well-known timestamps from different eras
HISTORICAL_TIMESTAMPS = [
1231006505, # genesis block (2009-01-03)
1354116278, # block 210000 — first halving (2012-11-28)
1468082773, # block 420000 — second halving (2016-07-09)
1588788036, # block 630000 — third halving (2020-05-11)
1713571767, # block 840000 — fourth halving (2024-04-20)
]
def test_historical_price(brk, mempool):
"""Historical price must have the same structure."""
path = "/api/v1/historical-price"
b = brk.get_json(path)
m = mempool.get_json(path)
show("GET", path, b, m, max_lines=15)
assert_same_structure(b, m)
assert "prices" in b
assert isinstance(b["prices"], list)
def test_historical_price_at_block_timestamps(brk, mempool, live):
"""Historical price at each discovered block's timestamp must match structure."""
for block in live.blocks:
info = brk.get_json(f"/api/block/{block.hash}")
ts = info["timestamp"]
path = f"/api/v1/historical-price?timestamp={ts}"
b = brk.get_json(path)
m = mempool.get_json(path)
show("GET", path, b, m)
assert_same_structure(b, m)
assert "prices" in b
assert len(b["prices"]) > 0
@pytest.mark.parametrize("ts", HISTORICAL_TIMESTAMPS, ids=[
"genesis", "halving1", "halving2", "halving3", "halving4",
])
def test_historical_price_at_era(brk, mempool, ts):
"""Historical price at well-known timestamps must match structure."""
path = f"/api/v1/historical-price?timestamp={ts}"
b = brk.get_json(path)
m = mempool.get_json(path)
show("GET", path, b, m)
assert_same_structure(b, m)
assert "prices" in b
assert len(b["prices"]) > 0

22
packages/brk_client/tests/mempool_compat/general/test_prices.py Normal file
View File

@@ -0,0 +1,22 @@
"""GET /api/v1/prices"""
from _lib import assert_same_structure, show
def test_prices(brk, mempool):
"""Current price must have the same structure."""
path = "/api/v1/prices"
b = brk.get_json(path)
m = mempool.get_json(path)
show("GET", path, b, m)
assert_same_structure(b, m)
assert "USD" in b
assert "time" in b
def test_prices_positive(brk, mempool):
"""USD price must be a positive number on both servers."""
path = "/api/v1/prices"
for label, client in [("brk", brk), ("mempool", mempool)]:
d = client.get_json(path)
assert d["USD"] > 0, f"{label} USD price is not positive: {d['USD']}"

23
packages/brk_client/tests/mempool_compat/mempool/test_mempool.py Normal file
View File

@@ -0,0 +1,23 @@
"""GET /api/mempool"""
from _lib import assert_same_structure, show
def test_mempool_info(brk, mempool):
"""Mempool stats must have the same keys and types."""
path = "/api/mempool"
b = brk.get_json(path)
m = mempool.get_json(path)
show("GET", path, b, m, max_lines=15)
assert_same_structure(b, m)
assert isinstance(b["count"], int)
assert isinstance(b["vsize"], int)
def test_mempool_info_positive(brk, mempool):
"""Both servers must report a non-empty mempool."""
path = "/api/mempool"
for label, client in [("brk", brk), ("mempool", mempool)]:
d = client.get_json(path)
assert d["count"] > 0, f"{label} mempool count is 0"
assert d["vsize"] > 0, f"{label} mempool vsize is 0"

25
packages/brk_client/tests/mempool_compat/mempool/test_recent.py Normal file
View File

@@ -0,0 +1,25 @@
"""GET /api/mempool/recent"""
from _lib import assert_same_structure, show
def test_mempool_recent(brk, mempool):
"""Recent mempool txs must have the same element structure."""
path = "/api/mempool/recent"
b = brk.get_json(path)
m = mempool.get_json(path)
show("GET", path, b, m)
assert isinstance(b, list) and isinstance(m, list)
assert len(b) > 0
if b and m:
assert_same_structure(b[0], m[0])
def test_mempool_recent_fields(brk, mempool):
"""Each recent tx must have txid, fee, vsize, value."""
path = "/api/mempool/recent"
for label, client in [("brk", brk), ("mempool", mempool)]:
txs = client.get_json(path)
for tx in txs[:3]:
for key in ["txid", "fee", "vsize", "value"]:
assert key in tx, f"{label} recent tx missing '{key}': {tx}"

46
packages/brk_client/tests/mempool_compat/mempool/test_txids.py Normal file
View File

@@ -0,0 +1,46 @@
"""GET /api/mempool/txids"""
from _lib import show
HEX = set("0123456789abcdef")
def test_mempool_txids_basic(brk, mempool):
"""Txid list must be a non-empty array of strings on both servers."""
path = "/api/mempool/txids"
b = brk.get_json(path)
m = mempool.get_json(path)
show("GET", path, f"({len(b)} txids)", f"({len(m)} txids)")
assert isinstance(b, list) and isinstance(m, list)
assert len(b) > 0, "brk mempool has no txids"
assert isinstance(b[0], str) and len(b[0]) == 64
def test_mempool_txids_format(brk):
"""Every txid in brk's mempool list must be a 64-char lowercase hex string."""
b = brk.get_json("/api/mempool/txids")
show("GET", "/api/mempool/txids", f"({len(b)} txids)", "—")
bad = [t for t in b if not (isinstance(t, str) and len(t) == 64 and set(t.lower()) <= HEX)]
assert not bad, f"{len(bad)} malformed txid(s), e.g. {bad[0] if bad else None!r}"
def test_mempool_txids_unique(brk):
"""Brk's mempool txid list must not contain duplicates."""
b = brk.get_json("/api/mempool/txids")
show("GET", "/api/mempool/txids", f"({len(b)} txids)", "—")
assert len(b) == len(set(b)), (
f"duplicate txids: {len(b) - len(set(b))} duplicates out of {len(b)}"
)
def test_mempool_txids_count_matches_summary(brk):
"""`/api/mempool/txids` length must match `/api/mempool`'s `count` field."""
txids = brk.get_json("/api/mempool/txids")
summary = brk.get_json("/api/mempool")
show("GET", "/api/mempool/txids", f"len={len(txids)}", f"count={summary.get('count')}")
# Allow a small drift (1-2) since the mempool is updated asynchronously
# between the two fetches.
assert abs(len(txids) - summary["count"]) <= 5, (
f"txids={len(txids)} vs /api/mempool.count={summary['count']}"
)

17
packages/brk_client/tests/mempool_compat/mining/conftest.py Normal file
View File

@@ -0,0 +1,17 @@
"""Mining-specific fixtures shared by every mining test in this folder."""
import pytest
@pytest.fixture(scope="module")
def pool_slugs(mempool):
"""Top 3 active pool slugs from the last week."""
data = mempool.get_json("/api/v1/mining/pools/1w")
pools = data.get("pools", []) if isinstance(data, dict) else []
slugs = [p["slug"] for p in pools if p.get("blockCount", 0) > 0][:3]
return slugs or ["foundryusa"]
@pytest.fixture(scope="module")
def pool_slug(pool_slugs):
return pool_slugs[0]

15
packages/brk_client/tests/mempool_compat/mining/test_blocks_fee_rates.py Normal file
View File

@@ -0,0 +1,15 @@
"""GET /api/v1/mining/blocks/fee-rates/{time_period}"""
import pytest
from _lib import assert_same_structure, show, summary
@pytest.mark.parametrize("period", ["24h", "3d", "1w", "1m", "3m", "6m", "1y"])
def test_mining_blocks_fee_rates(brk, mempool, period):
"""Block fee-rate percentiles must have the same element structure."""
path = f"/api/v1/mining/blocks/fee-rates/{period}"
b = brk.get_json(path)
m = mempool.get_json(path)
show("GET", path, summary(b), summary(m))
assert_same_structure(b, m)

15
packages/brk_client/tests/mempool_compat/mining/test_blocks_fees.py Normal file
View File

@@ -0,0 +1,15 @@
"""GET /api/v1/mining/blocks/fees/{time_period}"""
import pytest
from _lib import assert_same_structure, show, summary
@pytest.mark.parametrize("period", ["24h", "3d", "1w", "1m", "3m", "6m", "1y"])
def test_mining_blocks_fees(brk, mempool, period):
"""Average block fees must have the same element structure."""
path = f"/api/v1/mining/blocks/fees/{period}"
b = brk.get_json(path)
m = mempool.get_json(path)
show("GET", path, summary(b), summary(m))
assert_same_structure(b, m)

15
packages/brk_client/tests/mempool_compat/mining/test_blocks_rewards.py Normal file
View File

@@ -0,0 +1,15 @@
"""GET /api/v1/mining/blocks/rewards/{time_period}"""
import pytest
from _lib import assert_same_structure, show, summary
@pytest.mark.parametrize("period", ["24h", "3d", "1w", "1m", "3m", "6m", "1y"])
def test_mining_blocks_rewards(brk, mempool, period):
"""Average block rewards must have the same element structure."""
path = f"/api/v1/mining/blocks/rewards/{period}"
b = brk.get_json(path)
m = mempool.get_json(path)
show("GET", path, summary(b), summary(m))
assert_same_structure(b, m)

15
packages/brk_client/tests/mempool_compat/mining/test_blocks_sizes_weights.py Normal file
View File

@@ -0,0 +1,15 @@
"""GET /api/v1/mining/blocks/sizes-weights/{time_period}"""
import pytest
from _lib import assert_same_structure, show, summary
@pytest.mark.parametrize("period", ["24h", "3d", "1w", "1m", "3m", "6m", "1y"])
def test_mining_blocks_sizes_weights(brk, mempool, period):
"""Block sizes and weights must have the same structure."""
path = f"/api/v1/mining/blocks/sizes-weights/{period}"
b = brk.get_json(path)
m = mempool.get_json(path)
show("GET", path, summary(b), summary(m))
assert_same_structure(b, m)

15
packages/brk_client/tests/mempool_compat/mining/test_blocks_timestamp.py Normal file
View File

@@ -0,0 +1,15 @@
"""GET /api/v1/mining/blocks/timestamp/{timestamp}"""
from _lib import assert_same_structure, show
def test_mining_blocks_timestamp(brk, mempool, live):
"""Block lookup by timestamp must have the same structure for various eras."""
for block in live.blocks:
info = brk.get_json(f"/api/block/{block.hash}")
ts = info["timestamp"]
path = f"/api/v1/mining/blocks/timestamp/{ts}"
b = brk.get_json(path)
m = mempool.get_json(path)
show("GET", path, b, m)
assert_same_structure(b, m)

15
packages/brk_client/tests/mempool_compat/mining/test_difficulty_adjustments.py Normal file
View File

@@ -0,0 +1,15 @@
"""GET /api/v1/mining/difficulty-adjustments/{time_period}"""
import pytest
from _lib import assert_same_structure, show, summary
@pytest.mark.parametrize("period", ["24h", "3d", "1w", "1m", "3m", "6m", "1y", "2y", "3y"])
def test_mining_difficulty_adjustments(brk, mempool, period):
"""Historical difficulty adjustments must have the same structure."""
path = f"/api/v1/mining/difficulty-adjustments/{period}"
b = brk.get_json(path)
m = mempool.get_json(path)
show("GET", path, summary(b), summary(m))
assert_same_structure(b, m)

15
packages/brk_client/tests/mempool_compat/mining/test_hashrate.py Normal file
View File

@@ -0,0 +1,15 @@
"""GET /api/v1/mining/hashrate/{time_period}"""
import pytest
from _lib import assert_same_structure, show, summary
@pytest.mark.parametrize("period", ["24h", "3d", "1w", "1m", "3m", "6m", "1y", "2y", "3y"])
def test_mining_hashrate(brk, mempool, period):
"""Network hashrate + difficulty must have the same structure."""
path = f"/api/v1/mining/hashrate/{period}"
b = brk.get_json(path)
m = mempool.get_json(path)
show("GET", path, summary(b), summary(m))
assert_same_structure(b, m)

15
packages/brk_client/tests/mempool_compat/mining/test_hashrate_pools.py Normal file
View File

@@ -0,0 +1,15 @@
"""GET /api/v1/mining/hashrate/pools/{time_period}"""
import pytest
from _lib import assert_same_structure, show, summary
@pytest.mark.parametrize("period", ["24h", "3d", "1w", "1m", "3m", "1y"])
def test_mining_hashrate_pools(brk, mempool, period):
"""Per-pool hashrate must have the same structure."""
path = f"/api/v1/mining/hashrate/pools/{period}"
b = brk.get_json(path)
m = mempool.get_json(path)
show("GET", path, summary(b), summary(m))
assert_same_structure(b, m)

13
packages/brk_client/tests/mempool_compat/mining/test_pool.py Normal file
View File

@@ -0,0 +1,13 @@
"""GET /api/v1/mining/pool/{slug}"""
from _lib import assert_same_structure, show, summary
def test_mining_pool_detail(brk, mempool, pool_slugs):
"""Pool detail must have the same structure for top pools."""
for slug in pool_slugs:
path = f"/api/v1/mining/pool/{slug}"
b = brk.get_json(path)
m = mempool.get_json(path)
show("GET", path, summary(b), summary(m))
assert_same_structure(b, m)

15
packages/brk_client/tests/mempool_compat/mining/test_pool_blocks.py Normal file
View File

@@ -0,0 +1,15 @@
"""GET /api/v1/mining/pool/{slug}/blocks"""
from _lib import assert_same_structure, show
def test_mining_pool_blocks(brk, mempool, pool_slugs):
"""Recent blocks by pool must have the same element structure."""
for slug in pool_slugs:
path = f"/api/v1/mining/pool/{slug}/blocks"
b = brk.get_json(path)
m = mempool.get_json(path)
show("GET", path, f"({len(b)} blocks)", f"({len(m)} blocks)")
assert isinstance(b, list) and isinstance(m, list)
if b and m:
assert_same_structure(b[0], m[0])

15
packages/brk_client/tests/mempool_compat/mining/test_pool_blocks_height.py Normal file
View File

@@ -0,0 +1,15 @@
"""GET /api/v1/mining/pool/{slug}/blocks/{height}"""
from _lib import assert_same_structure, show
def test_mining_pool_blocks_at_height(brk, mempool, pool_slug, live):
"""Pool blocks before various heights must have the same element structure."""
for block in live.blocks[::2]: # every other block, to keep run-time bounded
path = f"/api/v1/mining/pool/{pool_slug}/blocks/{block.height}"
b = brk.get_json(path)
m = mempool.get_json(path)
show("GET", path, f"({len(b)} blocks)", f"({len(m)} blocks)")
assert isinstance(b, list) and isinstance(m, list)
if b and m:
assert_same_structure(b[0], m[0])

13
packages/brk_client/tests/mempool_compat/mining/test_pool_hashrate.py Normal file
View File

@@ -0,0 +1,13 @@
"""GET /api/v1/mining/pool/{slug}/hashrate"""
from _lib import assert_same_structure, show, summary
def test_mining_pool_hashrate(brk, mempool, pool_slugs):
"""Pool hashrate history must have the same structure for top pools."""
for slug in pool_slugs:
path = f"/api/v1/mining/pool/{slug}/hashrate"
b = brk.get_json(path)
m = mempool.get_json(path)
show("GET", path, summary(b), summary(m))
assert_same_structure(b, m)

45
packages/brk_client/tests/mempool_compat/mining/test_pools.py Normal file
View File

@@ -0,0 +1,45 @@
"""GET /api/v1/mining/pools"""
from _lib import assert_same_structure, show
def test_mining_pools_list_structure(brk, mempool):
"""Pool list must have the same element structure."""
path = "/api/v1/mining/pools"
b = brk.get_json(path)
m = mempool.get_json(path)
show(
"GET", path,
b[:3] if isinstance(b, list) else b,
m[:3] if isinstance(m, list) else m,
)
assert_same_structure(b, m)
def _pools(data):
"""`pools` may live at the root or inside an envelope across versions."""
if isinstance(data, list):
return data
return data.get("pools", []) if isinstance(data, dict) else []
def test_mining_pools_list_fields(brk):
"""Each pool entry must carry slug and name (period-less endpoint omits stats)."""
b = _pools(brk.get_json("/api/v1/mining/pools"))
show("GET", "/api/v1/mining/pools", f"({len(b)} pools)", "—")
assert b, "no pools in brk's response"
required = {"slug", "name"}
for p in b[:5]:
missing = required - set(p.keys())
assert not missing, f"pool {p.get('slug', '?')} missing fields: {missing}"
assert isinstance(p["name"], str) and p["name"]
def test_mining_pools_slugs_unique(brk):
"""Pool slugs must be unique across the response."""
b = _pools(brk.get_json("/api/v1/mining/pools"))
slugs = [p["slug"] for p in b]
show("GET", "/api/v1/mining/pools", f"({len(slugs)} slugs)", "—")
assert len(slugs) == len(set(slugs)), (
f"duplicate slugs: {len(slugs) - len(set(slugs))}"
)

15
packages/brk_client/tests/mempool_compat/mining/test_pools_period.py Normal file
View File

@@ -0,0 +1,15 @@
"""GET /api/v1/mining/pools/{time_period}"""
import pytest
from _lib import assert_same_structure, show, summary
@pytest.mark.parametrize("period", ["24h", "3d", "1w", "1m", "3m", "6m", "1y", "2y", "3y", "all"])
def test_mining_pools_by_period(brk, mempool, period):
"""Pool stats for a time period must have the same structure."""
path = f"/api/v1/mining/pools/{period}"
b = brk.get_json(path)
m = mempool.get_json(path)
show("GET", path, summary(b), summary(m))
assert_same_structure(b, m)

15
packages/brk_client/tests/mempool_compat/mining/test_reward_stats.py Normal file
View File

@@ -0,0 +1,15 @@
"""GET /api/v1/mining/reward-stats/{block_count}"""
import pytest
from _lib import assert_same_structure, show
@pytest.mark.parametrize("block_count", [10, 100, 500])
def test_mining_reward_stats(brk, mempool, block_count):
"""Reward stats must have the same structure."""
path = f"/api/v1/mining/reward-stats/{block_count}"
b = brk.get_json(path)
m = mempool.get_json(path)
show("GET", path, b, m)
assert_same_structure(b, m)

12
packages/brk_client/tests/mempool_compat/transactions/test_cpfp.py Normal file
View File

@@ -0,0 +1,12 @@
"""GET /api/v1/cpfp/{txid}"""
from _lib import assert_same_structure, show
def test_cpfp(brk, mempool, block):
"""CPFP info structure must match for a confirmed tx."""
path = f"/api/v1/cpfp/{block.txid}"
b = brk.get_json(path)
m = mempool.get_json(path)
show("GET", path, b, m)
assert_same_structure(b, m)

40
packages/brk_client/tests/mempool_compat/transactions/test_post_tx.py Normal file
View File

@@ -0,0 +1,40 @@
"""POST /api/tx (broadcast)
We can't actually broadcast a real transaction in a test, so we send a
clearly malformed payload and verify both servers reject it with 4xx. The
goal is to confirm the endpoint exists and behaves like a transaction
broadcaster — not to push live transactions.
"""
from _lib import show
def test_post_tx_invalid_hex(brk, mempool):
"""Both servers must reject an obviously invalid hex payload with 4xx."""
path = "/api/tx"
bad_hex = "deadbeef" # too short to be a valid serialized transaction
b = brk.session.post(f"{brk.base_url}{path}", data=bad_hex, timeout=15)
mempool._wait()
m = mempool.session.post(f"{mempool.base_url}{path}", data=bad_hex, timeout=15)
show("POST", path, f"brk={b.status_code}", f"mempool={m.status_code}")
assert 400 <= b.status_code < 500, (
f"brk POST /api/tx with garbage should 4xx, got {b.status_code}: {b.text!r}"
)
assert 400 <= m.status_code < 500, (
f"mempool POST /api/tx with garbage should 4xx, got {m.status_code}: {m.text!r}"
)
def test_post_tx_empty_body(brk, mempool):
"""Both servers must reject an empty body with 4xx."""
path = "/api/tx"
b = brk.session.post(f"{brk.base_url}{path}", data="", timeout=15)
mempool._wait()
m = mempool.session.post(f"{mempool.base_url}{path}", data="", timeout=15)
show("POST", path, f"brk={b.status_code}", f"mempool={m.status_code}")
assert 400 <= b.status_code < 500
assert 400 <= m.status_code < 500

56
packages/brk_client/tests/mempool_compat/transactions/test_transaction_times.py Normal file
View File

@@ -0,0 +1,56 @@
"""GET /api/v1/transaction-times?txId[]=..."""
from _lib import show
def test_transaction_times_few(brk, mempool, live):
"""First-seen timestamps must match for a few txids."""
txids = [b.txid for b in live.blocks[:3]]
params = [("txId[]", t) for t in txids]
path = "/api/v1/transaction-times"
b = brk.get_json(path, params=params)
m = mempool.get_json(path, params=params)
show("GET", f"{path}?txId[]={{{len(txids)} txids}}", b, m)
assert isinstance(b, list) and isinstance(m, list)
assert len(b) == len(m) == len(txids)
assert b == m, f"timestamps differ: brk={b} vs mempool={m}"
def test_transaction_times_many(brk, mempool, live):
"""A larger batch (covering all sample blocks + coinbases) must match exactly."""
txids = [b.txid for b in live.blocks] + [b.coinbase_txid for b in live.blocks]
params = [("txId[]", t) for t in txids]
path = "/api/v1/transaction-times"
b = brk.get_json(path, params=params)
m = mempool.get_json(path, params=params)
show("GET", f"{path}?txId[]={{{len(txids)} txids}}", f"({len(b)})", f"({len(m)})")
assert len(b) == len(m) == len(txids)
assert b == m, f"timestamps differ: brk={b} vs mempool={m}"
def test_transaction_times_single(brk, mempool, live):
"""A single-element batch must return a 1-element list with the same value."""
txid = live.sample_txid
params = [("txId[]", txid)]
path = "/api/v1/transaction-times"
b = brk.get_json(path, params=params)
m = mempool.get_json(path, params=params)
show("GET", f"{path}?txId[]={txid[:16]}...", b, m)
assert isinstance(b, list) and isinstance(m, list)
assert len(b) == len(m) == 1
assert b == m, f"single timestamp differs: brk={b} vs mempool={m}"
def test_transaction_times_empty(brk, mempool):
"""An empty batch must be rejected (any non-2xx) on both servers.
mempool.space returns 500 — technically a server-side bug (it should be a
4xx since the request itself is malformed) — so we don't insist on exact
status parity, only that neither server silently treats it as valid input.
"""
path = "/api/v1/transaction-times"
b_resp = brk.get_raw(path)
m_resp = mempool.get_raw(path)
show("GET", path, f"brk={b_resp.status_code}", f"mempool={m_resp.status_code}")
assert not b_resp.ok, f"brk accepted empty batch with {b_resp.status_code}: {b_resp.text!r}"
assert not m_resp.ok, f"mempool accepted empty batch with {m_resp.status_code}"

21
packages/brk_client/tests/mempool_compat/transactions/test_tx.py Normal file
View File

@@ -0,0 +1,21 @@
"""GET /api/tx/{txid}"""
from _lib import assert_same_values, show
def test_tx_by_id(brk, mempool, block):
"""Full transaction data must match for a confirmed tx."""
path = f"/api/tx/{block.txid}"
b = brk.get_json(path)
m = mempool.get_json(path)
show("GET", path, b, m)
assert_same_values(b, m, exclude={"sigops"})
def test_tx_coinbase(brk, mempool, block):
"""Coinbase transaction must match."""
path = f"/api/tx/{block.coinbase_txid}"
b = brk.get_json(path)
m = mempool.get_json(path)
show("GET", path, b, m)
assert_same_values(b, m, exclude={"sigops"})

12
packages/brk_client/tests/mempool_compat/transactions/test_tx_hex.py Normal file
View File

@@ -0,0 +1,12 @@
"""GET /api/tx/{txid}/hex"""
from _lib import show
def test_tx_hex(brk, mempool, block):
"""Raw transaction hex must be identical."""
path = f"/api/tx/{block.txid}/hex"
b = brk.get_text(path)
m = mempool.get_text(path)
show("GET", path, b[:80] + "...", m[:80] + "...")
assert b == m

12
packages/brk_client/tests/mempool_compat/transactions/test_tx_merkle_proof.py Normal file
View File

@@ -0,0 +1,12 @@
"""GET /api/tx/{txid}/merkle-proof"""
from _lib import assert_same_values, show
def test_tx_merkle_proof(brk, mempool, block):
"""Merkle inclusion proof must match."""
path = f"/api/tx/{block.txid}/merkle-proof"
b = brk.get_json(path)
m = mempool.get_json(path)
show("GET", path, b, m)
assert_same_values(b, m)

12
packages/brk_client/tests/mempool_compat/transactions/test_tx_merkleblock_proof.py Normal file
View File

@@ -0,0 +1,12 @@
"""GET /api/tx/{txid}/merkleblock-proof"""
from _lib import show
def test_tx_merkleblock_proof(brk, mempool, block):
"""BIP37 merkleblock proof hex must be identical."""
path = f"/api/tx/{block.txid}/merkleblock-proof"
b = brk.get_text(path)
m = mempool.get_text(path)
show("GET", path, b[:80] + "...", m[:80] + "...")
assert b == m

38
packages/brk_client/tests/mempool_compat/transactions/test_tx_outspend.py Normal file
View File

@@ -0,0 +1,38 @@
"""GET /api/tx/{txid}/outspend/{vout}"""
from _lib import assert_same_values, show
def test_tx_outspend_first(brk, mempool, block):
"""Spending status of vout 0 must match exactly."""
path = f"/api/tx/{block.txid}/outspend/0"
b = brk.get_json(path)
m = mempool.get_json(path)
show("GET", path, b, m)
assert_same_values(b, m)
def test_tx_outspend_last(brk, mempool, block):
"""Spending status of the last vout must also match exactly."""
tx = mempool.get_json(f"/api/tx/{block.txid}")
last_vout = len(tx["vout"]) - 1
path = f"/api/tx/{block.txid}/outspend/{last_vout}"
b = brk.get_json(path)
m = mempool.get_json(path)
show("GET", path, b, m)
assert_same_values(b, m)
def test_tx_outspend_out_of_range(brk, mempool, block):
"""A vout index past the last output must produce the same response on both servers.
Both servers return `{"spent": false}` rather than 4xx — they don't bound-check
the vout index. The compat property is that they agree.
"""
tx = mempool.get_json(f"/api/tx/{block.txid}")
bad_vout = len(tx["vout"]) + 100
path = f"/api/tx/{block.txid}/outspend/{bad_vout}"
b = brk.get_json(path)
m = mempool.get_json(path)
show("GET", path, b, m)
assert b == m, f"out-of-range outspend disagrees: brk={b} vs mempool={m}"

12
packages/brk_client/tests/mempool_compat/transactions/test_tx_outspends.py Normal file
View File

@@ -0,0 +1,12 @@
"""GET /api/tx/{txid}/outspends"""
from _lib import assert_same_values, show
def test_tx_outspends(brk, mempool, block):
"""Spending status of all outputs must match exactly."""
path = f"/api/tx/{block.txid}/outspends"
b = brk.get_json(path)
m = mempool.get_json(path)
show("GET", path, b, m)
assert_same_values(b, m)

12
packages/brk_client/tests/mempool_compat/transactions/test_tx_raw.py Normal file
View File

@@ -0,0 +1,12 @@
"""GET /api/tx/{txid}/raw"""
from _lib import show
def test_tx_raw(brk, mempool, block):
"""Raw transaction bytes must be identical."""
path = f"/api/tx/{block.txid}/raw"
b = brk.get_bytes(path)
m = mempool.get_bytes(path)
show("GET", path, f"<{len(b)} bytes>", f"<{len(m)} bytes>")
assert b == m

16
packages/brk_client/tests/mempool_compat/transactions/test_tx_rbf.py Normal file
View File

@@ -0,0 +1,16 @@
"""GET /api/v1/tx/{txid}/rbf
For confirmed transactions both servers return an empty/null replacement
set; the structure is what's load-bearing here.
"""
from _lib import assert_same_structure, show
def test_tx_rbf_for_confirmed(brk, mempool, block):
"""RBF replacement timeline structure must match for a confirmed tx."""
path = f"/api/v1/tx/{block.txid}/rbf"
b = brk.get_json(path)
m = mempool.get_json(path)
show("GET", path, b, m)
assert_same_structure(b, m)

12
packages/brk_client/tests/mempool_compat/transactions/test_tx_status.py Normal file
View File

@@ -0,0 +1,12 @@
"""GET /api/tx/{txid}/status"""
from _lib import assert_same_values, show
def test_tx_status(brk, mempool, block):
"""Confirmation status must match for a confirmed tx."""
path = f"/api/tx/{block.txid}/status"
b = brk.get_json(path)
m = mempool.get_json(path)
show("GET", path, b, m)
assert_same_values(b, m)

14
packages/brk_client/tests/test_basic.py
View File

@@ -57,3 +57,17 @@ def test_fetch_typed_series():
print(e)
f = client.series.prices.ohlc.usd.by.day1().tail(10).fetch()
print(f)
def test_endpoint_len():
client = BrkClient("http://localhost:3110")
n = client.series.prices.split.close.usd.by.day1().len()
assert isinstance(n, int)
assert n > 0
def test_endpoint_version():
client = BrkClient("http://localhost:3110")
v = client.series.prices.split.close.usd.by.day1().version()
assert isinstance(v, int)
assert v >= 1

8
packages/brk_client/tests/test_metric_data.py
View File

@@ -18,7 +18,6 @@ def day1_metric():
version=1,
index="day1",
type="n",
total=100,
start=0,
end=5,
stamp="2024-01-01T00:00:00Z",
@@ -33,7 +32,6 @@ def height_metric():
version=1,
index="height",
type="n",
total=1000,
start=800000,
end=800005,
stamp="2024-01-01T00:00:00Z",
@@ -48,7 +46,6 @@ def month1_metric():
version=1,
index="month1",
type="n",
total=200,
start=0,
end=3,
stamp="2024-01-01T00:00:00Z",
@@ -63,7 +60,6 @@ def hour1_metric():
version=1,
index="hour1",
type="n",
total=200000,
start=0,
end=3,
stamp="2024-01-01T00:00:00Z",
@@ -78,7 +74,6 @@ def week1_metric():
version=1,
index="week1",
type="n",
total=800,
start=0,
end=3,
stamp="2024-01-01T00:00:00Z",
@@ -93,7 +88,6 @@ def year1_metric():
version=1,
index="year1",
type="n",
total=20,
start=0,
end=3,
stamp="2024-01-01T00:00:00Z",
@@ -108,7 +102,6 @@ def day3_metric():
version=1,
index="day3",
type="n",
total=2000,
start=0,
end=3,
stamp="2024-01-01T00:00:00Z",
@@ -123,7 +116,6 @@ def empty_metric():
version=1,
index="day1",
type="n",
total=100,
start=5,
end=5,
stamp="2024-01-01T00:00:00Z",