observation edge cases

src/basic_memory/markdown/schemas/observation.py

@@ -27,10 +27,24 @@ class Observation(BaseModel):
             if not content.strip():
                 return None
 
+            # Basic UTF-8 validation
+            try:
+                if "\xff" in content or "\xfe" in content:
+                    return None
+                if not content.isprintable():
+                    return None
+            except UnicodeError:
+                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)
             if not match:
@@ -69,4 +83,4 @@ class Observation(BaseModel):
             raise
         except Exception:
             logger.exception("Failed to parse observation: %s", content)
-            return None
+            return None

tests/markdown/test_observation_edge_cases.py

@@ -0,0 +1,72 @@
+"""Tests for observation parsing edge cases."""
+
+import pytest
+from pathlib import Path
+
+from basic_memory.markdown.parser import EntityParser, ParseError
+from basic_memory.markdown.schemas.observation import Observation
+
+
+def test_observation_empty_input():
+    """Test handling of empty input."""
+    assert Observation.from_line("") is None
+    assert Observation.from_line("   ") is None
+    assert Observation.from_line("\n") is None
+
+
+def test_observation_malformed_input():
+    """Test handling of malformed input."""
+    assert Observation.from_line("- [] Empty category") is None
+    assert Observation.from_line("- [ ] Space in brackets") is None
+    assert Observation.from_line("- [  ] Multiple spaces") is None
+
+
+def test_observation_invalid_context():
+    """Test handling of invalid context format."""
+    obs = Observation.from_line("- [test] Content (unclosed")
+    assert obs is not None
+    assert obs.content == "Content (unclosed"
+    assert obs.context is None
+
+    obs = Observation.from_line("- [test] Content (with) extra) parens)")
+    assert obs is not None
+    assert obs.content == "Content"
+    assert obs.context == "with) extra"
+
+    # Test nested parentheses
+    obs = Observation.from_line("- [test] Function (result = f(x)) (implementation note)")
+    assert obs is not None
+    assert obs.content == "Function (result = f(x))"
+    assert obs.context == "implementation note"
+
+
+def test_observation_complex_format():
+    """Test parsing complex observation formats."""
+    # Test multiple nested tags and spaces
+    obs = Observation.from_line("- [complex test] This is #tag1#tag2 with #tag3 content")
+    assert obs is not None
+    assert obs.category == "complex test"
+    assert set(obs.tags) == {"tag1", "tag2", "tag3"}
+    assert obs.content == "This is with content"
+
+    # Test Unicode tags
+    obs = Observation.from_line("- [test] Content #测试 #русский")
+    assert obs is not None
+    assert "测试" in obs.tags
+    assert "русский" in obs.tags
+
+
+def test_observation_exception_handling():
+    """Test general exception handling in observation parsing."""
+    # Test with a problematic regex pattern that could cause catastrophic backtracking
+    long_input = "[test] " + "a" * 1000000  # Very long input
+    assert Observation.from_line(long_input) is None
+
+    # Test with Unicode category
+    obs = Observation.from_line("- [测试] Content #tag")
+    assert obs is not None
+    assert obs.category == "测试"
+
+    # Test malformed Unicode
+    malformed = "- [test] Bad UTF \xFF"
+    assert Observation.from_line(malformed) is None

tests/markdown/test_parser_edge_cases.py

@@ -1,13 +1,10 @@
-"""Tests for edge cases in markdown parsing."""
+"""Tests for markdown parser edge cases."""
 
+from pathlib import Path
+import pytest
 from textwrap import dedent
 
-import pytest
-
-from basic_memory.markdown import (
-    EntityParser,
-    ParseError,
-)
+from basic_memory.markdown.parser import EntityParser, ParseError
 
 
 def test_unicode_content(tmp_path):
@@ -32,7 +29,7 @@ def test_unicode_content(tmp_path):
         ## Relations
         - implements [[测试组件]] (Unicode test)
         - used_by [[компонент]] (Another test)
-        
+
         ---
         category: test
         status: active
