Skip to main content
Glama
test_incremental_updates.py7.15 kB
#!/usr/bin/env python3 """Test script to validate incremental Y.js updates. This script tests that the new transact_document() pattern correctly: 1. Captures state BEFORE changes 2. Generates INCREMENTAL diffs (not full state) 3. Produces smaller update sizes for surgical edits vs full replacement Usage: cd mnemosyne-mcp uv run python scripts/test_incremental_updates.py """ import pycrdt from neem.hocuspocus.document import DocumentWriter, DocumentReader def test_incremental_vs_full_update(): """Compare update sizes: incremental append vs full replacement.""" print("=" * 60) print("TEST: Incremental vs Full Update Size Comparison") print("=" * 60) # Create a document with some initial content doc = pycrdt.Doc() writer = DocumentWriter(doc) # Add 10 paragraphs of content initial_content = "\n".join( f"<paragraph>This is paragraph {i} with some content.</paragraph>" for i in range(10) ) writer.replace_all_content(initial_content) reader = DocumentReader(doc) print(f"\nInitial document has {writer.get_block_count()} blocks") print(f"Initial content preview: {reader.to_xml()[:200]}...") # Now test incremental append print("\n--- Test 1: Incremental Append ---") # Capture state BEFORE changes (this is what transact_document does) old_state = doc.get_state() # Make a small change: append one paragraph writer.append_block("<paragraph>New paragraph added incrementally!</paragraph>") # Get INCREMENTAL update (diff from old state) incremental_update = doc.get_update(old_state) # Get FULL state update (what the old code did) full_update = doc.get_update() print(f"Incremental update size: {len(incremental_update)} bytes") print(f"Full state update size: {len(full_update)} bytes") print(f"Ratio: {len(full_update) / len(incremental_update):.1f}x larger for full state") # The incremental update should be MUCH smaller assert len(incremental_update) < len(full_update), "Incremental should be smaller!" print("✓ Incremental update is smaller than full state update") # Verify the incremental update can be applied to a fresh doc print("\n--- Test 2: Apply Incremental Update to Fresh Doc ---") # Test this properly by simulating a sync scenario # Doc A has content, Doc B starts empty, they sync doc_a = pycrdt.Doc() doc_b = pycrdt.Doc() # Doc A creates initial content writer_a = DocumentWriter(doc_a) writer_a.replace_all_content("<paragraph>Hello from Doc A</paragraph>") # Sync A -> B (initial sync) - get full state from A full_state_a = doc_a.get_update() doc_b.apply_update(full_state_a) reader_b = DocumentReader(doc_b) print(f"Doc B after initial sync: {reader_b.to_xml()}") assert "Hello from Doc A" in reader_b.to_xml() print("✓ Initial sync works") # Now Doc A makes an incremental change old_state_a = doc_a.get_state() writer_a.append_block("<paragraph>Incremental update from Doc A</paragraph>") incremental_a = doc_a.get_update(old_state_a) print(f"\nIncremental update size: {len(incremental_a)} bytes") # Apply incremental update to Doc B doc_b.apply_update(incremental_a) reader_b = DocumentReader(doc_b) print(f"Doc B after incremental sync: {reader_b.to_xml()}") assert "Incremental update from Doc A" in reader_b.to_xml() print("✓ Incremental sync works") print("\n" + "=" * 60) print("All tests passed!") print("=" * 60) def test_surgical_edits(): """Test various surgical edit methods.""" print("\n" + "=" * 60) print("TEST: Surgical Edit Methods") print("=" * 60) doc = pycrdt.Doc() writer = DocumentWriter(doc) reader = DocumentReader(doc) # Test append_block print("\n--- append_block ---") old_state = doc.get_state() writer.append_block("<paragraph>First paragraph</paragraph>") update = doc.get_update(old_state) print(f"Append block update size: {len(update)} bytes") print(f"Content: {reader.to_xml()}") assert writer.get_block_count() == 1 print("✓ append_block works") # Test insert_block_at print("\n--- insert_block_at ---") old_state = doc.get_state() writer.insert_block_at(0, "<paragraph>Inserted at beginning</paragraph>") update = doc.get_update(old_state) print(f"Insert block update size: {len(update)} bytes") print(f"Content: {reader.to_xml()}") assert writer.get_block_count() == 2 assert "Inserted at beginning" in reader.to_xml()[:100] print("✓ insert_block_at works") # Test delete_block_at print("\n--- delete_block_at ---") old_state = doc.get_state() writer.delete_block_at(0) update = doc.get_update(old_state) print(f"Delete block update size: {len(update)} bytes") print(f"Content: {reader.to_xml()}") assert writer.get_block_count() == 1 assert "Inserted at beginning" not in reader.to_xml() print("✓ delete_block_at works") # Test get_block_count print("\n--- get_block_count ---") writer.append_block("<paragraph>Second</paragraph>") writer.append_block("<paragraph>Third</paragraph>") assert writer.get_block_count() == 3 print(f"Block count: {writer.get_block_count()}") print("✓ get_block_count works") print("\n" + "=" * 60) print("All surgical edit tests passed!") print("=" * 60) def test_formatted_content(): """Test that formatted content (marks) work with surgical edits.""" print("\n" + "=" * 60) print("TEST: Formatted Content with Marks") print("=" * 60) doc = pycrdt.Doc() writer = DocumentWriter(doc) reader = DocumentReader(doc) # Test adding formatted content print("\n--- Adding formatted paragraph ---") old_state = doc.get_state() writer.append_block( '<paragraph>Hello <strong>bold</strong> and <em>italic</em> text</paragraph>' ) update = doc.get_update(old_state) print(f"Update size: {len(update)} bytes") print(f"Content: {reader.to_xml()}") print("✓ Formatted content added") # Test adding a heading print("\n--- Adding heading ---") old_state = doc.get_state() writer.append_block('<heading level="2">Section Header</heading>') update = doc.get_update(old_state) print(f"Update size: {len(update)} bytes") print(f"Content: {reader.to_xml()}") print("✓ Heading added") # Test adding a link print("\n--- Adding link ---") old_state = doc.get_state() writer.append_block( '<paragraph>Check out <a href="https://example.com">this link</a>!</paragraph>' ) update = doc.get_update(old_state) print(f"Update size: {len(update)} bytes") print(f"Content: {reader.to_xml()}") print("✓ Link added") print("\n" + "=" * 60) print("All formatted content tests passed!") print("=" * 60) if __name__ == "__main__": test_incremental_vs_full_update() test_surgical_edits() test_formatted_content() print("\n" + "=" * 60) print("🎉 ALL TESTS PASSED!") print("=" * 60)

Latest Blog Posts

MCP directory API

We provide all the information about MCP servers via our MCP API.

curl -X GET 'https://glama.ai/api/mcp/v1/servers/sophia-labs/mnemosyne-mcp'

If you have feedback or need assistance with the MCP directory API, please join our Discord server