Spaces:
Running on Zero
Running on Zero
File size: 5,301 Bytes
31c93b1 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 | """Quick functional test for all new modules."""
import asyncio
import tempfile
import pytest
from pathlib import Path
def test_config():
from hearthnet.config import default_config
cfg = default_config()
assert cfg.transport.port == 7080
assert cfg.bus.prefer_local is True
print(f" Config OK: port={cfg.transport.port}")
def test_identity():
try:
from hearthnet.identity.keys import generate, sign_payload, verify_payload
kp = generate()
payload = {"foo": "bar", "num": 1}
signed = sign_payload(payload, kp)
assert "signature" in signed
assert verify_payload(signed, kp.verify_key)
print(f" Identity OK: {kp.node_id_short}")
except Exception as e:
print(f" Identity SKIP (PyNaCl not installed?): {e}")
def test_events():
from hearthnet.events.log import EventLog
import gc
td = tempfile.mkdtemp()
try:
log = EventLog(Path(td) / "events.db", "test-community")
ev = log.append_local("community.created", "test-author", {"name": "Test"})
assert ev.lamport == 1
print(f" Events OK: lamport={ev.lamport}, id={ev.event_id[:12]}")
# Close the connection before cleanup
if hasattr(log, "_conn") and log._conn:
log._conn.close()
del log
gc.collect()
finally:
import shutil
try:
shutil.rmtree(td, ignore_errors=True)
except Exception:
pass
def test_blobs():
from hearthnet.blobs.chunker import chunk_blob
data = b"Hello HearthNet" * 1000
manifest, chunks = chunk_blob(data)
assert b"".join(chunks) == data
print(f" Blobs OK: cid={manifest.cid[:20]}, chunks={len(chunks)}")
@pytest.mark.asyncio
async def test_embedding():
from hearthnet.services.embedding.backends import SimpleHashBackend
backend = SimpleHashBackend()
embeddings = await backend.embed(["hello world", "test text"])
assert len(embeddings) == 2 and len(embeddings[0]) == 16
print(f" Embedding OK: {len(embeddings)} embeddings, dim={len(embeddings[0])}")
@pytest.mark.asyncio
async def test_marketplace():
from hearthnet.services.marketplace.service import MarketplaceService
from hearthnet.bus.capability import RouteRequest
svc = MarketplaceService()
req = RouteRequest(
capability="market.post", version_req=(1, 0),
body={"input": {"title": "Test", "body": "Hello", "category": "info"}},
caller="test-node", trace_id="trace1",
)
result = await svc.handle_post(req)
eid = result["output"]["event_id"]
print(f" Marketplace OK: event_id={eid[:12]}")
@pytest.mark.asyncio
async def test_chat():
from hearthnet.services.chat.service import ChatService
from hearthnet.bus.capability import RouteRequest
svc = ChatService("node-a")
req = RouteRequest(
capability="chat.send", version_req=(1, 0),
body={"input": {"recipient": "node-b", "body": "Hi there!"}},
caller="node-a", trace_id="t1",
)
result = await svc.send(req)
print(f" Chat OK: delivered={result['output']['delivered']}")
@pytest.mark.asyncio
async def test_rag():
from hearthnet.services.rag.service import RagService
from hearthnet.bus.capability import RouteRequest
svc = RagService(corpus="test")
req = RouteRequest(
capability="rag.ingest", version_req=(1, 0),
body={"input": {"text": "Water is essential for survival. Store clean water.", "title": "Survival Tips"}},
caller="test", trace_id="t1",
)
result = await svc.handle_ingest(req)
print(f" RAG ingest OK: chunks={result['output']['chunks_indexed']}")
req2 = RouteRequest(
capability="rag.query", version_req=(1, 0),
body={"input": {"query": "water survival", "k": 3}, "params": {"corpus": "test"}},
caller="test", trace_id="t2",
)
result2 = await svc.handle_query(req2)
chunks = result2["output"]["chunks"]
print(f" RAG query OK: found {len(chunks)} chunks")
def test_cli():
from hearthnet.cli import main
print(" CLI OK: commands registered")
def test_onboarding():
from hearthnet.ui.onboarding import encode_invite, decode_invite, InviteBlob, OnboardingError
from datetime import datetime, timezone, timedelta
exp = (datetime.now(timezone.utc) + timedelta(hours=24)).strftime("%Y-%m-%dT%H:%M:%SZ")
iat = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
blob = InviteBlob(
community_id="ed25519:test123",
community_name="Test Community",
inviter_node_id="ed25519:inviter",
invitee_node_id="",
issued_at=iat,
expires_at=exp,
signature="",
)
encoded = encode_invite(blob)
assert encoded.startswith("hn1:")
decoded = decode_invite(encoded)
assert decoded.community_name == "Test Community"
print(" Onboarding OK: invite encode/decode works")
if __name__ == "__main__":
print("Running functional tests...")
test_config()
test_identity()
test_events()
test_blobs()
asyncio.run(test_embedding())
asyncio.run(test_marketplace())
asyncio.run(test_chat())
asyncio.run(test_rag())
test_cli()
test_onboarding()
print("\nAll functional tests passed!")
|