cleveragents-core/features/steps/uko_indexer_coverage_steps.py

"""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)