OpenClaw Agent
f26d4d8f7f
- Multi-source content aggregation (Bilibili/YouTube/Zhihu) - Smart inbox management with AI quality scoring - Long video chunking analysis (30min+ auto-segmentation) - Topic-based organization - Git auto-sync to Obsidian - Template system with Jinja2 and fallback rendering - CLI commands: subscribe, inbox, process, skip, status, search - Comprehensive test suite with unit and integration tests - Documentation: README.md and SKILL.md Checkpoint 7-8 completed by Agent E
332 lines
13 KiB
Python
332 lines
13 KiB
Python
"""
|
|
Integration tests for the complete Obsidian KB pipeline.
|
|
Tests the full workflow: subscribe → fetch → analyze → generate → sync
|
|
"""
|
|
import unittest
|
|
import tempfile
|
|
import json
|
|
from pathlib import Path
|
|
from datetime import datetime
|
|
|
|
import sys
|
|
sys.path.insert(0, str(Path(__file__).parent.parent / 'src'))
|
|
|
|
from models import Content, ContentType, ContentStatus, ContentAnalysis
|
|
from storage import Storage
|
|
from template_renderer import TemplateRenderer
|
|
from config import Config
|
|
from content_chunker import ContentChunker
|
|
|
|
|
|
class TestPipeline(unittest.TestCase):
|
|
"""Integration tests for the complete pipeline."""
|
|
|
|
def setUp(self):
|
|
"""Set up test environment."""
|
|
self.temp_dir = tempfile.mkdtemp()
|
|
self.data_dir = Path(self.temp_dir) / "data"
|
|
self.vault_dir = Path(self.temp_dir) / "vault"
|
|
self.vault_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
# Initialize components
|
|
self.storage = Storage(base_path=str(self.data_dir))
|
|
self.config = Config(config_path=str(self.temp_dir / "config.yaml"))
|
|
self.config.set('git.vault_path', str(self.vault_dir))
|
|
|
|
templates_dir = Path(__file__).parent.parent / 'src' / 'templates'
|
|
self.renderer = TemplateRenderer(str(templates_dir))
|
|
self.chunker = ContentChunker()
|
|
|
|
def test_full_pipeline_short_video(self):
|
|
"""Test complete pipeline with short video."""
|
|
# 1. Create content (simulating subscription/fetch)
|
|
content = Content(
|
|
id="bvid_BV1test001",
|
|
source="bilibili",
|
|
title="Python Tips in 5 Minutes",
|
|
url="https://bilibili.com/video/BV1test001",
|
|
author="CodeMaster",
|
|
duration=300, # 5 minutes
|
|
content_type=ContentType.VIDEO,
|
|
status=ContentStatus.NEW,
|
|
tags=["python", "programming"]
|
|
)
|
|
|
|
# 2. Add to inbox
|
|
self.storage.add_to_inbox(content)
|
|
inbox = self.storage.get_inbox()
|
|
self.assertEqual(len(inbox), 1)
|
|
|
|
# 3. Simulate light scanning
|
|
content.status = ContentStatus.SCANNED
|
|
content.analysis = ContentAnalysis(
|
|
one_liner="Quick Python productivity tips",
|
|
summary="This video covers 5 essential Python tips for beginners.",
|
|
key_points=["Use list comprehensions", "Leverage built-in functions"],
|
|
tags=["python", "productivity"],
|
|
quality_score=85
|
|
)
|
|
self.storage.update_inbox_item(content)
|
|
|
|
# 4. Generate note
|
|
output_path = self.vault_dir / "inbox" / "python_tips.md"
|
|
result_path = self.renderer.render_to_file(content, str(output_path))
|
|
|
|
# 5. Verify output
|
|
self.assertTrue(Path(result_path).exists())
|
|
note_content = Path(result_path).read_text(encoding='utf-8')
|
|
|
|
# Verify content
|
|
self.assertIn("Python Tips in 5 Minutes", note_content)
|
|
self.assertIn("Quick Python productivity tips", note_content)
|
|
self.assertIn("#python", note_content)
|
|
|
|
# 6. Update status to completed
|
|
content.status = ContentStatus.GENERATED
|
|
content.note_path = str(result_path)
|
|
self.storage.update_inbox_item(content)
|
|
|
|
# 7. Verify final state
|
|
completed_item = self.storage.get_item_by_id(content.id)
|
|
self.assertEqual(completed_item.status, ContentStatus.GENERATED)
|
|
self.assertIsNotNone(completed_item.note_path)
|
|
|
|
def test_pipeline_long_video_with_chunking(self):
|
|
"""Test pipeline with long video requiring chunking."""
|
|
# 1. Create long video content
|
|
content = Content(
|
|
id="bvid_BV2long001",
|
|
source="bilibili",
|
|
title="Complete Python Course",
|
|
url="https://bilibili.com/video/BV2long001",
|
|
author="PythonAcademy",
|
|
duration=7200, # 2 hours
|
|
content_type=ContentType.VIDEO,
|
|
status=ContentStatus.NEW,
|
|
tags=["python", "tutorial", "course"]
|
|
)
|
|
|
|
# 2. Add to inbox
|
|
self.storage.add_to_inbox(content)
|
|
|
|
# 3. Determine chunking strategy
|
|
strategy = self.chunker.get_chunking_strategy(content.duration)
|
|
self.assertEqual(strategy['strategy'], 'multi_part')
|
|
self.assertGreater(strategy['chunk_count'], 2)
|
|
|
|
# 4. Simulate chunking analysis
|
|
# Create mock chunks
|
|
chunks = [
|
|
{"index": 0, "start": 0, "end": 1800, "summary": "Introduction to Python"},
|
|
{"index": 1, "start": 1800, "end": 3600, "summary": "Data structures"},
|
|
{"index": 2, "start": 3600, "end": 5400, "summary": "Functions and classes"},
|
|
{"index": 3, "start": 5400, "end": 7200, "summary": "Advanced topics"},
|
|
]
|
|
|
|
# 5. Simulate deep analysis with chunks
|
|
content.status = ContentStatus.GENERATED
|
|
content.analysis = ContentAnalysis(
|
|
one_liner="Comprehensive Python course from basics to advanced",
|
|
summary="A 2-hour course covering Python fundamentals and advanced concepts.",
|
|
key_points=["Python basics", "Data structures", "OOP", "Advanced patterns"],
|
|
tags=["python", "course", "programming"],
|
|
quality_score=95,
|
|
prerequisites=["Basic programming knowledge"],
|
|
outline=[
|
|
{"title": "Introduction", "time_range": "00:00-30:00", "summary": "Python basics"},
|
|
{"title": "Data Structures", "time_range": "30:00-60:00", "summary": "Lists, dicts, sets"},
|
|
{"title": "OOP", "time_range": "60:00-90:00", "summary": "Classes and objects"},
|
|
{"title": "Advanced", "time_range": "90:00-120:00", "summary": "Decorators, generators"},
|
|
],
|
|
pitfalls=["Python 2 vs 3 confusion", "Mutable default arguments"],
|
|
practice_advice=["Code along with examples", "Build projects"]
|
|
)
|
|
|
|
# 6. Generate tutorial note
|
|
output_path = self.vault_dir / "topics" / "programming" / "python_course.md"
|
|
result_path = self.renderer.render_to_file(content, str(output_path))
|
|
|
|
# 7. Verify tutorial template was used
|
|
note_content = Path(result_path).read_text(encoding='utf-8')
|
|
|
|
self.assertIn("Complete Python Course", note_content)
|
|
self.assertIn("Comprehensive Python course", note_content)
|
|
self.assertIn("前置知识", note_content) # Prerequisites section
|
|
self.assertIn("步骤详解", note_content) # Steps section
|
|
self.assertIn("常见坑点", note_content) # Pitfalls section
|
|
self.assertIn("实践建议", note_content) # Practice advice section
|
|
|
|
def test_pipeline_article(self):
|
|
"""Test pipeline with article content."""
|
|
# 1. Create article content
|
|
content = Content(
|
|
id="zhihu_article_001",
|
|
source="zhihu",
|
|
title="Understanding Neural Networks",
|
|
url="https://zhihu.com/question/123456",
|
|
author="AIResearcher",
|
|
content_type=ContentType.ARTICLE,
|
|
status=ContentStatus.NEW,
|
|
tags=["ai", "machine-learning"]
|
|
)
|
|
|
|
# 2. Add to inbox
|
|
self.storage.add_to_inbox(content)
|
|
|
|
# 3. Simulate analysis
|
|
content.status = ContentStatus.GENERATED
|
|
content.analysis = ContentAnalysis(
|
|
one_liner="Comprehensive guide to neural network fundamentals",
|
|
summary="Deep dive into how neural networks work.",
|
|
key_points=["Perceptrons", "Backpropagation", "Activation functions"],
|
|
tags=["ai", "neural-networks", "deep-learning"],
|
|
quality_score=90
|
|
)
|
|
|
|
# 4. Generate note
|
|
output_path = self.vault_dir / "inbox" / "neural_networks.md"
|
|
result_path = self.renderer.render_to_file(content, str(output_path))
|
|
|
|
# 5. Verify article template
|
|
note_content = Path(result_path).read_text(encoding='utf-8')
|
|
|
|
self.assertIn("Understanding Neural Networks", note_content)
|
|
self.assertIn("原文链接", note_content) # Original link section
|
|
|
|
def test_inbox_to_topic_workflow(self):
|
|
"""Test moving content from inbox to topics."""
|
|
# 1. Add content to inbox
|
|
content = Content(
|
|
id="bvid_BV3topic01",
|
|
source="bilibili",
|
|
title="React Tutorial",
|
|
url="https://bilibili.com/video/BV3topic01",
|
|
author="FrontendDev",
|
|
duration=1800,
|
|
tags=["react", "javascript", "frontend"]
|
|
)
|
|
content.analysis = ContentAnalysis(
|
|
one_liner="React fundamentals",
|
|
quality_score=88
|
|
)
|
|
|
|
self.storage.add_to_inbox(content)
|
|
|
|
# 2. Process and complete
|
|
content.status = ContentStatus.GENERATED
|
|
self.storage.update_inbox_item(content)
|
|
|
|
# 3. Move to topic (simulate)
|
|
topic_path = self.vault_dir / "topics" / "frontend" / "react_tutorial.md"
|
|
topic_path.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
result_path = self.renderer.render_to_file(content, str(topic_path))
|
|
content.note_path = str(result_path)
|
|
self.storage.update_inbox_item(content)
|
|
|
|
# 4. Verify
|
|
self.assertTrue(Path(result_path).exists())
|
|
self.assertIn("topics/frontend", content.note_path)
|
|
|
|
def test_skip_content(self):
|
|
"""Test skipping content."""
|
|
# 1. Add content
|
|
content = Content(
|
|
id="bvid_BV4skip001",
|
|
source="bilibili",
|
|
title="Low Quality Video",
|
|
url="https://bilibili.com/video/BV4skip001"
|
|
)
|
|
content.analysis = ContentAnalysis(quality_score=30) # Low quality
|
|
|
|
self.storage.add_to_inbox(content)
|
|
|
|
# 2. Skip it
|
|
content.status = ContentStatus.SKIPPED
|
|
self.storage.update_inbox_item(content)
|
|
|
|
# 3. Verify not in pending
|
|
pending = self.storage.get_pending_items()
|
|
skipped_ids = [c.id for c in self.storage.get_inbox() if c.status == ContentStatus.SKIPPED]
|
|
|
|
self.assertIn(content.id, skipped_ids)
|
|
self.assertNotIn(content, pending)
|
|
|
|
def test_batch_processing(self):
|
|
"""Test processing multiple items."""
|
|
# Add multiple items
|
|
contents = [
|
|
Content(id=f"batch_{i}", source="bilibili", title=f"Video {i}", url=f"https://example.com/{i}")
|
|
for i in range(5)
|
|
]
|
|
|
|
for content in contents:
|
|
self.storage.add_to_inbox(content)
|
|
|
|
# Process all pending
|
|
pending = self.storage.get_pending_items()
|
|
self.assertEqual(len(pending), 5)
|
|
|
|
for content in pending:
|
|
content.status = ContentStatus.GENERATED
|
|
content.analysis = ContentAnalysis(
|
|
one_liner=f"Summary for {content.title}",
|
|
quality_score=80
|
|
)
|
|
self.storage.update_inbox_item(content)
|
|
|
|
# Verify all processed
|
|
final_pending = self.storage.get_pending_items()
|
|
self.assertEqual(len(final_pending), 0)
|
|
|
|
|
|
class TestEndToEnd(unittest.TestCase):
|
|
"""End-to-end integration tests."""
|
|
|
|
def test_config_storage_renderer_integration(self):
|
|
"""Test integration of config, storage, and renderer."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
# Setup
|
|
data_dir = Path(tmpdir) / "data"
|
|
vault_dir = Path(tmpdir) / "vault"
|
|
config_path = Path(tmpdir) / "config.yaml"
|
|
|
|
config = Config(config_path=str(config_path))
|
|
config.set('git.vault_path', str(vault_dir))
|
|
config.set('chunking.chunk_duration_seconds', 600)
|
|
config.save()
|
|
|
|
storage = Storage(data_dir=str(data_dir))
|
|
|
|
templates_dir = Path(__file__).parent.parent / 'src' / 'templates'
|
|
renderer = TemplateRenderer(str(templates_dir))
|
|
|
|
# Create and process content
|
|
content = Content(
|
|
id="e2e_test_001",
|
|
source="bilibili",
|
|
title="E2E Test Video",
|
|
url="https://example.com/e2e",
|
|
duration=1200
|
|
)
|
|
content.analysis = ContentAnalysis(
|
|
one_liner="End to end test",
|
|
key_points=["Point 1", "Point 2"]
|
|
)
|
|
|
|
# Full workflow
|
|
storage.add_to_inbox(content)
|
|
output_path = Path(config.vault_path) / "e2e_output.md"
|
|
renderer.render_to_file(content, str(output_path))
|
|
content.status = ContentStatus.GENERATED
|
|
content.note_path = str(output_path)
|
|
storage.update_inbox_item(content)
|
|
|
|
# Verify
|
|
self.assertTrue(output_path.exists())
|
|
self.assertEqual(storage.get_item_by_id(content.id).status, ContentStatus.GENERATED)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|