From 246e4038b42d45bf05185fde84ef44cd08fe47f9 Mon Sep 17 00:00:00 2001 From: starmen1265 Date: Fri, 13 Mar 2026 17:15:02 +0800 Subject: [PATCH] test(01-01): add failing tests for WorkflowStateManager RED phase: 20 test cases for: - WorkflowState.is_step_completed (5 tests) - WorkflowState.mark_step_* methods (4 tests) - WorkflowStateManager atomic write (4 tests) - WorkflowStateManager backup fallback (4 tests) - WorkflowStateManager.get_or_create (3 tests) 11 tests fail due to missing kb_pipeline.state module. Co-Authored-By: Claude Opus 4.6 --- tests/test_pipeline/test_state.py | 401 ++++++++++++++++++++++++++++++ 1 file changed, 401 insertions(+) create mode 100644 tests/test_pipeline/test_state.py diff --git a/tests/test_pipeline/test_state.py b/tests/test_pipeline/test_state.py new file mode 100644 index 0000000..38f6c8a --- /dev/null +++ b/tests/test_pipeline/test_state.py @@ -0,0 +1,401 @@ +""" +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_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_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_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_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_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_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_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_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 + valid_state = WorkflowState( + content_id="backup-valid-001", + platform="bilibili", + source_url="https://www.bilibili.com/video/BV999" + ) + manager.save(valid_state) + + # Corrupt main file + with open(state_file, 'w', encoding='utf-8') as f: + f.write("{ invalid json content }") + + # Should fall back to backup + 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_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_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_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_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