/
devsec
/
anothertap
Обзор
Документация
Войти
/
devsec
/
anothertap
Код
Запросы
0
Задачи
Вики
Пакеты
0
Релизы
0
CI/CD
Аналитика
Безопасность
master
src/services/test_plan_manager.py
434 строки
14 KB
devsec
init project
23 авг 2025, 21:01
23 авг 2025, 21:01
5b0e5af
Код
Авторство
О чём код?
""" Test Plan Management Service for CRUD operations on test plans """ import json import os import logging from typing import List, Optional, Dict, Any from pathlib import Path from datetime import datetime from models.test_plan import TestPlan, PlanStatus from models.test_step import TestStep class TestPlanManager: """ Service for managing test plans with CRUD operations """ def __init__(self, data_directory: str = "data"): self.logger = logging.getLogger(__name__) self.data_directory = Path(data_directory) self.test_plans_file = self.data_directory / "test_plans.json" self._test_plans: Dict[str, TestPlan] = {} # Ensure data directory exists self.data_directory.mkdir(exist_ok=True) # Load existing test plans self.load_test_plans() def create_test_plan(self, test_plan: TestPlan) -> bool: """ Create a new test plan Args: test_plan: Test plan to create Returns: True if creation successful, False otherwise """ try: if test_plan.id in self._test_plans: self.logger.warning(f"Test plan with ID {test_plan.id} already exists") return False test_plan.created_date = datetime.now().isoformat() test_plan.modified_date = datetime.now().isoformat() self._test_plans[test_plan.id] = test_plan self.save_test_plans() self.logger.info(f"Created test plan: {test_plan.display_name}") return True except Exception as e: self.logger.error(f"Failed to create test plan: {e}") return False def get_test_plan(self, plan_id: str) -> Optional[TestPlan]: """ Get a test plan by ID Args: plan_id: ID of the test plan to retrieve Returns: Test plan if found, None otherwise """ return self._test_plans.get(plan_id) def get_all_test_plans(self) -> List[TestPlan]: """ Get all test plans Returns: List of all test plans """ return list(self._test_plans.values()) def update_test_plan(self, test_plan: TestPlan) -> bool: """ Update an existing test plan Args: test_plan: Test plan to update Returns: True if update successful, False otherwise """ try: if test_plan.id not in self._test_plans: self.logger.warning(f"Test plan with ID {test_plan.id} not found") return False test_plan.update_modified_date() self._test_plans[test_plan.id] = test_plan self.save_test_plans() self.logger.info(f"Updated test plan: {test_plan.display_name}") return True except Exception as e: self.logger.error(f"Failed to update test plan: {e}") return False def delete_test_plan(self, plan_id: str) -> bool: """ Delete a test plan Args: plan_id: ID of the test plan to delete Returns: True if deletion successful, False otherwise """ try: if plan_id not in self._test_plans: self.logger.warning(f"Test plan with ID {plan_id} not found") return False test_plan = self._test_plans[plan_id] del self._test_plans[plan_id] self.save_test_plans() self.logger.info(f"Deleted test plan: {test_plan.display_name}") return True except Exception as e: self.logger.error(f"Failed to delete test plan: {e}") return False def duplicate_test_plan(self, plan_id: str, new_name: Optional[str] = None) -> Optional[TestPlan]: """ Duplicate an existing test plan Args: plan_id: ID of the test plan to duplicate new_name: Name for the new test plan (optional) Returns: New test plan if successful, None otherwise """ try: original = self.get_test_plan(plan_id) if not original: return None # Create a deep copy plan_dict = original.to_dict() new_plan = TestPlan.from_dict(plan_dict) # Update properties for new plan import uuid new_plan.id = str(uuid.uuid4()) new_plan.name = new_name or f"{original.name} (Copy)" new_plan.created_date = datetime.now().isoformat() new_plan.modified_date = datetime.now().isoformat() new_plan.reset_execution_state() # Reset step IDs for step in new_plan.all_steps: step.id = str(uuid.uuid4()) step.reset_execution_state() if self.create_test_plan(new_plan): return new_plan else: return None except Exception as e: self.logger.error(f"Failed to duplicate test plan: {e}") return None def find_test_plans(self, **criteria) -> List[TestPlan]: """ Find test plans matching criteria Args: **criteria: Search criteria (name, category, tags, etc.) Returns: List of test plans matching criteria """ results = [] for plan in self._test_plans.values(): match = True for key, value in criteria.items(): if key == "tags": # Special handling for tags search if not any(tag.lower() in value.lower() for tag in plan.tags): match = False break elif key == "name_contains": # Special handling for name search if value.lower() not in plan.name.lower(): match = False break elif hasattr(plan, key): plan_value = getattr(plan, key) if isinstance(plan_value, str): if value.lower() not in plan_value.lower(): match = False break else: if plan_value != value: match = False break else: match = False break if match: results.append(plan) return results def get_test_plans_by_status(self, status: PlanStatus) -> List[TestPlan]: """ Get test plans by execution status Args: status: Plan status to filter by Returns: List of test plans with specified status """ return [plan for plan in self._test_plans.values() if plan.status == status] def get_test_plans_by_category(self, category: str) -> List[TestPlan]: """ Get test plans by category Args: category: Category to filter by Returns: List of test plans in specified category """ return [plan for plan in self._test_plans.values() if plan.category == category] def get_test_plans_using_device(self, device_id: str) -> List[TestPlan]: """ Get test plans that use a specific device Args: device_id: ID of the device Returns: List of test plans using the device """ results = [] for plan in self._test_plans.values(): if device_id in plan.get_required_devices(): results.append(plan) return results def validate_test_plan(self, plan_id: str) -> List[str]: """ Validate a test plan Args: plan_id: ID of the test plan to validate Returns: List of validation issues (empty if valid) """ plan = self.get_test_plan(plan_id) if not plan: return ["Test plan not found"] return plan.validate_plan() def export_test_plan(self, plan_id: str, file_path: str) -> bool: """ Export a test plan to JSON file Args: plan_id: ID of the test plan to export file_path: Path to export file Returns: True if export successful, False otherwise """ try: plan = self.get_test_plan(plan_id) if not plan: self.logger.error(f"Test plan {plan_id} not found") return False plan_data = plan.to_dict() with open(file_path, 'w', encoding='utf-8') as f: json.dump({ 'test_plan': plan_data, 'version': '1.0.0', 'exported_at': datetime.now().isoformat() }, f, indent=2, ensure_ascii=False) self.logger.info(f"Exported test plan '{plan.name}' to {file_path}") return True except Exception as e: self.logger.error(f"Failed to export test plan: {e}") return False def import_test_plan(self, file_path: str, overwrite: bool = False) -> Optional[TestPlan]: """ Import a test plan from JSON file Args: file_path: Path to import file overwrite: Whether to overwrite existing test plan with same ID Returns: Imported test plan if successful, None otherwise """ try: with open(file_path, 'r', encoding='utf-8') as f: data = json.load(f) plan_data = data.get('test_plan') if not plan_data: self.logger.error("No test plan data found in file") return None plan = TestPlan.from_dict(plan_data) if plan.id in self._test_plans and not overwrite: # Generate new ID if not overwriting import uuid plan.id = str(uuid.uuid4()) plan.name = f"{plan.name} (Imported)" plan.reset_execution_state() if self.create_test_plan(plan): self.logger.info(f"Imported test plan '{plan.name}' from {file_path}") return plan else: return None except Exception as e: self.logger.error(f"Failed to import test plan: {e}") return None def export_all_test_plans(self, file_path: str) -> bool: """ Export all test plans to JSON file Args: file_path: Path to export file Returns: True if export successful, False otherwise """ try: plans_data = [plan.to_dict() for plan in self._test_plans.values()] with open(file_path, 'w', encoding='utf-8') as f: json.dump({ 'test_plans': plans_data, 'version': '1.0.0', 'exported_at': datetime.now().isoformat() }, f, indent=2, ensure_ascii=False) self.logger.info(f"Exported {len(plans_data)} test plans to {file_path}") return True except Exception as e: self.logger.error(f"Failed to export test plans: {e}") return False def get_categories(self) -> List[str]: """ Get all unique categories from test plans Returns: List of unique categories """ categories = set() for plan in self._test_plans.values(): if plan.category: categories.add(plan.category) return sorted(list(categories)) def get_all_tags(self) -> List[str]: """ Get all unique tags from test plans Returns: List of unique tags """ tags = set() for plan in self._test_plans.values(): tags.update(plan.tags) return sorted(list(tags)) def save_test_plans(self): """Save all test plans to JSON file""" try: plans_data = [plan.to_dict() for plan in self._test_plans.values()] with open(self.test_plans_file, 'w', encoding='utf-8') as f: json.dump({ 'test_plans': plans_data, 'version': '1.0.0' }, f, indent=2, ensure_ascii=False) except Exception as e: self.logger.error(f"Failed to save test plans: {e}") def load_test_plans(self): """Load test plans from JSON file""" try: if not self.test_plans_file.exists(): self.logger.info("No test plans file found, starting with empty test plan list") return with open(self.test_plans_file, 'r', encoding='utf-8') as f: data = json.load(f) plans_data = data.get('test_plans', []) for plan_data in plans_data: plan = TestPlan.from_dict(plan_data) # Reset execution status on load plan.reset_execution_state() self._test_plans[plan.id] = plan self.logger.info(f"Loaded {len(plans_data)} test plans") except Exception as e: self.logger.error(f"Failed to load test plans: {e}")