"""Integration tests — end-to-end workflows, tampering detection, performance.""" import asyncio import json import threading import time from datetime import datetime, timezone from unittest.mock import patch from hdk import HDK, HDKConfig from hdk.certificate import CertificateGenerator from hdk.core.canary import CanaryLayer from hdk.core.dag import DAGStorage from hdk.core.hash_engine import HashEngine from hdk.hedera.anchor import OfflineAnchorService from hdk.models import EventType, HDKNode, NodeLevel from tests.conftest import TEST_KEY def test_offline_to_online(): """100 LLM calls offline, then flush pending anchors.""" dag = DAGStorage(":memory:") engine = HashEngine() canary = CanaryLayer(dag, key=TEST_KEY) canary.initialize() # Add 100 nodes for i in range(100): content_hash = engine.hash_content(f"content_{i}") self_hash = engine.hash_with_parents(content_hash, []) node = HDKNode( id=f"node-{i}", level=NodeLevel.QUERY, content_hash=content_hash, parent_chain=[], self_hash=self_hash, event_type=EventType.LLM_QUERY, model="gpt-4", ) canary.weave_into_node(node) node.node_hash = engine.create_node_hash(node) dag.add_node(node) service = OfflineAnchorService(dag, canary) assert service.get_pending_count() == 0 # Anchor offline anchor = asyncio.run(service.submit_anchor(force=True)) assert anchor is not None assert anchor.node_count == 100 assert service.get_pending_count() == 1 # pending (no Hedera) dag.close() def test_tampering_detection_counter(): """Attacker modifies read counter in SQLite → canary detects.""" dag = DAGStorage(":memory:") canary = CanaryLayer(dag, key=TEST_KEY) canary.initialize() # Normal reads canary.authorized_read() canary.authorized_read() checkpoint = canary.create_checkpoint() # Tamper with counter dag._conn.execute("UPDATE canary_state SET read_counter = 999 WHERE id = 1") dag._conn.commit() # Reload and verify canary2 = CanaryLayer(dag, key=TEST_KEY) assert canary2.verify(checkpoint["commitment"], checkpoint["counter"]) is False dag.close() def test_tampering_detection_node_hash(): """Attacker modifies content_hash in SQLite → verify_integrity catches.""" dag = DAGStorage(":memory:") engine = HashEngine() content_hash = engine.hash_content("original content") self_hash = engine.hash_with_parents(content_hash, []) node = HDKNode( id="node-1", level=NodeLevel.EVENT, content_hash=content_hash, parent_chain=[], self_hash=self_hash, event_type=EventType.CUSTOM, ) node.node_hash = engine.create_node_hash(node) dag.add_node(node) # Tamper dag._conn.execute( "UPDATE nodes SET content_hash = 'TAMPERED' WHERE id = 'node-1'" ) dag._conn.commit() valid, errors = dag.verify_integrity() assert valid is False assert len(errors) > 0 dag.close() def test_tampering_detection_merkle_root(): """Attacker modifies merkle_root in anchor → verification fails.""" dag = DAGStorage(":memory:") engine = HashEngine() canary = CanaryLayer(dag, key=TEST_KEY) canary.initialize() # Add nodes for i in range(5): ch = engine.hash_content(f"c_{i}") sh = engine.hash_with_parents(ch, []) n = HDKNode( id=f"n-{i}", level=NodeLevel.EVENT, content_hash=ch, parent_chain=[], self_hash=sh, event_type=EventType.CUSTOM, ) canary.weave_into_node(n) n.node_hash = engine.create_node_hash(n) dag.add_node(n) service = OfflineAnchorService(dag, canary) anchor = asyncio.run(service.submit_anchor(force=True)) original_root = anchor.merkle_root # Tamper merkle root dag._conn.execute( "UPDATE anchors SET merkle_root = 'FAKE_ROOT' WHERE anchor_sequence = 0" ) dag._conn.commit() # Read tampered anchor tampered = dag.get_last_anchor() assert tampered.merkle_root != original_root assert tampered.merkle_root == "FAKE_ROOT" valid, errors = dag.verify_integrity() assert valid is False assert any("merkle_root mismatch" in e for e in errors) dag.close() def test_certificate_roundtrip(): """Generate → JSON → parse → verify.""" hdk = HDK(config=HDKConfig(db_path=":memory:")) hdk.record_event(EventType.LLM_QUERY, "prompt") hdk.record_event(EventType.LLM_RESPONSE, "response") gen = CertificateGenerator(hdk._dag, hdk._engine) doc = "My final document" cert = gen.generate(doc, "author_123") # Serialize json_str = gen.to_json(cert) data = json.loads(json_str) # Verify fields survive roundtrip assert data["document_hash"] == cert.document_hash assert data["author_hash"] == cert.author_hash assert data["merkle_root"] == cert.merkle_root # Verify against document result = gen.verify_certificate(cert, doc) assert result["document_hash_valid"] is True assert result["overall"] is True hdk.close() def test_large_dag_performance(): """Add 10,000 nodes. merkle_root < 1s, verify < 5s, export < 2s.""" dag = DAGStorage(":memory:") engine = HashEngine() # Add 10k nodes for i in range(10_000): ch = engine.hash_content(f"content_{i}") sh = engine.hash_with_parents(ch, []) n = HDKNode( id=f"n-{i}", level=NodeLevel.EVENT, content_hash=ch, parent_chain=[], self_hash=sh, event_type=EventType.LLM_QUERY, ) n.node_hash = engine.create_node_hash(n) dag.add_node(n) assert dag.get_node_count() == 10_000 # Merkle root < 1s t0 = time.time() root = dag.compute_current_merkle_root() t_merkle = time.time() - t0 assert len(root) == 64 assert t_merkle < 1.0, f"merkle_root took {t_merkle:.2f}s" # Verify < 5s t0 = time.time() valid, errors = dag.verify_integrity() t_verify = time.time() - t0 assert valid is True assert t_verify < 5.0, f"verify took {t_verify:.2f}s" # Export < 2s t0 = time.time() exported = dag.export_dag() t_export = time.time() - t0 assert t_export < 2.0, f"export took {t_export:.2f}s" data = json.loads(exported) assert len(data["nodes"]) == 10_000 dag.close() def test_concurrent_writes(): """Multiple threads writing to DAG simultaneously — no corruption.""" dag = DAGStorage(":memory:") engine = HashEngine() errors = [] def writer(thread_id: int, count: int): try: for i in range(count): ch = engine.hash_content(f"t{thread_id}_c{i}") sh = engine.hash_with_parents(ch, []) n = HDKNode( id=f"t{thread_id}-{i}", level=NodeLevel.EVENT, content_hash=ch, parent_chain=[], self_hash=sh, event_type=EventType.CUSTOM, ) n.node_hash = engine.create_node_hash(n) dag.add_node(n) except Exception as e: errors.append(str(e)) threads = [threading.Thread(target=writer, args=(t, 50)) for t in range(4)] for t in threads: t.start() for t in threads: t.join() # All nodes written (SQLite serializes writes, so this should work) total = dag.get_node_count() assert total == 200, f"Expected 200, got {total}. Errors: {errors}" # Integrity still valid valid, errs = dag.verify_integrity() assert valid is True dag.close()