"""End-to-end workflow tests."""
import os
import shutil
import tempfile
from pathlib import Path
import pytest
from wassden.handlers import (
handle_analyze_changes,
handle_check_completeness,
handle_get_traceability,
handle_prompt_code,
handle_prompt_design,
handle_prompt_tasks,
handle_validate_design,
handle_validate_requirements,
handle_validate_tasks,
)
from wassden.types import Language, SpecDocuments
class TestFullWorkflow:
"""Test complete end-to-end workflows."""
@pytest.fixture(autouse=True)
def setup_temp_workspace(self):
"""Set up temporary workspace for each test."""
self.temp_dir = Path(tempfile.mkdtemp())
self.specs_dir = self.temp_dir / "specs"
self.specs_dir.mkdir()
# Change to temp directory for file operations
self.original_cwd = Path.cwd()
os.chdir(self.temp_dir)
yield
# Cleanup
os.chdir(self.original_cwd)
shutil.rmtree(self.temp_dir)
@pytest.mark.asyncio
async def test_complete_sdd_workflow(self):
"""Test complete Spec-Driven Development workflow."""
# Step 1-3: Completeness check, requirements creation and validation
req_file = await self._test_requirements_phase()
# Step 4-6: Design prompt, creation and validation
design_file = await self._test_design_phase(req_file)
# Step 7-9: Tasks prompt, creation and validation
tasks_file = await self._test_tasks_phase(req_file, design_file)
# Step 10-12: Implementation, traceability and change analysis
await self._test_implementation_phase(req_file, design_file, tasks_file)
async def _test_requirements_phase(self) -> Path:
"""Test requirements phase of workflow."""
# Step 1: Check completeness with sufficient information
completeness_result = await handle_check_completeness(self._get_sample_user_input(), Language.JAPANESE)
assert completeness_result.content
completeness_text = completeness_result.content[0].text
assert "requirements.md" in completeness_text
assert "EARS" in completeness_text
# Step 2: Create requirements manually
requirements_content = self._get_sample_requirements()
req_file = self.specs_dir / "requirements.md"
req_file.write_text(requirements_content)
# Step 3: Validate requirements
validation_specs = await SpecDocuments.from_paths(requirements_path=req_file, language=Language.JAPANESE)
result = await handle_validate_requirements(validation_specs)
validation_text = result.content[0].text
assert "✅" in validation_text
assert "要件数: 7" in validation_text
return req_file
async def _test_design_phase(self, req_file: Path) -> Path:
"""Test design phase of workflow."""
# Step 4: Generate design prompt
design_prompt_specs = await SpecDocuments.from_paths(requirements_path=req_file, language=Language.JAPANESE)
result = await handle_prompt_design(design_prompt_specs)
design_prompt_text = result.content[0].text
assert "design.md" in design_prompt_text
assert "アーキテクチャ" in design_prompt_text
# Step 5: Create design document
design_content = self._get_sample_design()
design_file = self.specs_dir / "design.md"
design_file.write_text(design_content)
# Step 6: Validate design
design_validation_specs = await SpecDocuments.from_paths(
requirements_path=req_file, design_path=design_file, language=Language.JAPANESE
)
result = await handle_validate_design(design_validation_specs)
design_validation_text = result.content[0].text
assert "✅" in design_validation_text
assert "参照要件数: 7" in design_validation_text
return design_file
async def _test_tasks_phase(self, req_file: Path, design_file: Path) -> Path:
"""Test tasks phase of workflow."""
# Step 7: Generate tasks prompt
tasks_prompt_specs = await SpecDocuments.from_paths(
requirements_path=req_file, design_path=design_file, language=Language.JAPANESE
)
result = await handle_prompt_tasks(tasks_prompt_specs)
tasks_prompt_text = result.content[0].text
assert "tasks.md" in tasks_prompt_text
assert "WBS" in tasks_prompt_text
# Step 8: Create tasks document
tasks_content = self._get_sample_tasks()
tasks_file = self.specs_dir / "tasks.md"
tasks_file.write_text(tasks_content)
# Step 9: Validate tasks
tasks_validation_specs = await SpecDocuments.from_paths(tasks_path=tasks_file, language=Language.JAPANESE)
result = await handle_validate_tasks(tasks_validation_specs)
tasks_validation_text = result.content[0].text
assert "✅" in tasks_validation_text
assert "タスク数: 9" in tasks_validation_text
return tasks_file
async def _test_implementation_phase(self, req_file: Path, design_file: Path, tasks_file: Path) -> None:
"""Test implementation phase of workflow."""
# Step 10: Generate implementation prompt
code_prompt_specs = await SpecDocuments.from_paths(
requirements_path=req_file, design_path=design_file, tasks_path=tasks_file, language=Language.JAPANESE
)
code_prompt_result = await handle_prompt_code(code_prompt_specs)
code_prompt_text = code_prompt_result.content[0].text
assert "実装" in code_prompt_text
assert "TASK-01-01" in code_prompt_text
# Step 11: Get traceability report
traceability_specs = await SpecDocuments.from_paths(
requirements_path=req_file, design_path=design_file, tasks_path=tasks_file, language=Language.JAPANESE
)
result = await handle_get_traceability(traceability_specs)
traceability_text = result.content[0].text
assert "トレーサビリティレポート" in traceability_text
assert "要件数: 7" in traceability_text
# Step 12: Analyze change impact
change_analysis_result = await handle_analyze_changes(
req_file, "Added REQ-08 for user profile management", Language.JAPANESE
)
change_analysis_text = change_analysis_result.content[0].text
assert "変更影響分析" in change_analysis_text
assert "REQ-08" in change_analysis_text
def _get_sample_user_input(self) -> str:
"""Get sample user input for testing."""
return """
Python FastAPI RESTful API for task management.
Technology: Python 3.12, FastAPI, SQLAlchemy, PostgreSQL
Users: Project managers and team members
Constraints: Must be cloud-deployable, response time < 1s
Scope: CRUD operations for tasks, user authentication, basic reporting
"""
def _get_sample_requirements(self) -> str:
"""Get sample requirements content for testing."""
return """
# Requirements Document
## 0. サマリー
タスク管理用のPython FastAPI RESTful APIシステム。
プロジェクトマネージャーとチームメンバーが使用。
クラウド展開可能で1秒以内のレスポンス時間を実現。
## 1. 用語集
- **API**: Application Programming Interface
- **CRUD**: Create, Read, Update, Delete operations
- **FastAPI**: Modern Python web framework
- **SQLAlchemy**: Python ORM
## 2. スコープ
### インスコープ
- タスクのCRUD操作
- ユーザー認証機能
- 基本的なレポート機能
### アウトオブスコープ
- 高度な分析機能
- リアルタイム通知
- モバイルアプリ
## 3. 制約
- Python 3.12以上使用
- FastAPI フレームワーク使用
- PostgreSQLデータベース使用
- クラウド展開可能であること
- レスポンス時間1秒以内
## 4. 非機能要件（NFR）
- **NFR-01**: APIレスポンス時間は1秒以内であること
- **NFR-02**: 同時接続100ユーザーまで対応すること
- **NFR-03**: データは暗号化して保存すること
- **NFR-04**: APIドキュメントが自動生成されること
- **NFR-05**: 99%以上の稼働率を維持すること
## 5. KPI / 受入基準
- **KPI-01**: API応答時間平均500ms以下
- **KPI-02**: システム稼働率99%以上
- **KPI-03**: ユーザー満足度4.0/5.0以上
## 6. 機能要件（EARS）
- **REQ-01**: システムは、認証されたユーザーがタスクを作成すること
- **REQ-02**: システムは、認証されたユーザーがタスク一覧を取得すること
- **REQ-03**: システムは、認証されたユーザーがタスクを更新すること
- **REQ-04**: システムは、認証されたユーザーがタスクを削除すること
- **REQ-05**: システムは、ユーザーがログインできること
- **REQ-06**: システムは、ユーザーがログアウトできること
- **REQ-07**: システムは、基本的なタスク統計を提供すること
## 7. テスト要件（Testing Requirements）
- **TR-01**: タスクCRUD操作のテスト要件
- **TR-02**: ユーザー認証のテスト要件
- **TR-03**: レポート機能のテスト要件
"""
def _get_sample_design(self) -> str:
"""Get sample design content for testing."""
return """
# Design Document
## 1. アーキテクチャ概要
RESTful API アーキテクチャを採用したタスク管理システム
[Requirements: REQ-01, REQ-02, REQ-03, REQ-04, REQ-05, REQ-06, REQ-07]
## 2. コンポーネント設計
- **auth-service**: ユーザー認証処理 [REQ-05, REQ-06]
- **task-service**: タスクCRUD操作 [REQ-01, REQ-02, REQ-03, REQ-04]
- **stats-service**: 統計情報提供 [REQ-07]
- **database-layer**: データ永続化層
- **api-gateway**: APIエンドポイント管理
## 3. データ設計
```python
class User:
id: int
username: str
email: str
hashed_password: str
class Task:
id: int
title: str
description: str
status: str
user_id: int
created_at: datetime
updated_at: datetime
```
## 4. API設計
- **POST /auth/login** → ユーザーログイン [REQ-05]
- **POST /auth/logout** → ユーザーログアウト [REQ-06]
- **POST /tasks** → タスク作成 [REQ-01]
- **GET /tasks** → タスク一覧取得 [REQ-02]
- **PUT /tasks/{id}** → タスク更新 [REQ-03]
- **DELETE /tasks/{id}** → タスク削除 [REQ-04]
- **GET /stats** → 統計情報取得 [REQ-07]
## 5. 非機能設計
- **パフォーマンス**: FastAPIとasync/awaitでレスポンス時間最適化
- **セキュリティ**: JWT認証、パスワードハッシュ化
- **可用性**: ヘルスチェックエンドポイント実装
- **ドキュメント**: FastAPIの自動ドキュメント生成
## 6. テスト戦略
### 6.1 単体テスト
各サービスクラスの個別テスト
### 6.2 統合テスト
API エンドポイントのテスト
### 6.3 E2Eテスト
完全なワークフローテスト
## 7. トレーサビリティ (必須)
- REQ-01 ⇔ **task-service**
- REQ-02 ⇔ **task-service**
- REQ-03 ⇔ **task-service**
- REQ-04 ⇔ **task-service**
- REQ-05 ⇔ **auth-service**
- REQ-06 ⇔ **auth-service**
- REQ-07 ⇔ **stats-service**
- TR-01 ⇔ **test-auth-flow**
- TR-02 ⇔ **test-task-crud**
- TR-03 ⇔ **test-stats-generation**
"""
def _get_sample_tasks(self) -> str:
"""Get sample tasks content for testing."""
return """
# Tasks Document
## 1. 概要
FastAPI タスク管理システムのWBS（Work Breakdown Structure）
## 2. タスク一覧
### Phase 1: 基盤構築
- [ ] **TASK-01-01**: 環境セットアップ
- **REQ**: [NFR-04]
- **DC**: **database-layer**
- **依存**: なし
- **受け入れ観点**:
- 観点1: Python、FastAPI、PostgreSQL環境が正常に構築される
- [ ] **TASK-01-02**: データベース設計実装
- **REQ**: [REQ-01, REQ-02, REQ-03, REQ-04]
- **DC**: **database-layer**
- **依存**: TASK-01-01
- **受け入れ観点**:
- 観点1: SQLAlchemyモデルが正しく定義される
### Phase 2: 認証機能実装
- [ ] **TASK-02-01**: ユーザー認証システム
- **REQ**: [REQ-05, REQ-06]
- **DC**: **auth-service**
- **依存**: TASK-01-02
- **受け入れ観点**:
- 観点1: JWT認証が正常に動作する
### Phase 3: コア機能実装
- [ ] **TASK-03-01**: タスクCRUD API
- **REQ**: [REQ-01, REQ-02, REQ-03, REQ-04]
- **DC**: **task-service**
- **依存**: TASK-02-01
- **受け入れ観点**:
- 観点1: CRUD操作が正常に動作する
- [ ] **TASK-03-02**: 統計情報API
- **REQ**: [REQ-07]
- **DC**: **stats-service**
- **依存**: TASK-03-01
- **受け入れ観点**:
- 観点1: 統計情報が正しく生成される
- [ ] **TASK-03-03**: API Gateway 設定
- **REQ**: [NFR-01]
- **DC**: **api-gateway**
- **依存**: TASK-03-01
- **受け入れ観点**:
- 観点1: APIルーティングが正常に動作する
### Phase 4: テスト・品質保証
- [ ] **TASK-04-01**: 認証フローテスト作成
- **REQ**: [TR-01]
- **DC**: **test-auth-flow**
- **依存**: TASK-02-01
- **受け入れ観点**:
- 観点1: 認証テストが正常に実行される
- [ ] **TASK-04-02**: タスクCRUDテスト作成
- **REQ**: [TR-02]
- **DC**: **test-task-crud**
- **依存**: TASK-03-01
- **受け入れ観点**:
- 観点1: CRUDテストが正常に実行される
- [ ] **TASK-04-03**: 統計生成テスト作成
- **REQ**: [TR-03]
- **DC**: **test-stats-generation**
- **依存**: TASK-03-02
- **受け入れ観点**:
- 観点1: 統計テストが正常に実行される
## 3. 依存関係
```
TASK-01-01 → TASK-01-02 → TASK-02-01 → TASK-03-01 → TASK-04-02
→ TASK-03-02 → TASK-04-03
→ TASK-03-03
→ TASK-04-01
```
## 4. マイルストーン
- **M1**: Phase 1完了 (環境・基盤整備完了)
- **M2**: Phase 2完了 (認証機能完了)
- **M3**: Phase 3完了 (コア機能完了)
- **M4**: リリース準備完了 (テスト・品質保証完了)
"""
@pytest.mark.asyncio
async def test_incomplete_input_workflow(self):
"""Test workflow with incomplete user input."""
# Step 1: Check completeness with insufficient information
completeness_result = await handle_check_completeness("Simple project", Language.JAPANESE)
completeness_text = completeness_result.content[0].text
# Should ask for more information
assert "不足している" in completeness_text or "不明" in completeness_text
assert "技術" in completeness_text or "ユーザー" in completeness_text
# Should provide questions to clarify requirements
question_indicators = ["？", "ですか", "何で", "誰で", "どの"]
assert any(indicator in completeness_text for indicator in question_indicators)
@pytest.mark.asyncio
async def test_validation_failure_workflow(self):
"""Test workflow with validation failures."""
# Create invalid requirements file
invalid_requirements = """
# Invalid Requirements
## 0. サマリー
Missing required sections and invalid IDs
## 1. 用語集
- **Term**: Definition
## 2. スコープ
Project scope
## 3. 制約
Constraints
## 4. 非機能要件（NFR）
- **NFR-01**: Performance requirement
## 5. KPI / 受入基準
- **KPI-01**: Success metric
## 6. 機能要件（EARS）
- **REQ-001**: Invalid ID format (should be REQ-XX)
- **REQ-1**: Also invalid format
- **INVALID-01**: Wrong prefix
"""
req_file = self.specs_dir / "requirements.md"
req_file.write_text(invalid_requirements)
# Validate requirements - should fail
validation_specs = await SpecDocuments.from_paths(requirements_path=req_file, language=Language.JAPANESE)
result = await handle_validate_requirements(validation_specs)
validation_text = result.content[0].text
assert "⚠️" in validation_text or "修正が必要" in validation_text
assert "REQ-001" in validation_text # Should detect invalid format
assert "REQ-1" in validation_text
assert "INVALID-01" in validation_text
@pytest.mark.asyncio
async def test_traceability_gaps_workflow(self):
"""Test workflow with traceability gaps."""
# Create requirements with REQ-01, REQ-02, REQ-03
requirements_content = """
## 機能要件（EARS）
- **REQ-01**: システムは、機能Aを提供すること
- **REQ-02**: システムは、機能Bを提供すること
- **REQ-03**: システムは、機能Cを提供すること
"""
# Create design that only references REQ-01 and REQ-02
design_content = """
## アーキテクチャ概要
System design [REQ-01, REQ-02]
## コンポーネント設計
- **component-a**: Function A [REQ-01]
- **component-b**: Function B [REQ-02]
## データ設計
Data model
## API設計
API endpoints
## 非機能設計
NFR implementation
## テスト戦略
Test plan
"""
req_file = self.specs_dir / "requirements.md"
design_file = self.specs_dir / "design.md"
req_file.write_text(requirements_content)
design_file.write_text(design_content)
# Validate design - should detect missing REQ-03
design_validation_specs = await SpecDocuments.from_paths(
requirements_path=req_file, design_path=design_file, language=Language.JAPANESE
)
result = await handle_validate_design(design_validation_specs)
validation_text = result.content[0].text
assert "⚠️" in validation_text or "修正が必要" in validation_text
assert "REQ-03" in validation_text # Should detect missing reference
# Get traceability report to show gaps
traceability_specs = await SpecDocuments.from_paths(
requirements_path=req_file,
design_path=design_file,
tasks_path=Path("nonexistent.md"),
)
result = await handle_get_traceability(traceability_specs)
traceability_text = result.content[0].text
assert "設計されていない要件" in traceability_text or "設計参照なし" in traceability_text
class TestWorkflowErrorHandling:
"""Test error handling in workflows."""
@pytest.fixture(autouse=True)
def setup_temp_workspace(self):
"""Set up temporary workspace."""
self.temp_dir = Path(tempfile.mkdtemp())
self.original_cwd = Path.cwd()
os.chdir(self.temp_dir)
yield
os.chdir(self.original_cwd)
shutil.rmtree(self.temp_dir)
@pytest.mark.asyncio
async def test_missing_files_workflow(self):
"""Test workflow with missing prerequisite files."""
# Try to validate requirements when file doesn't exist
validation_specs = await SpecDocuments.from_paths(
requirements_path=Path("nonexistent/requirements.md"), language=Language.JAPANESE
)
result = await handle_validate_requirements(validation_specs)
validation_text = result.content[0].text
assert "エラー" in validation_text
assert "見つかりません" in validation_text
# Try to generate design prompt when requirements don't exist
design_prompt_specs = await SpecDocuments.from_paths(
requirements_path=Path("nonexistent/requirements.md"), language=Language.JAPANESE
)
result = await handle_prompt_design(design_prompt_specs)
design_text = result.content[0].text
assert "エラー" in design_text
@pytest.mark.asyncio
async def test_corrupted_files_workflow(self):
"""Test workflow with corrupted/malformed files."""
# Create corrupted requirements file
specs_dir = Path("specs")
specs_dir.mkdir(exist_ok=True)
corrupted_file = specs_dir / "requirements.md"
corrupted_file.write_text("Completely invalid content with no structure")
# Validation should handle corrupted files gracefully
validation_specs = await SpecDocuments.from_paths(requirements_path=corrupted_file, language=Language.JAPANESE)
result = await handle_validate_requirements(validation_specs)
validation_text = result.content[0].text
# Should either show validation errors or handle gracefully
assert "修正" in validation_text or "エラー" in validation_text
