"""Edge-case coverage steps for UKO Indexer behave tests. Covers removal-failure paths, max_triples validation, reindex on fresh resource, index_graph internals (empty obj, rdfs:label, provenance), index_stubs edge cases, and index_backends protocol base methods. All steps prefixed with ``idx`` to avoid AmbiguousStep collisions with other feature files. """ from __future__ import annotations 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 UKOTriple from cleveragents.domain.models.acms.index_stubs import ( InMemoryGraphIndexBackend, InMemoryTextIndexBackend, InMemoryVectorIndexBackend, ) from cleveragents.domain.models.acms.provenance import ( ProvenancedTriple, ProvenanceMetadata, ) from features.mocks.uko_indexer_mocks import ( RemovalFailingGraphBackend, RemovalFailingTextBackend, RemovalFailingVectorBackend, SelectiveFailingGraphBackend, ) from features.steps.uko_indexer_common import ULID_1, _make_resource __all__: list[str] = [] # --------------------------------------------------------------------------- # Removal-failing backend steps # --------------------------------------------------------------------------- @given("idx a UKOIndexer with removal-failing graph backend") def step_indexer_removal_failing_graph(context: Any) -> None: context.idx_indexer = UKOIndexer( analyzer_registry=context.idx_registry, graph_backend=RemovalFailingGraphBackend(), text_backend=context.idx_text_backend, vector_backend=context.idx_vector_backend, content_reader=context.idx_content_reader, ) @given("idx a UKOIndexer with removal-failing text backend") def step_indexer_removal_failing_text(context: Any) -> None: context.idx_indexer = UKOIndexer( analyzer_registry=context.idx_registry, graph_backend=context.idx_graph_backend, text_backend=RemovalFailingTextBackend(), vector_backend=context.idx_vector_backend, content_reader=context.idx_content_reader, ) @given("idx a UKOIndexer with removal-failing vector backend") def step_indexer_removal_failing_vector(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=RemovalFailingVectorBackend(), content_reader=context.idx_content_reader, ) # --------------------------------------------------------------------------- # max_triples validation # --------------------------------------------------------------------------- @then("idx creating a UKOIndexer with max_triples 0 should raise ValueError") def step_max_triples_zero(context: Any) -> None: from cleveragents.domain.models.acms.analyzers import AnalyzerRegistry try: UKOIndexer( analyzer_registry=AnalyzerRegistry(), graph_backend=InMemoryGraphIndexBackend(), max_triples=0, ) assert False, "Expected ValueError" # noqa: B011 except ValueError as exc: assert "positive" in str(exc).lower() # --------------------------------------------------------------------------- # reindex_resource on fresh resource # --------------------------------------------------------------------------- @when('idx I reindex resource "{rid}" in project "{project}"') def step_reindex_resource(context: Any, rid: str, project: str) -> None: resource = _make_resource(rid) context.idx_result = context.idx_indexer.reindex_resource( resource, project=project, ) # --------------------------------------------------------------------------- # index_graph internals edge cases # --------------------------------------------------------------------------- @then("idx index_graph should skip triples with empty object") def step_index_graph_empty_obj(context: Any) -> None: from cleveragents.application.services.uko_indexer_internals import index_graph backend = InMemoryGraphIndexBackend() # Use model_construct to bypass the validator that requires at # least one of object_uri / object_value to be non-empty. This # tests the defensive guard inside index_graph. triple = UKOTriple.model_construct( subject_uri="urn:x", predicate="uko:name", object_uri="", object_value="", confidence=1.0, ) prov = ProvenanceMetadata(source_resource=ULID_1, source_path="f.py") pt = ProvenancedTriple.model_construct(triple=triple, provenance=prov) errors: list[str] = [] stored, subjects = index_graph( backend, "local/test", _make_resource(ULID_1), [pt], errors, "uko://resource/" + ULID_1, ) assert stored == 0 assert len(subjects) == 0 @then("idx index_graph should tolerate rdfs:label failure") def step_index_graph_label_failure(context: Any) -> None: from cleveragents.application.services.uko_indexer_internals import index_graph backend = SelectiveFailingGraphBackend(fail_on={"rdfs:label"}) triple = UKOTriple( subject_uri="urn:x", predicate="uko:name", object_uri="urn:target", object_value="Target Label", ) pt = ProvenancedTriple( triple=triple, provenance=ProvenanceMetadata(source_resource=ULID_1, source_path="f.py"), ) errors: list[str] = [] stored, _subjects = index_graph( backend, "local/test", _make_resource(ULID_1), [pt], errors, "uko://resource/" + ULID_1, ) assert stored == 1 assert any("label" in e.lower() for e in errors) @then("idx index_graph should tolerate provenance link failure") def step_index_graph_provenance_failure(context: Any) -> None: from cleveragents.application.services.uko_indexer_internals import index_graph backend = SelectiveFailingGraphBackend(fail_on={"uko:sourceResource"}) triple = UKOTriple( subject_uri="urn:x", predicate="uko:name", object_uri="urn:target", ) pt = ProvenancedTriple( triple=triple, provenance=ProvenanceMetadata(source_resource=ULID_1, source_path="f.py"), ) errors: list[str] = [] stored, _subjects = index_graph( backend, "local/test", _make_resource(ULID_1), [pt], errors, "uko://resource/" + ULID_1, ) assert stored == 1 assert any("provenance" in e.lower() for e in errors) # --------------------------------------------------------------------------- # index_stubs edge cases # --------------------------------------------------------------------------- @then("idx text search should stop at the limit") def step_text_search_limit(context: Any) -> None: backend = InMemoryTextIndexBackend() backend.index_document("p", "d1", "hello world", {}) backend.index_document("p", "d2", "hello again", {}) backend.index_document("p", "d3", "hello third", {}) results = backend.search("p", "hello", limit=2) assert len(results) == 2 @then("idx vector search_similar with empty embedding should raise ValueError") def step_vector_empty_embedding(context: Any) -> None: backend = InMemoryVectorIndexBackend() try: backend.search_similar("p", []) assert False, "Expected ValueError" # noqa: B011 except ValueError as exc: assert "non-empty" in str(exc).lower() @then("idx vector search_similar with limit 0 should raise ValueError") def step_vector_zero_limit(context: Any) -> None: backend = InMemoryVectorIndexBackend() try: backend.search_similar("p", [1.0], limit=0) assert False, "Expected ValueError" # noqa: B011 except ValueError as exc: assert "positive" in str(exc).lower() @then("idx vector search_similar filters by project") def step_vector_project_filter(context: Any) -> None: backend = InMemoryVectorIndexBackend() backend.index_embedding("proj_a", "d1", [1.0], {}) backend.index_embedding("proj_b", "d2", [1.0], {}) results = backend.search_similar("proj_a", [1.0], limit=10) assert len(results) == 1 assert results[0].doc_id == "d1" @then("idx graph remove_triples on missing project should be no-op") def step_graph_remove_missing_project(context: Any) -> None: backend = InMemoryGraphIndexBackend() backend.remove_triples("nonexistent", subject="s", predicate=None, obj=None) assert backend.triple_count() == 0 @then("idx graph triple_count without project should sum all projects") def step_graph_triple_count_all(context: Any) -> None: backend = InMemoryGraphIndexBackend() backend.add_triple("p1", "s1", "pred", "o1") backend.add_triple("p2", "s2", "pred", "o2") assert backend.triple_count(project=None) == 2 assert backend.triple_count(project="p1") == 1 # --------------------------------------------------------------------------- # index_backends protocol base method coverage # --------------------------------------------------------------------------- @then("idx all index backend protocol base methods should be callable") def step_protocol_base_methods(context: Any) -> None: from cleveragents.domain.models.acms.index_backends import ( GraphIndexBackend, TextIndexBackend, VectorIndexBackend, ) class _TextSub(TextIndexBackend): pass class _VectorSub(VectorIndexBackend): pass class _GraphSub(GraphIndexBackend): pass t = _TextSub() # type: ignore[abstract] t.index_document("p", "d", "c", {}) t.search("p", "q", limit=1) t.remove_document("p", "d") t.rebuild_index("p") v = _VectorSub() # type: ignore[abstract] v.index_embedding("p", "d", [1.0], {}) v.search_similar("p", [1.0], limit=1) v.remove_embedding("p", "d") g = _GraphSub() # type: ignore[abstract] g.add_triple("p", "s", "pred", "o") g.query("p", "SELECT ?s WHERE {}") g.remove_triples("p", subject="s", predicate=None, obj=None) # --------------------------------------------------------------------------- # min_relevance filtering (P3-3) # --------------------------------------------------------------------------- @then("idx vector search_similar with min_relevance above 1.0 raises ValueError") def step_vector_min_relevance_above(context: Any) -> None: backend = InMemoryVectorIndexBackend() backend.index_embedding("p", "d1", [1.0], {}) try: backend.search_similar("p", [1.0], limit=10, min_relevance=1.1) assert False, "Expected ValueError" # noqa: B011 except ValueError: pass @then("idx vector search_similar with min_relevance 1.0 includes exact matches") def step_vector_min_relevance_exact(context: Any) -> None: backend = InMemoryVectorIndexBackend() backend.index_embedding("p", "d1", [1.0], {}) results = backend.search_similar("p", [1.0], limit=10, min_relevance=1.0) assert len(results) == 1 assert results[0].doc_id == "d1" @then("idx vector search_similar with min_relevance 0.0 includes all matches") def step_vector_min_relevance_zero(context: Any) -> None: backend = InMemoryVectorIndexBackend() backend.index_embedding("p", "d1", [1.0], {}) backend.index_embedding("p", "d2", [2.0], {}) results = backend.search_similar("p", [1.0], limit=10, min_relevance=0.0) assert len(results) == 2 # --------------------------------------------------------------------------- # Stub size-limit enforcement (P2-16) # --------------------------------------------------------------------------- @then("idx text stub should raise RuntimeError when max_entries exceeded") def step_text_stub_size_limit(context: Any) -> None: backend = InMemoryTextIndexBackend(max_entries=2) backend.index_document("p", "d1", "content1", {}) backend.index_document("p", "d2", "content2", {}) try: backend.index_document("p", "d3", "content3", {}) assert False, "Expected RuntimeError" # noqa: B011 except RuntimeError as exc: assert "max_entries" in str(exc) @then("idx vector stub should raise RuntimeError when max_entries exceeded") def step_vector_stub_size_limit(context: Any) -> None: backend = InMemoryVectorIndexBackend(max_entries=2) backend.index_embedding("p", "d1", [1.0], {}) backend.index_embedding("p", "d2", [2.0], {}) try: backend.index_embedding("p", "d3", [3.0], {}) assert False, "Expected RuntimeError" # noqa: B011 except RuntimeError as exc: assert "max_entries" in str(exc) @then("idx graph stub should raise RuntimeError when max_entries exceeded") def step_graph_stub_size_limit(context: Any) -> None: backend = InMemoryGraphIndexBackend(max_entries=2) backend.add_triple("p", "s1", "pred", "o1") backend.add_triple("p", "s2", "pred", "o2") try: backend.add_triple("p", "s3", "pred", "o3") assert False, "Expected RuntimeError" # noqa: B011 except RuntimeError as exc: assert "max_entries" in str(exc) # --------------------------------------------------------------------------- # Provenance metadata persistence (P2-22, P2-27) # --------------------------------------------------------------------------- @then("idx index_graph should store provenance metadata triples") def step_index_graph_provenance_metadata(context: Any) -> None: from cleveragents.application.services.uko_indexer_internals import index_graph backend = InMemoryGraphIndexBackend() triple = UKOTriple( subject_uri="urn:x", predicate="uko:name", object_uri="urn:target", object_value="TargetLabel", confidence=0.85, ) prov = ProvenanceMetadata( source_resource=ULID_1, source_path="src/example.py", source_range="10-25", ) pt = ProvenancedTriple(triple=triple, provenance=prov) errors: list[str] = [] stored, _subjects = index_graph( backend, "local/test", _make_resource(ULID_1), [pt], errors, "uko://resource/" + ULID_1, ) assert stored == 1 # Query all triples for the project bindings = backend.query("local/test", "SELECT * WHERE { ?s ?p ?o }") # Build a predicate -> object map for the subject preds = {b["p"]: b["o"] for b in bindings if b["s"] == "urn:x"} # Data triple assert preds.get("uko:name") == "urn:target" # Provenance metadata assert preds.get("uko:sourceResource") == "uko://resource/" + ULID_1 assert preds.get("uko:sourcePath") == "src/example.py" assert preds.get("uko:sourceRange") == "10-25" assert "uko:validFrom" in preds # ISO timestamp assert preds.get("uko:isCurrent") == "true" # Confidence < 1.0 should be stored assert preds.get("uko:confidence") == "0.85" # --------------------------------------------------------------------------- # Concurrency — thread-safe indexing (#58) # --------------------------------------------------------------------------- @then("idx concurrent indexing of different resources should be thread-safe") def step_idx_concurrent_indexing(context: Any) -> None: import threading from cleveragents.domain.models.acms.python_analyzer import PythonAnalyzer from features.mocks.uko_indexer_mocks import InMemoryContentReader reg = context.idx_registry reg.register(PythonAnalyzer(), priority=10) reader = InMemoryContentReader() indexer = UKOIndexer( analyzer_registry=reg, graph_backend=context.idx_graph_backend, text_backend=context.idx_text_backend, vector_backend=context.idx_vector_backend, content_reader=reader, ) results: list[Any] = [None] * 4 errs: list[Exception] = [] rids = [f"01HQ8ZDRX5000000000000{i:04d}" for i in range(4)] def _do(i: int) -> None: rid = rids[i] reader.set_content(rid, f"def f{i}(): pass\n") try: results[i] = indexer.index_resource( _make_resource(rid, f"src/m{i}.py"), project="local/app", ) except Exception as e: errs.append(e) ts = [threading.Thread(target=_do, args=(i,)) for i in range(4)] for t in ts: t.start() for t in ts: t.join(timeout=10) assert not errs, f"Concurrent errors: {errs}" assert indexer.indexed_resource_count == 4