forked from cleveragents/cleveragents-core
283 lines
8.5 KiB
Python
283 lines
8.5 KiB
Python
"""Protocol stubs, UKOTriple validation, and direct internal-call steps.
|
|
|
|
Coverage scenarios that exercise Protocol ``raise NotImplementedError``
|
|
stubs, Pydantic validator error paths, and internal helper functions
|
|
called directly (bypassing the public API) to ensure slipcover traces
|
|
the code.
|
|
|
|
All steps prefixed with ``idx`` to avoid AmbiguousStep collisions.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from behave import then # type: ignore[import-untyped]
|
|
|
|
from cleveragents.domain.models.acms.analyzers import UKOTriple
|
|
from cleveragents.domain.models.acms.provenance import IndexResult
|
|
from features.steps.uko_indexer_common import ULID_1, _make_resource
|
|
|
|
__all__: list[str] = []
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Protocol stub NotImplementedError steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then(
|
|
"idx calling ContentReader.read_content directly should raise NotImplementedError"
|
|
)
|
|
def step_proto_content_reader(context: Any) -> None:
|
|
from cleveragents.application.services.uko_indexer_protocols import (
|
|
ContentReader,
|
|
)
|
|
|
|
class _Bare(ContentReader): # type: ignore[misc]
|
|
pass
|
|
|
|
bare = _Bare() # type: ignore[abstract]
|
|
try:
|
|
bare.read_content(_make_resource(ULID_1))
|
|
assert False, "Expected NotImplementedError" # noqa: B011
|
|
except NotImplementedError:
|
|
pass
|
|
|
|
|
|
@then(
|
|
"idx calling IndexLifecycleHook.on_indexed directly"
|
|
" should raise NotImplementedError"
|
|
)
|
|
def step_proto_hook_indexed(context: Any) -> None:
|
|
from cleveragents.application.services.uko_indexer_protocols import (
|
|
IndexLifecycleHook,
|
|
)
|
|
|
|
class _Bare(IndexLifecycleHook): # type: ignore[misc]
|
|
pass
|
|
|
|
bare = _Bare() # type: ignore[abstract]
|
|
try:
|
|
bare.on_indexed(IndexResult(resource_id=ULID_1))
|
|
assert False, "Expected NotImplementedError" # noqa: B011
|
|
except NotImplementedError:
|
|
pass
|
|
|
|
|
|
@then(
|
|
"idx calling IndexLifecycleHook.on_removed directly"
|
|
" should raise NotImplementedError"
|
|
)
|
|
def step_proto_hook_removed(context: Any) -> None:
|
|
from cleveragents.application.services.uko_indexer_protocols import (
|
|
IndexLifecycleHook,
|
|
)
|
|
|
|
class _Bare(IndexLifecycleHook): # type: ignore[misc]
|
|
pass
|
|
|
|
bare = _Bare() # type: ignore[abstract]
|
|
try:
|
|
bare.on_removed(ULID_1, "p")
|
|
assert False, "Expected NotImplementedError" # noqa: B011
|
|
except NotImplementedError:
|
|
pass
|
|
|
|
|
|
@then(
|
|
"idx calling IndexLifecycleHook.on_error directly should raise NotImplementedError"
|
|
)
|
|
def step_proto_hook_error(context: Any) -> None:
|
|
from cleveragents.application.services.uko_indexer_protocols import (
|
|
IndexLifecycleHook,
|
|
)
|
|
|
|
class _Bare(IndexLifecycleHook): # type: ignore[misc]
|
|
pass
|
|
|
|
bare = _Bare() # type: ignore[abstract]
|
|
try:
|
|
bare.on_error(ULID_1, "err")
|
|
assert False, "Expected NotImplementedError" # noqa: B011
|
|
except NotImplementedError:
|
|
pass
|
|
|
|
|
|
@then(
|
|
"idx calling AnalyzerProtocol.supported_extensions directly"
|
|
" should raise NotImplementedError"
|
|
)
|
|
def step_proto_analyzer_ext(context: Any) -> None:
|
|
from cleveragents.domain.models.acms.analyzers import AnalyzerProtocol
|
|
|
|
class _Bare(AnalyzerProtocol): # type: ignore[misc]
|
|
pass
|
|
|
|
bare = _Bare() # type: ignore[abstract]
|
|
try:
|
|
_ = bare.supported_extensions
|
|
assert False, "Expected NotImplementedError" # noqa: B011
|
|
except NotImplementedError:
|
|
pass
|
|
|
|
|
|
@then("idx calling AnalyzerProtocol.domain directly should raise NotImplementedError")
|
|
def step_proto_analyzer_domain(context: Any) -> None:
|
|
from cleveragents.domain.models.acms.analyzers import AnalyzerProtocol
|
|
|
|
class _Bare(AnalyzerProtocol): # type: ignore[misc]
|
|
pass
|
|
|
|
bare = _Bare() # type: ignore[abstract]
|
|
try:
|
|
_ = bare.domain
|
|
assert False, "Expected NotImplementedError" # noqa: B011
|
|
except NotImplementedError:
|
|
pass
|
|
|
|
|
|
@then("idx calling AnalyzerProtocol.analyze directly should raise NotImplementedError")
|
|
def step_proto_analyzer_analyze(context: Any) -> None:
|
|
from cleveragents.domain.models.acms.analyzers import AnalyzerProtocol
|
|
|
|
class _Bare(AnalyzerProtocol): # type: ignore[misc]
|
|
pass
|
|
|
|
bare = _Bare() # type: ignore[abstract]
|
|
try:
|
|
bare.analyze("content", "uko://r")
|
|
assert False, "Expected NotImplementedError" # noqa: B011
|
|
except NotImplementedError:
|
|
pass
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# UKOTriple validation edge cases
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then("idx creating UKOTriple with whitespace-only subject_uri should raise ValueError")
|
|
def step_triple_ws_subject(context: Any) -> None:
|
|
try:
|
|
UKOTriple(
|
|
subject_uri=" ",
|
|
predicate="p",
|
|
object_uri="uko://o",
|
|
)
|
|
assert False, "Expected ValueError" # noqa: B011
|
|
except ValueError:
|
|
pass
|
|
|
|
|
|
@then("idx creating UKOTriple with whitespace-only predicate should raise ValueError")
|
|
def step_triple_ws_predicate(context: Any) -> None:
|
|
try:
|
|
UKOTriple(
|
|
subject_uri="uko://s",
|
|
predicate=" ",
|
|
object_uri="uko://o",
|
|
)
|
|
assert False, "Expected ValueError" # noqa: B011
|
|
except ValueError:
|
|
pass
|
|
|
|
|
|
@then(
|
|
"idx creating UKOTriple with empty object_uri and empty object_value"
|
|
" should raise ValueError"
|
|
)
|
|
def step_triple_no_object(context: Any) -> None:
|
|
try:
|
|
UKOTriple(
|
|
subject_uri="uko://s",
|
|
predicate="rdf:type",
|
|
object_uri="",
|
|
object_value="",
|
|
)
|
|
assert False, "Expected ValueError" # noqa: B011
|
|
except ValueError:
|
|
pass
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# fire_on_removed direct call
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then("idx calling fire_on_removed with a raising hook should not raise")
|
|
def step_fire_on_removed_direct(context: Any) -> None:
|
|
from cleveragents.application.services.uko_indexer_internals import (
|
|
fire_on_removed,
|
|
)
|
|
|
|
class _RaisingHook:
|
|
def on_indexed(self, result: IndexResult) -> None:
|
|
pass
|
|
|
|
def on_removed(self, resource_id: str, project: str) -> None:
|
|
raise RuntimeError("boom")
|
|
|
|
def on_error(self, resource_id: str, error: str) -> None:
|
|
pass
|
|
|
|
# Should not raise — fire_on_removed swallows the exception
|
|
fire_on_removed(
|
|
_RaisingHook(), # type: ignore[arg-type]
|
|
ULID_1,
|
|
"local/test",
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# AnalyzerRegistry validation and listing steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then("idx registering a non-protocol object should raise TypeError")
|
|
def step_registry_non_protocol(context: Any) -> None:
|
|
from cleveragents.domain.models.acms.analyzers import AnalyzerRegistry
|
|
|
|
registry = AnalyzerRegistry()
|
|
try:
|
|
registry.register("not_an_analyzer") # type: ignore[arg-type]
|
|
assert False, "Expected TypeError" # noqa: B011
|
|
except TypeError:
|
|
pass
|
|
|
|
|
|
@then("idx registering an analyzer with empty extensions should raise ValueError")
|
|
def step_registry_empty_extensions(context: Any) -> None:
|
|
from cleveragents.domain.models.acms.analyzers import AnalyzerRegistry
|
|
|
|
class _NoExtensions:
|
|
supported_extensions: frozenset[str] = frozenset()
|
|
domain = "empty"
|
|
|
|
def analyze(self, content: str, resource_uri: str) -> list[UKOTriple]:
|
|
return []
|
|
|
|
registry = AnalyzerRegistry()
|
|
try:
|
|
registry.register(_NoExtensions()) # type: ignore[arg-type]
|
|
assert False, "Expected ValueError" # noqa: B011
|
|
except ValueError:
|
|
pass
|
|
|
|
|
|
@then("idx looking up empty extension should return None")
|
|
def step_registry_empty_ext(context: Any) -> None:
|
|
assert context.idx_registry.get_for_extension("") is None
|
|
|
|
|
|
@then('idx list_extensions should include ".py"')
|
|
def step_registry_list_ext(context: Any) -> None:
|
|
exts = context.idx_registry.list_extensions()
|
|
assert ".py" in exts, f".py not in {exts}"
|
|
|
|
|
|
@then("idx list_analyzers should return {n:d} analyzer")
|
|
def step_registry_list_analyzers(context: Any, n: int) -> None:
|
|
analyzers = context.idx_registry.list_analyzers()
|
|
assert len(analyzers) == n, f"Expected {n} analyzers, got {len(analyzers)}"
|