"""Validation edge cases + coverage scenarios for UKO Indexer behave tests.""" from __future__ import annotations import shutil import tempfile import threading from pathlib import Path from typing import Any from behave import given, then, when # type: ignore[import-untyped] from cleveragents.application.services.uko_indexer import UKOIndexer from cleveragents.domain.models.acms.analyzers import ( AnalyzerRegistry, UKOTriple, ) from cleveragents.domain.models.acms.index_backends import IndexedDocument from cleveragents.domain.models.acms.provenance import ( IndexResult, ProvenancedTriple, ProvenanceMetadata, ) from cleveragents.domain.models.acms.python_analyzer import PythonAnalyzer from cleveragents.domain.models.core.resource import Resource from features.mocks.uko_indexer_mocks import ( FailingContentReader, InMemoryContentReader, ) from features.steps.uko_indexer_common import ( DEFAULT_PROJECT, ULID_1, _make_resource, ) __all__: list[str] = [] @then("idx LocationContentReader should reject path with dotdot component") def step_lcr_dotdot(context: Any) -> None: from cleveragents.application.services.uko_indexer_protocols import ( LocationContentReader, ) reader = LocationContentReader() resource = _make_resource(ULID_1, location="../etc/passwd") try: reader.read_content(resource) assert False, "Expected ValueError" # noqa: B011 except ValueError as exc: assert "traversal" in str(exc).lower() @then("idx LocationContentReader should reject resource with no location") def step_lcr_no_location(context: Any) -> None: from cleveragents.application.services.uko_indexer_protocols import ( LocationContentReader, ) reader = LocationContentReader() resource = _make_resource(ULID_1, location=None) try: reader.read_content(resource) assert False, "Expected ValueError" # noqa: B011 except ValueError as exc: assert "no location" in str(exc).lower() @then("idx LocationContentReader should reject max_content_size of 0") def step_lcr_zero_size(context: Any) -> None: from cleveragents.application.services.uko_indexer_protocols import ( LocationContentReader, ) try: LocationContentReader(max_content_size=0) assert False, "Expected ValueError" # noqa: B011 except ValueError as exc: assert "positive" in str(exc).lower() @then("idx LocationContentReader should read a real file successfully") def step_lcr_read_real(context: Any) -> None: import tempfile from cleveragents.application.services.uko_indexer_protocols import ( LocationContentReader, ) with tempfile.NamedTemporaryFile( mode="w", suffix=".py", delete=False, encoding="utf-8" ) as f: f.write("x = 1\n") f.flush() path = f.name from pathlib import Path reader = LocationContentReader(base_dir=Path(path).parent) resource = _make_resource(ULID_1, location=path) content = reader.read_content(resource) assert content == "x = 1\n" Path(path).unlink() @then("idx LocationContentReader should reject non-existent file") def step_lcr_nonexistent(context: Any) -> None: from cleveragents.application.services.uko_indexer_protocols import ( LocationContentReader, ) reader = LocationContentReader() resource = _make_resource(ULID_1, location="/tmp/nonexistent_file_abc123.py") try: reader.read_content(resource) assert False, "Expected OSError" # noqa: B011 except OSError: pass @then("idx LocationContentReader should reject path escaping base_dir") def step_lcr_escape_base(context: Any) -> None: import tempfile from pathlib import Path from cleveragents.application.services.uko_indexer_protocols import ( LocationContentReader, ) # Create a temp dir to use as base_dir base = Path(tempfile.mkdtemp()) reader = LocationContentReader(base_dir=base) # Try to read /etc/hostname which is outside the base resource = _make_resource(ULID_1, location="/etc/hostname") try: reader.read_content(resource) assert False, "Expected ValueError" # noqa: B011 except ValueError as exc: assert "escapes" in str(exc).lower() base.rmdir() @then("idx LocationContentReader should reject a non-regular file") def step_lcr_non_regular_file(context: Any) -> None: import os import tempfile from pathlib import Path from cleveragents.application.services.uko_indexer_protocols import ( LocationContentReader, ) tmp_dir = Path(tempfile.mkdtemp()) fifo_path = tmp_dir / "testpipe" os.mkfifo(str(fifo_path)) reader = LocationContentReader(base_dir=tmp_dir) resource = _make_resource(ULID_1, location=str(fifo_path)) try: reader.read_content(resource) assert False, "Expected ValueError" # noqa: B011 except ValueError as exc: assert ( "regular file" in str(exc).lower() or "not a regular file" in str(exc).lower() ), str(exc) finally: fifo_path.unlink() tmp_dir.rmdir() @then("idx LocationContentReader should reject content exceeding max size") def step_lcr_max_content_size(context: Any) -> None: import tempfile from pathlib import Path from cleveragents.application.services.uko_indexer_protocols import ( LocationContentReader, ) tmp_dir = Path(tempfile.mkdtemp()) big_file = tmp_dir / "big.txt" big_file.write_text("x" * 100, encoding="utf-8") reader = LocationContentReader(base_dir=tmp_dir, max_content_size=10) resource = _make_resource(ULID_1, location=str(big_file)) try: reader.read_content(resource) assert False, "Expected ValueError" # noqa: B011 except ValueError as exc: assert "exceeds" in str(exc).lower(), str(exc) finally: big_file.unlink() tmp_dir.rmdir() @then("idx creating UKOIndexer with invalid graph_backend should raise TypeError") def step_idx_invalid_graph(context: Any) -> None: try: UKOIndexer( analyzer_registry=context.idx_registry, graph_backend="not_a_backend", # type: ignore[arg-type] ) assert False, "Expected TypeError" # noqa: B011 except TypeError: pass @then("idx the indexer analyzer_registry should be the same registry") def step_idx_registry_prop(context: Any) -> None: assert context.idx_indexer.analyzer_registry is context.idx_registry @given("idx a lifecycle hook that raises on on_indexed") def step_idx_hook_fail_indexed(context: Any) -> None: class _FailOnIndexed: def on_indexed(self, result: IndexResult) -> None: raise RuntimeError("hook boom") def on_removed(self, resource_id: str, project: str) -> None: pass def on_error(self, resource_id: str, error: str) -> None: pass context.idx_failing_hook = _FailOnIndexed() @given("idx a lifecycle hook that raises on on_removed") def step_idx_hook_fail_removed(context: Any) -> None: class _FailOnRemoved: def on_indexed(self, result: IndexResult) -> None: pass def on_removed(self, resource_id: str, project: str) -> None: raise RuntimeError("hook boom") def on_error(self, resource_id: str, error: str) -> None: pass context.idx_failing_hook = _FailOnRemoved() @given("idx a lifecycle hook that raises on on_error") def step_idx_hook_fail_error(context: Any) -> None: class _FailOnError: def on_indexed(self, result: IndexResult) -> None: pass def on_removed(self, resource_id: str, project: str) -> None: pass def on_error(self, resource_id: str, error: str) -> None: raise RuntimeError("hook boom") context.idx_failing_hook = _FailOnError() @given("idx a content reader that raises on read") def step_idx_failing_reader(context: Any) -> None: context.idx_content_reader = FailingContentReader() @when("idx I create an indexer with the failing hook") def step_idx_make_with_hook(context: Any) -> None: context.idx_indexer = UKOIndexer( analyzer_registry=context.idx_registry, graph_backend=context.idx_graph_backend, text_backend=context.idx_text_backend, vector_backend=context.idx_vector_backend, content_reader=context.idx_content_reader, lifecycle_hook=context.idx_failing_hook, ) @then('idx removing resource "{rid}" should not raise') def step_idx_remove_no_raise(context: Any, rid: str) -> None: context.idx_indexer.remove_resource(rid, project=DEFAULT_PROJECT) @then("idx the index result should have errors") def step_idx_result_has_errors(context: Any) -> None: assert len(context.idx_result.errors) > 0, "Expected errors" @given("idx an analyzer that raises RuntimeError") def step_idx_failing_analyzer(context: Any) -> None: class _FailingAnalyzer: domain = "failing" supported_extensions = frozenset({".py"}) def analyze( self, content: str, resource_uri: str, ) -> list[UKOTriple]: raise RuntimeError("analyzer boom") context.idx_registry = AnalyzerRegistry() context.idx_registry.register(_FailingAnalyzer()) # type: ignore[arg-type] @given("idx an analyzer that produces {n:d} triples") def step_idx_many_triples_analyzer(context: Any, n: int) -> None: class _ManyTriplesAnalyzer: domain = "many" supported_extensions = frozenset({".py"}) def analyze( self, content: str, resource_uri: str, ) -> list[UKOTriple]: return [ UKOTriple( subject_uri=f"uko://s{i}", predicate="rdf:type", object_uri="uko://Thing", ) for i in range(n) ] context.idx_registry = AnalyzerRegistry() context.idx_registry.register(_ManyTriplesAnalyzer()) # type: ignore[arg-type] @when("idx I create an indexer with max_triples {n:d}") def step_idx_make_max_triples(context: Any, n: int) -> None: context.idx_indexer = UKOIndexer( analyzer_registry=context.idx_registry, graph_backend=context.idx_graph_backend, text_backend=context.idx_text_backend, vector_backend=context.idx_vector_backend, content_reader=context.idx_content_reader, max_triples=n, ) @then("idx the index result triple_count should be at most {n:d}") def step_idx_triple_count_capped(context: Any, n: int) -> None: assert context.idx_result.triple_count <= n, ( f"Expected at most {n} triples, got {context.idx_result.triple_count}" ) @given("idx an AnalyzerRegistry with the PythonAnalyzer") def step_registry_with_python(context: Any) -> None: registry = AnalyzerRegistry() registry.register(PythonAnalyzer()) context.idx_registry = registry @when('idx I analyze content "{content}" with resource URI "{resource_uri}"') def step_analyze_content(context: Any, content: str, resource_uri: str) -> None: real_content = content.replace("\\n", "\n") analyzer = context.idx_registry.get_for_extension(".py") assert analyzer is not None context.idx_triples = analyzer.analyze(real_content, resource_uri) @when( "idx I wrap the first triple with provenance" ' for resource "{resource_id}" at path "{path}"' ) def step_wrap_provenance(context: Any, resource_id: str, path: str) -> None: assert len(context.idx_triples) > 0, "No triples to wrap" prov = ProvenanceMetadata( source_resource=resource_id, source_path=path, ) context.idx_pt = ProvenancedTriple( triple=context.idx_triples[0], provenance=prov, ) @given('idx a content reader with resource "{rid}" content "{content}"') def step_idx_content_reader_setup(context: Any, rid: str, content: str) -> None: reader = InMemoryContentReader() reader.set_content(rid, content.replace("\\n", "\n")) context.idx_content_reader = reader @when("idx I create an indexer") def step_idx_create_indexer(context: Any) -> None: context.idx_indexer = UKOIndexer( analyzer_registry=context.idx_registry, graph_backend=context.idx_graph_backend, text_backend=context.idx_text_backend, vector_backend=context.idx_vector_backend, content_reader=context.idx_content_reader, ) @when('idx I index resource "{rid}" in project "{project}"') @then('idx I index resource "{rid}" in project "{project}"') def step_idx_index_resource_generic(context: Any, rid: str, project: str) -> None: resource = Resource( resource_id=rid, resource_type_name="git-checkout", location="src/example.py", classification="physical", ) context.idx_result = context.idx_indexer.index_resource( resource, project=project, ) @then("idx creating IndexedDocument with empty doc_id should raise ValueError") def step_indexed_doc_empty_docid(context: Any) -> None: try: IndexedDocument(project="p", doc_id="", char_count=0) assert False, "Expected ValueError" # noqa: B011 except ValueError: pass @then("idx creating ProvenanceMetadata with invalid source_range should raise") def step_prov_invalid_source_range(context: Any) -> None: try: ProvenanceMetadata(source_resource="r1", source_range="not-valid") assert False, "Expected ValueError" # noqa: B011 except Exception: pass @then("idx creating ProvenanceMetadata with empty source_range should succeed") def step_prov_empty_source_range(context: Any) -> None: pm = ProvenanceMetadata(source_resource="r1", source_range="") assert pm.source_range == "" @then("idx file watcher should cancel prior dest timer on rapid sequence") def step_fw_cancel_dest_timer(context: Any) -> None: """File B has pending timer, then move A->B cancels it (line 375).""" from watchdog.events import FileModifiedEvent, FileMovedEvent from cleveragents.application.services.resource_file_watcher import ( ResourceFileWatcher, ) ev = threading.Event() def cb(rid: str, _p: str, _ct: str) -> None: ev.set() td = tempfile.mkdtemp() try: fa = Path(td) / "a.py" fb = Path(td) / "b.py" fa.write_text("x=1\n", encoding="utf-8") fb.write_text("y=1\n", encoding="utf-8") w = ResourceFileWatcher(on_change=cb, debounce_seconds=10.0) w.watch(fa, resource_id="01HQ8ZDRX50000000000000088", project="local/t") w.watch(fb, resource_id="01HQ8ZDRX50000000000000089", project="local/t") w.start() # Creates pending timer for B w._handle_fs_event(FileModifiedEvent(str(fb.resolve()))) assert str(fb.resolve()) in w._pending_timers # Move A -> B: should cancel B's pending timer (line 375) w._handle_fs_event(FileMovedEvent(str(fa.resolve()), str(fb.resolve()))) # The move creates a new timer for dest (B); wait for it w._debounce_seconds = 0.05 # speed up for test w._handle_fs_event(FileModifiedEvent(str(fb.resolve()))) assert ev.wait(timeout=2.0) w.stop() finally: shutil.rmtree(td, ignore_errors=True) @then("idx file watcher EventBus move event should include dest_path") def step_fw_eventbus_move_dest(context: Any) -> None: """Verify EventBus event for move includes dest_path in details.""" from watchdog.events import FileMovedEvent from cleveragents.application.services.resource_file_watcher import ( ResourceFileWatcher, ) from features.mocks.uko_indexer_mocks import TrackingEventBus td = tempfile.mkdtemp() try: fp = Path(td) / "src.py" fp.write_text("x=1\n", encoding="utf-8") dp = str((Path(td) / "dst.py").resolve()) bus = TrackingEventBus() w = ResourceFileWatcher(event_bus=bus, debounce_seconds=0.05) w.watch(fp, resource_id="01HQ8ZDRX50000000000000066", project="local/t") w.start() w._handle_fs_event(FileMovedEvent(str(fp.resolve()), dp)) assert bus.wait(timeout=2.0) details = bus.events[-1].details assert "dest_path" in details, f"dest_path missing from {details}" assert details["dest_path"] == dp w.stop() finally: shutil.rmtree(td, ignore_errors=True)