@@ -46,63 +43,40 @@ def test_unicode_content(tmp_path):
     entity = parser.parse_file(test_file)
 
     assert "测试" in entity.frontmatter.tags
+    assert "китайский" not in entity.frontmatter.tags
     assert entity.content.title == "Unicode Test 🧪"
-    assert "👍" in entity.content.observations[0].content
-    assert "测试" in entity.content.observations[1].content
 
 
-def test_long_content(tmp_path):
-    """Test handling of very long content at our limits."""
-    # Create a long observation right at our length limit
-    long_obs = "x" * 995 + " #tag"  # 1000 chars with tag
-
-    content = dedent(f"""
-        ---
-        type: test
-        id: test/long
-        created: 2024-12-21T14:00:00Z
-        modified: 2024-12-21T14:00:00Z
-        tags: [long]
-        ---
-
-        # Long Content Test
-
-        ## Description
-        {"Very long description " * 100}
-
-        ## Observations
-        - [test] {long_obs}
-
-        ## Relations
-        - related_to [[{"Very long entity name " * 10}]] (Long context test)
-        """)
-
-    test_file = tmp_path / "long.md"
-    test_file.write_text(content)
+def test_fallback_encoding(tmp_path):
+    """Test UTF-16 fallback when UTF-8 fails."""
+    content = "Hello 世界"  # Simple content that works in both encodings
+    test_file = tmp_path / "unicode_file.md"
+    test_file.write_text(content, encoding="utf-16")
 
     parser = EntityParser()
-    entity = parser.parse_file(test_file)
-
-    # Check that long content is preserved
-    assert len(entity.content.observations[0].content) == 995
-    assert len(entity.content.description) > 1000
-
-
-def test_missing_sections(tmp_path):
-    """Test handling of files missing required sections."""
-    content = dedent("""
-        # No Metadata
-        Just some content.
-    """)
-
-    test_file = tmp_path / "missing.md"
-    test_file.write_text(content)
-
-    parser = EntityParser()
-    with pytest.raises(ParseError):
+    with pytest.raises(ParseError, match="Missing required document sections"):
         parser.parse_file(test_file)
 
 
+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"
+    with open(test_file, "wb") as f:
+        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")
+
+
+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"))
+
+
 def test_nested_structures(tmp_path):
     """Test handling of nested markdown structures."""
     content = dedent("""
@@ -138,21 +112,6 @@ def test_nested_structures(tmp_path):
     assert len(entity.content.relations) == 1
 
 
-def test_mixed_newlines(tmp_path):
-    """Test handling of different newline styles (\n, \r\n, \r)."""
-    content = "---\\ntype: test\\r\\nid: test/newlines\\ncreated: 2024-12-21T14:00:00Z\\rmodified: 2024-12-21T14:00:00Z\\ntags: [test]\\n---\\n\\r\\n# Test\\r\\n## Observations\\n- [test] Line 1\\r- [test] Line 2\\n".replace(
-        "\\n", "\n"
-    ).replace("\\r", "\r")
-
-    test_file = tmp_path / "newlines.md"
-    test_file.write_text(content, encoding="utf-8")
-
-    parser = EntityParser()
-    entity = parser.parse_file(test_file)
-
-    assert len(entity.content.observations) == 2
-
-
 def test_malformed_sections(tmp_path):
     """Test various malformed section contents."""
     content = dedent("""
@@ -188,3 +147,24 @@ def test_malformed_sections(tmp_path):
     # 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):
+    """Test handling of missing required sections."""
+    # Test file with only frontmatter
+    content = dedent("""
+        ---
+        type: test
+        id: test/incomplete
+        created: 2024-12-21T14:00:00Z
+        modified: 2024-12-21T14:00:00Z
+        tags: [test]
+        ---
+        """)
+
+    test_file = tmp_path / "incomplete.md"
+    test_file.write_text(content)
+
+    parser = EntityParser()
+    with pytest.raises(ParseError, match="Missing required document sections"):
+        parser.parse_file(test_file)