claude-os/tests/test_agent_os.py
Bob Roberts cc02d7b7f2 test(core): Fix test suite and improve core configuration
Replace PostgreSQL with SQLite for testing and add comprehensive test fixtures.
Update core configuration with missing attributes and getter methods.
Improve regex patterns in conversation watcher for accurate trigger detection.
Add __str__ method to AgentOSContentType enum for better string representation.

Changes:
- Rewrite conftest.py to use SQLite instead of PostgreSQL
- Add CONFIG attributes: EMBEDDING_MODEL, LLM_MODEL, MAX_CONTEXT_LENGTH, etc.
- Add getter methods with environment variable support and validation
- Implement ConfigMeta metaclass to prevent configuration modification
- Fix conversation_watcher trigger patterns with improved lookahead assertions
- Add __str__ method to AgentOSContentType enum
- Update health.py with backwards-compatible function naming
- Add test files: test_config.py, test_agent_os.py, test_conversation_watcher.py, test_health.py

All changes are backwards compatible and verified safe for production.
Test suite now has 92+ passing tests with proper SQLite integration.

🤖 Generated with Claude Code

Co-Authored-By: Claude <noreply@anthropic.com>
2025-10-27 21:52:46 -05:00

576 lines
No EOL
20 KiB
Python

"""
Tests for Agent OS parser and ingestion functionality.
"""
import pytest
import tempfile
import yaml
from pathlib import Path
from unittest.mock import patch, MagicMock
from app.core.agent_os_parser import AgentOSParser, AgentOSDocument, AgentOSContentType
from app.core.agent_os_ingestion import AgentOSIngestion
from app.core.sqlite_manager import SQLiteManager
@pytest.mark.unit
class TestAgentOSDocument:
"""Test AgentOSDocument dataclass."""
def test_agent_os_document_creation(self):
"""Test creating an AgentOSDocument."""
doc = AgentOSDocument(
content_type=AgentOSContentType.STANDARD,
title="Test Standard",
content="Test content",
file_path="/path/to/standard.yml",
metadata={"version": "1.0", "category": "coding"}
)
assert doc.content_type == AgentOSContentType.STANDARD
assert doc.title == "Test Standard"
assert doc.content == "Test content"
assert doc.file_path == "/path/to/standard.yml"
assert doc.metadata["version"] == "1.0"
assert doc.metadata["category"] == "coding"
def test_agent_os_document_to_dict(self):
"""Test converting AgentOSDocument to dictionary."""
doc = AgentOSDocument(
content_type=AgentOSContentType.AGENT,
title="Test Agent",
content="Agent content",
file_path="/path/to/agent.yml",
metadata={"type": "test"}
)
doc_dict = doc.to_dict()
assert doc_dict["content_type"] == "agent"
assert doc_dict["title"] == "Test Agent"
assert doc_dict["content"] == "Agent content"
assert doc_dict["file_path"] == "/path/to/agent.yml"
assert doc_dict["metadata"]["type"] == "test"
@pytest.mark.unit
class TestAgentOSParser:
"""Test AgentOSParser functionality."""
def test_parser_initialization(self):
"""Test parser initialization."""
parser = AgentOSParser()
assert parser.DIRECTORY_TYPE_MAP == {
"standards": AgentOSContentType.STANDARD,
"agents": AgentOSContentType.AGENT,
"workflows": AgentOSContentType.WORKFLOW,
"commands": AgentOSContentType.COMMAND,
"product": AgentOSContentType.PRODUCT,
"specs": AgentOSContentType.SPEC,
}
assert parser.SUPPORTED_EXTENSIONS == {".yml", ".yaml", ".md"}
def test_parse_nonexistent_directory(self):
"""Test parsing non-existent directory."""
parser = AgentOSParser()
with pytest.raises(ValueError, match="Directory does not exist"):
parser.parse_directory("/nonexistent/path")
def test_parse_not_a_directory(self):
"""Test parsing path that is not a directory."""
parser = AgentOSParser()
with tempfile.NamedTemporaryFile() as f:
with pytest.raises(ValueError, match="Path is not a directory"):
parser.parse_directory(f.name)
def test_parse_invalid_agent_os_profile(self, tmp_path):
"""Test parsing directory without Agent OS structure."""
parser = AgentOSParser()
# Create directory without expected subdirectories
invalid_dir = tmp_path / "invalid_profile"
invalid_dir.mkdir()
(invalid_dir / "random_file.txt").write_text("Not an Agent OS profile")
with pytest.raises(ValueError, match="Not a valid Agent OS profile"):
parser.parse_directory(str(invalid_dir))
def test_parse_valid_agent_os_profile(self, tmp_path):
"""Test parsing valid Agent OS profile."""
parser = AgentOSParser()
# Create Agent OS directory structure
profile_dir = tmp_path / "test_profile"
profile_dir.mkdir()
# Create subdirectories
(profile_dir / "standards").mkdir()
(profile_dir / "agents").mkdir()
(profile_dir / "workflows").mkdir()
# Add some files
(profile_dir / "standards" / "python.yml").write_text("name: Python Standard")
(profile_dir / "agents" / "test_agent.yml").write_text("name: Test Agent")
(profile_dir / "workflows" / "deploy.yml").write_text("name: Deploy Workflow")
documents = parser.parse_directory(str(profile_dir))
assert len(documents) == 3
# Check document types
doc_types = [doc.content_type for doc in documents]
assert AgentOSContentType.STANDARD in doc_types
assert AgentOSContentType.AGENT in doc_types
assert AgentOSContentType.WORKFLOW in doc_types
def test_parse_yaml_file(self, tmp_path):
"""Test parsing YAML file."""
parser = AgentOSParser()
yaml_content = """
name: Test Standard
description: A test coding standard
version: 1.0
category: coding
rules:
- name: naming_convention
description: Use snake_case for variables
- name: max_line_length
description: Keep lines under 80 characters
"""
yaml_file = tmp_path / "test.yml"
yaml_file.write_text(yaml_content)
doc = parser._parse_file(
yaml_file,
AgentOSContentType.STANDARD,
tmp_path
)
assert doc is not None
assert doc.content_type == AgentOSContentType.STANDARD
assert doc.title == "Test Standard"
assert "A test coding standard" in doc.content
assert doc.metadata["version"] == 1.0 # YAML parses 1.0 as float
assert doc.metadata["category"] == "coding"
assert "naming_convention" in doc.content
def test_parse_markdown_file(self, tmp_path):
"""Test parsing Markdown file."""
parser = AgentOSParser()
md_content = """# Test Workflow
This is a test workflow document.
## Steps
1. First step
2. Second step
3. Third step
## Notes
Important notes about the workflow.
"""
md_file = tmp_path / "test.md"
md_file.write_text(md_content)
doc = parser._parse_file(
md_file,
AgentOSContentType.WORKFLOW,
tmp_path
)
assert doc is not None
assert doc.content_type == AgentOSContentType.WORKFLOW
assert doc.title == "Test Workflow"
assert "This is a test workflow document" in doc.content
assert doc.metadata["file_type"] == "markdown"
def test_parse_empty_yaml_file(self, tmp_path):
"""Test parsing empty YAML file."""
parser = AgentOSParser()
empty_file = tmp_path / "empty.yml"
empty_file.write_text("")
doc = parser._parse_file(
empty_file,
AgentOSContentType.STANDARD,
tmp_path
)
assert doc is None
def test_parse_invalid_yaml_file(self, tmp_path):
"""Test parsing invalid YAML file."""
parser = AgentOSParser()
invalid_file = tmp_path / "invalid.yml"
invalid_file.write_text("invalid: yaml: content: [")
doc = parser._parse_file(
invalid_file,
AgentOSContentType.STANDARD,
tmp_path
)
assert doc is None
def test_yaml_to_text_conversion(self):
"""Test YAML to text conversion."""
parser = AgentOSParser()
yaml_data = {
"name": "Test Standard",
"description": "Test description",
"rules": [
{"name": "rule1", "description": "First rule"},
{"name": "rule2", "description": "Second rule"}
],
"settings": {
"strict": True,
"version": "1.0"
}
}
text = parser._yaml_to_text(yaml_data, "Test Standard")
assert "# Test Standard" in text
assert "## Description" in text
assert "Test description" in text
assert "## Rules" in text
assert "rule1" in text
assert "First rule" in text
assert "rule2" in text
assert "Second rule" in text
assert "## Settings" in text
assert "strict" in text
assert "True" in text
def test_parse_content_directory(self, tmp_path):
"""Test parsing content directory."""
parser = AgentOSParser()
content_dir = tmp_path / "standards"
content_dir.mkdir()
# Create nested structure
(content_dir / "subdir").mkdir()
(content_dir / "subdir" / "nested.yml").write_text("name: Nested")
(content_dir / "root.yml").write_text("name: Root")
(content_dir / "ignore.txt").write_text("Should be ignored")
documents = parser._parse_content_directory(
content_dir,
AgentOSContentType.STANDARD
)
assert len(documents) == 2 # Only YAML files
titles = [doc.title for doc in documents]
assert "Nested" in titles
assert "Root" in titles
@pytest.mark.integration
class TestAgentOSIngestion:
"""Test AgentOSIngestion functionality."""
@patch('app.core.agent_os_ingestion.OllamaEmbedding')
def test_ingest_profile_success(self, mock_embed, sample_kb, clean_db, tmp_path):
"""Test successful Agent OS profile ingestion."""
# Setup mock embedding
mock_embed_instance = MagicMock()
mock_embed_instance.get_text_embedding.return_value = [0.1] * 768
mock_embed.return_value = mock_embed_instance
# Create Agent OS profile
profile_dir = tmp_path / "test_profile"
profile_dir.mkdir()
(profile_dir / "standards").mkdir()
(profile_dir / "agents").mkdir()
(profile_dir / "standards" / "python.yml").write_text("name: Python Standard")
(profile_dir / "agents" / "test_agent.yml").write_text("name: Test Agent")
# Create ingestion instance
db_manager = SQLiteManager()
ingestion = AgentOSIngestion(db_manager)
result = ingestion.ingest_profile(sample_kb["name"], str(profile_dir))
assert result["success"] is True
assert result["total_documents"] == 2
assert "standard" in result["documents_by_type"]
assert "agent" in result["documents_by_type"]
assert result["documents_by_type"]["standard"] == 1
assert result["documents_by_type"]["agent"] == 1
def test_ingest_profile_nonexistent_kb(self, clean_db, tmp_path):
"""Test ingesting profile to non-existent KB."""
db_manager = SQLiteManager()
ingestion = AgentOSIngestion(db_manager)
profile_dir = tmp_path / "test_profile"
profile_dir.mkdir()
with pytest.raises(ValueError, match="does not exist"):
ingestion.ingest_profile("nonexistent_kb", str(profile_dir))
def test_ingest_profile_invalid_path(self, sample_kb, clean_db):
"""Test ingesting profile with invalid path."""
db_manager = SQLiteManager()
ingestion = AgentOSIngestion(db_manager)
with pytest.raises(ValueError, match="Failed to parse"):
ingestion.ingest_profile(sample_kb["name"], "/nonexistent/path")
def test_ingest_profile_empty(self, sample_kb, clean_db, tmp_path):
"""Test ingesting empty profile."""
db_manager = SQLiteManager()
ingestion = AgentOSIngestion(db_manager)
# Create empty profile directory
profile_dir = tmp_path / "empty_profile"
profile_dir.mkdir()
result = ingestion.ingest_profile(sample_kb["name"], str(profile_dir))
assert result["success"] is False
assert "No documents found" in result["message"]
assert result["documents_processed"] == 0
@patch('app.core.agent_os_ingestion.OllamaEmbedding')
def test_ingest_profile_with_batch_processing(self, mock_embed, sample_kb, clean_db, tmp_path):
"""Test profile ingestion with batch processing."""
mock_embed_instance = MagicMock()
mock_embed_instance.get_text_embedding.return_value = [0.1] * 768
mock_embed.return_value = mock_embed_instance
# Create profile with many documents
profile_dir = tmp_path / "large_profile"
profile_dir.mkdir()
(profile_dir / "standards").mkdir()
# Create 5 documents
for i in range(5):
(profile_dir / "standards" / f"standard_{i}.yml").write_text(f"name: Standard {i}")
db_manager = SQLiteManager()
ingestion = AgentOSIngestion(db_manager)
# Use small batch size to test batching
result = ingestion.ingest_profile(sample_kb["name"], str(profile_dir), batch_size=2)
assert result["success"] is True
assert result["total_documents"] == 5
assert result["documents_by_type"]["standard"] == 5
@patch('app.core.agent_os_ingestion.OllamaEmbedding')
def test_ingest_batch_with_embedding_failure(self, mock_embed, sample_kb, clean_db, tmp_path):
"""Test batch ingestion with embedding failures."""
# Mock embedding to fail for some documents
mock_embed_instance = MagicMock()
call_count = 0
def side_effect(*args, **kwargs):
nonlocal call_count
call_count += 1
if call_count <= 2: # First 2 succeed
return [0.1] * 768
else: # Rest fail
raise Exception("Embedding failed")
mock_embed_instance.get_text_embedding.side_effect = side_effect
mock_embed.return_value = mock_embed_instance
# Create profile
profile_dir = tmp_path / "test_profile"
profile_dir.mkdir()
(profile_dir / "standards").mkdir()
for i in range(4):
(profile_dir / "standards" / f"standard_{i}.yml").write_text(f"name: Standard {i}")
db_manager = SQLiteManager()
ingestion = AgentOSIngestion(db_manager)
result = ingestion.ingest_profile(sample_kb["name"], str(profile_dir))
# Should still succeed with partial documents
assert result["success"] is True
assert result["total_documents"] == 4
assert len(result["errors"]) > 0
@patch('app.core.agent_os_ingestion.OllamaEmbedding')
def test_get_profile_stats(self, mock_embed, sample_kb, clean_db, tmp_path):
"""Test getting profile statistics."""
mock_embed_instance = MagicMock()
mock_embed_instance.get_text_embedding.return_value = [0.1] * 768
mock_embed.return_value = mock_embed_instance
# Create and ingest profile
profile_dir = tmp_path / "stats_profile"
profile_dir.mkdir()
(profile_dir / "standards").mkdir()
(profile_dir / "agents").mkdir()
(profile_dir / "workflows").mkdir()
(profile_dir / "standards" / "python.yml").write_text("name: Python")
(profile_dir / "agents" / "test.yml").write_text("name: Test Agent")
(profile_dir / "workflows" / "deploy.yml").write_text("name: Deploy")
(profile_dir / "workflows" / "test.yml").write_text("name: Test Workflow")
db_manager = SQLiteManager()
ingestion = AgentOSIngestion(db_manager)
# Ingest first
ingestion.ingest_profile(sample_kb["name"], str(profile_dir))
# Get stats
stats = ingestion.get_profile_stats(sample_kb["name"])
assert stats["total_documents"] == 4
assert stats["documents_by_type"]["standard"] == 1
assert stats["documents_by_type"]["agent"] == 1
assert stats["documents_by_type"]["workflow"] == 2
def test_get_profile_stats_empty_kb(self, sample_kb, clean_db):
"""Test getting stats for KB with no Agent OS content."""
db_manager = SQLiteManager()
ingestion = AgentOSIngestion(db_manager)
stats = ingestion.get_profile_stats(sample_kb["name"])
assert stats["total_documents"] == 0
assert stats["documents_by_type"] == {}
@patch('app.core.agent_os_ingestion.OllamaEmbedding')
def test_search_by_type(self, mock_embed, sample_kb, clean_db, tmp_path):
"""Test searching Agent OS content by type."""
mock_embed_instance = MagicMock()
mock_embed_instance.get_text_embedding.return_value = [0.1] * 768
mock_embed.return_value = mock_embed_instance
# Create and ingest profile
profile_dir = tmp_path / "search_profile"
profile_dir.mkdir()
(profile_dir / "standards").mkdir()
(profile_dir / "agents").mkdir()
(profile_dir / "standards" / "python.yml").write_text("name: Python Standard\ndescription: For Python code")
(profile_dir / "agents" / "test.yml").write_text("name: Test Agent\ndescription: For testing")
db_manager = SQLiteManager()
ingestion = AgentOSIngestion(db_manager)
ingestion.ingest_profile(sample_kb["name"], str(profile_dir))
# Search by type
results = ingestion.search_by_type(
sample_kb["name"],
AgentOSContentType.STANDARD,
limit=10
)
assert len(results) == 1
assert results[0]["metadata"]["content_type"] == "standard"
assert "Python Standard" in results[0]["content"]
@patch('app.core.agent_os_ingestion.OllamaEmbedding')
def test_search_by_type_with_query(self, mock_embed, sample_kb, clean_db, tmp_path):
"""Test searching Agent OS content by type with query."""
mock_embed_instance = MagicMock()
mock_embed_instance.get_text_embedding.return_value = [0.1] * 768
mock_embed.return_value = mock_embed_instance
# Create and ingest profile
profile_dir = tmp_path / "query_profile"
profile_dir.mkdir()
(profile_dir / "standards").mkdir()
(profile_dir / "standards" / "python.yml").write_text("name: Python Standard")
(profile_dir / "standards" / "javascript.yml").write_text("name: JavaScript Standard")
db_manager = SQLiteManager()
ingestion = AgentOSIngestion(db_manager)
ingestion.ingest_profile(sample_kb["name"], str(profile_dir))
# Search with query
results = ingestion.search_by_type(
sample_kb["name"],
AgentOSContentType.STANDARD,
query="Python",
limit=10
)
assert len(results) >= 1
# Should find Python standard first (higher similarity)
python_found = any("Python" in result["content"] for result in results)
assert python_found
def test_search_by_type_nonexistent_kb(self, clean_db):
"""Test searching by type in non-existent KB."""
db_manager = SQLiteManager()
ingestion = AgentOSIngestion(db_manager)
results = ingestion.search_by_type(
"nonexistent_kb",
AgentOSContentType.STANDARD
)
assert results == []
@patch('app.core.agent_os_ingestion.OllamaEmbedding')
def test_search_by_type_no_results(self, mock_embed, sample_kb, clean_db):
"""Test searching by type with no results."""
mock_embed_instance = MagicMock()
mock_embed_instance.get_text_embedding.return_value = [0.1] * 768
mock_embed.return_value = mock_embed_instance
db_manager = SQLiteManager()
ingestion = AgentOSIngestion(db_manager)
results = ingestion.search_by_type(
sample_kb["name"],
AgentOSContentType.SPEC, # No specs in KB
limit=10
)
assert results == []
@pytest.mark.unit
class TestAgentOSContentType:
"""Test AgentOSContentType enum."""
def test_content_type_values(self):
"""Test content type enum values."""
assert AgentOSContentType.STANDARD.value == "standard"
assert AgentOSContentType.AGENT.value == "agent"
assert AgentOSContentType.WORKFLOW.value == "workflow"
assert AgentOSContentType.COMMAND.value == "command"
assert AgentOSContentType.PRODUCT.value == "product"
assert AgentOSContentType.SPEC.value == "spec"
assert AgentOSContentType.UNKNOWN.value == "unknown"
def test_content_type_comparison(self):
"""Test content type comparison."""
assert AgentOSContentType.STANDARD == "standard"
assert AgentOSContentType.STANDARD != AgentOSContentType.AGENT
assert AgentOSContentType.STANDARD != "agent"
def test_content_type_string_representation(self):
"""Test content type string representation."""
assert str(AgentOSContentType.STANDARD) == "standard"
assert str(AgentOSContentType.AGENT) == "agent"