""" Comprehensive test: all HTTP API endpoints + live network fetch from kubo nodes. """ import json import os import tempfile import trio from py_ipfs_lite.peer import Peer from py_ipfs_lite.config import Config # Well-known CID: "hello world" text file WELL_KNOWN_CID = "QmT78zSuBmuS4z925WZfrqQ1qHaJ56DQaTfyMUF7F8ff5o" WELL_KNOWN_CONTENT = b"hello world\n" async def test_endpoints(peer: Peer): """Test all HTTP API endpoints via the service layer.""" print("=" * 60) print("TESTING API ENDPOINTS") print("=" * 60) passed = 0 failed = 0 def ok(name): nonlocal passed passed += 1 print(f" PASS: {name}") def fail(name, e): nonlocal failed failed += 1 print(f" FAIL: {name} — {e}") # 1. Version try: from py_ipfs_lite.services import node_service v = node_service.get_version_info() assert "Version" in v ok("version") except Exception as e: fail("version", e) # 2. ID try: ident = await node_service.get_identity(peer) assert ident.id assert len(ident.addresses) > 0 ok(f"id ({len(ident.addresses)} addrs)") except Exception as e: fail("id", e) # 3. Add file (bytes) try: from py_ipfs_lite.services import files_service async def _gen(): yield b"test file content for endpoint testing" result = await files_service.add_file_from_stream(peer, "test.txt", _gen()) assert result.cid test_cid = result.cid ok(f"add (cid={test_cid[:20]}...)") except Exception as e: fail("add", e) test_cid = None # 4. Cat (get file) try: assert test_cid, "need CID from add step" data = await files_service.get_file_bytes(peer, test_cid) assert data == b"test file content for endpoint testing" ok(f"cat ({len(data)} bytes)") except Exception as e: fail("cat", e) # 5. Block stat try: from py_ipfs_lite.services import block_service stat = await block_service.stat_block(peer, test_cid) assert stat.size > 0 ok(f"block/stat (size={stat.size})") except Exception as e: fail("block/stat", e) # 6. Block get try: raw = await block_service.get_block(peer, test_cid) assert len(raw) > 0 ok(f"block/get ({len(raw)} bytes)") except Exception as e: fail("block/get", e) # 7. DAG put try: from py_ipfs_lite.services import dag_service node_data = {"hello": "world", "num": 42} dag_result = await dag_service.put_node(peer, node_data, codec="dag-json") assert dag_result.cid dag_cid = dag_result.cid ok(f"dag/put (cid={dag_cid[:20]}...)") except Exception as e: fail("dag/put", e) dag_cid = None # 8. DAG get try: assert dag_cid, "need CID from dag/put step" get_result = await dag_service.get_node(peer, dag_cid) assert get_result.node_data["hello"] == "world" assert get_result.node_data["num"] == 42 ok("dag/get") except Exception as e: fail("dag/get", e) # 9. Pin add try: from py_ipfs_lite.services import pin_service await pin_service.add_pin(peer, test_cid, recursive=True) ok("pin/add") except Exception as e: fail("pin/add", e) # 10. Pin ls try: pins = await pin_service.list_pins(peer, "all") assert test_cid in pins ok(f"pin/ls ({len(pins)} pins)") except Exception as e: fail("pin/ls", e) # 11. Pin rm try: await pin_service.remove_pin(peer, test_cid) ok("pin/rm") except Exception as e: fail("pin/rm", e) # 12. Repo stat try: from py_ipfs_lite.services import repo_service stat = await repo_service.get_repo_stat(peer) assert stat.num_objects >= 0 ok(f"repo/stat (objects={stat.num_objects})") except Exception as e: fail("repo/stat", e) # 13. Repo version try: ver = await repo_service.get_repo_version(peer) ok(f"repo/version ({ver})") except Exception as e: fail("repo/version", e) # 14. Refs local try: refs = await repo_service.list_local_refs(peer) ok(f"refs/local ({len(refs)} refs)") except Exception as e: fail("refs/local", e) # 15. Swarm peers try: from py_ipfs_lite.services import swarm_service peers = await swarm_service.list_connected_peers(peer) ok(f"swarm/peers ({peers.count} connected)") except Exception as e: fail("swarm/peers", e) # 16. Repo GC try: await pin_service.run_gc(peer) ok("repo/gc") except Exception as e: fail("repo/gc", e) # 17. Block rm try: await block_service.remove_block(peer, test_cid) ok("block/rm") except Exception as e: fail("block/rm", e) print(f"\n Results: {passed} passed, {failed} failed out of {passed + failed}") return failed == 0 async def test_network_fetch(peer: Peer): """Fetch a well-known file from the live IPFS network.""" print("\n" + "=" * 60) print("TESTING NETWORK FETCH FROM KUBO NODES") print("=" * 60) from py_ipfs_lite.cli import DEFAULT_BOOTSTRAP_PEERS # 1. Bootstrap print(f"\n1. Bootstrapping to {len(DEFAULT_BOOTSTRAP_PEERS)} peers...") await peer.bootstrap(DEFAULT_BOOTSTRAP_PEERS) await trio.sleep(10) conns = peer.host.get_network().connections print(f" Connected to {len(conns)} peers") if len(conns) == 0: print(" ERROR: No peers connected — cannot fetch from network") return False # 2. Fetch well-known CID print(f"\n2. Fetching CID: {WELL_KNOWN_CID}") try: result = await peer.get_file(WELL_KNOWN_CID, timeout=120) data = await result.read() print(f" Fetched {len(data)} bytes") print(f" Content: {data!r}") assert data == WELL_KNOWN_CONTENT, ( f"Content mismatch! Expected {WELL_KNOWN_CONTENT!r}, got {data!r}" ) print(" PASS: Content matches expected value") except Exception as e: print(f" FAIL: {type(e).__name__}: {e}") import traceback traceback.print_exc() return False # 3. Test seeking print("\n3. Testing SeekableReader seek...") try: result.seek(0) partial = await result.read(5) assert partial == b"hello", f"Seek+read failed: got {partial!r}" print(" PASS: SeekableReader seek works") except Exception as e: print(f" FAIL: {e}") return False # 4. Fetch a different well-known CID (the IPFS docs website unixfs node) # This CID is for a small DAG-PB node print("\n4. Fetching a second well-known CID (QmYwAPJzv5CZsnA625s3Xf2nemtYgPpHdWEz79ojWnPbdG - IPFS docs)...") try: result2 = await peer.get_file( "QmYwAPJzv5CZsnA625s3Xf2nemtYgPpHdWEz79ojWnPbdG", timeout=60 ) data2 = await result2.read() print(f" Fetched {len(data2)} bytes") print(" PASS: Second CID fetched successfully") except Exception as e: print(f" FAIL: {e}") # Not fatal — the first fetch was the important one print("\n" + "=" * 60) print("ALL NETWORK TESTS PASSED") print("=" * 60) return True async def main(): config = Config( offline=False, reprovide_interval_seconds=-1, default_timeout=120.0, ) peer = Peer(config, listen_addrs=["/ip4/0.0.0.0/tcp/0"]) await peer.start() try: all_ok = True # Test API endpoints endpoints_ok = await test_endpoints(peer) all_ok = all_ok and endpoints_ok # Test network fetch network_ok = await test_network_fetch(peer) all_ok = all_ok and network_ok print("\n" + "=" * 60) if all_ok: print("ALL TESTS PASSED") else: print("SOME TESTS FAILED") print("=" * 60) finally: await peer.close() if __name__ == "__main__": trio.run(main)