diff --git a/pyproject.toml b/pyproject.toml
index 47c9f3d..2ac981a 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -116,6 +116,9 @@ testpaths = ["tests"]
python_files = ["test_*.py"]
python_classes = ["Test*"]
python_functions = ["test_*"]
+markers = [
+ "integration: tests that load real models (slow, requires network)",
+]
addopts = [
"--cov=cordon",
"--cov-report=term-missing",
diff --git a/src/cordon/analysis/thresholder.py b/src/cordon/analysis/thresholder.py
index 7327c29..f34f6d9 100644
--- a/src/cordon/analysis/thresholder.py
+++ b/src/cordon/analysis/thresholder.py
@@ -38,14 +38,13 @@ def select_significant(
scores = np.array([sw.score for sw in scored_windows])
- # Calculate percentile thresholds
- # e.g., min=0.05 (exclude top 5%) -> 95th percentile
- # e.g., max=0.15 (include up to 15%) -> 85th percentile
- upper_percentile = (1 - config.anomaly_range_min) * 100
- lower_percentile = (1 - config.anomaly_range_max) * 100
-
- upper_threshold = np.percentile(scores, upper_percentile)
- lower_threshold = np.percentile(scores, lower_percentile)
+ # anomaly_range_min = fraction to exclude from top (e.g., 0.05 = exclude top 5%)
+ # anomaly_range_max = cumulative fraction to include (e.g., 0.15 = include up to top 15%)
+ exclusion_cutoff = (1 - config.anomaly_range_min) * 100
+ inclusion_floor = (1 - config.anomaly_range_max) * 100
+
+ upper_threshold = np.percentile(scores, exclusion_cutoff)
+ lower_threshold = np.percentile(scores, inclusion_floor)
# Select windows in the range: lower <= score < upper
selected = [
diff --git a/src/cordon/cli.py b/src/cordon/cli.py
index 5f61461..02ed715 100644
--- a/src/cordon/cli.py
+++ b/src/cordon/cli.py
@@ -116,6 +116,12 @@ def parse_args() -> argparse.Namespace:
default=None,
help="Device for embedding and scoring (default: auto-detect)",
)
+ config_group.add_argument(
+ "--max-line-length",
+ type=int,
+ default=None,
+ help="Maximum characters per line before splitting into virtual lines (default: no limit)",
+ )
config_group.add_argument(
"--scoring-batch-size",
type=int,
@@ -300,6 +306,7 @@ def _main_impl() -> None:
try:
config = AnalysisConfig(
window_size=args.window_size,
+ max_line_length=args.max_line_length,
k_neighbors=args.k_neighbors,
anomaly_percentile=anomaly_percentile,
anomaly_range_min=anomaly_range_min,
diff --git a/src/cordon/core/__init__.py b/src/cordon/core/__init__.py
index eacd86a..b790836 100644
--- a/src/cordon/core/__init__.py
+++ b/src/cordon/core/__init__.py
@@ -2,10 +2,15 @@
from cordon.core.types import (
AnalysisResult,
Embedder,
+ Formatter,
MergedBlock,
+ Merger,
+ Reader,
ScoredWindow,
Scorer,
+ Segmenter,
TextWindow,
+ Thresholder,
)
__all__ = [
@@ -15,5 +20,10 @@
"MergedBlock",
"AnalysisResult",
"Embedder",
+ "Formatter",
+ "Merger",
+ "Reader",
"Scorer",
+ "Segmenter",
+ "Thresholder",
]
diff --git a/src/cordon/core/config.py b/src/cordon/core/config.py
index 6b6e8c6..d5e61d4 100644
--- a/src/cordon/core/config.py
+++ b/src/cordon/core/config.py
@@ -3,12 +3,49 @@
from typing import Literal
+@dataclass(frozen=True)
+class WindowConfig:
+ """Configuration for window segmentation.
+
+ Attributes:
+ window_size: Number of lines per sliding window.
+ max_line_length: Maximum characters per line before splitting
+ into virtual lines. None disables splitting.
+ """
+
+ window_size: int = 4
+ max_line_length: int | None = None
+
+
+@dataclass(frozen=True)
+class ScoringConfig:
+ """Configuration for anomaly scoring.
+
+ Attributes:
+ k_neighbors: Number of neighbors for k-NN density scoring.
+ anomaly_percentile: Fraction of windows to retain (e.g. 0.1 = top 10%).
+ anomaly_range_min: Lower bound for range mode.
+ anomaly_range_max: Upper bound for range mode.
+ device: Compute device for scoring. None for auto-detect.
+ scoring_batch_size: Batch size for k-NN scoring queries.
+ """
+
+ k_neighbors: int = 5
+ anomaly_percentile: float = 0.1
+ anomaly_range_min: float | None = None
+ anomaly_range_max: float | None = None
+ device: Literal["cuda", "mps", "cpu"] | None = None
+ scoring_batch_size: int | None = None
+
+
@dataclass(frozen=True)
class AnalysisConfig:
"""Global configuration for the analysis pipeline.
Attributes:
window_size: Number of lines per sliding window.
+ max_line_length: Maximum characters per line before splitting into
+ virtual lines. None disables splitting.
k_neighbors: Number of neighbors for k-NN density scoring.
anomaly_percentile: Fraction of windows to retain (e.g. 0.1 = top 10%).
anomaly_range_min: Lower bound for range mode (e.g. 0.05 = exclude top 5%).
@@ -33,6 +70,7 @@ class AnalysisConfig:
"""
window_size: int = 4
+ max_line_length: int | None = None
k_neighbors: int = 5
anomaly_percentile: float = 0.1
anomaly_range_min: float | None = None
@@ -66,6 +104,8 @@ def _validate_core_params(self) -> None:
raise ValueError("k_neighbors must be >= 1")
if not 0.0 <= self.anomaly_percentile <= 1.0:
raise ValueError("anomaly_percentile must be between 0.0 and 1.0")
+ if self.max_line_length is not None and self.max_line_length < 1:
+ raise ValueError("max_line_length must be >= 1 if set")
if self.batch_size < 1:
raise ValueError("batch_size must be >= 1")
if self.scoring_batch_size is not None and self.scoring_batch_size < 1:
diff --git a/src/cordon/core/types.py b/src/cordon/core/types.py
index 456c822..7c1497c 100644
--- a/src/cordon/core/types.py
+++ b/src/cordon/core/types.py
@@ -1,5 +1,6 @@
from collections.abc import Iterable, Iterator, Sequence
from dataclasses import dataclass
+from pathlib import Path
from typing import TYPE_CHECKING, Any, Protocol
import numpy as np
@@ -141,3 +142,51 @@ def score_windows(
List of scored windows
"""
...
+
+
+class Reader(Protocol):
+ """Protocol for log file readers."""
+
+ def read_lines(self, file_path: Path) -> Iterator[tuple[int, str]]:
+ """Read lines from a file with line number tracking."""
+ ...
+
+
+class Segmenter(Protocol):
+ """Protocol for line-to-window segmentation."""
+
+ def segment(
+ self, lines: Iterator[tuple[int, str]], config: "AnalysisConfig"
+ ) -> Iterator[TextWindow]:
+ """Segment lines into text windows."""
+ ...
+
+
+class Thresholder(Protocol):
+ """Protocol for selecting significant windows."""
+
+ def select_significant(
+ self, scored_windows: Sequence[ScoredWindow], config: "AnalysisConfig"
+ ) -> list[ScoredWindow]:
+ """Select windows above significance threshold."""
+ ...
+
+
+class Merger(Protocol):
+ """Protocol for merging overlapping windows."""
+
+ def merge_windows(self, scored_windows: Sequence[ScoredWindow]) -> list[MergedBlock]:
+ """Merge overlapping scored windows into contiguous blocks."""
+ ...
+
+
+class Formatter(Protocol):
+ """Protocol for formatting merged blocks into output."""
+
+ def format_blocks(
+ self,
+ merged_blocks: Sequence[MergedBlock],
+ lines: Sequence[tuple[int, str]],
+ ) -> str:
+ """Format merged blocks into output string."""
+ ...
diff --git a/src/cordon/embedding/__init__.py b/src/cordon/embedding/__init__.py
index 5b95a72..7afff73 100644
--- a/src/cordon/embedding/__init__.py
+++ b/src/cordon/embedding/__init__.py
@@ -1,11 +1,18 @@
"""Embedding module for log analysis."""
-from typing import TYPE_CHECKING
+import importlib
+from typing import TYPE_CHECKING, cast
if TYPE_CHECKING:
from cordon.core.config import AnalysisConfig
from cordon.core.types import Embedder
+_BACKEND_REGISTRY: dict[str, str] = {
+ "remote": "cordon.embedding.remote.RemoteEmbedder",
+ "llama-cpp": "cordon.embedding.llama_cpp.LlamaCppEmbedder",
+ "sentence-transformers": "cordon.embedding.transformer.TransformerEmbedder",
+}
+
def create_embedder(config: "AnalysisConfig") -> "Embedder":
"""Factory function to create the appropriate embedder for a config.
@@ -14,27 +21,22 @@ def create_embedder(config: "AnalysisConfig") -> "Embedder":
config: Analysis configuration with backend selection.
Returns:
- Embedder instance matching the configured backend.
+ Embedder instance implementing the Embedder protocol.
Raises:
ValueError: If the backend is not recognized.
"""
- if config.backend == "remote":
- from cordon.embedding.remote import RemoteEmbedder
-
- return RemoteEmbedder(config)
-
- if config.backend == "llama-cpp":
- from cordon.embedding.llama_cpp import LlamaCppEmbedder
-
- return LlamaCppEmbedder(config)
-
- if config.backend == "sentence-transformers":
- from cordon.embedding.transformer import TransformerEmbedder
-
- return TransformerEmbedder(config)
-
- raise ValueError(f"Unknown backend: {config.backend}")
+ class_path = _BACKEND_REGISTRY.get(config.backend)
+ if class_path is None:
+ raise ValueError(
+ f"Unknown backend: {config.backend!r}. "
+ f"Available: {', '.join(sorted(_BACKEND_REGISTRY))}"
+ )
+
+ module_path, class_name = class_path.rsplit(".", 1)
+ module = importlib.import_module(module_path)
+ embedder_class = getattr(module, class_name)
+ return cast("Embedder", embedder_class(config))
__all__ = ["create_embedder"]
diff --git a/src/cordon/pipeline.py b/src/cordon/pipeline.py
index 464557d..de7f365 100644
--- a/src/cordon/pipeline.py
+++ b/src/cordon/pipeline.py
@@ -4,12 +4,21 @@
import numpy as np
from cordon.analysis.scorer import DensityAnomalyScorer
-from cordon.analysis.thresholder import Thresholder
+from cordon.analysis.thresholder import Thresholder as ThresholderImpl
from cordon.core.config import AnalysisConfig
-from cordon.core.types import AnalysisResult, ScoredWindow
+from cordon.core.types import (
+ AnalysisResult,
+ Formatter,
+ Merger,
+ Reader,
+ ScoredWindow,
+ Scorer,
+ Segmenter,
+ Thresholder,
+)
from cordon.embedding import create_embedder
from cordon.ingestion.reader import LogFileReader
-from cordon.postprocess.formatter import OutputFormatter
+from cordon.postprocess.formatter import XmlFormatter
from cordon.postprocess.merger import IntervalMerger
from cordon.segmentation.windower import SlidingWindowSegmenter
@@ -22,14 +31,36 @@ class SemanticLogAnalyzer:
anomalies highlighted.
"""
- def __init__(self, config: AnalysisConfig | None = None) -> None:
- """Initialize the analyzer with configuration.
+ def __init__(
+ self,
+ config: AnalysisConfig | None = None,
+ *,
+ reader: Reader | None = None,
+ segmenter: Segmenter | None = None,
+ scorer: Scorer | None = None,
+ thresholder: Thresholder | None = None,
+ merger: Merger | None = None,
+ formatter: Formatter | None = None,
+ ) -> None:
+ """Initialize the analyzer with configuration and optional custom components.
Args:
config: Analysis configuration (uses defaults if None).
+ reader: Custom reader implementation.
+ segmenter: Custom segmenter implementation.
+ scorer: Custom scorer implementation.
+ thresholder: Custom thresholder implementation.
+ merger: Custom merger implementation.
+ formatter: Custom formatter implementation.
"""
self.config = config if config is not None else AnalysisConfig()
self._embedder = create_embedder(self.config)
+ self._reader = reader if reader is not None else LogFileReader()
+ self._segmenter = segmenter if segmenter is not None else SlidingWindowSegmenter()
+ self._scorer = scorer if scorer is not None else DensityAnomalyScorer()
+ self._thresholder = thresholder if thresholder is not None else ThresholderImpl()
+ self._merger = merger if merger is not None else IntervalMerger()
+ self._formatter = formatter if formatter is not None else XmlFormatter()
def analyze_file(self, file_path: Path) -> str:
"""Analyze a log file and return formatted output.
@@ -55,37 +86,31 @@ def analyze_file_detailed(self, file_path: Path) -> AnalysisResult:
start_time = time.time()
# stage 1: ingestion
- reader = LogFileReader()
- lines_list = list(reader.read_lines(file_path))
+ lines_list = list(self._reader.read_lines(file_path))
total_lines = len(lines_list)
# stage 2: segmentation
- segmenter = SlidingWindowSegmenter()
- windows = segmenter.segment(iter(lines_list), self.config)
+ windows = self._segmenter.segment(iter(lines_list), self.config)
# stage 3: vectorization
embedded = list(self._embedder.embed_windows(windows))
total_windows = len(embedded)
# stage 4: scoring
- scorer = DensityAnomalyScorer()
- scored = scorer.score_windows(embedded, self.config)
+ scored = self._scorer.score_windows(embedded, self.config)
del embedded
# stage 5: thresholding
- thresholder = Thresholder()
- significant = thresholder.select_significant(scored, self.config)
+ significant = self._thresholder.select_significant(scored, self.config)
significant_windows = len(significant)
# stage 6: merging
- merger = IntervalMerger()
- merged = merger.merge_windows(significant)
+ merged = self._merger.merge_windows(significant)
merged_blocks = len(merged)
del significant
# stage 7: formatting
- formatter = OutputFormatter()
- output = formatter.format_blocks(merged, lines_list)
+ output = self._formatter.format_blocks(merged, lines_list)
# calculate statistics
processing_time = time.time() - start_time
diff --git a/src/cordon/postprocess/__init__.py b/src/cordon/postprocess/__init__.py
index 6042896..d3ddc9b 100644
--- a/src/cordon/postprocess/__init__.py
+++ b/src/cordon/postprocess/__init__.py
@@ -1,4 +1,4 @@
-from cordon.postprocess.formatter import OutputFormatter
+from cordon.postprocess.formatter import XmlFormatter
from cordon.postprocess.merger import IntervalMerger
-__all__ = ["IntervalMerger", "OutputFormatter"]
+__all__ = ["IntervalMerger", "XmlFormatter"]
diff --git a/src/cordon/postprocess/formatter.py b/src/cordon/postprocess/formatter.py
index c074cc6..90f85d2 100644
--- a/src/cordon/postprocess/formatter.py
+++ b/src/cordon/postprocess/formatter.py
@@ -4,7 +4,7 @@
from cordon.core.types import MergedBlock
-class OutputFormatter:
+class XmlFormatter:
"""Generate XML-tagged output with original line content.
This formatter wraps each merged block in XML tags that specify
diff --git a/src/cordon/postprocess/merger.py b/src/cordon/postprocess/merger.py
index 3c6dee1..a04e4b4 100644
--- a/src/cordon/postprocess/merger.py
+++ b/src/cordon/postprocess/merger.py
@@ -1,8 +1,18 @@
from collections.abc import Sequence
+from typing import NamedTuple
from cordon.core.types import MergedBlock, ScoredWindow
+class _WindowInterval(NamedTuple):
+ """Internal representation of a window's line interval for merging."""
+
+ start: int
+ end: int
+ window_id: int
+ score: float
+
+
class IntervalMerger:
"""Merge overlapping line ranges into contiguous blocks.
@@ -15,42 +25,38 @@ def merge_windows(self, scored_windows: Sequence[ScoredWindow]) -> list[MergedBl
"""Merge overlapping windows into contiguous blocks.
Args:
- scored_windows: Sequence of scored windows to merge
+ scored_windows: Sequence of scored windows to merge.
Returns:
- List of merged blocks with no overlaps
+ List of merged blocks with no overlaps.
"""
if not scored_windows:
return []
- # convert to intervals: (start, end, window_id, score)
intervals = [
- (
- sw.window.start_line,
- sw.window.end_line,
- sw.window.window_id,
- sw.score,
+ _WindowInterval(
+ start=sw.window.start_line,
+ end=sw.window.end_line,
+ window_id=sw.window.window_id,
+ score=sw.score,
)
for sw in scored_windows
]
- intervals.sort(key=lambda interval: interval[0])
+ intervals.sort(key=lambda interval: interval.start)
- # initialize merge state with first interval
merged: list[MergedBlock] = []
- current_start, current_end, first_id, first_score = intervals[0]
- contributing_ids = [first_id]
- max_score = first_score
+ first = intervals[0]
+ current_start = first.start
+ current_end = first.end
+ contributing_ids = [first.window_id]
+ max_score = first.score
- # sweep through remaining intervals
- for start, end, window_id, score in intervals[1:]:
- # check if overlapping or adjacent (lines N and N+1 are adjacent)
- if start <= current_end + 1:
- # extend current block
- current_end = max(current_end, end)
- contributing_ids.append(window_id)
- max_score = max(max_score, score)
+ for interval in intervals[1:]:
+ if interval.start <= current_end + 1:
+ current_end = max(current_end, interval.end)
+ contributing_ids.append(interval.window_id)
+ max_score = max(max_score, interval.score)
else:
- # gap found - save current block and start new one
merged.append(
MergedBlock(
start_line=current_start,
@@ -59,12 +65,11 @@ def merge_windows(self, scored_windows: Sequence[ScoredWindow]) -> list[MergedBl
max_score=max_score,
)
)
- current_start = start
- current_end = end
- contributing_ids = [window_id]
- max_score = score
+ current_start = interval.start
+ current_end = interval.end
+ contributing_ids = [interval.window_id]
+ max_score = interval.score
- # append final block
merged.append(
MergedBlock(
start_line=current_start,
diff --git a/src/cordon/segmentation/windower.py b/src/cordon/segmentation/windower.py
index 28ecad5..2b3bb5f 100644
--- a/src/cordon/segmentation/windower.py
+++ b/src/cordon/segmentation/windower.py
@@ -1,4 +1,3 @@
-from collections import deque
from collections.abc import Iterator
from cordon.core.config import AnalysisConfig
@@ -10,7 +9,8 @@ class SlidingWindowSegmenter:
This segmenter creates non-overlapping chunks of text from a stream of lines.
Each window maintains references to its original line numbers for downstream
- processing.
+ processing. Optionally splits long lines into virtual lines based on a
+ character limit.
"""
def segment(
@@ -18,49 +18,75 @@ def segment(
) -> Iterator[TextWindow]:
"""Segment lines into non-overlapping text windows.
+ When max_line_length is set, lines exceeding that limit are split into
+ multiple virtual lines, each counting toward window_size. The window's
+ start_line/end_line still reference real file line numbers.
+
Args:
- lines: Iterator of (line_number, line_content) tuples
- config: Analysis configuration with window_size
+ lines: Iterator of (line_number, line_content) tuples.
+ config: Analysis configuration with window_size and
+ optional max_line_length.
Yields:
- TextWindow instances with content and line tracking
+ TextWindow instances with content and line tracking.
"""
window_size = config.window_size
+ max_line_length = config.max_line_length
- # use deque without maxlen to handle variable-length buffers
- buffer: deque[tuple[int, str]] = deque()
+ buffer: list[tuple[int, str]] = []
window_id = 0
for line_num, line_text in lines:
- buffer.append((line_num, line_text))
-
- # when buffer reaches window size, yield a window
- if len(buffer) == window_size:
- start_line = buffer[0][0]
- end_line = buffer[-1][0]
- content = "\n".join(text for _, text in buffer)
-
- yield TextWindow(
- content=content,
- start_line=start_line,
- end_line=end_line,
- window_id=window_id,
- )
-
- window_id += 1
-
- # clear buffer for next non-overlapping window
- buffer.clear()
-
- # handle final partial window
- if len(buffer) > 0:
- start_line = buffer[0][0]
- end_line = buffer[-1][0]
- content = "\n".join(text for _, text in buffer)
-
- yield TextWindow(
- content=content,
- start_line=start_line,
- end_line=end_line,
- window_id=window_id,
- )
+ chunks = self._split_line(line_num, line_text, max_line_length)
+
+ for chunk_line_num, chunk_text in chunks:
+ buffer.append((chunk_line_num, chunk_text))
+
+ if len(buffer) == window_size:
+ yield self._flush_buffer(buffer, window_id)
+ window_id += 1
+ buffer = []
+
+ if buffer:
+ yield self._flush_buffer(buffer, window_id)
+
+ @staticmethod
+ def _flush_buffer(buffer: list[tuple[int, str]], window_id: int) -> TextWindow:
+ """Create a TextWindow from buffered lines.
+
+ Args:
+ buffer: List of (line_number, line_content) tuples.
+ window_id: Unique identifier for this window.
+
+ Returns:
+ A TextWindow constructed from the buffer contents.
+ """
+ start_line = buffer[0][0]
+ end_line = buffer[-1][0]
+ content = "\n".join(text for _, text in buffer)
+ return TextWindow(
+ content=content,
+ start_line=start_line,
+ end_line=end_line,
+ window_id=window_id,
+ )
+
+ @staticmethod
+ def _split_line(line_num: int, line_text: str, max_length: int | None) -> list[tuple[int, str]]:
+ """Split a line into chunks if it exceeds max_length.
+
+ Args:
+ line_num: The real file line number.
+ line_text: The line content.
+ max_length: Maximum characters per chunk, or None to disable.
+
+ Returns:
+ List of (line_number, chunk_text) tuples. All chunks share
+ the same line_number since they originate from the same file line.
+ """
+ if max_length is None or len(line_text) <= max_length:
+ return [(line_num, line_text)]
+
+ return [
+ (line_num, line_text[i : i + max_length]) for i in range(0, len(line_text), max_length)
+ ]
diff --git a/tests/conftest.py b/tests/conftest.py
new file mode 100644
index 0000000..a8f6650
--- /dev/null
+++ b/tests/conftest.py
@@ -0,0 +1,55 @@
+"""Shared test fixtures for the cordon test suite."""
+
+from typing import Any
+
+import numpy as np
+import numpy.typing as npt
+import pytest
+
+from cordon.core.config import AnalysisConfig
+from cordon.core.types import MergedBlock, ScoredWindow, TextWindow
+
+
+@pytest.fixture
+def default_config() -> AnalysisConfig:
+ """Create a default AnalysisConfig for testing."""
+ return AnalysisConfig(device="cpu")
+
+
+@pytest.fixture
+def sample_windows() -> list[TextWindow]:
+ """Create a list of sample TextWindow instances."""
+ return [
+ TextWindow(
+ content=f"content for window {i}",
+ start_line=i * 4 + 1,
+ end_line=(i + 1) * 4,
+ window_id=i,
+ )
+ for i in range(5)
+ ]
+
+
+@pytest.fixture
+def sample_embeddings() -> npt.NDArray[np.floating[Any]]:
+ """Create sample normalized embedding vectors."""
+ rng = np.random.default_rng(42)
+ embeddings = rng.standard_normal((5, 384)).astype(np.float32)
+ norms = np.linalg.norm(embeddings, axis=1, keepdims=True)
+ return embeddings / norms
+
+
+@pytest.fixture
+def sample_scored_windows(sample_windows: list[TextWindow]) -> list[ScoredWindow]:
+ """Create sample scored windows."""
+ scores = [0.1, 0.5, 0.9, 0.3, 0.7]
+ return [ScoredWindow(window=w, score=s) for w, s in zip(sample_windows, scores, strict=True)]
+
+
+@pytest.fixture
+def sample_merged_blocks() -> list[MergedBlock]:
+ """Create sample merged blocks."""
+ return [
+ MergedBlock(start_line=1, end_line=4, original_windows=(0,), max_score=0.5),
+ MergedBlock(start_line=10, end_line=16, original_windows=(2, 3), max_score=0.9),
+ ]
diff --git a/tests/test_core.py b/tests/test_core.py
index 206ce54..5332b73 100644
--- a/tests/test_core.py
+++ b/tests/test_core.py
@@ -156,3 +156,14 @@ def test_range_mode_with_default_percentile(self) -> None:
)
assert config.anomaly_range_min == 0.05
assert config.anomaly_range_max == 0.15
+
+ def test_max_line_length_validation(self) -> None:
+ """Test that max_line_length must be >= 1 if set."""
+ with pytest.raises(ValueError, match="max_line_length"):
+ AnalysisConfig(max_line_length=0)
+ with pytest.raises(ValueError, match="max_line_length"):
+ AnalysisConfig(max_line_length=-1)
+ config = AnalysisConfig(max_line_length=None)
+ assert config.max_line_length is None
+ config = AnalysisConfig(max_line_length=500)
+ assert config.max_line_length == 500
diff --git a/tests/test_llama_cpp.py b/tests/test_llama_cpp.py
index 5316702..5517cf9 100644
--- a/tests/test_llama_cpp.py
+++ b/tests/test_llama_cpp.py
@@ -1,6 +1,9 @@
"""Unit tests for llama.cpp embedder backend."""
+from __future__ import annotations
+
import logging
+from typing import TYPE_CHECKING
import numpy as np
import pytest
@@ -8,6 +11,9 @@
from cordon.core.config import AnalysisConfig
from cordon.core.types import TextWindow
+if TYPE_CHECKING:
+ from cordon.embedding.llama_cpp import LlamaCppEmbedder
+
class TestLlamaCppEmbedderConfiguration:
"""Tests for LlamaCppEmbedder configuration and initialization."""
@@ -119,7 +125,7 @@ def model_path(self) -> str:
return model_path
@pytest.fixture
- def embedder(self, model_path: str):
+ def embedder(self, model_path: str) -> LlamaCppEmbedder:
"""Create a LlamaCppEmbedder instance for testing."""
pytest.importorskip("llama_cpp")
diff --git a/tests/test_pipeline_unit.py b/tests/test_pipeline_unit.py
new file mode 100644
index 0000000..548c711
--- /dev/null
+++ b/tests/test_pipeline_unit.py
@@ -0,0 +1,86 @@
+"""Unit tests for pipeline with mocked components."""
+
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+from cordon.core.config import AnalysisConfig
+from cordon.pipeline import SemanticLogAnalyzer
+
+
+class TestPipelineDI:
+ """Test dependency injection in SemanticLogAnalyzer."""
+
+ @patch("cordon.pipeline.create_embedder")
+ def test_custom_reader(self, mock_create: MagicMock, default_config: AnalysisConfig) -> None:
+ """Test that a custom reader is used when provided."""
+ mock_embedder = MagicMock()
+ mock_create.return_value = mock_embedder
+ mock_embedder.embed_windows.return_value = iter([])
+
+ mock_reader = MagicMock()
+ mock_reader.read_lines.return_value = iter([(1, "test line")])
+
+ analyzer = SemanticLogAnalyzer(default_config, reader=mock_reader)
+ analyzer.analyze_file_detailed(Path("dummy.log"))
+
+ mock_reader.read_lines.assert_called_once()
+
+ @patch("cordon.pipeline.create_embedder")
+ def test_custom_formatter(self, mock_create: MagicMock, default_config: AnalysisConfig) -> None:
+ """Test that a custom formatter is used when provided."""
+ mock_embedder = MagicMock()
+ mock_create.return_value = mock_embedder
+ mock_embedder.embed_windows.return_value = iter([])
+
+ mock_reader = MagicMock()
+ mock_reader.read_lines.return_value = iter([(1, "line")])
+
+ mock_formatter = MagicMock()
+ mock_formatter.format_blocks.return_value = ""
+
+ analyzer = SemanticLogAnalyzer(default_config, reader=mock_reader, formatter=mock_formatter)
+ result = analyzer.analyze_file_detailed(Path("dummy.log"))
+
+ mock_formatter.format_blocks.assert_called_once()
+ assert result.output == ""
+
+ @patch("cordon.pipeline.create_embedder")
+ def test_default_components_used_when_none_provided(
+ self, mock_create: MagicMock, default_config: AnalysisConfig
+ ) -> None:
+ """Test that default concrete classes are used when no custom components given."""
+ from cordon.analysis.scorer import DensityAnomalyScorer
+ from cordon.analysis.thresholder import Thresholder as ThresholderImpl
+ from cordon.ingestion.reader import LogFileReader
+ from cordon.postprocess.formatter import XmlFormatter
+ from cordon.postprocess.merger import IntervalMerger
+ from cordon.segmentation.windower import SlidingWindowSegmenter
+
+ mock_create.return_value = MagicMock()
+
+ analyzer = SemanticLogAnalyzer(default_config)
+
+ assert isinstance(analyzer._reader, LogFileReader)
+ assert isinstance(analyzer._segmenter, SlidingWindowSegmenter)
+ assert isinstance(analyzer._scorer, DensityAnomalyScorer)
+ assert isinstance(analyzer._thresholder, ThresholderImpl)
+ assert isinstance(analyzer._merger, IntervalMerger)
+ assert isinstance(analyzer._formatter, XmlFormatter)
+
+ @patch("cordon.pipeline.create_embedder")
+ def test_custom_scorer(self, mock_create: MagicMock, default_config: AnalysisConfig) -> None:
+ """Test that a custom scorer is used when provided."""
+ mock_embedder = MagicMock()
+ mock_create.return_value = mock_embedder
+ mock_embedder.embed_windows.return_value = iter([])
+
+ mock_reader = MagicMock()
+ mock_reader.read_lines.return_value = iter([(1, "line")])
+
+ mock_scorer = MagicMock()
+ mock_scorer.score_windows.return_value = []
+
+ analyzer = SemanticLogAnalyzer(default_config, reader=mock_reader, scorer=mock_scorer)
+ analyzer.analyze_file_detailed(Path("dummy.log"))
+
+ mock_scorer.score_windows.assert_called_once()
diff --git a/tests/test_postprocess.py b/tests/test_postprocess.py
index 9c56df8..d4f9301 100644
--- a/tests/test_postprocess.py
+++ b/tests/test_postprocess.py
@@ -1,5 +1,5 @@
from cordon.core.types import MergedBlock, ScoredWindow, TextWindow
-from cordon.postprocess.formatter import OutputFormatter
+from cordon.postprocess.formatter import XmlFormatter
from cordon.postprocess.merger import IntervalMerger
@@ -119,15 +119,15 @@ def test_merge_single_window(self) -> None:
assert merged[0].end_line == 5
-class TestOutputFormatter:
- """Tests for OutputFormatter class."""
+class TestXmlFormatter:
+ """Tests for XmlFormatter class."""
def test_format_single_block(self) -> None:
"""Test formatting a single block."""
lines = [(1, "line 1"), (2, "line 2"), (3, "line 3")]
blocks = [MergedBlock(start_line=1, end_line=2, original_windows=(0,), max_score=0.8)]
- formatter = OutputFormatter()
+ formatter = XmlFormatter()
output = formatter.format_blocks(blocks, lines)
assert '' in output
@@ -146,7 +146,7 @@ def test_format_multiple_blocks(self) -> None:
MergedBlock(start_line=5, end_line=7, original_windows=(1,), max_score=0.9),
]
- formatter = OutputFormatter()
+ formatter = XmlFormatter()
output = formatter.format_blocks(blocks, lines)
assert '' in output
@@ -161,7 +161,7 @@ def test_format_empty_blocks(self) -> None:
lines = [(1, "line 1")]
blocks: list[MergedBlock] = []
- formatter = OutputFormatter()
+ formatter = XmlFormatter()
output = formatter.format_blocks(blocks, lines)
assert output == '\n'
@@ -175,7 +175,7 @@ def test_format_escapes_xml_special_chars(self) -> None:
]
blocks = [MergedBlock(start_line=1, end_line=3, original_windows=(0,), max_score=0.8)]
- formatter = OutputFormatter()
+ formatter = XmlFormatter()
output = formatter.format_blocks(blocks, lines)
assert '' in output
@@ -198,7 +198,7 @@ def test_format_multi_block_content_correctness(self) -> None:
MergedBlock(start_line=7, end_line=8, original_windows=(1,), max_score=0.9),
]
- formatter = OutputFormatter()
+ formatter = XmlFormatter()
output = formatter.format_blocks(blocks, lines)
assert "line 7" in output
@@ -215,7 +215,7 @@ def test_format_truncated_file(self) -> None:
MergedBlock(start_line=2, end_line=10, original_windows=(0,), max_score=0.8),
]
- formatter = OutputFormatter()
+ formatter = XmlFormatter()
output = formatter.format_blocks(blocks, lines)
assert "line 2" in output
@@ -230,7 +230,7 @@ def test_format_unsorted_block_input(self) -> None:
MergedBlock(start_line=2, end_line=3, original_windows=(0,), max_score=0.7),
]
- formatter = OutputFormatter()
+ formatter = XmlFormatter()
output = formatter.format_blocks(blocks, lines)
block_2_pos = output.index(' MagicMock:
"""Create a mock response from litellm.embedding()."""
mock_response = MagicMock()
mock_response.data = [
@@ -60,7 +66,7 @@ def mock_litellm_response(self):
return mock_response
@pytest.fixture
- def embedder(self):
+ def embedder(self) -> RemoteEmbedder:
"""Create a RemoteEmbedder instance for testing."""
from cordon.embedding.remote import RemoteEmbedder
diff --git a/tests/test_segmentation.py b/tests/test_segmentation.py
index 443894f..d4c08c0 100644
--- a/tests/test_segmentation.py
+++ b/tests/test_segmentation.py
@@ -92,3 +92,57 @@ def test_window_ids_incremental(self) -> None:
for i, window in enumerate(windows):
assert window.window_id == i
+
+ def test_split_long_line_at_limit(self) -> None:
+ """Test that a line exactly at max_line_length is not split."""
+ config = AnalysisConfig(window_size=2, max_line_length=10)
+ lines = iter([(1, "0123456789"), (2, "abcdefghij")])
+ segmenter = SlidingWindowSegmenter()
+ windows = list(segmenter.segment(lines, config))
+ assert len(windows) == 1
+ assert windows[0].start_line == 1
+ assert windows[0].end_line == 2
+
+ def test_split_long_line_exceeds_limit(self) -> None:
+ """Test that a line exceeding max_line_length is split into virtual lines."""
+ config = AnalysisConfig(window_size=3, max_line_length=5)
+ lines = iter([(1, "abcdefghijklmno")])
+ segmenter = SlidingWindowSegmenter()
+ windows = list(segmenter.segment(lines, config))
+ assert len(windows) == 1
+ assert windows[0].start_line == 1
+ assert windows[0].end_line == 1
+ assert "abcde" in windows[0].content
+ assert "fghij" in windows[0].content
+ assert "klmno" in windows[0].content
+
+ def test_split_creates_multiple_windows(self) -> None:
+ """Test that a very long line creates multiple windows."""
+ config = AnalysisConfig(window_size=2, max_line_length=5)
+ lines = iter([(1, "abcdefghijklmnopqrst")])
+ segmenter = SlidingWindowSegmenter()
+ windows = list(segmenter.segment(lines, config))
+ assert len(windows) == 2
+ assert windows[0].start_line == 1
+ assert windows[0].end_line == 1
+ assert windows[1].start_line == 1
+ assert windows[1].end_line == 1
+
+ def test_no_split_when_under_limit(self) -> None:
+ """Test that short lines are not split."""
+ config = AnalysisConfig(window_size=2, max_line_length=100)
+ lines = iter([(1, "short"), (2, "lines")])
+ segmenter = SlidingWindowSegmenter()
+ windows = list(segmenter.segment(lines, config))
+ assert len(windows) == 1
+ assert "short" in windows[0].content
+ assert "lines" in windows[0].content
+
+ def test_no_split_when_disabled(self) -> None:
+ """Test that max_line_length=None disables splitting."""
+ config = AnalysisConfig(window_size=1, max_line_length=None)
+ lines = iter([(1, "a" * 10000)])
+ segmenter = SlidingWindowSegmenter()
+ windows = list(segmenter.segment(lines, config))
+ assert len(windows) == 1
+ assert len(windows[0].content) == 10000
diff --git a/tests/test_transformer.py b/tests/test_transformer.py
index c934824..3c02a06 100644
--- a/tests/test_transformer.py
+++ b/tests/test_transformer.py
@@ -1,6 +1,9 @@
"""Unit tests for sentence-transformers embedder backend."""
-from unittest.mock import patch
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+from unittest.mock import MagicMock, patch
import numpy as np
import pytest
@@ -8,10 +11,14 @@
from cordon.core.config import AnalysisConfig
from cordon.core.types import TextWindow
+if TYPE_CHECKING:
+ from cordon.embedding.transformer import TransformerEmbedder
+
class TestTransformerEmbedderConfiguration:
"""Tests for TransformerEmbedder configuration and initialization."""
+ @pytest.mark.integration
def test_default_backend_creates_transformer(self) -> None:
"""Test that default backend creates TransformerEmbedder."""
from cordon.embedding import create_embedder
@@ -23,6 +30,7 @@ def test_default_backend_creates_transformer(self) -> None:
assert isinstance(embedder, TransformerEmbedder)
+ @pytest.mark.integration
def test_explicit_sentence_transformers_backend(self) -> None:
"""Test explicit sentence-transformers backend selection."""
from cordon.embedding import create_embedder
@@ -34,6 +42,7 @@ def test_explicit_sentence_transformers_backend(self) -> None:
assert isinstance(embedder, TransformerEmbedder)
+ @pytest.mark.integration
def test_device_detection(self) -> None:
"""Test device detection logic."""
from cordon.embedding.transformer import TransformerEmbedder
@@ -43,6 +52,7 @@ def test_device_detection(self) -> None:
assert embedder.device == "cpu"
+ @pytest.mark.integration
def test_custom_model_name(self) -> None:
"""Test custom model name configuration."""
from cordon.embedding.transformer import TransformerEmbedder
@@ -53,7 +63,7 @@ def test_custom_model_name(self) -> None:
assert embedder.config.model_name == "all-MiniLM-L6-v2"
@patch("cordon.embedding.transformer.SentenceTransformer")
- def test_model_loading_failure(self, mock_st) -> None:
+ def test_model_loading_failure(self, mock_st: MagicMock) -> None:
"""Test that model loading failure raises RuntimeError."""
mock_st.side_effect = OSError("Model not found")
with pytest.raises(RuntimeError, match="Failed to load"):
@@ -62,11 +72,12 @@ def test_model_loading_failure(self, mock_st) -> None:
TransformerEmbedder(AnalysisConfig(device="cpu"))
+@pytest.mark.integration
class TestTransformerEmbedderEmbedding:
"""Tests for TransformerEmbedder embedding functionality."""
@pytest.fixture
- def embedder(self):
+ def embedder(self) -> TransformerEmbedder:
"""Create a TransformerEmbedder instance for testing."""
from cordon.embedding.transformer import TransformerEmbedder
@@ -176,6 +187,7 @@ def test_batching(self, embedder) -> None:
assert embedding.shape[0] > 0
+@pytest.mark.integration
class TestTransformerEmbedderIntegration:
"""Integration tests with the full analysis pipeline."""
@@ -190,3 +202,42 @@ def test_factory_creates_transformer_by_default(self) -> None:
assert isinstance(embedder, TransformerEmbedder)
assert embedder.config.backend == "sentence-transformers"
+
+
+class TestTransformerEmbedderUnit:
+ """Unit tests for TransformerEmbedder with mocked SentenceTransformer."""
+
+ @pytest.fixture
+ @patch("cordon.embedding.transformer.SentenceTransformer")
+ def embedder(self, mock_st: MagicMock) -> TransformerEmbedder:
+ """Create a TransformerEmbedder with a mocked model."""
+ from cordon.embedding.transformer import TransformerEmbedder
+
+ mock_model = MagicMock()
+ rng = np.random.default_rng(0)
+ raw = rng.standard_normal((1, 384)).astype(np.float32)
+ mock_model.encode.return_value = raw / np.linalg.norm(raw, axis=1, keepdims=True)
+ mock_st.return_value = mock_model
+
+ config = AnalysisConfig(device="cpu", batch_size=2)
+ return TransformerEmbedder(config)
+
+ def test_embed_returns_normalized_vectors(self, embedder: TransformerEmbedder) -> None:
+ """Test that mocked embedding produces normalized output."""
+ window = TextWindow(content="test content", start_line=1, end_line=1, window_id=0)
+ results = list(embedder.embed_windows([window]))
+
+ assert len(results) == 1
+ _, embedding = results[0]
+ assert isinstance(embedding, np.ndarray)
+ assert embedding.dtype == np.float32
+ norm = np.linalg.norm(embedding)
+ assert np.isclose(norm, 1.0, atol=1e-6)
+
+ def test_embed_preserves_window_identity(self, embedder: TransformerEmbedder) -> None:
+ """Test that embedding preserves the original window object."""
+ window = TextWindow(content="log entry", start_line=5, end_line=5, window_id=3)
+ results = list(embedder.embed_windows([window]))
+
+ result_window, _ = results[0]
+ assert result_window == window