243 lines
8.2 KiB
Python
243 lines
8.2 KiB
Python
"""Property-basierte Tests für den Wissensspeicher: Artefakt-Ingestion.
|
|
|
|
**Validates: Requirements 3.2, 3.4, 3.6, 3.11, 3.13, 3.17, 3.21**
|
|
|
|
Property 5: Wissensartefakt-Ingestion erzeugt vollständige Metadaten im Index
|
|
- For any gültiges Wissensartefakt aus einem beliebigen Kontext, nach der Verarbeitung
|
|
durch die ETL-Pipeline muss es im YAML-Index erscheinen mit korrektem YAML-Frontmatter
|
|
(mindestens Typ, Titel, Tags, Quellkontext, Erstelldatum, Content-Hash) und der Pfad
|
|
muss der Scope-basierten Ordnerstruktur entsprechen.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
from hypothesis import given, settings
|
|
from hypothesis import strategies as st
|
|
|
|
from monorepo.knowledge.etl import ETLPipeline
|
|
from monorepo.knowledge.index import YAMLIndex
|
|
from monorepo.knowledge.sources.markdown import MarkdownSource
|
|
from monorepo.models import ArtifactType, Context
|
|
|
|
|
|
# --- Constants ---
|
|
|
|
# Valid artifact types from the model
|
|
VALID_ARTIFACT_TYPES = [t.value for t in ArtifactType]
|
|
|
|
# Valid contexts (non-shared, as source contexts for artifacts)
|
|
VALID_CONTEXTS = [c.value for c in Context if c != Context.SHARED]
|
|
|
|
|
|
# --- Strategies ---
|
|
|
|
|
|
@st.composite
|
|
def valid_tag(draw: st.DrawFn) -> str:
|
|
"""Generates a valid tag: lowercase alphanumeric with optional hyphens."""
|
|
return draw(
|
|
st.from_regex(r"[a-z][a-z0-9\-]{0,14}[a-z0-9]", fullmatch=True)
|
|
)
|
|
|
|
|
|
@st.composite
|
|
def valid_artifact_name(draw: st.DrawFn) -> str:
|
|
"""Generates a valid artifact filename (kebab-case, no extension)."""
|
|
return draw(
|
|
st.from_regex(r"[a-z][a-z0-9\-]{1,20}[a-z0-9]", fullmatch=True)
|
|
)
|
|
|
|
|
|
@st.composite
|
|
def valid_title(draw: st.DrawFn) -> str:
|
|
"""Generates a valid artifact title (non-empty, printable)."""
|
|
return draw(
|
|
st.text(
|
|
alphabet=st.characters(
|
|
whitelist_categories=("L", "N", "Z"),
|
|
whitelist_characters="-_ ",
|
|
),
|
|
min_size=3,
|
|
max_size=60,
|
|
).filter(lambda t: t.strip() != "")
|
|
)
|
|
|
|
|
|
@st.composite
|
|
def valid_markdown_artifact(draw: st.DrawFn) -> dict[str, object]:
|
|
"""Generates a complete valid Markdown artifact with frontmatter.
|
|
|
|
Returns a dict with:
|
|
- name: filename without .md extension
|
|
- type: artifact type (decision, note, etc.)
|
|
- title: artifact title
|
|
- tags: list of tags
|
|
- content: markdown body content
|
|
- context: source context
|
|
"""
|
|
name = draw(valid_artifact_name())
|
|
artifact_type = draw(st.sampled_from(VALID_ARTIFACT_TYPES))
|
|
title = draw(valid_title())
|
|
tags = draw(st.lists(valid_tag(), min_size=1, max_size=5, unique=True))
|
|
context = draw(st.sampled_from(VALID_CONTEXTS))
|
|
|
|
# Generate some markdown body content
|
|
body_line = draw(
|
|
st.text(
|
|
alphabet=st.characters(
|
|
whitelist_categories=("L", "N", "Z"),
|
|
whitelist_characters="-_., ",
|
|
),
|
|
min_size=10,
|
|
max_size=100,
|
|
).filter(lambda t: t.strip() != "")
|
|
)
|
|
content = f"\n# {title}\n\n{body_line}\n"
|
|
|
|
return {
|
|
"name": name,
|
|
"type": artifact_type,
|
|
"title": title,
|
|
"tags": tags,
|
|
"content": content,
|
|
"context": context,
|
|
}
|
|
|
|
|
|
def _write_markdown_file(directory: Path, artifact: dict[str, object]) -> Path:
|
|
"""Writes a Markdown file with YAML frontmatter to the given directory.
|
|
|
|
Returns the path to the written file.
|
|
"""
|
|
name: str = artifact["name"] # type: ignore[assignment]
|
|
artifact_type: str = artifact["type"] # type: ignore[assignment]
|
|
title: str = artifact["title"] # type: ignore[assignment]
|
|
tags: list[str] = artifact["tags"] # type: ignore[assignment]
|
|
content: str = artifact["content"] # type: ignore[assignment]
|
|
|
|
# Build YAML frontmatter
|
|
tags_str = ", ".join(tags)
|
|
frontmatter = (
|
|
f"---\n"
|
|
f"type: {artifact_type}\n"
|
|
f"title: \"{title}\"\n"
|
|
f"tags: [{tags_str}]\n"
|
|
f"---\n"
|
|
)
|
|
|
|
file_path = directory / f"{name}.md"
|
|
file_path.write_text(frontmatter + content, encoding="utf-8")
|
|
return file_path
|
|
|
|
|
|
# --- Property Test ---
|
|
|
|
|
|
class TestProperty5ArtifactIngestionMetadata:
|
|
"""Property 5: Wissensartefakt-Ingestion erzeugt vollständige Metadaten im Index.
|
|
|
|
**Validates: Requirements 3.2, 3.4, 3.6, 3.11, 3.13, 3.17, 3.21**
|
|
|
|
For any valid knowledge artifact from any context, after processing by the
|
|
ETL pipeline, it must appear in the YAML index with:
|
|
- id set
|
|
- title matches
|
|
- type matches
|
|
- tags match
|
|
- scope == context
|
|
- content_hash starts with "sha256:"
|
|
- path follows scope-based structure ({context}/{type}/{name}.md)
|
|
"""
|
|
|
|
@given(artifact=valid_markdown_artifact())
|
|
@settings(max_examples=50, deadline=5000)
|
|
def test_ingested_artifact_has_complete_metadata_in_index(
|
|
self, artifact: dict[str, object], tmp_path_factory: object
|
|
) -> None:
|
|
"""After ingestion, the artifact appears in the YAML index with all required metadata."""
|
|
# Use a unique temp directory for each test invocation
|
|
import tempfile
|
|
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
tmp = Path(tmpdir)
|
|
source_dir = tmp / "source"
|
|
source_dir.mkdir()
|
|
store_path = tmp / "store"
|
|
store_path.mkdir()
|
|
|
|
# Extract artifact data
|
|
name: str = artifact["name"] # type: ignore[assignment]
|
|
artifact_type: str = artifact["type"] # type: ignore[assignment]
|
|
title: str = artifact["title"] # type: ignore[assignment]
|
|
tags: list[str] = artifact["tags"] # type: ignore[assignment]
|
|
context: str = artifact["context"] # type: ignore[assignment]
|
|
|
|
# Write the artifact as a Markdown file
|
|
_write_markdown_file(source_dir, artifact)
|
|
|
|
# Set up ETL pipeline
|
|
index = YAMLIndex(store_path / "_index.yaml")
|
|
index.load()
|
|
pipeline = ETLPipeline(index=index, store_path=store_path)
|
|
source = MarkdownSource(directory=source_dir)
|
|
|
|
# Run ingestion
|
|
result = pipeline.ingest(source, context)
|
|
|
|
# --- Assertions ---
|
|
# The artifact was processed successfully
|
|
assert result.processed == 1, (
|
|
f"Expected 1 processed artifact, got {result.processed}. "
|
|
f"Errors: {result.errors}"
|
|
)
|
|
assert result.updated == 1, (
|
|
f"Expected 1 updated artifact, got {result.updated}."
|
|
)
|
|
|
|
# The artifact appears in the YAML index
|
|
expected_id = f"{context}/{artifact_type}/{name}"
|
|
entry = index.get_entry(expected_id)
|
|
assert entry is not None, (
|
|
f"Artifact '{expected_id}' not found in index. "
|
|
f"Available entries: {list(index.entries.keys())}"
|
|
)
|
|
|
|
# id is set
|
|
assert entry.id == expected_id
|
|
|
|
# title matches
|
|
assert entry.title == title
|
|
|
|
# type matches
|
|
assert entry.type == artifact_type
|
|
|
|
# tags match (order-independent)
|
|
assert set(entry.tags) == set(tags), (
|
|
f"Tags mismatch: expected {tags}, got {entry.tags}"
|
|
)
|
|
|
|
# scope == context
|
|
assert entry.scope == context, (
|
|
f"Scope mismatch: expected '{context}', got '{entry.scope}'"
|
|
)
|
|
|
|
# content_hash starts with "sha256:"
|
|
assert entry.content_hash.startswith("sha256:"), (
|
|
f"Content hash should start with 'sha256:', got '{entry.content_hash}'"
|
|
)
|
|
# Content hash has actual hex digest after prefix
|
|
hex_part = entry.content_hash[len("sha256:"):]
|
|
assert len(hex_part) == 64, (
|
|
f"SHA-256 hex digest should be 64 chars, got {len(hex_part)}"
|
|
)
|
|
|
|
# path follows scope-based structure: {context}/{type}/{name}.md
|
|
# Normalize path separators (Windows uses backslashes in Path objects)
|
|
normalized_path = entry.path.replace("\\", "/")
|
|
expected_path = f"{context}/{artifact_type}/{name}.md"
|
|
assert normalized_path == expected_path, (
|
|
f"Path mismatch: expected '{expected_path}', got '{entry.path}'"
|
|
)
|