starmen1265
766283366e
- Remove kb-bilibili-metadata/ and kb-zhihu-metadata/ directories - Create kb_workflow/__init__.py to export main and process_command - Update test_cli/test_process_command.py imports for new package structure Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
143 lines
5.1 KiB
Python
143 lines
5.1 KiB
Python
"""Tests for CLI 'process' subcommand
|
|
|
|
Tests:
|
|
1. `kb-workflow process --help` shows usage
|
|
2. `kb-workflow process <url>` calls Pipeline.run()
|
|
3. Invalid URL returns appropriate error
|
|
4. --output-dir flag sets custom output directory
|
|
"""
|
|
|
|
import pytest
|
|
import sys
|
|
from unittest.mock import patch, MagicMock, AsyncMock
|
|
from pathlib import Path
|
|
|
|
|
|
class TestProcessCommandHelp:
|
|
"""Test 1: `kb-workflow process --help` shows usage"""
|
|
|
|
def test_process_help_shows_usage(self, capsys):
|
|
"""`kb-workflow process --help` shows usage information"""
|
|
from kb_workflow import kb_workflow
|
|
|
|
# Mock sys.argv
|
|
with patch.object(sys, 'argv', ['kb-workflow', 'process', '--help']):
|
|
with pytest.raises(SystemExit) as exc_info:
|
|
kb_workflow.main()
|
|
# --help exits with 0
|
|
assert exc_info.value.code == 0
|
|
|
|
captured = capsys.readouterr()
|
|
# Should show process command in help
|
|
assert 'process' in captured.out.lower() or 'url' in captured.out.lower()
|
|
|
|
|
|
class TestProcessCommandCallsPipeline:
|
|
"""Test 2: `kb-workflow process <url>` calls Pipeline.run()"""
|
|
|
|
def test_process_url_calls_pipeline_run(self, capsys):
|
|
"""`kb-workflow process <url>` calls Pipeline.run()"""
|
|
from kb_workflow import kb_workflow
|
|
|
|
with patch.object(kb_workflow, 'process_command') as mock_process:
|
|
mock_process.return_value = {"success": True, "content_dir": "/tmp/content"}
|
|
|
|
with patch.object(sys, 'argv', ['kb-workflow', 'process', 'https://www.bilibili.com/video/BV1test']):
|
|
kb_workflow.main()
|
|
|
|
# process_command should be called with the URL
|
|
mock_process.assert_called_once()
|
|
call_args = mock_process.call_args
|
|
assert 'https://www.bilibili.com/video/BV1test' in str(call_args)
|
|
|
|
def test_process_url_with_output_dir(self, capsys):
|
|
"""`kb-workflow process <url> --output-dir /custom` passes output_dir"""
|
|
from kb_workflow import kb_workflow
|
|
|
|
with patch.object(kb_workflow, 'process_command') as mock_process:
|
|
mock_process.return_value = {"success": True, "content_dir": "/custom"}
|
|
|
|
with patch.object(sys, 'argv', [
|
|
'kb-workflow', 'process',
|
|
'https://www.bilibili.com/video/BV1test',
|
|
'--output-dir', '/custom/output'
|
|
]):
|
|
kb_workflow.main()
|
|
|
|
call_kwargs = mock_process.call_args[1]
|
|
assert call_kwargs.get('output_dir') == '/custom/output'
|
|
|
|
|
|
class TestProcessCommandInvalidUrl:
|
|
"""Test 3: Invalid URL returns appropriate error"""
|
|
|
|
def test_process_invalid_url_returns_error(self, capsys):
|
|
"""`kb-workflow process <invalid-url>` returns appropriate error"""
|
|
from kb_workflow import kb_workflow
|
|
|
|
with patch.object(kb_workflow, 'process_command') as mock_process:
|
|
mock_process.return_value = {
|
|
"success": False,
|
|
"error": "Failed to parse URL: invalid-url"
|
|
}
|
|
|
|
with patch.object(sys, 'argv', ['kb-workflow', 'process', 'invalid-url']):
|
|
kb_workflow.main()
|
|
|
|
# process_command should be called
|
|
mock_process.assert_called_once()
|
|
|
|
|
|
class TestProcessCommandFlags:
|
|
"""Test 4: --output-dir flag sets custom output directory"""
|
|
|
|
def test_output_dir_flag_sets_custom_directory(self):
|
|
"""--output-dir flag sets custom output directory"""
|
|
from kb_workflow import kb_workflow
|
|
|
|
with patch.object(kb_workflow, 'process_command') as mock_process:
|
|
mock_process.return_value = {"success": True}
|
|
|
|
with patch.object(sys, 'argv', [
|
|
'kb-workflow', 'process',
|
|
'https://www.bilibili.com/video/BV1test',
|
|
'-o', '/tmp/custom'
|
|
]):
|
|
kb_workflow.main()
|
|
|
|
call_kwargs = mock_process.call_args[1]
|
|
assert call_kwargs.get('output_dir') == '/tmp/custom'
|
|
|
|
def test_skip_completed_default_true(self):
|
|
"""skip_completed defaults to True"""
|
|
from kb_workflow import kb_workflow
|
|
|
|
with patch.object(kb_workflow, 'process_command') as mock_process:
|
|
mock_process.return_value = {"success": True}
|
|
|
|
with patch.object(sys, 'argv', [
|
|
'kb-workflow', 'process',
|
|
'https://www.bilibili.com/video/BV1test'
|
|
]):
|
|
kb_workflow.main()
|
|
|
|
call_kwargs = mock_process.call_args[1]
|
|
assert call_kwargs.get('skip_completed') is True
|
|
|
|
def test_no_skip_flag_disables_skip_completed(self):
|
|
"""--no-skip flag disables skip_completed"""
|
|
from kb_workflow import kb_workflow
|
|
|
|
with patch.object(kb_workflow, 'process_command') as mock_process:
|
|
mock_process.return_value = {"success": True}
|
|
|
|
with patch.object(sys, 'argv', [
|
|
'kb-workflow', 'process',
|
|
'https://www.bilibili.com/video/BV1test',
|
|
'--no-skip'
|
|
]):
|
|
kb_workflow.main()
|
|
|
|
call_kwargs = mock_process.call_args[1]
|
|
assert call_kwargs.get('skip_completed') is False
|