# 测试用例生成器：基于 IR 生成测试用例集 import json import logging import uuid from datetime import datetime, timezone from server.config import settings from server.core.llm_provider.router import router from server.services.ir_engine.generator import ir_generator logger = logging.getLogger("testflow") DEFAULT_GEN_PROMPT = """你是一位资深的软件测试工程师。基于以下中间表示（IR），请生成完整的测试用例集。 测试设计原则： 1. 每个功能点至少覆盖一条正向用例和一条异常用例 2. 对于 P0 功能点，额外覆盖边界值测试 3. 用例步骤使用 Given-When-Then 结构 4. 预期结果必须明确可验证 输出格式：纯 JSON 数组，每个测试用例包含以下字段： - id: 唯一标识符（字符串） - module: 模块名 - feature: 功能点名称 - case_title: 用例标题 - preconditions: 前置条件 - steps: 测试步骤（Given-When-Then 格式） - expected_result: 预期结果 - priority: 用例优先级 (P0/P1/P2) - tags: 标签数组（如：["正向", "异常", "边界"]） ## IR 内容： {{ ir_content }} 请只输出 JSON 数组，不要用 ```json 代码块包裹。""" class TestCaseGenerator: """Generate test case sets from IR YAML using LLM.""" def __init__(self): self._tc_store: dict[str, dict] = {} async def generate(self, ir_id: str, skill_name: str = "default") -> dict: """Generate test cases from an IR version.""" # Get IR content ir_version = await ir_generator.get_ir(ir_id) if not ir_version: return {"error": f"IR version not found: {ir_id}", "cases": []} # Handle both old YAML and new pipeline JSON formats ir_content = ir_version.get("yaml_content", "") ir_json = ir_version.get("ir_json") # If we have the new pipeline format (rules array), convert to testcases directly if ir_json and ir_json.get("rules"): logger.info("[TC] 从 %d 条 IR 规则转换测试用例...", len(ir_json["rules"])) cases = self._rules_to_cases(ir_json["rules"], ir_json.get("feature", "")) tc_set_id = str(uuid.uuid4())[:8] tc_set = self._store_tc(tc_set_id, ir_id, cases) # Log distribution p0 = sum(1 for c in cases if c.get("priority") == "P0") p1 = sum(1 for c in cases if c.get("priority") == "P1") logger.info("[TC] ✅ 生成 %d 条用例 (P0:%d P1:%d P2:%d)", len(cases), p0, p1, len(cases) - p0 - p1) return tc_set # Fallback: use LLM with skill prompt if not ir_content: return {"error": "IR has no content", "cases": [], "tc_set_id": ""} # Load skill prompt skill = ir_generator.load_skill(skill_name) prompt_template = skill.get("gen_cases_prompt", DEFAULT_GEN_PROMPT) prompt = prompt_template.replace("{{ ir_content }}", ir_content) logger.info("[TC] 调用 LLM 生成测试用例 for IR %s with skill %s", ir_id, skill_name) client = router.get_text_client() try: raw = client.chat( model=router.text_model, messages=[{"role": "user", "content": prompt}], temperature=0.0, ) except Exception as e: logger.error("Test case generation failed: %s", e) return {"error": str(e), "cases": [], "tc_set_id": ""} # Parse JSON response cases = self._parse_json(raw) # Create test case set tc_set_id = str(uuid.uuid4())[:8] tc_set = { "tc_set_id": tc_set_id, "ir_id": ir_id, "cases": cases, "case_count": len(cases), "created_at": datetime.now(timezone.utc).isoformat(), } self._tc_store[tc_set_id] = tc_set # Save to output output_dir = settings.OUTPUT_DIR / ir_version.get("prd_id", "unknown") / "tc_sets" output_dir.mkdir(parents=True, exist_ok=True) tc_path = output_dir / f"tc_{tc_set_id}.json" with open(tc_path, "w", encoding="utf-8") as f: json.dump(tc_set, f, ensure_ascii=False, indent=2) usg = client.usage logger.info("Test cases generated: id=%s, count=%d, tokens: prompt=%d completion=%d total=%d", tc_set_id, len(cases), usg["prompt_tokens"], usg["completion_tokens"], usg["total_tokens"]) return tc_set async def get_tc_set(self, tc_set_id: str) -> dict | None: """Get a test case set by ID.""" return self._tc_store.get(tc_set_id) @staticmethod def _parse_json(raw: str) -> list[dict]: """Parse JSON from LLM response, handling code fences.""" text = raw.strip() if text.startswith("```json") or text.startswith("```"): first_nl = text.find("\n") if first_nl != -1: text = text[first_nl + 1:] if text.endswith("```"): text = text[:-3] text = text.strip() try: parsed = json.loads(text) if isinstance(parsed, list): return parsed if isinstance(parsed, dict) and "cases" in parsed: return parsed["cases"] return [parsed] if isinstance(parsed, dict) else [] except json.JSONDecodeError: logger.warning("Failed to parse test case JSON, returning empty") return [] def _rules_to_cases(self, rules: list[dict], feature: str) -> list[dict]: """Convert pipeline IR rules to test case format. TC IDs: TC-{feature_id}-{seq}""" import re feature_id = re.sub(r'[^A-Z]', '', feature.upper())[:6] or "FEAT" if len(feature_id) < 2: feature_id = (feature_id + "FEAT")[:4] cases = [] for i, rule in enumerate(rules): tc_id = f"TC-{feature_id}-{i+1:03d}" desc = rule.get("description", "") # Build steps string from trigger + actions trigger = rule.get("trigger", {}) conditions = trigger.get("conditions", []) cond_texts = [] for c in conditions: unit = f" {c.get('unit', '')}" if c.get('unit') else "" cond_texts.append(f"{c.get('signal','?')} {c.get('operator','?')} {c.get('value','?')}{unit}") given = "Given " + (", ".join(cond_texts) if cond_texts else "触发条件满足") actions = rule.get("actions", []) action_texts = [] for a in actions: if a.get("type") == "user_interaction": action_texts.append(f"{a.get('description','')}('{a.get('content','')}')") else: action_texts.append(a.get("description", "")) then = "Then " + ("; ".join(action_texts) if action_texts else "执行动作") steps = f"{given}\nWhen 条件触发\n{then}" # Determine tags tags = ["正向"] priority = rule.get("priority", "P2") if priority == "P0": tags.append("冒烟") precond = rule.get("precondition", {}) if precond: tags.append(f"{precond.get('app_type', '')} {precond.get('app_state', '')}".strip()) cases.append({ "id": tc_id, "ir_rule_id": rule.get("rule_id", ""), "module": feature, "feature": rule.get("description", desc)[:40], "case_title": desc[:60], "preconditions": json.dumps(precond, ensure_ascii=False) if precond else "", "steps": steps, "expected_result": action_texts[-1] if action_texts else desc, "priority": priority, "tags": [t for t in tags if t], }) return cases def _store_tc(self, tc_set_id: str, ir_id: str, cases: list[dict]) -> dict: """Store a test case set and return its metadata.""" tc_set = { "tc_set_id": tc_set_id, "ir_id": ir_id, "cases": cases, "case_count": len(cases), "created_at": datetime.now(timezone.utc).isoformat(), } self._tc_store[tc_set_id] = tc_set return tc_set tc_generator = TestCaseGenerator()