global: fixes

This commit is contained in:
nym21
2026-05-03 00:57:22 +02:00
parent 2b8a0a8cf7
commit 9cb5f2c880
13 changed files with 120 additions and 74 deletions

View File

@@ -49,67 +49,44 @@ def test_address_txs_shape_dynamic(brk, mempool, live_addrs):
@pytest.mark.parametrize("addr", STATIC_ADDRS)
def test_address_txs_ordering(brk, addr):
"""All entries must be confirmed and heights monotonically non-increasing."""
"""Response is mempool-prefix (unconfirmed, newest-first) + chain-suffix (confirmed, height-desc)."""
b = brk.get_address_txs(addr)
if not b:
pytest.skip(f"{addr} has no txs in brk")
for tx in b:
assert tx["status"]["confirmed"] is True, (
f"{addr} returned unconfirmed tx {tx['txid']} (this endpoint is chain-only on brk)"
)
heights = [tx["status"]["block_height"] for tx in b]
confirmed_flags = [tx["status"]["confirmed"] for tx in b]
assert confirmed_flags == sorted(confirmed_flags), (
f"{addr}: confirmed flags must be False*..*True* (mempool prefix then chain), got "
f"{confirmed_flags[:10]}..."
)
chain = [tx for tx in b if tx["status"]["confirmed"]]
heights = [tx["status"]["block_height"] for tx in chain]
assert heights == sorted(heights, reverse=True), (
f"{addr} not newest-first by height: {heights[:5]}..."
f"{addr} chain segment not newest-first by height: {heights[:5]}..."
)
@pytest.mark.parametrize("addr", STATIC_ADDRS)
def test_address_txs_limit(brk, addr):
"""Hard cap of 50 confirmed txs per call."""
"""Hard cap of 50 entries per call (mempool first, chain fills remainder)."""
b = brk.get_address_txs(addr)
assert len(b) <= 50, f"{addr} returned {len(b)} txs, exceeds 50-cap"
@pytest.mark.parametrize("addr", STABLE_ADDRS)
def test_address_txs_top_match_stable(brk, mempool, addr):
"""For inactive historical addresses, brk and mempool agree on first-page order."""
b_txids = [t["txid"] for t in brk.get_address_txs(addr)]
m_txids = [t["txid"] for t in mempool.get_json(f"/api/address/{addr}/txs")]
assert b_txids == m_txids, (
f"{addr} first-page txid order diverges:\n"
f" brk: {b_txids[:5]}...\n"
f" mempool: {m_txids[:5]}..."
)
def test_address_txs_pagination(brk, mempool):
"""`after_txid` returns a fresh, strictly-older page; matches mempool.space."""
addr = "3D2oetdNuZUqQHPJmcMDDHYoqkyNVsFk9r"
first = brk.get_address_txs(addr)
assert len(first) == 50, f"expected full first page, got {len(first)}"
last_txid = first[-1]["txid"]
last_height = first[-1]["status"]["block_height"]
second = brk.get_address_txs(addr, after_txid=last_txid)
assert second, "second page must be non-empty for a 5700-tx address"
first_txids = {t["txid"] for t in first}
second_txids = {t["txid"] for t in second}
assert not (first_txids & second_txids), "pagination must not return overlapping txs"
for tx in second:
assert tx["status"]["block_height"] <= last_height, (
f"page 2 tx {tx['txid']} at height {tx['status']['block_height']} "
f"exceeds page-1 tail height {last_height}"
)
m_second = mempool.get_json(f"/api/address/{addr}/txs?after_txid={last_txid}")
b_ids = [t["txid"] for t in second]
m_ids = [t["txid"] for t in m_second]
assert b_ids == m_ids, (
f"page-2 order diverges from mempool:\n"
f" brk: {b_ids[:5]}...\n"
f" mempool: {m_ids[:5]}..."
"""For inactive historical addresses, the confirmed tail must agree exactly with mempool.space."""
b_chain = [t["txid"] for t in brk.get_address_txs(addr) if t["status"]["confirmed"]]
m_chain = [
t["txid"]
for t in mempool.get_json(f"/api/address/{addr}/txs")
if t["status"]["confirmed"]
]
assert b_chain == m_chain, (
f"{addr} confirmed-tail txid order diverges:\n"
f" brk: {b_chain[:5]}...\n"
f" mempool: {m_chain[:5]}..."
)

View File

@@ -69,8 +69,8 @@ def test_difficulty_adjustment_invariants(brk):
assert 1_000 <= d["timeAvg"] <= 3_600_000
assert 1_000 <= d["adjustedTimeAvg"] <= 3_600_000
# remainingTime is constructed as remainingBlocks * timeAvg in brk.
assert d["remainingTime"] == d["remainingBlocks"] * d["timeAvg"]
# remainingTime is remainingBlocks * adjustedTimeAvg (matches mempool.space).
assert d["remainingTime"] == d["remainingBlocks"] * d["adjustedTimeAvg"]
assert d["estimatedRetargetDate"] > now_ms
assert d["previousTime"] * 1000 < now_ms

View File

@@ -55,3 +55,26 @@ def test_cpfp_malformed_short(brk, bad):
with pytest.raises(BrkError) as exc_info:
brk.get_text(f"/api/v1/cpfp/{bad}")
assert exc_info.value.status == 400
def test_cpfp_mempool_unconfirmed(brk, mempool):
"""Unconfirmed mempool tx: brk and mempool.space agree on cpfp shape."""
txids = mempool.get_json("/api/mempool/txids")
if not txids:
pytest.skip("mempool.space mempool currently empty")
for txid in txids[:50]:
try:
b = brk.get_cpfp(txid)
except BrkError:
continue
try:
m = mempool.get_json(f"/api/v1/cpfp/{txid}")
except Exception:
continue
show("GET", f"/api/v1/cpfp/{txid}", b, m)
assert_same_structure(b, m)
assert isinstance(b.get("ancestors"), list)
assert isinstance(b.get("descendants", []), list)
return
pytest.skip("no shared unconfirmed tx between brk and mempool.space")

View File

@@ -49,3 +49,29 @@ def test_tx_status_malformed_unknown(brk):
with pytest.raises(BrkError) as exc_info:
brk.get_text(f"/api/tx/{bad}/status")
assert exc_info.value.status == 404
def test_tx_status_mempool_unconfirmed(brk, mempool):
"""Unconfirmed mempool tx: status must be confirmed=false with no block fields."""
txids = mempool.get_json("/api/mempool/txids")
if not txids:
pytest.skip("mempool.space mempool currently empty")
for txid in txids[:25]:
try:
b = brk.get_tx_status(txid)
except BrkError:
continue
if b.get("confirmed"):
continue
try:
m = mempool.get_json(f"/api/tx/{txid}/status")
except Exception:
continue
if m.get("confirmed"):
continue
show("GET", f"/api/tx/{txid}/status", b, m)
assert_same_values(b, m)
assert b["confirmed"] is False
return
pytest.skip("no shared unconfirmed tx between brk and mempool.space")