123 lines
4 KiB
Python
123 lines
4 KiB
Python
#!/usr/bin/env python3
|
|
"""Unit test: wasabi-0001 CoinJoinCoinSelector AnonScoreTxSourceBiasedShuffle O(N^3) -> O(N^2)"""
|
|
|
|
import sys
|
|
|
|
class FakeCoin:
|
|
def __init__(self, id_, txid, anon):
|
|
self.id = id_
|
|
self.txid = txid
|
|
self.anon = anon
|
|
|
|
def defective_shuffle(coins):
|
|
"""Defective: list scan for TransactionId dedup, O(N^3)"""
|
|
comps = 0
|
|
ordered = []
|
|
for _ in range(len(coins)):
|
|
ordered_ids = {c.id for c in ordered}
|
|
remaining = sorted([c for c in coins if c.id not in ordered_ids], key=lambda x: x.anon)
|
|
alternating = []
|
|
alt_txids = []
|
|
ordered_txids_list = [c.txid for c in ordered]
|
|
skipped = []
|
|
for c in remaining:
|
|
in_alt = False
|
|
for a in alternating:
|
|
comps += 1
|
|
if a.txid == c.txid:
|
|
in_alt = True
|
|
break
|
|
in_ord = False
|
|
if not in_alt:
|
|
for o in ordered:
|
|
comps += 1
|
|
if o.txid == c.txid:
|
|
in_ord = True
|
|
break
|
|
if in_alt or in_ord:
|
|
skipped.append(c)
|
|
else:
|
|
alternating.append(c)
|
|
alternating.extend(skipped)
|
|
coin = alternating[0]
|
|
ordered.append(coin)
|
|
return ordered, comps
|
|
|
|
def patched_shuffle(coins):
|
|
"""Patched: HashSet for TransactionId dedup, O(N^2)"""
|
|
comps = 0
|
|
ordered = []
|
|
ordered_txids = set()
|
|
for _ in range(len(coins)):
|
|
ordered_ids = {c.id for c in ordered}
|
|
remaining = sorted([c for c in coins if c.id not in ordered_ids], key=lambda x: x.anon)
|
|
alternating = []
|
|
alt_txids = set()
|
|
skipped = []
|
|
for c in remaining:
|
|
comps += 1 # set lookup is O(1)
|
|
if c.txid in alt_txids or c.txid in ordered_txids:
|
|
skipped.append(c)
|
|
else:
|
|
alternating.append(c)
|
|
alt_txids.add(c.txid)
|
|
alternating.extend(skipped)
|
|
coin = alternating[0]
|
|
ordered.append(coin)
|
|
ordered_txids.add(coin.txid)
|
|
return ordered, comps
|
|
|
|
passed = 0
|
|
failed = 0
|
|
|
|
# Test 1: Correctness
|
|
coins = [FakeCoin(i, i // 3, float(20 - i)) for i in range(20)]
|
|
d, _ = defective_shuffle(coins)
|
|
p, _ = patched_shuffle(coins)
|
|
if [c.id for c in d] == [c.id for c in p]:
|
|
print("PASS test1_correctness: identical output order"); passed += 1
|
|
else:
|
|
print("FAIL test1_correctness: output order differs"); failed += 1
|
|
|
|
# Test 2: Performance N=100
|
|
coins = [FakeCoin(i, i // 5, float(100 - i)) for i in range(100)]
|
|
_, dc = defective_shuffle(coins)
|
|
_, pc = patched_shuffle(coins)
|
|
ratio = dc / max(pc, 1)
|
|
print(f" N=100: defective={dc} patched={pc} ratio={ratio:.1f}x")
|
|
if ratio > 10.0:
|
|
print("PASS test2_perf_n100: >10x fewer comparisons"); passed += 1
|
|
else:
|
|
print(f"FAIL test2_perf_n100: ratio {ratio:.1f}x not >10x"); failed += 1
|
|
|
|
# Test 3: Performance N=200
|
|
coins = [FakeCoin(i, i // 5, float(200 - i)) for i in range(200)]
|
|
_, dc = defective_shuffle(coins)
|
|
_, pc = patched_shuffle(coins)
|
|
ratio = dc / max(pc, 1)
|
|
print(f" N=200: defective={dc} patched={pc} ratio={ratio:.1f}x")
|
|
if ratio > 15.0:
|
|
print("PASS test3_perf_n200: >15x fewer comparisons"); passed += 1
|
|
else:
|
|
print(f"FAIL test3_perf_n200: ratio {ratio:.1f}x not >15x"); failed += 1
|
|
|
|
# Test 4: Single coin
|
|
coins = [FakeCoin(0, 0, 1.0)]
|
|
d, _ = defective_shuffle(coins)
|
|
p, _ = patched_shuffle(coins)
|
|
if len(d) == 1 and len(p) == 1 and d[0].id == p[0].id:
|
|
print("PASS test4_single_coin: correct for N=1"); passed += 1
|
|
else:
|
|
print("FAIL test4_single_coin"); failed += 1
|
|
|
|
# Test 5: All same TransactionId
|
|
coins = [FakeCoin(i, 42, float(i)) for i in range(50)]
|
|
d, _ = defective_shuffle(coins)
|
|
p, _ = patched_shuffle(coins)
|
|
if [c.id for c in d] == [c.id for c in p]:
|
|
print("PASS test5_same_txid: identical when all txids equal"); passed += 1
|
|
else:
|
|
print("FAIL test5_same_txid"); failed += 1
|
|
|
|
print(f"\n{passed}/{passed + failed} tests passed")
|
|
sys.exit(1 if failed > 0 else 0)
|