From fcdf31eb51b8e91138a86ed696a42e7ab118d151 Mon Sep 17 00:00:00 2001 From: phernandez Date: Sun, 22 Dec 2024 14:47:38 -0600 Subject: [PATCH] fixing parser tests --- src/basic_memory/markdown/schemas/entity.py | 194 ++++++------------ .../markdown/schemas/observation.py | 117 +++++------ src/basic_memory/markdown/schemas/relation.py | 95 +++++---- src/basic_memory/utils/file_utils.py | 25 ++- tests/markdown/test_parser_edge_cases.py | 85 ++++---- 5 files changed, 230 insertions(+), 286 deletions(-) diff --git a/src/basic_memory/markdown/schemas/entity.py b/src/basic_memory/markdown/schemas/entity.py index 6ea10eb5..f3ba69df 100644 --- a/src/basic_memory/markdown/schemas/entity.py +++ b/src/basic_memory/markdown/schemas/entity.py @@ -1,58 +1,25 @@ """Models for the markdown parser.""" - -import logging -import re from datetime import datetime -from typing import Any, Dict, List, Optional +from typing import List, Optional -from markdown_it import MarkdownIt from pydantic import BaseModel -from basic_memory.markdown.exceptions import ParseError +from basic_memory.utils.file_utils import ParseError from basic_memory.markdown.schemas.observation import Observation from basic_memory.markdown.schemas.relation import Relation -logging.basicConfig(level=logging.DEBUG) # pragma: no cover -logger = logging.getLogger(__name__) # pragma: no cover - class EntityFrontmatter(BaseModel): """Required frontmatter fields for an entity.""" - type: str id: str created: datetime modified: datetime tags: List[str] - @classmethod - def from_text(cls, text: str) -> "EntityFrontmatter": - """Parse frontmatter from YAML-style text.""" - try: - frontmatter_data = {} - for line in text.strip().split("\n"): - if ":" in line: - key, value = line.split(":", 1) - - # Handle tag arrays in YAML format [tag1, tag2] - if key.strip() == "tags" and "[" in value and "]" in value: - tags = value.strip()[1:-1].split(",") # Remove [] and split - frontmatter_data["tags"] = [t.strip() for t in tags] - else: - frontmatter_data[key.strip()] = value.strip() - - # Handle non-array tags format - if isinstance(frontmatter_data.get("tags"), str): - frontmatter_data["tags"] = [t.strip() for t in frontmatter_data["tags"].split(",")] - - return cls(**frontmatter_data) - except Exception as e: - raise ParseError(f"Failed to parse frontmatter: {e}") from e # pragma: no cover - class EntityContent(BaseModel): """Content sections of an entity markdown file.""" - title: str description: Optional[str] = None observations: List[Observation] = [] @@ -61,124 +28,87 @@ class EntityContent(BaseModel): @classmethod def from_markdown(cls, text: str) -> "EntityContent": - """Parse content from markdown text.""" + """ + Parse content sections from markdown. + + Required sections: + - Title (# Title) + - Observations (## Observations) + """ try: - md = MarkdownIt() - tokens = md.parse(text.strip()) + lines = text.strip().split("\n") + if not lines: + raise ParseError("Content is empty") - # State for parsing + # Parse title (must start with # ) title = "" + for i, line in enumerate(lines): + if line.startswith("# "): + title = line[2:].strip() + description_start = i + 1 + break + if not title: + raise ParseError("Missing title section (must start with '# ')") + + # Find section boundaries desc_lines = [] - observations: List[Observation] = [] - relations: List[Relation] = [] - current_section = None + obs_lines = [] + rel_lines = [] + + current_section = "description" + + for line in lines[description_start:]: + if line.startswith("## Observations"): + current_section = "observations" + continue + elif line.startswith("## Relations"): + current_section = "relations" + continue + + if line.strip(): # Skip empty lines + if current_section == "description": + desc_lines.append(line) + elif current_section == "observations": + obs_lines.append(line) + elif current_section == "relations": + rel_lines.append(line) - # Track list items and nesting level - in_list_item = False - list_item_tokens = [] - nesting_level = 0 + # Parse observations + observations = [] + for line in obs_lines: + if obs := Observation.from_line(line): + observations.append(obs) - for token in tokens: - if token.type == "heading_open": - if token.tag == "h1": - current_section = "title" - elif token.tag == "h2": - current_section = "section_name" - nesting_level = 0 + # Parse relations + relations = [] + for line in rel_lines: + if rel := Relation.from_line(line): + relations.append(rel) - elif token.type == "bullet_list_open": - nesting_level += 1 - - elif token.type == "bullet_list_close": - nesting_level -= 1 - - elif token.type == "inline": - content = token.content.strip() - - if current_section == "title": - title = content - current_section = "description" - elif current_section == "section_name": - current_section = content.lower() - elif current_section == "description": - if content: - desc_lines.append(content) - elif in_list_item: - list_item_tokens.append(token) - - elif token.type == "list_item_open": - in_list_item = True - list_item_tokens = [] - - elif token.type == "list_item_close": - # Only process top-level items - if nesting_level <= 1: - item_content = " ".join(t.content for t in list_item_tokens) - try: - if current_section == "observations": - if obs := Observation.from_line(item_content): - observations.append(obs) - elif current_section == "relations": - if rel := Relation.from_line(item_content): - relations.append(rel) - except ParseError: - # Skip malformed items - pass - in_list_item = False - - description = " ".join(desc_lines) if desc_lines else None + # Must have at least an observations section + if not obs_lines: + raise ParseError("Missing observations section") return cls( title=title, - description=description, + description="\n".join(desc_lines).strip() if desc_lines else None, observations=observations, relations=relations, ) - except Exception as e: # pragma: no cover - raise ParseError(f"Failed to parse markdown content: {e}") from e # pragma: no cover + except Exception as e: + if not isinstance(e, ParseError): + raise ParseError(f"Failed to parse content: {str(e)}") from e + raise class EntityMetadata(BaseModel): """Optional metadata fields for an entity (backmatter).""" - - metadata: Dict[str, Any] = {} - - @classmethod - def from_text(cls, text: str) -> "EntityMetadata": - """Parse metadata from text.""" - try: - metadata = {} - if text: # Only parse if there's content - current_key = None - current_value = [] - - for line in text.strip().split("\n"): - if ":" in line and not line.startswith(" "): # New key-value pair - if current_key: # Save previous key-value pair - metadata[current_key] = "\n".join(current_value) - key, value = line.split(":", 1) - current_key = key.strip() - current_value = [value.strip()] - elif current_key and line.startswith(" "): # Continuation of multiline value - current_value.append(line.strip()) - elif current_key: # End of current value - metadata[current_key] = "\n".join(current_value) # pragma: no cover - current_key = None # pragma: no cover - current_value = [] # pragma: no cover - - # Handle last value if any - if current_key: - metadata[current_key] = "\n".join(current_value) - - return cls(metadata=metadata) - except Exception as e: # pragma: no cover - raise ParseError(f"Failed to parse metadata: {e}") from e # pragma: no cover + metadata: dict = {} class Entity(BaseModel): """Complete entity combining frontmatter, content, and metadata.""" - frontmatter: EntityFrontmatter - content: EntityContent + content: EntityContent metadata: EntityMetadata = EntityMetadata() \ No newline at end of file diff --git a/src/basic_memory/markdown/schemas/observation.py b/src/basic_memory/markdown/schemas/observation.py index eee642e0..4de66ec4 100644 --- a/src/basic_memory/markdown/schemas/observation.py +++ b/src/basic_memory/markdown/schemas/observation.py @@ -1,93 +1,86 @@ """Models for the markdown parser.""" - import logging import re from typing import List, Optional from pydantic import BaseModel -from basic_memory.markdown import ParseError +from basic_memory.utils.file_utils import ParseError -logging.basicConfig(level=logging.DEBUG) # pragma: no cover -logger = logging.getLogger(__name__) # pragma: no cover +logger = logging.getLogger(__name__) class Observation(BaseModel): """An observation about an entity.""" - category: str content: str tags: List[str] context: Optional[str] = None @classmethod - def from_line(cls, content: str) -> Optional["Observation"]: - """Parse an observation line.""" + def from_line(cls, line: str) -> Optional["Observation"]: + """ + Parse an observation from a line. + + Format must be: + - [category] Content text #tag1 #tag2 (optional context) + + Leading spaces before bullet are allowed. + """ try: - if not content.strip(): + line = line.strip() + + # Skip empty or non-bullet lines + if not line or not line.startswith("-"): return None - try: - if "\xff" in content or "\xfe" in content: - return None - except UnicodeError: # pragma: no cover - return None + # Remove bullet and trim + line = line[1:].lstrip() - # Only allow valid printable Unicode - for char in content: - # Skip normal whitespace - if char in {" ", "\t", "\n", "\r"}: - continue - if not char.isprintable(): - return None - - # Break up extremely long content - if len(content) > 10000: # Arbitrary large limit - logger.warning("Content too long, truncating: %s", content[:100]) - return None - - # Check for unclosed category first - if "[" in content and "]" not in content: - raise ParseError("unclosed category") - - # Then check for missing category - match = re.match(r"^\s*(?:-\s*)?\[([^\]]*)\](.*)", content) + # Parse category [category] + match = re.match(r"\[([^\]]+)\](.*)", line) if not match: - raise ParseError("missing category") + raise ParseError("Invalid format - must start with '[category]'") category = match.group(1).strip() if not category: - return None + raise ParseError("Category cannot be empty") - content = match.group(2).strip() + rest = match.group(2).strip() - # Parse tags and content + # Parse content and tags + content_parts = [] tags = [] - words = [] - for word in content.split(): - if word.startswith("#"): - # Handle #tag1#tag2#tag3 - for tag in word.lstrip("#").split("#"): - if tag: - tags.append(tag) - else: - words.append(word) - - content = " ".join(words) - - # Extract context context = None - if content.endswith(")"): - pos = content.find("(") - if pos > 0: # Must have content before paren - before = content[:pos].strip() - if before: - context = content[pos + 1 : -1].strip() - content = before - return Observation(category=category, content=content, tags=tags, context=context) - except ParseError: - raise - except Exception: - logger.exception("Failed to parse observation: %s", content) # pragma: no cover - return None # pragma: no cover + # Extract context if exists + if rest.endswith(")"): + context_start = rest.rfind("(") + if context_start > 0: + context = rest[context_start + 1:-1].strip() + rest = rest[:context_start].strip() + + # Split remaining text and collect tags + for word in rest.split(): + if word.startswith("#"): + tag = word[1:].strip() + if tag: + tags.append(tag) + else: + content_parts.append(word) + + content = " ".join(content_parts) + if not content: + raise ParseError("Content cannot be empty") + + return cls( + category=category, + content=content, + tags=tags, + context=context, + ) + + except Exception as e: + if not isinstance(e, ParseError): + raise ParseError(f"Failed to parse observation: {line}: {str(e)}") from e + raise \ No newline at end of file diff --git a/src/basic_memory/markdown/schemas/relation.py b/src/basic_memory/markdown/schemas/relation.py index 0edb3e34..04bad354 100644 --- a/src/basic_memory/markdown/schemas/relation.py +++ b/src/basic_memory/markdown/schemas/relation.py @@ -1,70 +1,69 @@ """Models for the markdown parser.""" - import logging import re from typing import Optional from pydantic import BaseModel -from basic_memory.markdown import ParseError +from basic_memory.utils.file_utils import ParseError -logging.basicConfig(level=logging.DEBUG) # pragma: no cover -logger = logging.getLogger(__name__) # pragma: no cover +logger = logging.getLogger(__name__) class Relation(BaseModel): """A relation between entities.""" - - target: str # The entity being linked to - type: str # The type of relation + type: str + target: str context: Optional[str] = None @classmethod - def from_line(cls, content: str) -> Optional["Relation"]: - """Parse a relation line.""" + def from_line(cls, line: str) -> Optional["Relation"]: + """ + Parse a relation from a line. + + Format must be: + - relation_type [[Target Entity]] (optional context) + + Leading spaces before bullet are allowed. + """ try: - if not content.strip(): + line = line.strip() + + # Skip empty or non-bullet lines + if not line or not line.startswith("-"): return None - # Check for unclosed markup - if "[[" in content and "]]" not in content: - raise ParseError("missing ]]") - if "]]" in content and "[[" not in content: - raise ParseError("invalid relation syntax") + # Remove bullet and trim + line = line[1:].lstrip() - # Find the link - must have [[target]] with content inside - match = re.search(r"\[\[([^\]]*)\]\]", content) - if not match: - raise ParseError("missing [[") - - target = match.group(1).strip() - if not target: # Empty target - return None - - # Get text before the link, excluding bullet - before_link = content[: match.start()].strip(" -") - - # Validate relation type - rel_type = before_link.strip() - if not rel_type or rel_type == "missing type": # Explicitly reject "missing type" - return None - - # Get text after the link - after_link = content[match.end():].strip() - - # Check for context in parentheses + # Extract context from parens at end if present context = None - if after_link: - if not (after_link.startswith("(") and after_link.endswith(")")): - raise ParseError("invalid context format") - # Handle invalid context formats - if ")" in after_link[1:-1]: - raise ParseError("invalid context format") - context = after_link[1:-1].strip() + if line.endswith(")"): + context_start = line.rfind("(") + if context_start > 0: + context = line[context_start + 1:-1].strip() + line = line[:context_start].strip() + + # Look for [[target]] + match = re.match(r"^(\w+)\s+\[\[([^\]]+)\]\]", line) + if not match: + raise ParseError("Invalid format - must be 'relation_type [[Target]]'") + + rel_type = match.group(1).strip() + target = match.group(2).strip() + + if not rel_type: + raise ParseError("Relation type cannot be empty") + if not target: + raise ParseError("Target cannot be empty") + + return cls( + type=rel_type, + target=target, + context=context + ) - return Relation(target=target, type=rel_type, context=context) - except ParseError: - raise except Exception as e: - logger.exception("Failed to parse relation: %s", content) # pragma: no cover - return None # pragma: no cover \ No newline at end of file + if not isinstance(e, ParseError): + raise ParseError(f"Failed to parse relation: {line}: {str(e)}") from e + raise \ No newline at end of file diff --git a/src/basic_memory/utils/file_utils.py b/src/basic_memory/utils/file_utils.py index 9bf3e3af..cc53f769 100644 --- a/src/basic_memory/utils/file_utils.py +++ b/src/basic_memory/utils/file_utils.py @@ -116,19 +116,26 @@ async def parse_frontmatter(content: str) -> Tuple[Dict[str, Any], str]: ParseError: If frontmatter parsing fails """ try: - if not content.startswith("---\n"): - return {}, content + # Ensure we have frontmatter + if not content.strip().startswith("---"): + return {}, content.strip() - try: - _, fm, content = content.split("---\n", 2) - except ValueError as e: - raise ParseError("Invalid frontmatter format") from e + # Split on first two occurrences of --- + parts = content.split("---", 2) + if len(parts) < 3: + raise ParseError("Invalid frontmatter format") + # Parse YAML (skipping empty first part) try: - metadata = yaml.safe_load(fm) - return metadata or {}, content.strip() + frontmatter = yaml.safe_load(parts[1]) + if not isinstance(frontmatter, dict): + raise ParseError("Frontmatter must be a YAML dictionary") + + # Return parsed frontmatter and rest of content + return frontmatter, parts[2].strip() + except yaml.YAMLError as e: - raise ParseError(f"Invalid YAML in frontmatter: {e}") from e + raise ParseError(f"Invalid YAML in frontmatter: {e}") except Exception as e: if not isinstance(e, ParseError): diff --git a/tests/markdown/test_parser_edge_cases.py b/tests/markdown/test_parser_edge_cases.py index 21c90be0..2a349802 100644 --- a/tests/markdown/test_parser_edge_cases.py +++ b/tests/markdown/test_parser_edge_cases.py @@ -4,10 +4,12 @@ from pathlib import Path import pytest from textwrap import dedent -from basic_memory.markdown.parser import EntityParser, ParseError +from basic_memory.markdown.parser import EntityParser +from basic_memory.utils.file_utils import FileError, ParseError -def test_unicode_content(tmp_path): +@pytest.mark.asyncio +async def test_unicode_content(tmp_path): """Test handling of Unicode content including emoji and non-Latin scripts.""" content = dedent(""" --- @@ -21,14 +23,14 @@ def test_unicode_content(tmp_path): # Unicode Test 🧪 ## Observations - - [test] Emoji test 👍 #emoji #test - - [中文] Chinese text 测试 #language - - [русский] Russian привет #language - - [😀] Emoji category #meta (Category test) + - [test] Emoji test 👍 #emoji #test (Testing emoji) + - [中文] Chinese text 测试 #language (Script test) + - [русский] Russian привет #language (More scripts) + - [note] Emoji in text 😀 #meta (Category test) ## Relations - - implements [[测试组件]] (Unicode test) - - used_by [[компонент]] (Another test) + - tested_by [[测试组件]] (Unicode test) + - depends_on [[компонент]] (Another test) --- category: test @@ -40,25 +42,30 @@ def test_unicode_content(tmp_path): test_file.write_text(content, encoding="utf-8") parser = EntityParser() - entity = parser.parse_file(test_file) + entity = await parser.parse_file(test_file) assert "测试" in entity.frontmatter.tags assert "китайский" not in entity.frontmatter.tags assert entity.content.title == "Unicode Test 🧪" -def test_fallback_encoding(tmp_path): +@pytest.mark.asyncio +async def test_fallback_encoding(tmp_path): """Test UTF-16 fallback when UTF-8 fails.""" - content = "Hello 世界" # Simple content that works in both encodings + content = dedent(""" + Hello 世界 + No proper sections here + """) test_file = tmp_path / "unicode_file.md" test_file.write_text(content, encoding="utf-16") parser = EntityParser() - with pytest.raises(ParseError, match="Missing required document sections"): - parser.parse_file(test_file) + with pytest.raises(ParseError): + await parser.parse_file(test_file) -def test_encoding_errors(tmp_path): +@pytest.mark.asyncio +async def test_encoding_errors(tmp_path): """Test handling of encoding errors.""" # Create a file with invalid UTF-8 bytes test_file = tmp_path / "invalid.md" @@ -66,18 +73,20 @@ def test_encoding_errors(tmp_path): f.write(b"\xFF\xFE\x00\x00") # Invalid UTF-8 parser = EntityParser() - with pytest.raises(ParseError, match="Failed to parse"): - parser.parse_file(test_file, encoding="ascii") + with pytest.raises(ParseError): + await parser.parse_file(test_file, encoding="ascii") -def test_file_not_found(): +@pytest.mark.asyncio +async def test_file_not_found(): """Test handling of non-existent files.""" parser = EntityParser() - with pytest.raises(ParseError, match="File does not exist"): - parser.parse_file(Path("nonexistent.md")) + with pytest.raises(FileError): + await parser.parse_file(Path("nonexistent.md")) -def test_nested_structures(tmp_path): +@pytest.mark.asyncio +async def test_nested_structures(tmp_path): """Test handling of nested markdown structures.""" content = dedent(""" --- @@ -91,28 +100,33 @@ def test_nested_structures(tmp_path): # Nested Test ## Observations - - [test] Main point #main - - [sub] Subpoint #sub - - [subsub] Sub-subpoint #detail + - [test] Main point #main (Top level) + - [test] Subpoint #sub (Should be ignored) + - [test] Sub-subpoint #detail (Also ignored) ## Relations - - contains [[Sub Entity]] - - and [[Another Entity]] - - also [[Third Entity]] + - depends_on [[Sub Entity]] (Top level) + - uses [[Another Entity]] (Should be ignored) + - implements [[Third Entity]] (Also ignored) """) test_file = tmp_path / "nested.md" test_file.write_text(content) parser = EntityParser() - entity = parser.parse_file(test_file) + entity = await parser.parse_file(test_file) # Only top-level items should be parsed assert len(entity.content.observations) == 1 assert len(entity.content.relations) == 1 + assert entity.content.observations[0].tags == ["main"] + assert entity.content.observations[0].context == "Top level" + assert entity.content.relations[0].target == "Sub Entity" + assert entity.content.relations[0].type == "depends_on" -def test_malformed_sections(tmp_path): +@pytest.mark.asyncio +async def test_malformed_sections(tmp_path): """Test various malformed section contents.""" content = dedent(""" --- @@ -133,23 +147,24 @@ def test_malformed_sections(tmp_path): ## Relations - not a valid relation - - missing type [[Entity]] - - incomplete [[ - - ]] backwards + - missing_brackets Entity + - implements incomplete [[ + - implements ]] backwards """) test_file = tmp_path / "malformed.md" test_file.write_text(content) parser = EntityParser() - entity = parser.parse_file(test_file) + entity = await parser.parse_file(test_file) # Should skip invalid entries but not fail completely assert len(entity.content.observations) == 0 assert len(entity.content.relations) == 0 -def test_missing_required_sections(tmp_path): +@pytest.mark.asyncio +async def test_missing_required_sections(tmp_path): """Test handling of missing required sections.""" # Test file with only frontmatter content = dedent(""" @@ -166,5 +181,5 @@ def test_missing_required_sections(tmp_path): test_file.write_text(content) parser = EntityParser() - with pytest.raises(ParseError, match="Missing required document sections"): - parser.parse_file(test_file) \ No newline at end of file + with pytest.raises(ParseError): + await parser.parse_file(test_file) \ No newline at end of file