ankn
a5f8fb49ab
Bahn: aisupport, Analyse-O2C-C2S, awesome-bahn-mcp-servers, beam-mcp,
Confluence_Bot, db-planet-mcp-server, O2C-Harness, project-audit,
Projekt-KIQ-HP, teamlandkarte-mcp
Dhive: Jury-Voting
Privat: CV, NoteGraph (NOTE: NoteGraph needs complete redo after consolidation)
Shared: AI-Orchestrator, OrgMyLife, power_skills_and_more
Shared/references: symphony (read-only)
Bahn repos remain available as independent remotes - this monorepo
pulls them in via subtree, the originals are untouched.
807 lines
28 KiB
Python
807 lines
28 KiB
Python
"""Property-based tests for NoteGraph Ingestion using Hypothesis.
|
|
|
|
All 18 correctness properties from the design document are tested here.
|
|
Each test uses @settings(max_examples=100) as specified.
|
|
"""
|
|
|
|
import json
|
|
import re
|
|
import string
|
|
import tempfile
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
import yaml
|
|
from hypothesis import given, settings, assume, HealthCheck
|
|
from hypothesis import strategies as st
|
|
|
|
from ingestion.config import IngestionConfig
|
|
from ingestion.discovery import SUPPORTED_EXTENSIONS, discover_files
|
|
from ingestion.enrichment.models import ActionItem, EnrichmentResult, Entity
|
|
from ingestion.extraction.text import PlainTextExtractor
|
|
from ingestion.integrations.git import commit_imported_notes
|
|
from ingestion.integrations.orgmylife import OrgMyLifeClient
|
|
from ingestion.linking.linker import generate_wiki_links, _entity_to_wikilink
|
|
from ingestion.output.naming import generate_filename, resolve_collision
|
|
from ingestion.output.renderer import render_note
|
|
from ingestion.output.router import CATEGORY_FOLDERS, route_to_folder
|
|
from ingestion.watcher import should_ignore
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Common settings for all property tests
|
|
# ---------------------------------------------------------------------------
|
|
pbt_settings = settings(
|
|
max_examples=100,
|
|
suppress_health_check=[HealthCheck.function_scoped_fixture],
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Strategies
|
|
# ---------------------------------------------------------------------------
|
|
|
|
# Strategy for supported file extensions
|
|
supported_ext_st = st.sampled_from(sorted(SUPPORTED_EXTENSIONS))
|
|
|
|
# Strategy for unsupported file extensions
|
|
unsupported_ext_st = st.text(
|
|
alphabet=string.ascii_lowercase, min_size=2, max_size=5
|
|
).map(lambda s: f".{s}").filter(lambda ext: ext not in SUPPORTED_EXTENSIONS)
|
|
|
|
# Strategy for valid filenames (no path separators, no null bytes, no reserved chars)
|
|
filename_chars = st.characters(
|
|
whitelist_categories=("L", "N"),
|
|
)
|
|
filename_st = st.text(alphabet=filename_chars, min_size=1, max_size=20).map(
|
|
lambda s: s.strip() or "file"
|
|
)
|
|
|
|
# Strategy for valid UTF-8 text content (no null bytes, non-empty)
|
|
utf8_text_st = st.text(
|
|
alphabet=st.characters(blacklist_categories=("Cs",), blacklist_characters="\x00"),
|
|
min_size=1,
|
|
max_size=2000,
|
|
)
|
|
|
|
# Strategy for entity types
|
|
entity_type_st = st.sampled_from(["person", "project", "date", "action_item"])
|
|
|
|
# Strategy for confidence scores
|
|
confidence_st = st.floats(min_value=0.0, max_value=1.0, allow_nan=False)
|
|
|
|
# Strategy for person/project names (non-empty, printable, with at least one letter)
|
|
name_st = st.text(
|
|
alphabet=st.characters(whitelist_categories=("L", "N", "Zs")),
|
|
min_size=2,
|
|
max_size=40,
|
|
).filter(lambda s: s.strip() and any(c.isalpha() for c in s))
|
|
|
|
# Strategy for valid providers
|
|
provider_st = st.sampled_from(["openai", "anthropic", "google", "mistral", "ollama"])
|
|
|
|
# Strategy for providers requiring API keys
|
|
keyed_provider_st = st.sampled_from(["openai", "anthropic", "google", "mistral"])
|
|
|
|
# Strategy for categories
|
|
category_st = st.sampled_from(["meeting", "project", "decision", "inbox"])
|
|
|
|
# Strategy for dates in YYYY-MM-DD format
|
|
date_st = st.dates().map(lambda d: d.strftime("%Y-%m-%d"))
|
|
|
|
# Strategy for note titles (must produce a non-empty slug)
|
|
title_st = st.text(
|
|
alphabet=st.characters(whitelist_categories=("L", "N", "Zs")),
|
|
min_size=2,
|
|
max_size=60,
|
|
).filter(lambda s: s.strip() and any(c.isalnum() for c in s))
|
|
|
|
# Strategy for action item descriptions
|
|
action_desc_st = st.text(
|
|
alphabet=st.characters(whitelist_categories=("L", "N", "Zs", "P")),
|
|
min_size=3,
|
|
max_size=100,
|
|
).filter(lambda s: s.strip())
|
|
|
|
# Strategy for source types
|
|
source_type_st = st.text(
|
|
alphabet=string.ascii_lowercase + "-",
|
|
min_size=3,
|
|
max_size=20,
|
|
).filter(lambda s: s.strip("-") and not s.startswith("-") and not s.endswith("-"))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Property 1 (Task 3.2): File discovery filtering
|
|
# ---------------------------------------------------------------------------
|
|
# Feature: notegraph-ingestion, Property 1: File discovery returns only supported extensions
|
|
|
|
|
|
@pbt_settings
|
|
@given(
|
|
supported_files=st.lists(
|
|
st.tuples(filename_st, supported_ext_st), min_size=0, max_size=10
|
|
),
|
|
unsupported_files=st.lists(
|
|
st.tuples(filename_st, unsupported_ext_st), min_size=0, max_size=10
|
|
),
|
|
)
|
|
def test_discovery_returns_only_supported_extensions(
|
|
tmp_path: Path, supported_files, unsupported_files
|
|
):
|
|
"""For any directory tree with arbitrary extensions, discovery returns exactly
|
|
those with supported extensions (.pdf, .jpg, .jpeg, .png, .md, .txt, .docx)."""
|
|
with tempfile.TemporaryDirectory() as td:
|
|
td_path = Path(td)
|
|
|
|
# Create files with supported extensions
|
|
expected = set()
|
|
for i, (name, ext) in enumerate(supported_files):
|
|
fp = td_path / f"s{i}_{name}{ext}"
|
|
fp.write_text("content", encoding="utf-8")
|
|
expected.add(fp)
|
|
|
|
# Create files with unsupported extensions
|
|
for i, (name, ext) in enumerate(unsupported_files):
|
|
fp = td_path / f"u{i}_{name}{ext}"
|
|
fp.write_text("content", encoding="utf-8")
|
|
|
|
result = discover_files(td_path)
|
|
result_set = set(result)
|
|
|
|
# All returned files must have supported extensions
|
|
for f in result:
|
|
assert f.suffix.lower() in SUPPORTED_EXTENSIONS
|
|
|
|
# All supported files we created must be in the result
|
|
for f in expected:
|
|
assert f in result_set
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Property 2 (Task 9.2): Bulk import resilience
|
|
# ---------------------------------------------------------------------------
|
|
# Feature: notegraph-ingestion, Property 2: Failed files do not block remaining processing
|
|
|
|
|
|
@pbt_settings
|
|
@given(
|
|
fail_indices=st.lists(st.integers(min_value=0, max_value=4), unique=True, max_size=3),
|
|
)
|
|
def test_bulk_import_resilience(tmp_path: Path, fail_indices, mock_config):
|
|
"""For any batch where a subset fails, remaining files are processed successfully."""
|
|
from ingestion.pipeline import process_batch
|
|
|
|
with tempfile.TemporaryDirectory() as td:
|
|
td_path = Path(td)
|
|
|
|
# Create 5 text files
|
|
files = []
|
|
for i in range(5):
|
|
f = td_path / f"note_{i}.txt"
|
|
f.write_text(f"Content of note {i}", encoding="utf-8")
|
|
files.append(f)
|
|
|
|
# Normalize fail indices to valid range
|
|
fail_set = {i for i in fail_indices if 0 <= i < 5}
|
|
|
|
# Mock enrichment agent that fails for specific indices
|
|
call_count = {"n": 0}
|
|
|
|
def mock_enrich(text):
|
|
idx = call_count["n"]
|
|
call_count["n"] += 1
|
|
if idx in fail_set:
|
|
raise RuntimeError(f"Simulated failure for file {idx}")
|
|
return EnrichmentResult(
|
|
title=f"Note {idx}",
|
|
category="inbox",
|
|
tags=[],
|
|
entities=[],
|
|
action_items=[],
|
|
summary=None,
|
|
)
|
|
|
|
mock_agent = MagicMock()
|
|
mock_agent.enrich.side_effect = mock_enrich
|
|
|
|
# Set up output directory
|
|
notes_dir = td_path / "notes"
|
|
notes_dir.mkdir()
|
|
(notes_dir / "inbox").mkdir()
|
|
mock_config.notes_dir = str(notes_dir)
|
|
|
|
with patch("ingestion.pipeline.EnrichmentAgent", return_value=mock_agent):
|
|
result = process_batch(
|
|
paths=files,
|
|
config=mock_config,
|
|
dry_run=False,
|
|
no_commit=True,
|
|
)
|
|
|
|
expected_success = 5 - len(fail_set)
|
|
assert result.success == expected_success
|
|
assert result.failed == len(fail_set)
|
|
assert result.total == 5
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Property 3 (Task 2.6): Text extraction identity
|
|
# ---------------------------------------------------------------------------
|
|
# Feature: notegraph-ingestion, Property 3: Text extraction preserves content for text-based formats
|
|
|
|
|
|
@pbt_settings
|
|
@given(content=utf8_text_st)
|
|
def test_text_extraction_identity(tmp_path: Path, content: str):
|
|
"""For any valid UTF-8 string written to a .md or .txt file, extraction returns
|
|
identical content (modulo platform newline normalization)."""
|
|
with tempfile.TemporaryDirectory() as td:
|
|
td_path = Path(td)
|
|
extractor = PlainTextExtractor()
|
|
|
|
for ext in (".md", ".txt"):
|
|
fp = td_path / f"test{ext}"
|
|
fp.write_text(content, encoding="utf-8")
|
|
result = extractor.extract(fp)
|
|
# Python's read_text uses universal newlines (translates \r\n and \r to \n)
|
|
# This is expected platform behavior, so we compare after normalization
|
|
expected = content.replace("\r\n", "\n").replace("\r", "\n")
|
|
assert result.text == expected
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Property 4 (Task 5.4): Entity parsing valid structure
|
|
# ---------------------------------------------------------------------------
|
|
# Feature: notegraph-ingestion, Property 4: Entity parsing extracts all entities with valid structure
|
|
|
|
|
|
@pbt_settings
|
|
@given(
|
|
entities=st.lists(
|
|
st.fixed_dictionaries({
|
|
"type": entity_type_st,
|
|
"value": name_st,
|
|
"confidence": confidence_st,
|
|
}),
|
|
min_size=1,
|
|
max_size=10,
|
|
),
|
|
)
|
|
def test_entity_parsing_valid_structure(entities):
|
|
"""For any valid enrichment response JSON, each parsed entity has valid type,
|
|
non-empty value, confidence in [0.0, 1.0]."""
|
|
data = {
|
|
"title": "Test",
|
|
"category": "inbox",
|
|
"tags": [],
|
|
"entities": entities,
|
|
"action_items": [],
|
|
}
|
|
|
|
result = EnrichmentResult.model_validate(data)
|
|
|
|
for entity in result.entities:
|
|
assert entity.type in ("person", "project", "date", "action_item")
|
|
assert len(entity.value) > 0
|
|
assert 0.0 <= entity.confidence <= 1.0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Property 5 (Task 5.4): Confidence threshold filtering
|
|
# ---------------------------------------------------------------------------
|
|
# Feature: notegraph-ingestion, Property 5: Confidence threshold filtering
|
|
|
|
|
|
@pbt_settings
|
|
@given(
|
|
scores=st.lists(confidence_st, min_size=1, max_size=20),
|
|
)
|
|
def test_confidence_threshold_filtering(scores):
|
|
"""For any list of entities with arbitrary confidence scores, filtered output
|
|
contains exactly those with confidence >= 0.7."""
|
|
entities = [
|
|
Entity(type="person", value=f"Person {i}", confidence=score)
|
|
for i, score in enumerate(scores)
|
|
]
|
|
|
|
threshold = 0.7
|
|
filtered = [e for e in entities if e.confidence >= threshold]
|
|
|
|
# Verify the filter matches what the enrichment agent does
|
|
from ingestion.enrichment.agent import EnrichmentAgent
|
|
|
|
agent = MagicMock(spec=EnrichmentAgent)
|
|
agent.confidence_threshold = threshold
|
|
agent._filter_by_confidence = EnrichmentAgent._filter_by_confidence.__get__(agent)
|
|
|
|
result = agent._filter_by_confidence(entities)
|
|
|
|
assert len(result) == len(filtered)
|
|
for entity in result:
|
|
assert entity.confidence >= threshold
|
|
|
|
# Verify no entity below threshold is included
|
|
for entity in entities:
|
|
if entity.confidence < threshold:
|
|
assert entity not in result
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Property 6 (Task 7.3): Wiki-link format per entity type
|
|
# ---------------------------------------------------------------------------
|
|
# Feature: notegraph-ingestion, Property 6: Wiki-link generation follows correct format per entity type
|
|
|
|
|
|
@pbt_settings
|
|
@given(name=name_st)
|
|
def test_wikilink_format_person(name: str):
|
|
"""For any person name → [[people/slugified-name]]."""
|
|
from slugify import slugify
|
|
|
|
entity = Entity(type="person", value=name, confidence=0.9)
|
|
link = _entity_to_wikilink(entity)
|
|
|
|
expected_slug = slugify(name)
|
|
assume(expected_slug) # skip if slugify produces empty string
|
|
|
|
assert link == f"[[people/{expected_slug}]]"
|
|
|
|
|
|
@pbt_settings
|
|
@given(name=name_st)
|
|
def test_wikilink_format_project(name: str):
|
|
"""For any project name → [[projects/slugified-name]]."""
|
|
from slugify import slugify
|
|
|
|
entity = Entity(type="project", value=name, confidence=0.9)
|
|
link = _entity_to_wikilink(entity)
|
|
|
|
expected_slug = slugify(name)
|
|
assume(expected_slug) # skip if slugify produces empty string
|
|
|
|
assert link == f"[[projects/{expected_slug}]]"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Property 7 (Task 7.3): Only first occurrence linked
|
|
# ---------------------------------------------------------------------------
|
|
# Feature: notegraph-ingestion, Property 7: Only first entity occurrence is linked
|
|
|
|
|
|
@pbt_settings
|
|
@given(
|
|
name=name_st,
|
|
n_occurrences=st.integers(min_value=2, max_value=5),
|
|
)
|
|
def test_only_first_occurrence_linked(name: str, n_occurrences: int):
|
|
"""For any text with N >= 2 occurrences of the same entity, only the first
|
|
is converted to a wiki-link."""
|
|
from slugify import slugify
|
|
|
|
slug = slugify(name)
|
|
assume(slug) # skip if slugify produces empty string
|
|
|
|
# Build text with multiple occurrences separated by unique text
|
|
separator = ". Then we discussed "
|
|
text = separator.join([name] * n_occurrences)
|
|
|
|
entity = Entity(type="person", value=name, confidence=0.9)
|
|
result = generate_wiki_links(text, [entity])
|
|
|
|
expected_link = f"[[people/{slug}]]"
|
|
|
|
# First occurrence should be a wiki-link
|
|
assert expected_link in result
|
|
|
|
# Count wiki-links — should be exactly 1
|
|
link_count = result.count(expected_link)
|
|
assert link_count == 1, f"Expected 1 wiki-link, found {link_count}"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Property 8 (Task 8.4): Frontmatter round-trip validity
|
|
# ---------------------------------------------------------------------------
|
|
# Feature: notegraph-ingestion, Property 8: Frontmatter round-trip validity
|
|
|
|
|
|
@pbt_settings
|
|
@given(
|
|
title=title_st,
|
|
category=category_st,
|
|
tags=st.lists(st.text(alphabet=string.ascii_lowercase, min_size=2, max_size=10), max_size=5),
|
|
people=st.lists(name_st, min_size=0, max_size=3),
|
|
projects=st.lists(name_st, min_size=0, max_size=3),
|
|
)
|
|
def test_frontmatter_roundtrip_validity(
|
|
tmp_path: Path, title, category, tags, people, projects
|
|
):
|
|
"""For any valid EnrichmentResult, render to markdown then parse frontmatter →
|
|
valid YAML with required fields."""
|
|
with tempfile.TemporaryDirectory() as td:
|
|
td_path = Path(td)
|
|
|
|
entities = []
|
|
for p in people:
|
|
entities.append(Entity(type="person", value=p, confidence=0.9))
|
|
for proj in projects:
|
|
entities.append(Entity(type="project", value=proj, confidence=0.9))
|
|
entities.append(Entity(type="date", value="2024-06-15", confidence=0.99))
|
|
|
|
enrichment = EnrichmentResult(
|
|
title=title,
|
|
category=category,
|
|
tags=tags,
|
|
entities=entities,
|
|
action_items=[],
|
|
summary=None,
|
|
)
|
|
|
|
source_file = td_path / "source.pdf"
|
|
source_file.write_text("", encoding="utf-8")
|
|
|
|
rendered = render_note(enrichment, "Body text here.", source_file)
|
|
|
|
# Parse frontmatter
|
|
assert rendered.startswith("---\n")
|
|
parts = rendered.split("---\n", 2)
|
|
assert len(parts) >= 3, "Expected YAML frontmatter delimiters"
|
|
|
|
frontmatter_yaml = parts[1]
|
|
parsed = yaml.safe_load(frontmatter_yaml)
|
|
|
|
# Required fields
|
|
assert "title" in parsed
|
|
assert "date" in parsed
|
|
assert "source" in parsed
|
|
assert "file" in parsed["source"]
|
|
assert "imported" in parsed["source"]
|
|
|
|
# Title matches
|
|
assert parsed["title"] == title
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Property 9 (Task 4.2): Provider routing from configuration
|
|
# ---------------------------------------------------------------------------
|
|
# Feature: notegraph-ingestion, Property 9: Provider routing from configuration
|
|
|
|
|
|
@pbt_settings
|
|
@given(provider=provider_st)
|
|
def test_provider_routing_from_configuration(provider: str):
|
|
"""For any valid provider name (openai, anthropic, google, mistral, ollama),
|
|
configuring LLM_PROVIDER produces the correct model string."""
|
|
config = IngestionConfig(
|
|
llm_provider=provider,
|
|
llm_model="test-model",
|
|
openai_api_key="fake",
|
|
anthropic_api_key="fake",
|
|
google_api_key="fake",
|
|
mistral_api_key="fake",
|
|
)
|
|
|
|
model_string = config.llm_model_string
|
|
|
|
if provider == "ollama":
|
|
assert model_string == "ollama/test-model"
|
|
else:
|
|
assert model_string == f"{provider}/test-model"
|
|
|
|
# Verify the provider prefix is correct
|
|
assert model_string.startswith(f"{provider}/")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Property 10 (Task 4.3): Missing API key error
|
|
# ---------------------------------------------------------------------------
|
|
# Feature: notegraph-ingestion, Property 10: Missing API key produces descriptive error
|
|
|
|
|
|
@pbt_settings
|
|
@given(provider=keyed_provider_st)
|
|
def test_missing_api_key_error(provider: str):
|
|
"""For any provider requiring an API key, when the key is empty,
|
|
validate_api_key() raises SystemExit naming the missing variable."""
|
|
config = IngestionConfig(
|
|
llm_provider=provider,
|
|
llm_model="test-model",
|
|
openai_api_key="",
|
|
anthropic_api_key="",
|
|
google_api_key="",
|
|
mistral_api_key="",
|
|
)
|
|
|
|
env_var_map = {
|
|
"openai": "OPENAI_API_KEY",
|
|
"anthropic": "ANTHROPIC_API_KEY",
|
|
"google": "GOOGLE_API_KEY",
|
|
"mistral": "MISTRAL_API_KEY",
|
|
}
|
|
|
|
with pytest.raises(SystemExit) as exc_info:
|
|
config.validate_api_key()
|
|
|
|
error_msg = str(exc_info.value)
|
|
assert env_var_map[provider] in error_msg
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Property 11 (Task 8.4): Action items in output
|
|
# ---------------------------------------------------------------------------
|
|
# Feature: notegraph-ingestion, Property 11: Action items appear in output markdown section
|
|
|
|
|
|
@pbt_settings
|
|
@given(
|
|
descriptions=st.lists(action_desc_st, min_size=1, max_size=5),
|
|
)
|
|
def test_action_items_in_output(tmp_path: Path, descriptions):
|
|
"""For any list of action items, rendered markdown contains `## Action Items`
|
|
section listing every item."""
|
|
with tempfile.TemporaryDirectory() as td:
|
|
td_path = Path(td)
|
|
|
|
action_items = [ActionItem(description=desc) for desc in descriptions]
|
|
|
|
enrichment = EnrichmentResult(
|
|
title="Test Note",
|
|
category="inbox",
|
|
tags=[],
|
|
entities=[Entity(type="date", value="2024-01-01", confidence=0.99)],
|
|
action_items=action_items,
|
|
summary=None,
|
|
)
|
|
|
|
source_file = td_path / "source.txt"
|
|
source_file.write_text("", encoding="utf-8")
|
|
|
|
rendered = render_note(enrichment, "Body text.", source_file)
|
|
|
|
assert "## Action Items" in rendered
|
|
|
|
for desc in descriptions:
|
|
assert desc in rendered
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Property 12 (Task 15.2): OrgMyLife task payload
|
|
# ---------------------------------------------------------------------------
|
|
# Feature: notegraph-ingestion, Property 12: OrgMyLife task payload correctness
|
|
|
|
|
|
@pbt_settings
|
|
@given(
|
|
description=action_desc_st,
|
|
note_path=st.text(
|
|
alphabet=string.ascii_lowercase + "/.-_", min_size=5, max_size=50
|
|
).filter(lambda s: s.strip() and not s.startswith("/")),
|
|
)
|
|
def test_orgmylife_task_payload(description: str, note_path: str):
|
|
"""For any action item description and note path, the task creation payload
|
|
has correct title and source_url."""
|
|
import httpx
|
|
|
|
client = OrgMyLifeClient(base_url="https://api.example.com", api_key="test-key")
|
|
|
|
# Capture the payload sent to httpx.post
|
|
captured_payload = {}
|
|
|
|
def mock_post(url, json=None, headers=None, timeout=None):
|
|
captured_payload.update(json)
|
|
response = MagicMock()
|
|
response.status_code = 201
|
|
response.raise_for_status = MagicMock()
|
|
response.json.return_value = {"id": 1, "title": json["title"]}
|
|
return response
|
|
|
|
with patch.object(httpx, "post", side_effect=mock_post):
|
|
client.create_task(title=description, source_url=note_path)
|
|
|
|
assert captured_payload["title"] == description
|
|
assert captured_payload["source_url"] == note_path
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Property 13 (Task 8.4): Category-based folder routing
|
|
# ---------------------------------------------------------------------------
|
|
# Feature: notegraph-ingestion, Property 13: Category-based folder routing
|
|
|
|
|
|
@pbt_settings
|
|
@given(category=category_st)
|
|
def test_category_based_folder_routing(tmp_path: Path, category: str):
|
|
"""For any category (meeting/project/decision/inbox), output path is within
|
|
the correct folder."""
|
|
with tempfile.TemporaryDirectory() as td:
|
|
notes_dir = Path(td) / "notes"
|
|
notes_dir.mkdir()
|
|
|
|
result = route_to_folder(category, notes_dir)
|
|
|
|
expected_folder = CATEGORY_FOLDERS.get(category, "inbox")
|
|
assert result == notes_dir / expected_folder
|
|
assert result.exists()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Property 14 (Task 8.4): Filename date-slug pattern
|
|
# ---------------------------------------------------------------------------
|
|
# Feature: notegraph-ingestion, Property 14: Output filename follows date-slug pattern
|
|
|
|
|
|
@pbt_settings
|
|
@given(date=date_st, title=title_st)
|
|
def test_filename_date_slug_pattern(date: str, title: str):
|
|
"""For any date and title, filename matches YYYY-MM-DD-slugified-title.md."""
|
|
filename = generate_filename(title, date)
|
|
|
|
# Must end with .md
|
|
assert filename.endswith(".md")
|
|
|
|
# Must start with the date
|
|
assert filename.startswith(date)
|
|
|
|
# Pattern: YYYY-MM-DD-slug.md (slug is lowercase alphanumeric + hyphens)
|
|
pattern = r"^\d{4}-\d{2}-\d{2}-[a-z0-9]([a-z0-9-]*[a-z0-9])?\.md$"
|
|
assert re.match(pattern, filename), f"Filename '{filename}' doesn't match pattern"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Property 15 (Task 8.4): Filename collision avoidance
|
|
# ---------------------------------------------------------------------------
|
|
# Feature: notegraph-ingestion, Property 15: Filename collision avoidance
|
|
|
|
|
|
@pbt_settings
|
|
@given(
|
|
title=title_st,
|
|
n_existing=st.integers(min_value=1, max_value=5),
|
|
)
|
|
def test_filename_collision_avoidance(tmp_path: Path, title: str, n_existing: int):
|
|
"""When a file with the same name exists, a numeric suffix is appended."""
|
|
with tempfile.TemporaryDirectory() as td:
|
|
td_path = Path(td)
|
|
|
|
filename = generate_filename(title, "2024-01-15")
|
|
|
|
# Create the original file
|
|
(td_path / filename).write_text("existing", encoding="utf-8")
|
|
|
|
# Create collision files with suffixes
|
|
stem = filename.rsplit(".md", 1)[0]
|
|
for i in range(2, n_existing + 1):
|
|
(td_path / f"{stem}-{i}.md").write_text("existing", encoding="utf-8")
|
|
|
|
# Resolve collision
|
|
resolved = resolve_collision(td_path, filename)
|
|
|
|
# The resolved path must not exist yet
|
|
assert not resolved.exists()
|
|
|
|
# The resolved path must have a numeric suffix
|
|
expected_suffix = f"-{n_existing + 1}.md"
|
|
assert str(resolved).endswith(expected_suffix)
|
|
|
|
# The resolved path must be in the same directory
|
|
assert resolved.parent == td_path
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Property 16 (Task 14.2): Temporary file filtering
|
|
# ---------------------------------------------------------------------------
|
|
# Feature: notegraph-ingestion, Property 16: Temporary file filtering
|
|
|
|
|
|
@pbt_settings
|
|
@given(
|
|
filename=st.text(
|
|
alphabet=st.characters(
|
|
whitelist_categories=("L", "N", "P"),
|
|
blacklist_characters="/\\:\x00",
|
|
),
|
|
min_size=1,
|
|
max_size=50,
|
|
),
|
|
)
|
|
def test_temporary_file_filtering(filename: str):
|
|
"""For any filename, should_ignore returns true iff name starts with '.'
|
|
or ends with '.tmp'."""
|
|
result = should_ignore(filename)
|
|
expected = filename.startswith(".") or filename.endswith(".tmp")
|
|
assert result == expected, f"should_ignore('{filename}') = {result}, expected {expected}"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Property 17 (Task 9.2): Dry-run no file writes
|
|
# ---------------------------------------------------------------------------
|
|
# Feature: notegraph-ingestion, Property 17: Dry-run produces no file writes
|
|
|
|
|
|
@pbt_settings
|
|
@given(
|
|
n_files=st.integers(min_value=1, max_value=5),
|
|
)
|
|
def test_dry_run_no_file_writes(tmp_path: Path, n_files: int, mock_config):
|
|
"""For any input with dry_run=True, zero files are created."""
|
|
from ingestion.pipeline import process_batch
|
|
|
|
with tempfile.TemporaryDirectory() as td:
|
|
td_path = Path(td)
|
|
|
|
# Create input files
|
|
files = []
|
|
for i in range(n_files):
|
|
f = td_path / f"input_{i}.txt"
|
|
f.write_text(f"Content {i}", encoding="utf-8")
|
|
files.append(f)
|
|
|
|
# Set up output directory
|
|
notes_dir = td_path / "output_notes"
|
|
notes_dir.mkdir()
|
|
(notes_dir / "inbox").mkdir()
|
|
mock_config.notes_dir = str(notes_dir)
|
|
|
|
# Mock the enrichment agent
|
|
def mock_enrich(text):
|
|
return EnrichmentResult(
|
|
title="Test",
|
|
category="inbox",
|
|
tags=[],
|
|
entities=[],
|
|
action_items=[],
|
|
summary=None,
|
|
)
|
|
|
|
mock_agent = MagicMock()
|
|
mock_agent.enrich.side_effect = mock_enrich
|
|
|
|
with patch("ingestion.pipeline.EnrichmentAgent", return_value=mock_agent):
|
|
result = process_batch(
|
|
paths=files,
|
|
config=mock_config,
|
|
dry_run=True,
|
|
no_commit=True,
|
|
)
|
|
|
|
# Count files in the output notes directory (should be 0 new markdown files)
|
|
output_files = list(notes_dir.rglob("*.md"))
|
|
assert len(output_files) == 0, f"Expected 0 files written, found {len(output_files)}"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Property 18 (Task 16.2): Git commit message format
|
|
# ---------------------------------------------------------------------------
|
|
# Feature: notegraph-ingestion, Property 18: Git commit message format
|
|
|
|
|
|
@pbt_settings
|
|
@given(
|
|
count=st.integers(min_value=1, max_value=1000),
|
|
source_type=source_type_st,
|
|
)
|
|
def test_git_commit_message_format(count: int, source_type: str):
|
|
"""For any count N and source type, message matches
|
|
'ingestion: import N notes from [source-type]'."""
|
|
files = [Path(f"/fake/note_{i}.md") for i in range(count)]
|
|
captured_messages = []
|
|
|
|
def mock_run(cmd, **kwargs):
|
|
if cmd[0] == "git" and cmd[1] == "commit":
|
|
# Extract -m argument
|
|
msg_idx = cmd.index("-m") + 1
|
|
captured_messages.append(cmd[msg_idx])
|
|
result = MagicMock()
|
|
result.returncode = 0
|
|
result.stdout = ""
|
|
result.stderr = ""
|
|
return result
|
|
|
|
with patch("subprocess.run", side_effect=mock_run):
|
|
commit_imported_notes(files, source_type=source_type)
|
|
|
|
assert len(captured_messages) == 1
|
|
expected = f"ingestion: import {count} notes from {source_type}"
|
|
assert captured_messages[0] == expected
|