diff --git a/application/tests/harvester_test/artifact_registry_test.py b/application/tests/harvester_test/artifact_registry_test.py new file mode 100644 index 000000000..39d0c54f7 --- /dev/null +++ b/application/tests/harvester_test/artifact_registry_test.py @@ -0,0 +1,99 @@ +import unittest +from datetime import datetime + +from application.utils.harvester.artifact_registry import ArtifactRegistry +from application.utils.harvester.models import ArtifactRegistryRecord + + +class ArtifactRegistryTests(unittest.TestCase): + def test_insert_record(self): + registry = ArtifactRegistry() + + record = ArtifactRegistryRecord( + artifact_id="art:test:file.md", + repository="OWASP/ASVS", + locator_path="file.md", + content_hash="abc", + last_commit_sha="123", + last_pipeline_run="run1", + last_processed_at=datetime.now(), + status="new", + ) + + registry.upsert(record) + + self.assertTrue(registry.exists(record.artifact_id)) + + def test_get_record(self): + registry = ArtifactRegistry() + + record = ArtifactRegistryRecord( + artifact_id="art:test:file.md", + repository="OWASP/ASVS", + locator_path="file.md", + content_hash="abc", + last_commit_sha="123", + last_pipeline_run="run1", + last_processed_at=datetime.now(), + status="new", + ) + + registry.upsert(record) + + stored = registry.get(record.artifact_id) + assert stored is not None + + self.assertEqual(stored.content_hash, "abc") + + def test_update_record(self): + registry = ArtifactRegistry() + + record = ArtifactRegistryRecord( + artifact_id="art:test:file.md", + repository="OWASP/ASVS", + locator_path="file.md", + content_hash="abc", + last_commit_sha="123", + last_pipeline_run="run1", + last_processed_at=datetime.now(), + status="new", + ) + + registry.upsert(record) + + record.content_hash = "xyz" + record.status = "updated" + + registry.upsert(record) + + stored = registry.get(record.artifact_id) + assert stored is not None + + self.assertEqual(stored.content_hash, "xyz") + self.assertEqual(stored.status, "updated") + + def test_all_records(self): + registry = ArtifactRegistry() + + for i in range(3): + registry.upsert( + ArtifactRegistryRecord( + artifact_id=f"art:{i}", + repository="repo", + locator_path=f"{i}.md", + content_hash=str(i), + last_commit_sha="sha", + last_pipeline_run="run", + last_processed_at=datetime.now(), + status="new", + ) + ) + + self.assertEqual( + len(registry.all()), + 3, + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/application/tests/harvester_test/checkpoint_manager_test.py b/application/tests/harvester_test/checkpoint_manager_test.py new file mode 100644 index 000000000..a3ddb7cea --- /dev/null +++ b/application/tests/harvester_test/checkpoint_manager_test.py @@ -0,0 +1,68 @@ +import unittest +from datetime import datetime + +from application.utils.harvester.checkpoint_manager import CheckpointManager +from application.utils.harvester.models import CheckpointRecord + + +class CheckpointManagerTests(unittest.TestCase): + def test_save_checkpoint(self): + manager = CheckpointManager() + + checkpoint = CheckpointRecord( + repository="OWASP/ASVS", + pipeline_run_id="run1", + last_processed_commit="abc123", + status="running", + updated_at=datetime.now(), + ) + + manager.save(checkpoint) + + self.assertIsNotNone(manager.get("OWASP/ASVS")) + + def test_update_commit(self): + manager = CheckpointManager() + + checkpoint = CheckpointRecord( + repository="OWASP/ASVS", + pipeline_run_id="run1", + last_processed_commit="abc123", + status="running", + updated_at=datetime.now(), + ) + + manager.save(checkpoint) + + manager.update_commit( + "OWASP/ASVS", + "deadbeef", + ) + + stored = manager.get("OWASP/ASVS") + assert stored is not None + + self.assertEqual(stored.last_processed_commit, "deadbeef") + + def test_mark_completed(self): + manager = CheckpointManager() + + checkpoint = CheckpointRecord( + repository="OWASP/ASVS", + pipeline_run_id="run1", + last_processed_commit="abc123", + status="running", + updated_at=datetime.now(), + ) + + manager.save(checkpoint) + manager.mark_completed("OWASP/ASVS") + + stored = manager.get("OWASP/ASVS") + assert stored is not None + + self.assertEqual(stored.status, "completed") + + +if __name__ == "__main__": + unittest.main() diff --git a/application/tests/harvester_test/content_hash_test.py b/application/tests/harvester_test/content_hash_test.py new file mode 100644 index 000000000..60d84108f --- /dev/null +++ b/application/tests/harvester_test/content_hash_test.py @@ -0,0 +1,35 @@ +import unittest + +from application.utils.harvester.content_hash import ( + generate_content_hash, +) + + +class ContentHashTests(unittest.TestCase): + def test_same_text_same_hash(self): + text = "Hello World" + + self.assertEqual( + generate_content_hash(text), + generate_content_hash(text), + ) + + def test_different_text_different_hash(self): + self.assertNotEqual( + generate_content_hash("Hello"), + generate_content_hash("World"), + ) + + def test_empty_string(self): + digest = generate_content_hash("") + + self.assertEqual(len(digest), 64) + + def test_hash_is_hex(self): + digest = generate_content_hash("OpenCRE") + + int(digest, 16) + + +if __name__ == "__main__": + unittest.main() diff --git a/application/tests/harvester_test/deduplication_metrics_test.py b/application/tests/harvester_test/deduplication_metrics_test.py new file mode 100644 index 000000000..13e4e1152 --- /dev/null +++ b/application/tests/harvester_test/deduplication_metrics_test.py @@ -0,0 +1,35 @@ +import unittest + +from application.utils.harvester.deduplication_metrics import DeduplicationMetrics +from application.utils.harvester.models import DeduplicationStatus + + +class DeduplicationMetricsTests(unittest.TestCase): + def test_records_new_document(self): + metrics = DeduplicationMetrics() + + metrics.record(DeduplicationStatus.NEW) + + self.assertEqual(metrics.total_artifacts_scanned, 1) + self.assertEqual(metrics.artifacts_new, 1) + self.assertEqual(metrics.artifacts_emitted, 1) + + def test_records_updated_document(self): + metrics = DeduplicationMetrics() + + metrics.record(DeduplicationStatus.UPDATED) + + self.assertEqual(metrics.artifacts_updated, 1) + self.assertEqual(metrics.artifacts_emitted, 1) + + def test_records_unchanged_document(self): + metrics = DeduplicationMetrics() + + metrics.record(DeduplicationStatus.UNCHANGED) + + self.assertEqual(metrics.artifacts_unchanged, 1) + self.assertEqual(metrics.artifacts_skipped, 1) + + +if __name__ == "__main__": + unittest.main() diff --git a/application/tests/harvester_test/diff_parser_test.py b/application/tests/harvester_test/diff_parser_test.py index a4444e8d9..877d70a09 100644 --- a/application/tests/harvester_test/diff_parser_test.py +++ b/application/tests/harvester_test/diff_parser_test.py @@ -10,6 +10,11 @@ TEST_COMMITTED_AT = datetime.now(UTC) +TEST_REPOSITORY = "OWASP/ASVS" +TEST_COMMIT_SHA = "abc123" +TEST_COMMITTED_AT = datetime.now(UTC) + + class DiffParserTests(unittest.TestCase): def test_single_file_diff(self): parser = DiffParser() diff --git a/application/tests/harvester_test/diff_pipeline_test.py b/application/tests/harvester_test/diff_pipeline_test.py index 07160f170..e5420a4ed 100644 --- a/application/tests/harvester_test/diff_pipeline_test.py +++ b/application/tests/harvester_test/diff_pipeline_test.py @@ -1,8 +1,8 @@ from datetime import UTC, datetime +import os import subprocess import time import unittest -import os from application.utils.harvester.diff_normalizer import DiffNormalizer from application.utils.harvester.diff_parser import DiffParser @@ -19,7 +19,6 @@ class DiffPipelineBenchmark(unittest.TestCase): """ def test_pipeline_benchmark(self): - if os.getenv("OPENCRE_RUN_NETWORK_TESTS") != "1": self.skipTest("Network benchmark disabled") @@ -29,9 +28,7 @@ def test_pipeline_benchmark(self): "master", ) client.sync() - head_commit = client.get_current_commit_sha() - previous_commit = subprocess.run( [ "git", @@ -51,23 +48,18 @@ def test_pipeline_benchmark(self): normalizer = DiffNormalizer() start = time.perf_counter() - diff = retriever.get_diff( previous_commit, head_commit, ) - blocks = parser.parse( diff, repository="OWASP/ASVS", commit_sha=head_commit, committed_at=datetime.now(UTC), ) - normalizer.normalize(blocks) - elapsed = time.perf_counter() - start print(f"\nPipeline took {elapsed:.3f}s") - self.assertLess(elapsed, 5) diff --git a/application/tests/harvester_test/diff_retriever_test.py b/application/tests/harvester_test/diff_retriever_test.py index 502ac2d39..f8c3ea142 100644 --- a/application/tests/harvester_test/diff_retriever_test.py +++ b/application/tests/harvester_test/diff_retriever_test.py @@ -1,7 +1,7 @@ import unittest from unittest.mock import MagicMock -from unittest.mock import patch from unittest.mock import call +from unittest.mock import patch from application.utils.harvester.diff_retriever import ( DiffRetriever, @@ -16,10 +16,8 @@ def test_get_diff(self, mock_run): MagicMock(stdout="def456\n"), MagicMock(stdout=b"diff --git a/README.md b/README.md\n"), ] - client = MagicMock() client.get_local_path.return_value = "/tmp/repo" - retriever = DiffRetriever(client) diff = retriever.get_diff( @@ -31,7 +29,6 @@ def test_get_diff(self, mock_run): diff, "diff --git a/README.md b/README.md\n", ) - mock_run.assert_has_calls( [ call( @@ -88,7 +85,6 @@ def test_large_diff_raises(self, mock_run): client = MagicMock() client.get_local_path.return_value = "/tmp/repo" - retriever = DiffRetriever(client) with self.assertRaises(ValueError): diff --git a/application/tests/harvester_test/document_builder_test.py b/application/tests/harvester_test/document_builder_test.py new file mode 100644 index 000000000..ba2313c08 --- /dev/null +++ b/application/tests/harvester_test/document_builder_test.py @@ -0,0 +1,65 @@ +import unittest +from datetime import datetime + +from application.utils.harvester.document_builder import ( + DocumentBuilder, +) +from application.utils.harvester.models import ( + DiffBlock, +) + + +class DocumentBuilderTests(unittest.TestCase): + def test_build_document(self): + block = DiffBlock( + file_path="README.md", + repository="OWASP/ASVS", + commit_sha="abc123", + committed_at=datetime.now(), + added_lines=["Hello"], + ) + + document = DocumentBuilder().build( + block, + "# Title\n\nHello", + pipeline_run_id="20260714T120000Z", + ) + + self.assertEqual( + document.schema_version, + "0.2.0", + ) + + self.assertEqual( + document.artifact_id, + "art:OWASP/ASVS:README.md", + ) + + self.assertEqual( + document.pipeline_run_id, + "20260714T120000Z", + ) + + self.assertEqual( + document.text, + "# Title\n\nHello", + ) + + self.assertEqual( + document.source.repository, + "OWASP/ASVS", + ) + + self.assertEqual( + document.locator.path, + "README.md", + ) + + self.assertEqual( + len(document.heading_structure), + 1, + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/application/tests/harvester_test/document_deduplicator_test.py b/application/tests/harvester_test/document_deduplicator_test.py new file mode 100644 index 000000000..3aa4c56f6 --- /dev/null +++ b/application/tests/harvester_test/document_deduplicator_test.py @@ -0,0 +1,71 @@ +import unittest +from datetime import datetime + +from application.utils.harvester.artifact_registry import ArtifactRegistry +from application.utils.harvester.document_deduplicator import ( + DocumentDeduplicator, +) +from application.utils.harvester.models import ( + DeduplicationStatus, + Document, + Locator, + SourceInfo, +) + + +class DocumentDeduplicatorTests(unittest.TestCase): + def create_document(self, text: str) -> Document: + return Document( + schema_version="0.2.0", + artifact_id="art:test:file.md", + pipeline_run_id="run1", + text=text, + source=SourceInfo( + type="github", + repository="OWASP/ASVS", + commit_sha="abc123", + committed_at=datetime.now(), + ), + locator=Locator( + kind="repo_path", + id="file.md", + path="file.md", + ), + heading_structure=[], + span=None, + ) + + def test_new_document(self): + registry = ArtifactRegistry() + + deduplicator = DocumentDeduplicator(registry) + result = deduplicator.process(self.create_document("hello")) + self.assertEqual(result, DeduplicationStatus.NEW) + + def test_unchanged_document(self): + registry = ArtifactRegistry() + + deduplicator = DocumentDeduplicator(registry) + document = self.create_document("hello") + + deduplicator.process(document) + result = deduplicator.process(document) + + self.assertEqual(result, DeduplicationStatus.UNCHANGED) + + def test_updated_document(self): + registry = ArtifactRegistry() + + deduplicator = DocumentDeduplicator(registry) + + deduplicator.process(self.create_document("hello")) + + result = deduplicator.process( + self.create_document("changed"), + ) + + self.assertEqual(result, DeduplicationStatus.UPDATED) + + +if __name__ == "__main__": + unittest.main() diff --git a/application/tests/harvester_test/document_validator_test.py b/application/tests/harvester_test/document_validator_test.py new file mode 100644 index 000000000..c445d7245 --- /dev/null +++ b/application/tests/harvester_test/document_validator_test.py @@ -0,0 +1,85 @@ +import unittest +from datetime import datetime + +from application.utils.harvester.document_validator import ( + DocumentValidator, +) +from application.utils.harvester.models import ( + Document, + HeadingNode, + Locator, + SourceInfo, +) + + +def make_document() -> Document: + return Document( + schema_version="0.2.0", + artifact_id="art:OWASP/ASVS:README.md", + pipeline_run_id="20260714T120000Z", + text="# Title", + source=SourceInfo( + type="github", + repository="OWASP/ASVS", + commit_sha="abc123", + committed_at=datetime.now(), + ), + locator=Locator( + kind="repo_path", + id="README.md", + path="README.md", + ), + heading_structure=[ + HeadingNode( + level=1, + text="Title", + start_line=1, + end_line=1, + ) + ], + span=None, + ) + + +class DocumentValidatorTests(unittest.TestCase): + def test_valid_document(self): + validator = DocumentValidator() + + self.assertTrue(validator.validate(make_document())) + + def test_missing_artifact_id(self): + validator = DocumentValidator() + + document = make_document() + document.artifact_id = "" + + self.assertFalse(validator.validate(document)) + + def test_missing_text(self): + validator = DocumentValidator() + + document = make_document() + document.text = "" + + self.assertFalse(validator.validate(document)) + + def test_invalid_source_type(self): + validator = DocumentValidator() + + document = make_document() + document.source.type = "gitlab" + + self.assertFalse(validator.validate(document)) + + def test_non_markdown_document_is_valid(self): + validator = DocumentValidator() + + document = make_document() + document.heading_structure = [] + document.text = '{"hello": "world"}' + + self.assertTrue(validator.validate(document)) + + +if __name__ == "__main__": + unittest.main() diff --git a/application/tests/harvester_test/git_repository_client_test.py b/application/tests/harvester_test/git_repository_client_test.py index 8bbff6ab8..d012594c9 100644 --- a/application/tests/harvester_test/git_repository_client_test.py +++ b/application/tests/harvester_test/git_repository_client_test.py @@ -4,6 +4,7 @@ import tempfile from pathlib import Path +from unittest.mock import MagicMock from application.utils.harvester.git_repository_client import ( GitRepositoryClient, ) @@ -153,6 +154,33 @@ def test_clone_runs_git_command(self, mock_run): mock_run.assert_called() + @patch("application.utils.harvester.git_repository_client.subprocess.run") + def test_get_file_at_commit(self, mock_run): + + mock_run.return_value = MagicMock(stdout="# Hello\nWorld\n") + + client = GitRepositoryClient("OWASP", "ASVS", "master") + + client.get_local_path = MagicMock(return_value="/tmp/repo") + + content = client.get_file_at_commit("abc123", "README.md") + + self.assertEqual(content, "# Hello\nWorld\n") + + mock_run.assert_called_once_with( + [ + "git", + "-C", + "/tmp/repo", + "show", + "abc123:README.md", + ], + capture_output=True, + text=True, + check=True, + timeout=30, + ) + if __name__ == "__main__": unittest.main() diff --git a/application/tests/harvester_test/heading_extractor_test.py b/application/tests/harvester_test/heading_extractor_test.py new file mode 100644 index 000000000..e1411b9cd --- /dev/null +++ b/application/tests/harvester_test/heading_extractor_test.py @@ -0,0 +1,107 @@ +import unittest + +from application.utils.harvester.heading_extractor import ( + HeadingExtractor, +) + + +class HeadingExtractorTests(unittest.TestCase): + def test_single_heading(self): + text = """ +# Title + +Hello + +World +""" + + headings = HeadingExtractor().extract(text) + + self.assertEqual(len(headings), 1) + + self.assertEqual(headings[0].text, "Title") + self.assertEqual(headings[0].level, 1) + self.assertEqual(headings[0].start_line, 2) + + def test_nested_headings(self): + text = """ +# Root + +## Child One + +content + +## Child Two + +more + +# Second Root +""" + + headings = HeadingExtractor().extract(text) + + self.assertEqual(len(headings), 4) + + self.assertEqual(headings[0].text, "Root") + self.assertEqual(headings[1].text, "Child One") + self.assertEqual(headings[2].text, "Child Two") + self.assertEqual(headings[3].text, "Second Root") + + def test_heading_ranges(self): + text = """ +# Root + +text + +## Child + +child + +# Next +""" + + headings = HeadingExtractor().extract(text) + self.assertEqual(headings[0].end_line, 9) + self.assertEqual(headings[1].end_line, 9) + self.assertEqual(headings[2].end_line, 10) + + def test_ignore_non_headings(self): + text = """ +Hello + +###Heading + +####NoSpace + +## Valid Heading +""" + + headings = HeadingExtractor().extract(text) + self.assertEqual(len(headings), 1) + self.assertEqual(headings[0].text, "Valid Heading") + + def test_heading_stops_at_same_level(self): + text = """ +# Root + +## A + +### X + +## B + + content + """ + + headings = HeadingExtractor().extract(text) + + self.assertEqual(headings[1].text, "A") + self.assertEqual(headings[2].text, "X") + self.assertEqual(headings[3].text, "B") + + self.assertEqual(headings[1].end_line, 7) + self.assertEqual(headings[2].end_line, 7) + + +if __name__ == "__main__": + unittest.main() diff --git a/application/tests/harvester_test/incremental_pipeline_test.py b/application/tests/harvester_test/incremental_pipeline_test.py new file mode 100644 index 000000000..2b10814ed --- /dev/null +++ b/application/tests/harvester_test/incremental_pipeline_test.py @@ -0,0 +1,76 @@ +import unittest +from datetime import datetime + +from application.utils.harvester.artifact_registry import ArtifactRegistry +from application.utils.harvester.checkpoint_manager import CheckpointManager +from application.utils.harvester.document_deduplicator import ( + DocumentDeduplicator, +) +from application.utils.harvester.incremental_pipeline import ( + IncrementalPipeline, +) +from application.utils.harvester.models import ( + Document, + Locator, + SourceInfo, +) + + +class IncrementalPipelineTests(unittest.TestCase): + def make_document(self, text: str) -> Document: + + return Document( + schema_version="0.2.0", + artifact_id="art:test:file.md", + pipeline_run_id="run1", + text=text, + source=SourceInfo( + type="github", + repository="OWASP/ASVS", + commit_sha="abc123", + committed_at=datetime.now(), + ), + locator=Locator( + kind="repo_path", + id="file.md", + path="file.md", + ), + heading_structure=[], + span=None, + ) + + def test_only_new_and_updated_are_emitted(self): + registry = ArtifactRegistry() + + dedup = DocumentDeduplicator( + registry, + ) + + checkpoints = CheckpointManager() + + pipeline = IncrementalPipeline( + dedup, + checkpoints, + ) + + docs = [ + self.make_document("hello"), + self.make_document("hello"), + self.make_document("changed"), + ] + + emitted = pipeline.process( + "OWASP/ASVS", + "run1", + docs, + ) + + self.assertEqual(len(emitted), 2) + checkpoint = checkpoints.get("OWASP/ASVS") + + assert checkpoint is not None + self.assertEqual(checkpoint.status, "completed") + + +if __name__ == "__main__": + unittest.main() diff --git a/application/utils/harvester/__init__.py b/application/utils/harvester/__init__.py index 9961aae16..c73f833fc 100644 --- a/application/utils/harvester/__init__.py +++ b/application/utils/harvester/__init__.py @@ -21,21 +21,46 @@ from .filtering_metrics import FilteringMetricsCollector from .diff_retriever import DiffRetriever + from .filtering_benchmark import ( FilteringBenchmark, FilteringBenchmarkResult, ) +from .heading_extractor import ( + HeadingExtractor, + HeadingNode, +) + +from .document_builder import DocumentBuilder +from .document_validator import DocumentValidator +from .content_hash import generate_content_hash +from .artifact_registry import ArtifactRegistry +from .document_deduplicator import DocumentDeduplicator +from .checkpoint_manager import CheckpointManager +from .incremental_pipeline import IncrementalPipeline +from .deduplication_metrics import DeduplicationMetrics + __all__ = [ + "ArtifactRegistry", "build_repository_cache_path", + "CheckpointManager", "ChunkingConfig", "ConfigLoaderError", + "DeduplicationMetrics", "DiffRetriever", + "DocumentBuilder", + "DocumentDeduplicator", + "DocumentValidator", "GitRepositoryClient", "FileFilter", "FilteringMetricsCollector", "FilteringBenchmark", "FilteringBenchmarkResult", + "generate_content_hash", + "HeadingExtractor", + "HeadingNode", + "IncrementalPipeline", "PathRules", "PollingConfig", "RepositoryClient", diff --git a/application/utils/harvester/artifact_id.py b/application/utils/harvester/artifact_id.py new file mode 100644 index 000000000..fa3ed46f1 --- /dev/null +++ b/application/utils/harvester/artifact_id.py @@ -0,0 +1,11 @@ +def generate_artifact_id(repository: str, file_path: str) -> str: + """ + Generate a stable artifact identifier for a repository file. + + Example: + repository = "OWASP/ASVS" + file_path = "5.0/en/0x01-Frontispiece.md" + + -> art:OWASP/ASVS:5.0/en/0x01-Frontispiece.md + """ + return f"art:{repository}:{file_path}" diff --git a/application/utils/harvester/artifact_registry.py b/application/utils/harvester/artifact_registry.py new file mode 100644 index 000000000..621f8e567 --- /dev/null +++ b/application/utils/harvester/artifact_registry.py @@ -0,0 +1,24 @@ +from datetime import datetime +from .models import ArtifactRegistryRecord + + +class ArtifactRegistry: + """ + In-memory registry for artifact deduplication. + """ + + def __init__(self): + self._records: dict[str, ArtifactRegistryRecord] = {} + + def get(self, artifact_id: str) -> ArtifactRegistryRecord | None: + return self._records.get(artifact_id) + + def exists(self, artifact_id: str) -> bool: + return artifact_id in self._records + + def upsert(self, record: ArtifactRegistryRecord) -> None: + record.last_processed_at = datetime.now() + self._records[record.artifact_id] = record + + def all(self) -> list[ArtifactRegistryRecord]: + return list(self._records.values()) diff --git a/application/utils/harvester/checkpoint_manager.py b/application/utils/harvester/checkpoint_manager.py new file mode 100644 index 000000000..c9f0fdd3c --- /dev/null +++ b/application/utils/harvester/checkpoint_manager.py @@ -0,0 +1,36 @@ +from datetime import datetime + +from .models import CheckpointRecord + + +class CheckpointManager: + """ + Stores pipeline checkpoints for incremental processing. + """ + + def __init__(self): + self._checkpoints: dict[str, CheckpointRecord] = {} + + def save(self, checkpoint: CheckpointRecord) -> None: + self._checkpoints[checkpoint.repository] = checkpoint + + def get(self, repository: str) -> CheckpointRecord | None: + return self._checkpoints.get(repository) + + def update_commit(self, repository: str, commit_sha: str) -> None: + checkpoint = self._checkpoints.get(repository) + + if checkpoint is None: + return + + checkpoint.last_processed_commit = commit_sha + checkpoint.updated_at = datetime.now() + + def mark_completed(self, repository: str) -> None: + checkpoint = self._checkpoints.get(repository) + + if checkpoint is None: + return + + checkpoint.status = "completed" + checkpoint.updated_at = datetime.now() diff --git a/application/utils/harvester/content_hash.py b/application/utils/harvester/content_hash.py new file mode 100644 index 000000000..ed7bd8bda --- /dev/null +++ b/application/utils/harvester/content_hash.py @@ -0,0 +1,13 @@ +import hashlib + + +def generate_content_hash(text: str) -> str: + """ + Generate a deterministic SHA-256 hash for document content. + + Used for artifact-level deduplication. + """ + + return hashlib.sha256( + text.encode("utf-8"), + ).hexdigest() diff --git a/application/utils/harvester/deduplication_metrics.py b/application/utils/harvester/deduplication_metrics.py new file mode 100644 index 000000000..a3843a7fc --- /dev/null +++ b/application/utils/harvester/deduplication_metrics.py @@ -0,0 +1,29 @@ +from dataclasses import dataclass +from .models import DeduplicationStatus + + +@dataclass(slots=True) +class DeduplicationMetrics: + total_artifacts_scanned: int = 0 + + artifacts_new: int = 0 + artifacts_updated: int = 0 + artifacts_unchanged: int = 0 + + artifacts_emitted: int = 0 + artifacts_skipped: int = 0 + + def record(self, status: DeduplicationStatus) -> None: + self.total_artifacts_scanned += 1 + + if status is DeduplicationStatus.NEW: + self.artifacts_new += 1 + self.artifacts_emitted += 1 + + elif status is DeduplicationStatus.UPDATED: + self.artifacts_updated += 1 + self.artifacts_emitted += 1 + + elif status is DeduplicationStatus.UNCHANGED: + self.artifacts_unchanged += 1 + self.artifacts_skipped += 1 diff --git a/application/utils/harvester/diff_retriever.py b/application/utils/harvester/diff_retriever.py index 7efd45560..665d10d19 100644 --- a/application/utils/harvester/diff_retriever.py +++ b/application/utils/harvester/diff_retriever.py @@ -8,13 +8,10 @@ class DiffRetriever: """ - Retrieves unified git diffs between two commits. This class is responsible only for retrieving raw diff text. - Parsing and normalization are handled by downstream components. - """ MAX_DIFF_SIZE_BYTES = 50 * 1024 * 1024 @@ -35,7 +32,6 @@ def get_diff(self, base_commit: str, target_commit: str = "HEAD") -> str: Raises: subprocess.CalledProcessError: If git diff fails. - ValueError: If the diff exceeds the configured size limit. """ @@ -44,7 +40,6 @@ def get_diff(self, base_commit: str, target_commit: str = "HEAD") -> str: base_commit, target_commit, ) - base_commit = self._resolve_commit(base_commit) target_commit = self._resolve_commit(target_commit) @@ -70,9 +65,7 @@ def get_diff(self, base_commit: str, target_commit: str = "HEAD") -> str: raise diff_bytes = result.stdout - diff_size = len(diff_bytes) - if diff_size > self.MAX_DIFF_SIZE_BYTES: raise ValueError( f"Diff size ({diff_size} bytes) exceeds " diff --git a/application/utils/harvester/document_builder.py b/application/utils/harvester/document_builder.py new file mode 100644 index 000000000..9988c9f13 --- /dev/null +++ b/application/utils/harvester/document_builder.py @@ -0,0 +1,46 @@ +from .artifact_id import generate_artifact_id +from .heading_extractor import HeadingExtractor +from .models import ( + DiffBlock, + Document, + Locator, + SourceInfo, +) + + +class DocumentBuilder: + """ + Builds structured Document objects from parsed diffs. + + This bridges raw git diff ingestion and downstream + semantic chunking. + """ + + SCHEMA_VERSION = "0.2.0" + + def build(self, block: DiffBlock, full_text: str, pipeline_run_id: str) -> Document: + artifact_id = generate_artifact_id( + block.repository, + block.file_path, + ) + + headings = HeadingExtractor().extract(full_text) + + return Document( + schema_version=self.SCHEMA_VERSION, + artifact_id=artifact_id, + pipeline_run_id=pipeline_run_id, + text=full_text, + heading_structure=headings, + source=SourceInfo( + type="github", + repository=block.repository, + commit_sha=block.commit_sha, + committed_at=block.committed_at, + ), + locator=Locator( + kind="repo_path", + id=block.file_path, + path=block.file_path, + ), + ) diff --git a/application/utils/harvester/document_deduplicator.py b/application/utils/harvester/document_deduplicator.py new file mode 100644 index 000000000..8150d994d --- /dev/null +++ b/application/utils/harvester/document_deduplicator.py @@ -0,0 +1,59 @@ +from .artifact_registry import ArtifactRegistry +from .content_hash import generate_content_hash +from .models import ( + ArtifactRegistryRecord, + DeduplicationStatus, + Document, +) +from datetime import datetime + + +class DocumentDeduplicator: + """ + Performs artifact-level deduplication. + + Documents are classified as: + - NEW + - UPDATED + - UNCHANGED + """ + + def __init__(self, registry: ArtifactRegistry): + self._registry = registry + + def process(self, document: Document) -> DeduplicationStatus: + content_hash = generate_content_hash(document.text) + + existing = self._registry.get(document.artifact_id) + + if existing is None: + self._registry.upsert( + ArtifactRegistryRecord( + artifact_id=document.artifact_id, + repository=document.source.repository, + locator_path=document.locator.path, + content_hash=content_hash, + last_commit_sha=document.source.commit_sha, + last_pipeline_run=document.pipeline_run_id, + last_processed_at=datetime.now(), + status=DeduplicationStatus.NEW.value, + ) + ) + + return DeduplicationStatus.NEW + + if existing.content_hash == content_hash: + existing.status = DeduplicationStatus.UNCHANGED.value + + self._registry.upsert(existing) + + return DeduplicationStatus.UNCHANGED + + existing.content_hash = content_hash + existing.last_commit_sha = document.source.commit_sha + existing.last_pipeline_run = document.pipeline_run_id + existing.status = DeduplicationStatus.UPDATED.value + + self._registry.upsert(existing) + + return DeduplicationStatus.UPDATED diff --git a/application/utils/harvester/document_validator.py b/application/utils/harvester/document_validator.py new file mode 100644 index 000000000..9e099a4aa --- /dev/null +++ b/application/utils/harvester/document_validator.py @@ -0,0 +1,42 @@ +from .models import Document + + +class DocumentValidator: + """ + Validates structured Document objects before indexing. + + Ensures every required metadata field has been populated. + """ + + def validate(self, document: Document) -> bool: + if not document.schema_version: + return False + + if not document.artifact_id.startswith("art:"): + return False + + if not document.pipeline_run_id: + return False + + if not document.text: + return False + + if document.source.type != "github": + return False + + if not document.source.repository: + return False + + if not document.source.commit_sha: + return False + + if document.source.committed_at is None: + return False + + if document.locator.kind != "repo_path": + return False + + if not document.locator.path: + return False + + return True diff --git a/application/utils/harvester/git_repository_client.py b/application/utils/harvester/git_repository_client.py index 468be2665..b650b74f7 100644 --- a/application/utils/harvester/git_repository_client.py +++ b/application/utils/harvester/git_repository_client.py @@ -284,3 +284,34 @@ def is_valid_repository(self, repository_path: Path) -> bool: def verify_repository_integrity(self) -> bool: return self.is_valid_repository(self.local_path) + + def get_file_at_commit(self, commit_sha: str, file_path: str) -> str: + """ + Retrieve the contents of a file at a specific commit. + + Args: + commit_sha: + Commit to read from. + + file_path: + Repository-relative file path. + + Returns: + File contents as a string. + """ + + result = subprocess.run( + [ + "git", + "-C", + str(self.get_local_path()), + "show", + f"{commit_sha}:{file_path}", + ], + capture_output=True, + text=True, + check=True, + timeout=30, + ) + + return result.stdout diff --git a/application/utils/harvester/heading_extractor.py b/application/utils/harvester/heading_extractor.py new file mode 100644 index 000000000..7941a805b --- /dev/null +++ b/application/utils/harvester/heading_extractor.py @@ -0,0 +1,47 @@ +from dataclasses import dataclass +from .models import HeadingNode + + +class HeadingExtractor: + """ + Extracts Markdown headings and their line ranges. + + Heading ranges extend until the next heading of the same + or higher level, or the end of the document. + """ + + def extract(self, text: str) -> list[HeadingNode]: + lines = text.splitlines() + + headings: list[HeadingNode] = [] + + for line_number, line in enumerate(lines, start=1): + stripped = line.lstrip() + + if not stripped.startswith("#"): + continue + + hashes = len(stripped) - len(stripped.lstrip("#")) + + if hashes == 0: + continue + + if len(stripped) > hashes and stripped[hashes] != " ": + continue + + headings.append( + HeadingNode( + level=hashes, + text=stripped[hashes:].strip(), + start_line=line_number, + end_line=len(lines), + ) + ) + + for index, heading in enumerate(headings): + for next_heading in headings[index + 1 :]: + if next_heading.level <= heading.level: + heading.end_line = next_heading.start_line - 1 + break + + return headings diff --git a/application/utils/harvester/incremental_pipeline.py b/application/utils/harvester/incremental_pipeline.py new file mode 100644 index 000000000..52a96c766 --- /dev/null +++ b/application/utils/harvester/incremental_pipeline.py @@ -0,0 +1,64 @@ +from datetime import datetime +from .checkpoint_manager import CheckpointManager +from .deduplication_metrics import DeduplicationMetrics +from .document_deduplicator import ( + DeduplicationStatus, + DocumentDeduplicator, +) +from .models import ( + CheckpointRecord, + Document, +) + + +class IncrementalPipeline: + """ + Coordinates document deduplication and checkpoint updates. + + Only NEW or UPDATED documents are emitted downstream. + """ + + def __init__( + self, deduplicator: DocumentDeduplicator, checkpoint_manager: CheckpointManager + ): + + self._deduplicator = deduplicator + self._checkpoint_manager = checkpoint_manager + self.metrics = DeduplicationMetrics() + + def process( + self, repository: str, pipeline_run_id: str, documents: list[Document] + ) -> list[Document]: + + emitted: list[Document] = [] + metrics = DeduplicationMetrics() + + if documents: + self._checkpoint_manager.save( + CheckpointRecord( + repository=repository, + pipeline_run_id=pipeline_run_id, + last_processed_commit="", + status="running", + updated_at=datetime.now(), + ) + ) + + for document in documents: + status = self._deduplicator.process(document) + + metrics.record(status) + self._checkpoint_manager.update_commit( + repository, + document.source.commit_sha, + ) + + if status != DeduplicationStatus.UNCHANGED: + emitted.append(document) + + self._checkpoint_manager.mark_completed( + repository, + ) + + self.metrics = metrics + return emitted diff --git a/application/utils/harvester/models.py b/application/utils/harvester/models.py index 0eca718c9..4a899d5a3 100644 --- a/application/utils/harvester/models.py +++ b/application/utils/harvester/models.py @@ -1,6 +1,7 @@ from dataclasses import dataclass from datetime import datetime from pydantic import BaseModel +from enum import Enum @dataclass(slots=True) @@ -39,3 +40,81 @@ class DiffBlock: repository: str commit_sha: str committed_at: datetime | None = None + + +@dataclass(slots=True) +class SourceInfo: + type: str + repository: str + commit_sha: str + committed_at: datetime | None + + +@dataclass(slots=True) +class Locator: + kind: str + id: str + path: str + + +@dataclass(slots=True) +class SpanInfo: + heading_path: list[str] + start_line: int + end_line: int + index: int | None = None + total: int | None = None + start_char_idx: int | None = None + end_char_idx: int | None = None + + +@dataclass(slots=True) +class HeadingNode: + level: int + text: str + start_line: int + end_line: int + + +@dataclass(slots=True) +class Document: + schema_version: str + artifact_id: str + pipeline_run_id: str + text: str + source: SourceInfo + locator: Locator + heading_structure: list[HeadingNode] + span: SpanInfo | None = None + + +@dataclass(slots=True) +class ArtifactRegistryRecord: + """ + Tracks the processing state of an artifact. + Used for deduplication across pipeline runs. + """ + + artifact_id: str + repository: str + locator_path: str + content_hash: str + last_commit_sha: str + last_pipeline_run: str + last_processed_at: datetime + status: str + + +class DeduplicationStatus(str, Enum): + NEW = "new" + UPDATED = "updated" + UNCHANGED = "unchanged" + + +@dataclass(slots=True) +class CheckpointRecord: + repository: str + pipeline_run_id: str + last_processed_commit: str + status: str + updated_at: datetime