starmen1265
c9ca1b538d
- Remove sys.path manipulation from kb_workflow.py - Update CLI imports to use kb_workflow.pipeline prefix - Update all test file imports and patch paths Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
413 lines
15 KiB
Python
413 lines
15 KiB
Python
"""
|
|
Tests for state.py module - WorkflowStateManager.
|
|
|
|
Task 1 (RED): Failing tests for WorkflowStateManager implementation.
|
|
Tests cover:
|
|
- is_step_completed method
|
|
- mark_step_started/completed/failed methods
|
|
- Atomic write with backup rotation
|
|
- Backup fallback on corruption
|
|
- get_or_create method
|
|
"""
|
|
import json
|
|
import os
|
|
import pytest
|
|
from pathlib import Path
|
|
from datetime import datetime
|
|
|
|
from kb_workflow.pipeline.schemas import WorkflowState, StepRecord, StepStatus
|
|
|
|
|
|
class TestWorkflowStateIsStepCompleted:
|
|
"""Test is_step_completed method."""
|
|
|
|
def test_returns_true_for_completed_step(self):
|
|
"""is_step_completed returns True when step status is COMPLETED."""
|
|
state = WorkflowState(
|
|
content_id="test-content-001",
|
|
platform="bilibili",
|
|
source_url="https://www.bilibili.com/video/BV123456"
|
|
)
|
|
state.steps["fetch_metadata"] = StepRecord(
|
|
step_name="fetch_metadata",
|
|
status=StepStatus.COMPLETED,
|
|
completed_at=datetime.now()
|
|
)
|
|
|
|
assert state.is_step_completed("fetch_metadata") is True
|
|
|
|
def test_returns_false_for_pending_step(self):
|
|
"""is_step_completed returns False when step status is PENDING."""
|
|
state = WorkflowState(
|
|
content_id="test-content-002",
|
|
platform="bilibili",
|
|
source_url="https://www.bilibili.com/video/BV123456"
|
|
)
|
|
state.steps["download_video"] = StepRecord(
|
|
step_name="download_video",
|
|
status=StepStatus.PENDING
|
|
)
|
|
|
|
assert state.is_step_completed("download_video") is False
|
|
|
|
def test_returns_false_for_running_step(self):
|
|
"""is_step_completed returns False when step status is RUNNING."""
|
|
state = WorkflowState(
|
|
content_id="test-content-003",
|
|
platform="bilibili",
|
|
source_url="https://www.bilibili.com/video/BV123456"
|
|
)
|
|
state.steps["transcribe_audio"] = StepRecord(
|
|
step_name="transcribe_audio",
|
|
status=StepStatus.RUNNING,
|
|
started_at=datetime.now()
|
|
)
|
|
|
|
assert state.is_step_completed("transcribe_audio") is False
|
|
|
|
def test_returns_false_for_failed_step(self):
|
|
"""is_step_completed returns False when step status is FAILED."""
|
|
state = WorkflowState(
|
|
content_id="test-content-004",
|
|
platform="bilibili",
|
|
source_url="https://www.bilibili.com/video/BV123456"
|
|
)
|
|
state.steps["extract_keyframes"] = StepRecord(
|
|
step_name="extract_keyframes",
|
|
status=StepStatus.FAILED,
|
|
error_message="Video file not found"
|
|
)
|
|
|
|
assert state.is_step_completed("extract_keyframes") is False
|
|
|
|
def test_returns_false_for_nonexistent_step(self):
|
|
"""is_step_completed returns False when step doesn't exist."""
|
|
state = WorkflowState(
|
|
content_id="test-content-005",
|
|
platform="bilibili",
|
|
source_url="https://www.bilibili.com/video/BV123456"
|
|
)
|
|
|
|
assert state.is_step_completed("nonexistent_step") is False
|
|
|
|
|
|
class TestWorkflowStateMarkStep:
|
|
"""Test mark_step_started/completed/failed methods."""
|
|
|
|
def test_mark_step_started_creates_running_record(self):
|
|
"""mark_step_started creates a step record with RUNNING status."""
|
|
state = WorkflowState(
|
|
content_id="test-content-006",
|
|
platform="bilibili",
|
|
source_url="https://www.bilibili.com/video/BV123456"
|
|
)
|
|
|
|
state.mark_step_started("fetch_metadata")
|
|
|
|
assert "fetch_metadata" in state.steps
|
|
assert state.steps["fetch_metadata"].status == StepStatus.RUNNING
|
|
assert state.steps["fetch_metadata"].started_at is not None
|
|
|
|
def test_mark_step_completed_updates_status_and_timestamp(self):
|
|
"""mark_step_completed updates status to COMPLETED and sets completed_at."""
|
|
state = WorkflowState(
|
|
content_id="test-content-007",
|
|
platform="bilibili",
|
|
source_url="https://www.bilibili.com/video/BV123456"
|
|
)
|
|
state.mark_step_started("download_video")
|
|
|
|
state.mark_step_completed("download_video", output_summary="Downloaded 150MB")
|
|
|
|
assert state.steps["download_video"].status == StepStatus.COMPLETED
|
|
assert state.steps["download_video"].completed_at is not None
|
|
assert state.steps["download_video"].output_summary == "Downloaded 150MB"
|
|
|
|
def test_mark_step_completed_without_started(self):
|
|
"""mark_step_completed creates new record if step wasn't started."""
|
|
state = WorkflowState(
|
|
content_id="test-content-008",
|
|
platform="bilibili",
|
|
source_url="https://www.bilibili.com/video/BV123456"
|
|
)
|
|
|
|
state.mark_step_completed("quick_task", output_summary="Done immediately")
|
|
|
|
assert state.steps["quick_task"].status == StepStatus.COMPLETED
|
|
assert state.steps["quick_task"].completed_at is not None
|
|
|
|
def test_mark_step_failed_updates_status_and_error(self):
|
|
"""mark_step_failed updates status to FAILED and sets error_message."""
|
|
state = WorkflowState(
|
|
content_id="test-content-009",
|
|
platform="bilibili",
|
|
source_url="https://www.bilibili.com/video/BV123456"
|
|
)
|
|
state.mark_step_started("transcribe_audio")
|
|
|
|
state.mark_step_failed("transcribe_audio", error_message="FFmpeg not available")
|
|
|
|
assert state.steps["transcribe_audio"].status == StepStatus.FAILED
|
|
assert state.steps["transcribe_audio"].error_message == "FFmpeg not available"
|
|
assert state.steps["transcribe_audio"].completed_at is not None
|
|
|
|
|
|
class TestAtomicWrite:
|
|
"""Test WorkflowStateManager.save() atomic write pattern."""
|
|
|
|
def test_save_creates_state_file(self, tmp_path):
|
|
"""save creates workflow_state.json in content directory."""
|
|
from kb_workflow.pipeline.state import WorkflowStateManager
|
|
|
|
manager = WorkflowStateManager(content_dir=tmp_path)
|
|
state = WorkflowState(
|
|
content_id="test-atomic-001",
|
|
platform="bilibili",
|
|
source_url="https://www.bilibili.com/video/BV123456"
|
|
)
|
|
|
|
manager.save(state)
|
|
|
|
state_file = tmp_path / "workflow_state.json"
|
|
assert state_file.exists()
|
|
|
|
def test_save_creates_parent_directories(self, tmp_path):
|
|
"""save creates parent directories if they don't exist."""
|
|
from kb_workflow.pipeline.state import WorkflowStateManager
|
|
|
|
nested_dir = tmp_path / "deeply" / "nested" / "content"
|
|
manager = WorkflowStateManager(content_dir=nested_dir)
|
|
state = WorkflowState(
|
|
content_id="test-mkdir-001",
|
|
platform="bilibili",
|
|
source_url="https://www.bilibili.com/video/BV123456"
|
|
)
|
|
|
|
manager.save(state)
|
|
|
|
assert nested_dir.exists()
|
|
assert (nested_dir / "workflow_state.json").exists()
|
|
|
|
def test_save_rotates_backup(self, tmp_path):
|
|
"""save rotates backup (workflow_state.json -> workflow_state.json.bak)."""
|
|
from kb_workflow.pipeline.state import WorkflowStateManager
|
|
|
|
manager = WorkflowStateManager(content_dir=tmp_path)
|
|
state1 = WorkflowState(
|
|
content_id="first-save",
|
|
platform="bilibili",
|
|
source_url="https://www.bilibili.com/video/BV111"
|
|
)
|
|
state2 = WorkflowState(
|
|
content_id="second-save",
|
|
platform="bilibili",
|
|
source_url="https://www.bilibili.com/video/BV222"
|
|
)
|
|
|
|
manager.save(state1)
|
|
state_file = tmp_path / "workflow_state.json"
|
|
backup_file = tmp_path / "workflow_state.json.bak"
|
|
|
|
assert state_file.exists()
|
|
assert not backup_file.exists()
|
|
|
|
manager.save(state2)
|
|
|
|
assert state_file.exists()
|
|
assert backup_file.exists()
|
|
|
|
# Backup should contain first save
|
|
with open(backup_file, 'r', encoding='utf-8') as f:
|
|
backup_data = json.load(f)
|
|
assert backup_data["content_id"] == "first-save"
|
|
|
|
# Main should contain second save
|
|
with open(state_file, 'r', encoding='utf-8') as f:
|
|
main_data = json.load(f)
|
|
assert main_data["content_id"] == "second-save"
|
|
|
|
def test_no_temp_files_remain_after_save(self, tmp_path):
|
|
"""No temp files remain after successful save."""
|
|
from kb_workflow.pipeline.state import WorkflowStateManager
|
|
|
|
manager = WorkflowStateManager(content_dir=tmp_path)
|
|
state = WorkflowState(
|
|
content_id="test-cleanup",
|
|
platform="bilibili",
|
|
source_url="https://www.bilibili.com/video/BV123456"
|
|
)
|
|
|
|
manager.save(state)
|
|
|
|
temp_files = list(tmp_path.glob("*.tmp"))
|
|
assert len(temp_files) == 0
|
|
|
|
|
|
class TestBackupFallback:
|
|
"""Test WorkflowStateManager.load() backup fallback on corruption."""
|
|
|
|
def test_load_returns_state_for_valid_file(self, tmp_path):
|
|
"""load returns WorkflowState for valid file."""
|
|
from kb_workflow.pipeline.state import WorkflowStateManager
|
|
|
|
manager = WorkflowStateManager(content_dir=tmp_path)
|
|
original = WorkflowState(
|
|
content_id="valid-test-001",
|
|
platform="bilibili",
|
|
source_url="https://www.bilibili.com/video/BV123456"
|
|
)
|
|
|
|
manager.save(original)
|
|
loaded = manager.load()
|
|
|
|
assert loaded is not None
|
|
assert loaded.content_id == "valid-test-001"
|
|
assert loaded.platform == "bilibili"
|
|
|
|
def test_load_returns_none_for_missing_file(self, tmp_path):
|
|
"""load returns None when no state file exists."""
|
|
from kb_workflow.pipeline.state import WorkflowStateManager
|
|
|
|
manager = WorkflowStateManager(content_dir=tmp_path)
|
|
loaded = manager.load()
|
|
|
|
assert loaded is None
|
|
|
|
def test_load_falls_back_to_backup_on_corruption(self, tmp_path):
|
|
"""load falls back to backup when main file is corrupted."""
|
|
from kb_workflow.pipeline.state import WorkflowStateManager
|
|
|
|
manager = WorkflowStateManager(content_dir=tmp_path)
|
|
state_file = tmp_path / "workflow_state.json"
|
|
backup_file = tmp_path / "workflow_state.json.bak"
|
|
|
|
# Create valid backup by saving twice (backup is created on second save)
|
|
valid_state = WorkflowState(
|
|
content_id="backup-valid-001",
|
|
platform="bilibili",
|
|
source_url="https://www.bilibili.com/video/BV999"
|
|
)
|
|
manager.save(valid_state)
|
|
|
|
# Save again to create backup
|
|
valid_state2 = WorkflowState(
|
|
content_id="backup-valid-002",
|
|
platform="bilibili",
|
|
source_url="https://www.bilibili.com/video/BV888"
|
|
)
|
|
manager.save(valid_state2)
|
|
|
|
# Now backup exists with "backup-valid-001", main has "backup-valid-002"
|
|
assert backup_file.exists()
|
|
|
|
# Corrupt main file
|
|
with open(state_file, 'w', encoding='utf-8') as f:
|
|
f.write("{ invalid json content }")
|
|
|
|
# Should fall back to backup (which has backup-valid-001)
|
|
loaded = manager.load()
|
|
|
|
assert loaded is not None
|
|
assert loaded.content_id == "backup-valid-001"
|
|
|
|
def test_load_from_backup_when_main_missing(self, tmp_path):
|
|
"""load loads from backup when main file is missing."""
|
|
from kb_workflow.pipeline.state import WorkflowStateManager
|
|
|
|
manager = WorkflowStateManager(content_dir=tmp_path)
|
|
state_file = tmp_path / "workflow_state.json"
|
|
backup_file = tmp_path / "workflow_state.json.bak"
|
|
|
|
# Create state and save twice to create backup
|
|
state1 = WorkflowState(
|
|
content_id="backup-only-001",
|
|
platform="bilibili",
|
|
source_url="https://www.bilibili.com/video/BV111"
|
|
)
|
|
state2 = WorkflowState(
|
|
content_id="backup-only-002",
|
|
platform="bilibili",
|
|
source_url="https://www.bilibili.com/video/BV222"
|
|
)
|
|
manager.save(state1)
|
|
manager.save(state2)
|
|
|
|
# Delete main file, keep backup
|
|
state_file.unlink()
|
|
|
|
# Should load from backup
|
|
loaded = manager.load()
|
|
|
|
assert loaded is not None
|
|
assert loaded.content_id == "backup-only-001"
|
|
|
|
|
|
class TestStateManagerGetOrCreate:
|
|
"""Test WorkflowStateManager.get_or_create() method."""
|
|
|
|
def test_get_or_create_returns_existing_state(self, tmp_path):
|
|
"""get_or_create returns existing state when it exists."""
|
|
from kb_workflow.pipeline.state import WorkflowStateManager
|
|
|
|
manager = WorkflowStateManager(content_dir=tmp_path)
|
|
|
|
# Create and save state first
|
|
existing = WorkflowState(
|
|
content_id="existing-001",
|
|
platform="bilibili",
|
|
source_url="https://www.bilibili.com/video/BV123456"
|
|
)
|
|
existing.mark_step_completed("init")
|
|
manager.save(existing)
|
|
|
|
# get_or_create should return existing
|
|
state = manager.get_or_create(
|
|
content_id="existing-001",
|
|
platform="bilibili",
|
|
source_url="https://www.bilibili.com/video/BV123456"
|
|
)
|
|
|
|
assert state.content_id == "existing-001"
|
|
assert state.is_step_completed("init") is True
|
|
|
|
def test_get_or_create_creates_new_state(self, tmp_path):
|
|
"""get_or_create creates new state when it doesn't exist."""
|
|
from kb_workflow.pipeline.state import WorkflowStateManager
|
|
|
|
manager = WorkflowStateManager(content_dir=tmp_path)
|
|
|
|
state = manager.get_or_create(
|
|
content_id="new-001",
|
|
platform="zhihu",
|
|
source_url="https://zhuanlan.zhihu.com/p/123456"
|
|
)
|
|
|
|
assert state is not None
|
|
assert state.content_id == "new-001"
|
|
assert state.platform == "zhihu"
|
|
assert state.source_url == "https://zhuanlan.zhihu.com/p/123456"
|
|
assert len(state.steps) == 0
|
|
|
|
def test_get_or_create_is_idempotent(self, tmp_path):
|
|
"""get_or_create is idempotent - calling twice returns same state."""
|
|
from kb_workflow.pipeline.state import WorkflowStateManager
|
|
|
|
manager = WorkflowStateManager(content_dir=tmp_path)
|
|
|
|
state1 = manager.get_or_create(
|
|
content_id="idempotent-001",
|
|
platform="bilibili",
|
|
source_url="https://www.bilibili.com/video/BV123456"
|
|
)
|
|
state1.mark_step_completed("step1")
|
|
manager.save(state1)
|
|
|
|
state2 = manager.get_or_create(
|
|
content_id="idempotent-001",
|
|
platform="bilibili",
|
|
source_url="https://www.bilibili.com/video/BV123456"
|
|
)
|
|
|
|
assert state2.content_id == state1.content_id
|
|
assert state2.is_step_completed("step1") is True
|