gmakstutis/cognee
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

added

Browse source
This commit is contained in:
vasilije 2025-07-01 15:19:18 +02:00
parent 07f2afa69d
commit 1c727c73bb
2 changed files with 6 additions and 11 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
examples/python/incremental_loading_example.py
View file

@ -23,7 +23,7 @@ async def demonstrate_incremental_loading():
print("=" * 50) print("=" * 50)
# Initialize the incremental loader # Initialize the incremental loader
incremental_loader = IncrementalLoader(block_size=512) # 512 byte blocks for demo IncrementalLoader(block_size=512) # 512 byte blocks for demo
block_service = BlockHashService(block_size=512) block_service = BlockHashService(block_size=512)
# Create initial file content # Create initial file content

15
tests/test_incremental_loading.py
View file

@ -111,10 +111,10 @@ class TestIncrementalLoader:
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_should_process_new_file(self): async def test_should_process_new_file(self):
"""Test processing decision for new files""" """Test processing decision for new files"""
loader = IncrementalLoader() IncrementalLoader()
content = b"This is a new file that hasn't been seen before." content = b"This is a new file that hasn't been seen before."
file_obj = BytesIO(content) BytesIO(content)
# For a new file (no existing signature), should process # For a new file (no existing signature), should process
# Note: This test would need a mock database setup in real implementation # Note: This test would need a mock database setup in real implementation
@ -123,10 +123,10 @@ class TestIncrementalLoader:
def test_block_data_extraction(self): def test_block_data_extraction(self):
"""Test extraction of changed block data""" """Test extraction of changed block data"""
loader = IncrementalLoader(block_size=10) IncrementalLoader(block_size=10)
content = b"Block1____Block2____Block3____" content = b"Block1____Block2____Block3____"
file_obj = BytesIO(content) BytesIO(content)
# Create mock change info # Create mock change info
from cognee.modules.ingestion.incremental.block_hash_service import BlockInfo, FileSignature from cognee.modules.ingestion.incremental.block_hash_service import BlockInfo, FileSignature
@ -137,7 +137,7 @@ class TestIncrementalLoader:
BlockInfo(2, 34567, "hash3", 10, 20), BlockInfo(2, 34567, "hash3", 10, 20),
] ]
signature = FileSignature( FileSignature(
file_path="test", file_path="test",
file_size=30, file_size=30,
total_blocks=3, total_blocks=3,
@ -147,11 +147,6 @@ class TestIncrementalLoader:
signature_data=b"signature", signature_data=b"signature",
) )
change_info = {
"type": "incremental_changes",
"changed_blocks": [1], # Only middle block changed
"new_signature": signature,
}
# This would normally be called after should_process_file # This would normally be called after should_process_file
# Testing the block extraction logic # Testing the block extraction logic