
E2e Outside In Test Generator
- 86 installs
- 70 repo stars
- Updated July 26, 2026
- rysweet/amplihack
Helps with testing & qa tasks.
About
e2e-outside-in-test-generator is a Claude Code skill for testing & qa. It helps solo builders move faster with AI-assisted development.
- e2e-outside-in-test-generator
- Testing & QA
- AI-coding skill
E2e Outside In Test Generator by the numbers
- 86 all-time installs (skills.sh)
- +1 installs in the week ending Jul 26, 2026 (Skillselion tracking)
- Ranked #1,046 of 2,155 Testing & QA skills by installs in the Skillselion catalog
- Data as of Aug 2, 2026 (Skillselion catalog sync)
npx skills add https://github.com/rysweet/amplihack --skill e2e-outside-in-test-generatorAdd your badge
Show developers this skill is listed on Skillselion. Paste this into your README.
| Installs | 86 |
|---|---|
| repo stars | ★ 70 |
| Last updated | July 26, 2026 |
| Repository | rysweet/amplihack ↗ |
What it does
Helps with testing & qa tasks.
Files
E2E Outside-In Test Generator
Automatically generates comprehensive end-to-end Playwright tests for full-stack applications using outside-in testing methodology. Generates 40+ tests across 7 categories, finds real bugs, and validates through iterative fix loops.
LEVEL 1: Quick Start
Purpose
This skill analyzes your full-stack application and generates a complete Playwright test suite that:
- Validates user journeys from browser through backend
- Finds real bugs through systematic exploration
- Runs deterministically with no flakiness
- Integrates with CI out of the box
When to Use
Activate this skill when you:
- Need comprehensive browser testing for a full-stack app
- Want to validate critical user flows end-to-end
- Need regression coverage before major refactoring
- Are preparing for production deployment
Requirements: Your project must have both a frontend (Next.js, React, Vue, Angular) and backend API.
Quick Start
# In your project root
$ claude
> add e2e testsThe skill automatically:
1. Analyzes your application (routes, API endpoints, database schema) 2. Sets up infrastructure (Playwright config, test helpers, seed data) 3. Generates 40+ tests across 7 categories 4. Runs tests and fixes failures (up to 5 iterations) 5. Reports coverage and recommendations
Expected Output
e2e/
├── playwright.config.ts # Playwright configuration
├── test-helpers/
│ ├── auth.ts # Authentication helpers
│ ├── navigation.ts # Navigation helpers
│ ├── assertions.ts # Custom assertions
│ └── data-setup.ts # Test data management
├── fixtures/
│ ├── users.json # Test user data
│ ├── products.json # Test product data
│ └── seed.sql # Database seed script
├── happy-path/
│ ├── user-registration.spec.ts
│ ├── user-login.spec.ts
│ └── checkout-flow.spec.ts
├── edge-cases/
│ ├── invalid-inputs.spec.ts
│ └── boundary-conditions.spec.ts
├── error-handling/
│ ├── network-failures.spec.ts
│ └── validation-errors.spec.ts
├── performance/
│ ├── page-load-times.spec.ts
│ └── api-response-times.spec.ts
├── security/
│ ├── unauthorized-access.spec.ts
│ └── xss-protection.spec.ts
├── accessibility/
│ ├── keyboard-navigation.spec.ts
│ └── screen-reader.spec.ts
└── integration/
├── database-persistence.spec.ts
└── api-integration.spec.ts
Total: 42 tests across 7 categoriesRun Your Tests
# Run all tests
npx playwright test
# Run specific category
npx playwright test e2e/happy-path
# Run in headed mode
npx playwright test --headed
# Run in debug mode
npx playwright test --debugSuccess Criteria
After generation, your test suite achieves:
- ✓ 40+ tests across all 7 categories
- ✓ 100% pass rate after fix loop
- ✓ <2 minute total execution time
- ✓ ≥1 real bug discovered during generation
- ✓ Zero flakiness (deterministic test data)
LEVEL 2: Full Features
The 5 Phases
graph LR
A[Phase 1: Analysis] --> B[Phase 2: Infrastructure]
B --> C[Phase 3: Generation]
C --> D[Phase 4: Fix Loop]
D --> E[Phase 5: Coverage Audit]Phase 1: Stack Analysis
The skill performs deep application analysis:
Frontend Analysis:
- Detects framework (Next.js, React, Vue, Angular)
- Maps routes and pages
- Identifies navigation patterns
- Extracts interactive elements
Backend Analysis:
- Discovers API endpoints (REST/GraphQL)
- Maps data models and relationships
- Identifies authentication mechanisms
- Detects validation rules
Database Analysis:
- Extracts schema and relationships
- Identifies required fields
- Determines foreign key constraints
- Maps enum types
Example Analysis Output:
StackConfig(
frontend_framework="nextjs",
frontend_dir="app/",
backend_framework="fastapi",
api_base_url="http://localhost:8000/api",
database_type="postgresql",
auth_mechanism="jwt",
routes=[
Route(path="/", component="Home"),
Route(path="/login", component="Login"),
Route(path="/products", component="ProductList"),
Route(path="/products/:id", component="ProductDetail"),
Route(path="/checkout", component="Checkout")
],
api_endpoints=[
APIEndpoint(path="/auth/login", method="POST"),
APIEndpoint(path="/auth/register", method="POST"),
APIEndpoint(path="/products", method="GET"),
APIEndpoint(path="/products/:id", method="GET"),
APIEndpoint(path="/orders", method="POST")
],
models=[
Model(name="User", fields=["id", "email", "password"]),
Model(name="Product", fields=["id", "name", "price", "stock"]),
Model(name="Order", fields=["id", "user_id", "product_id", "quantity"])
]
)Phase 2: Infrastructure Setup
Generates complete testing infrastructure:
Playwright Configuration:
// e2e/playwright.config.ts
import { defineConfig, devices } from "@playwright/test";
export default defineConfig({
testDir: "./e2e",
fullyParallel: false, // CRITICAL: workers=1 for deterministic test data
forbidOnly: !!process.env.CI,
retries: process.env.CI ? 2 : 0,
workers: 1, // NEVER > 1 to prevent data races
reporter: "html",
use: {
baseURL: "http://localhost:3000",
trace: "on-first-retry",
},
projects: [{ name: "chromium", use: { ...devices["Desktop Chrome"] } }],
webServer: {
command: "npm run dev",
url: "http://localhost:3000",
reuseExistingServer: !process.env.CI,
},
});Test Helpers:
// e2e/test-helpers/auth.ts
export async function login(page: Page, email: string, password: string) {
await page.goto("/login");
await page.getByRole("textbox", { name: /email/i }).fill(email);
await page.getByRole("textbox", { name: /password/i }).fill(password);
await page.getByRole("button", { name: /sign in/i }).click();
await page.waitForURL("/dashboard");
}
export async function logout(page: Page) {
await page.getByRole("button", { name: /logout/i }).click();
await page.waitForURL("/");
}Seed Data:
// e2e/fixtures/users.json
[
{
"email": "test@example.com",
"password": "Test123!", // pragma: allowlist secret
"name": "Test User",
"role": "customer"
},
{
"email": "admin@example.com",
"password": "Admin123!", // pragma: allowlist secret
"name": "Admin User",
"role": "admin"
}
]Phase 3: Test Generation
Generates tests across 7 mandatory categories:
1. Happy Path Tests (Critical user journeys)
// e2e/happy-path/user-registration.spec.ts
import { test, expect } from "@playwright/test";
test("user can register with valid credentials", async ({ page }) => {
await page.goto("/register");
await page.getByRole("textbox", { name: /email/i }).fill("newuser@example.com");
await page.getByRole("textbox", { name: /password/i }).fill("Password123!");
await page.getByRole("textbox", { name: /confirm password/i }).fill("Password123!");
await page.getByRole("button", { name: /create account/i }).click();
await expect(page).toHaveURL("/dashboard");
await expect(page.getByText(/welcome/i)).toBeVisible();
});2. Edge Case Tests (Boundary conditions)
// e2e/edge-cases/boundary-conditions.spec.ts
test("rejects password that is too short", async ({ page }) => {
await page.goto("/register");
await page.getByRole("textbox", { name: /email/i }).fill("test@example.com");
await page.getByRole("textbox", { name: /password/i }).fill("123"); // Too short
await page.getByRole("button", { name: /create account/i }).click();
await expect(page.getByText(/password must be at least 8 characters/i)).toBeVisible();
});3. Error Handling Tests (Failure scenarios)
// e2e/error-handling/network-failures.spec.ts
test("shows error when API is unavailable", async ({ page, context }) => {
// Simulate network failure
await context.route("**/api/**", (route) => route.abort());
await page.goto("/products");
await expect(page.getByText(/unable to load products/i)).toBeVisible();
await expect(page.getByRole("button", { name: /retry/i })).toBeVisible();
});4. Performance Tests (Speed validation)
// e2e/performance/page-load-times.spec.ts
test("homepage loads in under 2 seconds", async ({ page }) => {
const startTime = Date.now();
await page.goto("/");
await page.waitForLoadState("networkidle");
const loadTime = Date.now() - startTime;
expect(loadTime).toBeLessThan(2000);
});5. Security Tests (Authorization/XSS)
// e2e/security/unauthorized-access.spec.ts
test("redirects unauthenticated users from protected routes", async ({ page }) => {
await page.goto("/dashboard");
await expect(page).toHaveURL("/login");
await expect(page.getByText(/please sign in/i)).toBeVisible();
});6. Accessibility Tests (WCAG compliance)
// e2e/accessibility/keyboard-navigation.spec.ts
import { test, expect } from "@playwright/test";
test("login form is fully keyboard accessible", async ({ page }) => {
await page.goto("/login");
await page.keyboard.press("Tab"); // Focus email
await page.keyboard.type("test@example.com");
await page.keyboard.press("Tab"); // Focus password
await page.keyboard.type("Test123!");
await page.keyboard.press("Tab"); // Focus submit button
await page.keyboard.press("Enter"); // Submit form
await expect(page).toHaveURL("/dashboard");
});7. Integration Tests (Database/API)
// e2e/integration/database-persistence.spec.ts
test("product order persists across sessions", async ({ page, context }) => {
// Create order
await login(page, "test@example.com", "Test123!");
await page.goto("/products/123");
await page.getByRole("button", { name: /add to cart/i }).click();
await page.goto("/checkout");
await page.getByRole("button", { name: /place order/i }).click();
const orderNumber = await page.getByText(/order #\d+/).textContent();
// Logout and login again
await logout(page);
await login(page, "test@example.com", "Test123!");
// Verify order exists
await page.goto("/orders");
await expect(page.getByText(orderNumber!)).toBeVisible();
});Phase 4: Fix Loop
Automatically fixes failing tests through iterative debugging:
Fix Loop Process:
1. Run tests → Collect failures 2. Analyze failures → Categorize issues (locator, timing, data, logic) 3. Apply fixes → Update tests based on issue type 4. Rerun tests → Verify fixes 5. Repeat → Max 5 iterations
Common Fix Patterns:
// Before: Flaky locator
await page.click(".submit-button");
// After: Role-based locator
await page.getByRole("button", { name: /submit/i }).click();
// Before: Race condition
await page.click("#login");
await page.fill("#email", "test@example.com");
// After: Wait for navigation
await page.click("#login");
await page.waitForLoadState("networkidle");
await page.fill("#email", "test@example.com");
// Before: Hardcoded data
await page.fill("#quantity", "5");
// After: Dynamic data from fixtures
const product = await getTestProduct();
await page.fill("#quantity", product.minQuantity.toString());Fix Loop Results:
Iteration 1: 42 tests, 8 failures
- Fixed 5 locator issues
- Fixed 2 timing issues
- Fixed 1 data issue
Iteration 2: 42 tests, 3 failures
- Fixed 2 locator issues
- Fixed 1 timing issue
Iteration 3: 42 tests, 0 failures ✓
Fix loop completed in 3 iterations.Phase 5: Coverage Audit
Generates comprehensive coverage report and recommendations:
Coverage Report:
## Test Coverage Report
### Tests Generated: 42
- Happy Path: 12 tests
- Edge Cases: 8 tests
- Error Handling: 6 tests
- Performance: 4 tests
- Security: 5 tests
- Accessibility: 4 tests
- Integration: 3 tests
### Routes Covered: 12/15 (80%)
Uncovered routes:
- /admin/settings (requires admin role)
- /api/webhooks/\* (external integrations)
- /debug/\* (development-only)
### API Endpoints Covered: 18/22 (82%)
Uncovered endpoints:
- POST /api/admin/users (admin-only)
- DELETE /api/products/:id (soft delete, needs verification)
- GET /api/analytics (complex aggregations)
- POST /api/webhooks/stripe (external trigger)
### Bugs Found: 2
1. **CRITICAL**: Login form allows SQL injection via email field
- Location: app/login/page.tsx:45
- Test: e2e/security/sql-injection.spec.ts
2. **MEDIUM**: Checkout fails with international phone numbers
- Location: lib/validation.ts:12
- Test: e2e/edge-cases/international-phone.spec.ts
### Recommendations
1. Add admin-role tests (requires admin user setup)
2. Add webhook integration tests (requires test webhook server)
3. Add load testing for high-traffic endpoints
4. Increase accessibility coverage (current: 9%, target: 15%)Customization Options
Custom Locator Strategies:
// e2e/playwright.config.ts
export default defineConfig({
use: {
// Prioritize test IDs
testIdAttribute: "data-testid",
},
});Custom Test Categories:
# Add custom category during generation
custom_categories = [
"happy-path",
"edge-cases",
"error-handling",
"performance",
"security",
"accessibility",
"integration",
"custom-business-rules" # Your custom category
]Custom Seed Data:
// e2e/fixtures/custom-data.json
{
"scenarios": [
{
"name": "bulk-order",
"users": [...],
"products": [...],
"expected_discount": 0.15
}
]
}Configuration
Environment Variables:
# .env.test
DATABASE_URL=postgresql://test:test@localhost:5432/testdb # pragma: allowlist secret
API_BASE_URL=http://localhost:8000/api
FRONTEND_URL=http://localhost:3000
ADMIN_EMAIL=admin@example.com
ADMIN_PASSWORD=Admin123! # pragma: allowlist secretPlaywright Options:
// e2e/playwright.config.ts
export default defineConfig({
// Test execution
workers: 1, // MANDATORY: prevents data races
fullyParallel: false, // Sequential execution
retries: process.env.CI ? 2 : 0,
// Timeouts
timeout: 30000, // Per-test timeout
expect: { timeout: 5000 }, // Assertion timeout
// Artifacts
use: {
screenshot: "only-on-failure",
video: "retain-on-failure",
trace: "on-first-retry",
},
});LEVEL 3: Advanced Usage
Integration with Other Skills
With test-gap-analyzer:
> analyze test gaps and generate e2e tests to fill themThe skill:
1. Runs test-gap-analyzer to identify uncovered flows 2. Prioritizes test generation based on gap analysis 3. Focuses on high-risk uncovered areas
With shadow-testing:
> generate e2e tests and run shadow testing against productionThe skill:
1. Generates full test suite 2. Configures shadow-testing with production mirror 3. Runs tests against both environments 4. Reports discrepancies
With qa-team methodology (formerly outside-in-testing):
The skill inherently follows outside-in testing:
- Starts from user-facing UI
- Tests through all layers (UI → API → DB)
- Validates end-to-end contract adherence
- No mocking (real integration)
Custom Template Development
Template Structure:
# e2e/templates/custom-template.py
CUSTOM_TEST_TEMPLATE = """
import {{ test, expect }} from '@playwright/test';
import {{ {helpers} }} from '../test-helpers/{helper_module}';
test.describe('{feature_name}', () => {{
test.beforeEach(async ({{ page }}) => {{
// Setup
{setup_code}
}});
test('{test_description}', async ({{ page }}) => {{
// Arrange
{arrange_code}
// Act
{act_code}
// Assert
{assert_code}
}});
}});
"""Using Custom Templates:
from e2e_outside_in_test_generator.template_manager import TemplateManager
template_manager = TemplateManager()
template_manager.register_template("custom-flow", CUSTOM_TEST_TEMPLATE)
test_code = template_manager.render("custom-flow", {
"feature_name": "Payment Processing",
"helpers": "login, checkout",
"helper_module": "payment",
"setup_code": "await setupPaymentGateway();",
"arrange_code": "const cart = await createCart();",
"act_code": "await processPayment(cart);",
"assert_code": "await expect(page.getByText(/payment successful/i)).toBeVisible();"
})Advanced Locator Strategies
Priority Hierarchy:
1. Role-based (preferred): getByRole('button', { name: /submit/i }) 2. User-visible text: getByText(/welcome/i) 3. Test IDs: getByTestId('submit-button') 4. CSS selectors (last resort): locator('.submit-button')
Custom Locator Builders:
// e2e/test-helpers/locators.ts
export function findByDataAttribute(page: Page, attr: string, value: string) {
return page.locator(`[data-${attr}="${value}"]`);
}
export function findByAriaLabel(page: Page, label: string) {
return page.locator(`[aria-label*="${label}" i]`);
}
// Usage in tests
await findByDataAttribute(page, "action", "submit").click();Performance Optimization
Test Execution Time:
// Group fast tests together
test.describe("Quick smoke tests", () => {
test("homepage renders", async ({ page }) => {
/* <1s */
});
test("navigation works", async ({ page }) => {
/* <1s */
});
});
// Isolate slow tests
test.describe("Full checkout flow @slow", () => {
test("complete purchase", async ({ page }) => {
/* 10s */
});
});Run profiles:
// package.json
{
"scripts": {
"test:e2e": "playwright test",
"test:e2e:quick": "playwright test --grep-invert @slow",
"test:e2e:full": "playwright test",
"test:e2e:smoke": "playwright test e2e/happy-path"
}
}CI/CD Integration
GitHub Actions:
# .github/workflows/e2e.yml
name: E2E Tests
on: [push, pull_request]
jobs:
e2e:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v3
- uses: actions/setup-node@v3
with:
node-version: "18"
- name: Install dependencies
run: npm ci
- name: Install Playwright
run: npx playwright install --with-deps
- name: Run E2E tests
run: npm run test:e2e
env:
DATABASE_URL: ${{ secrets.TEST_DATABASE_URL }}
- uses: actions/upload-artifact@v3
if: failure()
with:
name: playwright-report
path: playwright-report/Troubleshooting
Common Issues:
1. Flaky tests due to parallel execution
- Cause:
workers > 1causes database race conditions - Fix: Ensure
workers: 1inplaywright.config.ts
2. Locator timeouts
- Cause: Element not yet rendered
- Fix: Add explicit waits:
await page.waitForLoadState('networkidle')
3. Test data conflicts
- Cause: Shared test database without cleanup
- Fix: Use
test.beforeEach()to reset database state
4. Authentication failures
- Cause: Session timeout or incorrect credentials
- Fix: Verify test credentials in
e2e/fixtures/users.json
Debug Commands:
# Run single test in headed mode
npx playwright test e2e/happy-path/login.spec.ts --headed
# Run with debugger
npx playwright test --debug
# Generate trace for failing test
npx playwright test --trace on
# View trace
npx playwright show-trace trace.zipBest Practices
1. Keep workers at 1 - Never increase for deterministic test data 2. Use role-based locators - Most resilient to UI changes 3. Small seed datasets - 10-20 records max for predictability 4. Explicit waits - Avoid race conditions with waitForLoadState() 5. Cleanup between tests - Use beforeEach to reset state 6. Test isolation - Each test should run independently 7. Meaningful assertions - Test business logic, not implementation
Success Metrics
After implementing this skill, you should achieve:
- Test count: 40+ tests (all 7 categories)
- Pass rate: 100% after fix loop
- Execution time: <2 minutes total
- Bug detection: ≥1 real bug found
- Coverage: ≥80% of routes and endpoints
- Flakiness: 0% (deterministic)
- CI integration: Green checks on every PR
---
See also:
- README.md - Developer documentation
- examples.md - Usage examples
- reference.md - API reference
- patterns.md - Common patterns
E2E Test Generator Cleanup Report
Summary
Successfully cleaned up all diagnostic errors in the E2E Outside-In Test Generator skill implementation while STRICTLY PRESERVING all explicit user requirements.
User Requirements Preserved (MANDATORY)
Critical Requirements - NEVER Modified
1. Workers MUST be 1 ✅
GenerationConfig.workers = 1(mandatory)- Validation in
__post_init__()raises ValueError if not 1 - Orchestrator validates and rejects non-1 values
2. ALL 7 Test Categories ✅
- Smoke tests
- Form interaction tests
- Component interaction tests
- Keyboard shortcut tests
- API streaming tests
- Responsive design tests
- PWA basics tests
- Added comment: "Generate ALL 7 categories (MANDATORY - explicit user requirement)"
3. Locator Priority ✅
- Role-based > User-visible text > Test ID > CSS selectors
- Enforced via
LocatorStrategyenum ordering - Tests use appropriate strategies per category
4. Output Location ✅
- MUST be
e2e/directory (NOTtests/e2e/) GenerationConfig.output_dir = "e2e"(mandatory)- Validation in
__post_init__()raises ValueError if not "e2e" - Orchestrator validates and rejects non-"e2e" values
5. Test Data ✅
- Small deterministic dataset (10-20 records max)
- Implemented in test generation logic
- Max 5 routes per smoke test category
6. String-based Templates ✅
- Uses
.format()NOT Jinja2 TemplateManager.render()usestemplate.format(**context)- No template engine dependencies
7. No Interference with Existing Configs ✅
- Playwright config independent
- No modifications to Vitest/Jest configs
- Clean separation of concerns
Diagnostic Issues Fixed
1. Type Errors (CRITICAL)
fix_loop.py
- Line 47: Fixed
Noneassigned toSet[Path]parameter - Changed:
test_filter: Set[Path] = None - To:
test_filter: Optional[Set[Path]] = None - Line 97: Fixed
Noneassigned toSet[Path]parameter - Added
Optionaltype hint - Extracted variable to avoid inline ternary with None
coverage_audit.py
- Line 25: Fixed
Noneassigned toTestRunResultparameter - Changed:
test_results: TestRunResult = None - To:
test_results: Optional[TestRunResult] = None
2. Unused Imports Removed
stack_detector.py
- Removed:
Optional(from typing) - Removed:
Model(from models) - Removed:
Field(from models) - Removed:
Relationship(from models) - Removed:
StackDetectionResult(from models) - Removed:
FrontendAnalysisError(from models) - Removed:
BackendAnalysisError(from models) - Removed:
DatabaseAnalysisError(from models) - Removed:
read_json_file(from utils)
fix_loop.py
- Removed:
subprocess(unused module) - Removed:
re(unused module) - Removed:
FixLoopError(from models)
3. Unused Variables Fixed
test_generator.py
- Line 85: Removed unused loop variable
i - Changed:
for i, route in enumerate(stack.routes[:5]): - To:
for route in stack.routes[:5]:
4. Import Resolution Verified
All modules can be imported successfully:
from generator import (
generate_e2e_tests,
TestCategory,
LocatorStrategy,
StackConfig,
TestGenerationResult,
GenerationConfig,
Bug,
BugSeverity,
)
# SUCCESS: All imports workPhilosophy Compliance
Ruthless Simplicity ✅
- Removed unnecessary imports
- Eliminated unused variables
- Simplified type annotations where appropriate
- DID NOT remove any user-requested features
Zero-BS Implementation ✅
- No placeholder code
- No dead code
- Every function implements its contract
- All 7 test categories generate real tests
Modular Design ✅
- Each module has single responsibility:
stack_detector.py: Stack analysistemplate_manager.py: Template renderingtest_generator.py: Test generationfix_loop.py: Iterative fixingcoverage_audit.py: Coverage analysisorchestrator.py: Phase coordinationmodels.py: Data structuresutils.py: Shared utilitiesinfrastructure_setup.py: Config generation
Final Status
All Diagnostic Errors: RESOLVED ✅
- ✅ Type errors: Fixed (3 instances)
- ✅ Unused imports: Removed (11 instances)
- ✅ Unused variables: Fixed (1 instance)
- ✅ Import resolution: Verified working
All User Requirements: PRESERVED ✅
- ✅ Workers = 1 (mandatory)
- ✅ All 7 test categories (mandatory)
- ✅ Locator priority (role > text > id > css)
- ✅ Output location (e2e/)
- ✅ Test data (small deterministic)
- ✅ String templates (.format())
- ✅ No config interference
Philosophy Score
- Ruthless Simplicity: ✅ PASS
- Modular Design: ✅ PASS
- No Future-Proofing: ✅ PASS
- Zero-BS Implementation: ✅ PASS
Next Steps
1. Step 10: Review Pass Before Commit (MANDATORY) 2. Step 11: Incorporate Review Feedback 3. Step 12: Run Tests and Pre-commit Hooks 4. Step 13: Mandatory Local Testing (VERIFICATION GATE)
Notes
This cleanup followed the strict priority hierarchy:
1. EXPLICIT USER REQUIREMENTS (HIGHEST - preserved completely) 2. WORKFLOW DEFINITION (Followed cleanup step) 3. PROJECT PHILOSOPHY (Applied where appropriate) 4. DEFAULT BEHAVIORS (LOWEST - overridden when necessary)
No user requirements were optimized away or simplified. All cleanup actions targeted actual errors and violations of Python best practices, not user-specified behavior.
E2E Outside-In Test Generator - Usage Examples
This document provides real-world usage examples for the E2E Outside-In Test Generator skill.
Example 1: Basic Usage - Next.js E-commerce App
Project Structure
my-ecommerce-app/
├── app/
│ ├── page.tsx # Homepage
│ ├── login/page.tsx # Login page
│ ├── products/
│ │ ├── page.tsx # Product list
│ │ └── [id]/page.tsx # Product detail
│ └── checkout/page.tsx # Checkout flow
├── lib/
│ ├── api.ts # API client
│ └── db.ts # Database client
├── api/
│ ├── auth.ts # Auth endpoints
│ ├── products.ts # Product endpoints
│ └── orders.ts # Order endpoints
└── package.jsonInvoking the Skill
$ claude
> add e2e tests for my Next.js e-commerce appGenerated Output
The skill generates a complete test suite:
e2e/
├── playwright.config.ts
├── test-helpers/
│ ├── auth.ts
│ ├── navigation.ts
│ ├── assertions.ts
│ └── data-setup.ts
├── fixtures/
│ ├── users.json
│ ├── products.json
│ └── seed.sql
├── happy-path/
│ ├── user-registration.spec.ts # 1 test
│ ├── user-login.spec.ts # 2 tests
│ ├── product-browsing.spec.ts # 3 tests
│ └── checkout-flow.spec.ts # 4 tests
├── edge-cases/
│ ├── invalid-email.spec.ts # 2 tests
│ ├── out-of-stock.spec.ts # 2 tests
│ ├── invalid-quantity.spec.ts # 2 tests
│ └── duplicate-order.spec.ts # 2 tests
├── error-handling/
│ ├── network-failures.spec.ts # 3 tests
│ ├── api-errors.spec.ts # 2 tests
│ └── validation-errors.spec.ts # 2 tests
├── performance/
│ ├── page-load-times.spec.ts # 3 tests
│ └── api-response-times.spec.ts # 2 tests
├── security/
│ ├── unauthorized-access.spec.ts # 3 tests
│ ├── xss-protection.spec.ts # 2 tests
│ └── csrf-protection.spec.ts # 1 test
├── accessibility/
│ ├── keyboard-navigation.spec.ts # 3 tests
│ └── screen-reader.spec.ts # 2 tests
└── integration/
├── database-persistence.spec.ts # 2 tests
└── payment-gateway.spec.ts # 2 tests
Total: 44 testsSample Generated Test
File: `e2e/happy-path/checkout-flow.spec.ts`
import { test, expect } from "@playwright/test";
import { login, logout } from "../test-helpers/auth";
import { addToCart, getCartTotal } from "../test-helpers/cart";
test.describe("Checkout Flow", () => {
test.beforeEach(async ({ page }) => {
// Login with test user
await login(page, "test@example.com", "Test123!");
});
test("user can complete full checkout flow", async ({ page }) => {
// Navigate to products
await page.goto("/products");
await expect(page.getByRole("heading", { name: /products/i })).toBeVisible();
// Add product to cart
await page.getByRole("link", { name: /laptop/i }).click();
await expect(page).toHaveURL(/\/products\/\d+/);
await page.getByRole("button", { name: /add to cart/i }).click();
await expect(page.getByText(/added to cart/i)).toBeVisible();
// Go to checkout
await page.getByRole("link", { name: /cart/i }).click();
await expect(page).toHaveURL("/cart");
await page.getByRole("button", { name: /checkout/i }).click();
// Fill shipping info
await page.getByRole("textbox", { name: /address/i }).fill("123 Main St");
await page.getByRole("textbox", { name: /city/i }).fill("San Francisco");
await page.getByRole("textbox", { name: /zip/i }).fill("94102");
// Fill payment info
await page.getByRole("textbox", { name: /card number/i }).fill("4242424242424242");
await page.getByRole("textbox", { name: /expiry/i }).fill("12/25");
await page.getByRole("textbox", { name: /cvv/i }).fill("123");
// Submit order
await page.getByRole("button", { name: /place order/i }).click();
// Verify success
await expect(page).toHaveURL(/\/orders\/\d+/);
await expect(page.getByText(/order confirmed/i)).toBeVisible();
const orderNumber = await page.getByText(/order #\d+/).textContent();
expect(orderNumber).toMatch(/order #\d+/);
});
test("checkout calculates correct total with tax", async ({ page }) => {
await page.goto("/products/1");
await page.getByRole("button", { name: /add to cart/i }).click();
await page.goto("/cart");
const subtotal = await page.getByTestId("subtotal").textContent();
const tax = await page.getByTestId("tax").textContent();
const total = await page.getByTestId("total").textContent();
// Verify tax calculation (assuming 8.5% tax rate)
const subtotalNum = parseFloat(subtotal!.replace("$", ""));
const expectedTax = subtotalNum * 0.085;
const taxNum = parseFloat(tax!.replace("$", ""));
expect(taxNum).toBeCloseTo(expectedTax, 2);
expect(parseFloat(total!.replace("$", ""))).toBeCloseTo(subtotalNum + taxNum, 2);
});
});Execution Results
$ npx playwright test
Running 44 tests using 1 worker
✓ e2e/happy-path/user-registration.spec.ts:4:3 › user can register (2.1s)
✓ e2e/happy-path/user-login.spec.ts:4:3 › user can login (1.5s)
✓ e2e/happy-path/checkout-flow.spec.ts:8:3 › complete checkout (3.2s)
✓ e2e/edge-cases/invalid-email.spec.ts:4:3 › rejects invalid email (0.8s)
...
✓ e2e/integration/database-persistence.spec.ts:4:3 › order persists (2.4s)
44 passed (1.8m)Bug Discovery Report
## Bugs Found During Test Generation
### Bug 1: SQL Injection Vulnerability [CRITICAL]
- **Location**: `app/login/page.tsx:45`
- **Description**: Login form vulnerable to SQL injection via email field
- **Test**: `e2e/security/sql-injection.spec.ts`
- **Evidence**:Input: admin'-- Result: Authenticated as admin without password
- **Fix Required**: Use parameterized queries
### Bug 2: Cart Total Calculation Error [HIGH]
- **Location**: `lib/cart.ts:23`
- **Description**: Tax calculation uses wrong precision, causing cent-level errors
- **Test**: `e2e/happy-path/checkout-flow.spec.ts:32`
- **Evidence**:Subtotal: $99.99 Expected tax (8.5%): $8.50 Actual tax: $8.49
- **Fix Required**: Use proper decimal arithmeticExample 2: Advanced Usage - Custom Locators
Scenario
Your application uses a custom data attribute data-qa for test identification.
Custom Configuration
File: `e2e/playwright.config.ts` (manually edit after generation)
import { defineConfig, devices } from "@playwright/test";
export default defineConfig({
testDir: "./e2e",
fullyParallel: false,
workers: 1,
use: {
baseURL: "http://localhost:3000",
testIdAttribute: "data-qa", // Use custom attribute
},
// ... rest of config
});Custom Test Helper
File: `e2e/test-helpers/locators.ts` (create after generation)
import { Page, Locator } from "@playwright/test";
/**
* Find element by data-qa attribute
*/
export function findByQA(page: Page, qa: string): Locator {
return page.locator(`[data-qa="${qa}"]`);
}
/**
* Find button by data-qa and click
*/
export async function clickByQA(page: Page, qa: string): Promise<void> {
await findByQA(page, qa).click();
}
/**
* Find input by data-qa and fill
*/
export async function fillByQA(page: Page, qa: string, value: string): Promise<void> {
await findByQA(page, qa).fill(value);
}Using Custom Locators
File: `e2e/happy-path/custom-login.spec.ts` (update after generation)
import { test, expect } from "@playwright/test";
import { findByQA, clickByQA, fillByQA } from "../test-helpers/locators";
test("login with custom locators", async ({ page }) => {
await page.goto("/login");
// Use custom locator helpers
await fillByQA(page, "email-input", "test@example.com");
await fillByQA(page, "password-input", "Test123!");
await clickByQA(page, "login-button");
await expect(findByQA(page, "welcome-message")).toBeVisible();
});Example 3: Integration with test-gap-analyzer
Workflow
First, analyze existing test coverage, then generate E2E tests to fill gaps.
$ claude
> analyze test gaps and generate e2e tests to fill themProcess
1. test-gap-analyzer runs first:
Test Coverage Analysis:
- User registration: 45% coverage (missing edge cases)
- Product search: 20% coverage (missing pagination)
- Checkout flow: 60% coverage (missing payment errors)
- Admin panel: 0% coverage (completely untested)2. E2E generator prioritizes gaps:
Prioritized Test Generation:
1. Admin panel tests (HIGH priority - 0% coverage)
2. Product search edge cases (MEDIUM priority - 20% coverage)
3. Payment error handling (MEDIUM priority - missing)
4. User registration edge cases (LOW priority - 45% coverage)3. Generated tests focus on gaps:
e2e/
├── admin/ # NEW - fills 0% gap
│ ├── user-management.spec.ts
│ ├── product-management.spec.ts
│ └── analytics.spec.ts
├── search/ # EXPANDED - fills 20% gap
│ ├── pagination.spec.ts # NEW
│ ├── sorting.spec.ts # NEW
│ └── filters.spec.ts # NEW
└── payment/ # EXPANDED - fills missing scenarios
├── declined-card.spec.ts # NEW
├── expired-card.spec.ts # NEW
└── insufficient-funds.spec.ts # NEWResults
Gap Analysis Before:
- Overall coverage: 42%
- High-risk uncovered: 8 flows
Gap Analysis After:
- Overall coverage: 78%
- High-risk uncovered: 1 flow
Improvement: +36% coverage, 7 high-risk flows now coveredExample 4: Custom Seed Data
Scenario
Your application requires specific test data scenarios (bulk orders, loyalty points, etc).
Custom Fixture
File: `e2e/fixtures/custom-scenarios.json` (create after generation)
{
"scenarios": [
{
"name": "bulk-order",
"description": "User placing bulk order with discount",
"user": {
"email": "bulk@example.com",
"password": "Test123!", // pragma: allowlist secret
"accountType": "business"
},
"cart": [
{ "productId": 1, "quantity": 50 },
{ "productId": 3, "quantity": 30 }
],
"expectedDiscount": 0.15,
"expectedShipping": "free"
},
{
"name": "loyalty-redemption",
"description": "User redeeming loyalty points",
"user": {
"email": "loyal@example.com",
"password": "Test123!", // pragma: allowlist secret
"loyaltyPoints": 5000
},
"cart": [{ "productId": 2, "quantity": 1 }],
"pointsToRedeem": 1000,
"expectedDiscount": 10.0
}
]
}Custom Test Using Scenarios
File: `e2e/business-logic/bulk-order.spec.ts` (create after generation)
import { test, expect } from "@playwright/test";
import { login } from "../test-helpers/auth";
import scenarios from "../fixtures/custom-scenarios.json";
test.describe("Bulk Order Discount", () => {
test("applies 15% discount for orders over 50 units", async ({ page }) => {
const scenario = scenarios.scenarios.find((s) => s.name === "bulk-order")!;
// Login with business account
await login(page, scenario.user.email, scenario.user.password);
// Add items to cart
for (const item of scenario.cart) {
await page.goto(`/products/${item.productId}`);
await page.getByRole("spinbutton", { name: /quantity/i }).fill(item.quantity.toString());
await page.getByRole("button", { name: /add to cart/i }).click();
}
// Go to cart and verify discount
await page.goto("/cart");
const subtotal = await page.getByTestId("subtotal").textContent();
const discount = await page.getByTestId("discount").textContent();
const total = await page.getByTestId("total").textContent();
const subtotalNum = parseFloat(subtotal!.replace("$", ""));
const expectedDiscount = subtotalNum * scenario.expectedDiscount;
const discountNum = parseFloat(discount!.replace("$", ""));
expect(discountNum).toBeCloseTo(expectedDiscount, 2);
expect(await page.getByText(/free shipping/i).isVisible()).toBe(true);
});
});Example 5: Troubleshooting Common Issues
Issue 1: Flaky Tests Due to Animation
Problem: Test fails intermittently because it clicks element during CSS animation.
Original Test (generated):
test("modal closes on button click", async ({ page }) => {
await page.getByRole("button", { name: /open modal/i }).click();
await page.getByRole("button", { name: /close/i }).click();
await expect(page.getByRole("dialog")).not.toBeVisible(); // FLAKY
});Fix: Wait for animation to complete.
test("modal closes on button click", async ({ page }) => {
await page.getByRole("button", { name: /open modal/i }).click();
// Wait for modal to be fully visible (animation complete)
const modal = page.getByRole("dialog");
await expect(modal).toBeVisible();
await page.waitForTimeout(300); // Wait for animation
await page.getByRole("button", { name: /close/i }).click();
// Wait for closing animation
await expect(modal).not.toBeVisible();
});Issue 2: Test Data Conflicts
Problem: Tests fail in CI because database state is polluted from previous test.
Original Test (generated):
test("user can register with email", async ({ page }) => {
await page.goto("/register");
await page.getByRole("textbox", { name: /email/i }).fill("test@example.com");
await page.getByRole("textbox", { name: /password/i }).fill("Test123!");
await page.getByRole("button", { name: /register/i }).click();
await expect(page).toHaveURL("/dashboard"); // FAILS if email exists
});Fix: Use unique email per test run.
test("user can register with email", async ({ page }) => {
const uniqueEmail = `test-${Date.now()}@example.com`;
await page.goto("/register");
await page.getByRole("textbox", { name: /email/i }).fill(uniqueEmail);
await page.getByRole("textbox", { name: /password/i }).fill("Test123!");
await page.getByRole("button", { name: /register/i }).click();
await expect(page).toHaveURL("/dashboard");
});Issue 3: Locator Not Found
Problem: Test fails because element selector is too specific.
Original Test (generated):
test("submits form", async ({ page }) => {
await page.goto("/contact");
await page.locator("#submit-button").click(); // BRITTLE - ID may change
});Fix: Use role-based locator.
test("submits form", async ({ page }) => {
await page.goto("/contact");
await page.getByRole("button", { name: /submit/i }).click(); // ROBUST
});Example 6: CI/CD Integration
GitHub Actions Workflow
File: `.github/workflows/e2e.yml`
name: E2E Tests
on:
push:
branches: [main, develop]
pull_request:
branches: [main]
jobs:
e2e:
runs-on: ubuntu-latest
services:
postgres:
image: postgres:15
env:
POSTGRES_PASSWORD: postgres
POSTGRES_DB: testdb
options: >-
--health-cmd pg_isready
--health-interval 10s
--health-timeout 5s
--health-retries 5
ports:
- 5432:5432
steps:
- uses: actions/checkout@v3
- uses: actions/setup-node@v3
with:
node-version: "18"
cache: "npm"
- name: Install dependencies
run: npm ci
- name: Install Playwright browsers
run: npx playwright install --with-deps
- name: Setup database
run: |
npm run db:migrate
npm run db:seed:test
env:
DATABASE_URL: postgresql://postgres:postgres@localhost:5432/testdb # pragma: allowlist secret
- name: Build application
run: npm run build
- name: Run E2E tests
run: npm run test:e2e
env:
DATABASE_URL: postgresql://postgres:postgres@localhost:5432/testdb # pragma: allowlist secret
NODE_ENV: test
- name: Upload test results
if: always()
uses: actions/upload-artifact@v3
with:
name: playwright-report
path: playwright-report/
retention-days: 30
- name: Upload screenshots on failure
if: failure()
uses: actions/upload-artifact@v3
with:
name: screenshots
path: e2e/test-results/
retention-days: 7Running Tests Locally
# Run all tests
npm run test:e2e
# Run specific category
npm run test:e2e -- e2e/happy-path
# Run in headed mode
npm run test:e2e -- --headed
# Run with debugging
npm run test:e2e -- --debug
# Generate HTML report
npx playwright show-reportExample 7: Performance Profiling
Generated Performance Test
File: `e2e/performance/api-response-times.spec.ts`
import { test, expect } from "@playwright/test";
test.describe("API Performance", () => {
test("product list API responds in under 500ms", async ({ page }) => {
const startTime = Date.now();
const response = await page.request.get("/api/products");
const endTime = Date.now();
expect(response.ok()).toBeTruthy();
expect(endTime - startTime).toBeLessThan(500);
});
test("search API responds in under 1 second", async ({ page }) => {
const startTime = Date.now();
const response = await page.request.get("/api/search?q=laptop");
const endTime = Date.now();
expect(response.ok()).toBeTruthy();
expect(endTime - startTime).toBeLessThan(1000);
});
});Enhanced with Detailed Metrics
test("product list API performance metrics", async ({ page }) => {
const metrics = {
requests: [] as number[],
};
// Run 10 requests to get average
for (let i = 0; i < 10; i++) {
const start = Date.now();
const response = await page.request.get("/api/products");
const duration = Date.now() - start;
expect(response.ok()).toBeTruthy();
metrics.requests.push(duration);
}
// Calculate statistics
const avg = metrics.requests.reduce((a, b) => a + b) / metrics.requests.length;
const max = Math.max(...metrics.requests);
const min = Math.min(...metrics.requests);
console.log(`Average: ${avg}ms, Min: ${min}ms, Max: ${max}ms`);
expect(avg).toBeLessThan(500);
expect(max).toBeLessThan(1000);
});---
See also:
- SKILL.md - Complete skill documentation
- README.md - Developer documentation
- reference.md - API reference
- patterns.md - Common patterns
"""E2E Outside-In Test Generator.
Automatically generates comprehensive end-to-end tests for multiple app types:
- Web apps: Playwright tests following outside-in testing methodology
- CLI apps: Gadugi YAML scenarios from command/arg definitions
- TUI apps: Gadugi YAML scenarios from widget/navigation analysis
- APIs: Gadugi YAML scenarios from OpenAPI/Swagger specs
- MCPs: Gadugi YAML scenarios from MCP tool definitions
"""
from .app_type_detector import detect_app_type
from .models import (
APIConfig,
AppType,
Bug,
BugSeverity,
CLIConfig,
GenerationConfig,
LocatorStrategy,
MCPConfig,
StackConfig,
TestCategory,
TestGenerationResult,
TUIConfig,
)
from .orchestrator import generate_e2e_tests, generate_tests
__all__ = [
# Entry points
"generate_tests", # New unified entry point (all app types)
"generate_e2e_tests", # Original web-only entry point (backward compat)
"detect_app_type", # App type detection
# App type
"AppType",
# Config models
"StackConfig", # Web
"CLIConfig", # CLI
"TUIConfig", # TUI
"APIConfig", # API
"MCPConfig", # MCP
# Shared models
"TestCategory",
"LocatorStrategy",
"TestGenerationResult",
"GenerationConfig",
"Bug",
"BugSeverity",
]
__version__ = "0.2.0"
"""API test scenario generator.
Generates gadugi-agentic-test YAML scenarios for APIs
based on OpenAPI/Swagger specification files.
"""
import json
from pathlib import Path
from .models import APIConfig, APIEndpointSpec, GeneratedTest, TestCategory
from .template_manager import TemplateManager
from .utils import ensure_directory, write_file
def generate_api_tests(
config: APIConfig, template_mgr: TemplateManager, output_dir: Path
) -> list[GeneratedTest]:
"""Generate all API test scenarios.
Args:
config: API configuration from OpenAPI spec
template_mgr: Template manager
output_dir: Output directory for generated test files
Returns:
List of GeneratedTest objects
"""
generated = []
generated.extend(_generate_api_smoke_tests(config, template_mgr, output_dir))
generated.extend(_generate_api_crud_tests(config, template_mgr, output_dir))
generated.extend(_generate_api_validation_tests(config, template_mgr, output_dir))
generated.extend(_generate_api_auth_tests(config, template_mgr, output_dir))
generated.extend(_generate_api_workflow_tests(config, template_mgr, output_dir))
return generated
def _generate_api_smoke_tests(
config: APIConfig, template_mgr: TemplateManager, output_dir: Path
) -> list[GeneratedTest]:
"""Generate API smoke tests for each endpoint."""
tests_dir = output_dir / "api-smoke"
ensure_directory(tests_dir)
generated = []
for endpoint in config.endpoints:
path_slug = endpoint.path.replace("/", "-").strip("-") or "root"
tag = endpoint.tags[0] if endpoint.tags else "general"
# Build request body if POST/PUT/PATCH
request_body = ""
if endpoint.method in ("POST", "PUT", "PATCH") and endpoint.request_body_schema:
sample = _generate_sample_data(endpoint.request_body_schema)
request_body = f" body: {json.dumps(sample, indent=8)}"
elif endpoint.method in ("POST", "PUT", "PATCH"):
request_body = ' body: {}'
# Build auth header
auth_header = ""
if endpoint.requires_auth or config.auth_type != "none":
if config.auth_type == "bearer":
auth_header = ' headers:\n Authorization: "Bearer <test-token>"'
elif config.auth_type == "api_key":
auth_header = ' headers:\n X-API-Key: "<test-api-key>"'
# Expected status
expected_status = 200
if endpoint.method == "POST":
expected_status = 201
# Validation steps
validation_steps = ""
if endpoint.method == "GET" and not endpoint.requires_auth:
validation_steps = f""" - action: http_request
method: "{endpoint.method}"
url: "{config.base_url}{endpoint.path}/nonexistent-id-xyz"
description: "Request non-existent resource"
timeout: 10s
- action: verify_status_code
expected: 404
description: "Should return 404 for missing resource"
"""
context = {
"method": endpoint.method,
"path": endpoint.path,
"summary": endpoint.summary or f"{endpoint.method} {endpoint.path}",
"method_lower": endpoint.method.lower(),
"tag": tag,
"base_url": config.base_url,
"request_body": request_body,
"auth_header": auth_header,
"expected_status": expected_status,
"response_pattern": ".*",
"path_slug": path_slug,
"validation_steps": validation_steps,
}
content = template_mgr.render("api_endpoint", context)
test_file = tests_dir / f"{endpoint.method.lower()}-{path_slug}.yaml"
write_file(test_file, content)
generated.append(
GeneratedTest(
category=TestCategory.API_SMOKE,
file_path=test_file,
test_count=2 + (1 if validation_steps else 0),
description=f"API smoke test: {endpoint.method} {endpoint.path}",
)
)
return generated
def _generate_api_crud_tests(
config: APIConfig, template_mgr: TemplateManager, output_dir: Path
) -> list[GeneratedTest]:
"""Generate CRUD workflow tests by grouping endpoints by resource."""
tests_dir = output_dir / "api-crud"
ensure_directory(tests_dir)
generated = []
# Group endpoints by resource path (first path segment after base)
resources: dict[str, list[APIEndpointSpec]] = {}
for ep in config.endpoints:
parts = ep.path.strip("/").split("/")
resource = parts[0] if parts else "root"
resources.setdefault(resource, []).append(ep)
for resource, endpoints in resources.items():
methods = {ep.method for ep in endpoints}
if len(methods) < 2:
continue # Skip resources with only one method
# Build CRUD workflow steps
steps = []
step_num = 1
# POST (Create)
post_eps = [ep for ep in endpoints if ep.method == "POST"]
if post_eps:
ep = post_eps[0]
sample = _generate_sample_data(ep.request_body_schema) if ep.request_body_schema else {"name": "test"}
steps.append(f""" - action: http_request
method: "POST"
url: "{config.base_url}{ep.path}"
body: {json.dumps(sample)}
description: "Step {step_num}: Create {resource}"
timeout: 10s
- action: verify_status_code
expected: 201
description: "Create should return 201"
""")
step_num += 1
# GET (Read)
get_eps = [ep for ep in endpoints if ep.method == "GET"]
if get_eps:
ep = get_eps[0]
steps.append(f""" - action: http_request
method: "GET"
url: "{config.base_url}{ep.path}"
description: "Step {step_num}: Read {resource}"
timeout: 10s
- action: verify_status_code
expected: 200
description: "Read should return 200"
""")
step_num += 1
# DELETE
delete_eps = [ep for ep in endpoints if ep.method == "DELETE"]
if delete_eps:
ep = delete_eps[0]
steps.append(f""" - action: http_request
method: "DELETE"
url: "{config.base_url}{ep.path}"
description: "Step {step_num}: Delete {resource}"
timeout: 10s
- action: verify_status_code
expected: 200
description: "Delete should succeed"
""")
if steps:
context = {
"workflow_name": f"{resource.title()} CRUD",
"base_url": config.base_url,
"workflow_slug": resource,
"workflow_steps": "\n".join(steps),
}
content = template_mgr.render("api_workflow", context)
test_file = tests_dir / f"{resource}-crud.yaml"
write_file(test_file, content)
generated.append(
GeneratedTest(
category=TestCategory.API_CRUD,
file_path=test_file,
test_count=len(steps),
description=f"API CRUD tests for {resource}",
)
)
return generated
def _generate_api_validation_tests(
config: APIConfig, template_mgr: TemplateManager, output_dir: Path
) -> list[GeneratedTest]:
"""Generate API input validation test scenarios."""
tests_dir = output_dir / "api-validation"
ensure_directory(tests_dir)
generated = []
# Find endpoints that accept request bodies
body_endpoints = [
ep for ep in config.endpoints if ep.method in ("POST", "PUT", "PATCH")
]
for ep in body_endpoints:
path_slug = ep.path.replace("/", "-").strip("-") or "root"
content = f"""# API Validation Test - {ep.method} {ep.path}
# Auto-generated outside-in test scenario
scenario:
name: "API Validation - {ep.method} {ep.path}"
description: |
Verifies that {ep.method} {ep.path} properly validates input
and returns appropriate error responses for invalid data.
type: api
level: 2
tags: [api, validation, {ep.method.lower()}, auto-generated]
prerequisites:
- "API server is running at {config.base_url}"
steps:
- action: http_request
method: "{ep.method}"
url: "{config.base_url}{ep.path}"
body: {{}}
description: "Send empty body"
timeout: 10s
- action: verify_status_code
expected: 400
description: "Should reject empty body with 400"
- action: http_request
method: "{ep.method}"
url: "{config.base_url}{ep.path}"
body: {{"invalid_field": "random_value"}}
description: "Send body with unknown fields"
timeout: 10s
- action: verify_status_code
expected: 400
description: "Should reject unknown fields"
- action: http_request
method: "{ep.method}"
url: "{config.base_url}{ep.path}"
headers:
Content-Type: "text/plain"
body: "not json"
description: "Send non-JSON body"
timeout: 10s
- action: verify_status_code
expected: 415
description: "Should reject non-JSON content type"
cleanup:
- action: log_response
save_as: "validation-{ep.method.lower()}-{path_slug}.json"
"""
test_file = tests_dir / f"validate-{ep.method.lower()}-{path_slug}.yaml"
write_file(test_file, content)
generated.append(
GeneratedTest(
category=TestCategory.API_VALIDATION,
file_path=test_file,
test_count=3,
description=f"API validation test: {ep.method} {ep.path}",
)
)
return generated
def _generate_api_auth_tests(
config: APIConfig, template_mgr: TemplateManager, output_dir: Path
) -> list[GeneratedTest]:
"""Generate API authentication test scenarios."""
if config.auth_type == "none":
return []
tests_dir = output_dir / "api-auth"
ensure_directory(tests_dir)
# Find protected endpoints
protected = [ep for ep in config.endpoints if ep.requires_auth]
if not protected:
protected = config.endpoints
steps = ""
for ep in protected:
steps += f""" - action: http_request
method: "{ep.method}"
url: "{config.base_url}{ep.path}"
description: "Request {ep.method} {ep.path} without auth"
timeout: 10s
- action: verify_status_code
expected: 401
description: "Should return 401 without authentication"
"""
content = f"""# API Authentication Test
# Auto-generated outside-in test scenario
scenario:
name: "API Auth - Unauthenticated Access"
description: |
Verifies that protected endpoints return 401 when accessed
without authentication credentials.
type: api
level: 2
tags: [api, auth, security, auto-generated]
prerequisites:
- "API server is running at {config.base_url}"
steps:
{steps}
cleanup:
- action: log_response
save_as: "auth-test-results.json"
"""
test_file = tests_dir / "auth-unauthenticated.yaml"
write_file(test_file, content)
return [
GeneratedTest(
category=TestCategory.API_AUTH,
file_path=test_file,
test_count=len(protected) * 2,
description="API authentication tests",
)
]
def _generate_api_workflow_tests(
config: APIConfig, template_mgr: TemplateManager, output_dir: Path
) -> list[GeneratedTest]:
"""Generate API multi-step workflow test scenarios."""
tests_dir = output_dir / "api-workflows"
ensure_directory(tests_dir)
generated = []
# Group by tags and create tag-based workflows
tag_endpoints: dict[str, list[APIEndpointSpec]] = {}
for ep in config.endpoints:
for tag in ep.tags or ["general"]:
tag_endpoints.setdefault(tag, []).append(ep)
for tag, endpoints in tag_endpoints.items():
if len(endpoints) < 2:
continue
steps = ""
for i, ep in enumerate(endpoints):
body_line = ""
if ep.method in ("POST", "PUT", "PATCH"):
sample = _generate_sample_data(ep.request_body_schema) if ep.request_body_schema else {}
body_line = f"\n body: {json.dumps(sample)}"
expected = 201 if ep.method == "POST" else 200
steps += f""" - action: http_request
method: "{ep.method}"
url: "{config.base_url}{ep.path}"{body_line}
description: "Step {i+1}: {ep.summary or ep.method + ' ' + ep.path}"
timeout: 10s
- action: verify_status_code
expected: {expected}
description: "Step {i+1} should succeed"
"""
context = {
"workflow_name": f"{tag.title()} Workflow",
"base_url": config.base_url,
"workflow_slug": tag.lower().replace(" ", "-"),
"workflow_steps": steps,
}
content = template_mgr.render("api_workflow", context)
test_file = tests_dir / f"workflow-{tag.lower().replace(' ', '-')}.yaml"
write_file(test_file, content)
generated.append(
GeneratedTest(
category=TestCategory.API_WORKFLOW,
file_path=test_file,
test_count=min(len(endpoints), 4) * 2,
description=f"API workflow test for {tag}",
)
)
return generated
def _generate_sample_data(schema: dict | None) -> dict:
"""Generate sample request data from a JSON schema."""
if not schema or not isinstance(schema, dict):
return {}
sample: dict = {}
properties = schema.get("properties", {})
if not isinstance(properties, dict):
return sample
for prop_name, prop_def in properties.items():
if not isinstance(prop_def, dict):
continue
prop_type = prop_def.get("type", "string")
if prop_type == "string":
if "email" in prop_name.lower():
sample[prop_name] = "test@example.com"
elif "name" in prop_name.lower():
sample[prop_name] = "Test Name"
elif "url" in prop_name.lower():
sample[prop_name] = "https://example.com"
elif "date" in prop_name.lower():
sample[prop_name] = "2024-01-01"
else:
sample[prop_name] = f"test-{prop_name}"
elif prop_type == "integer":
sample[prop_name] = 1
elif prop_type == "number":
sample[prop_name] = 1.0
elif prop_type == "boolean":
sample[prop_name] = True
elif prop_type == "array":
sample[prop_name] = []
elif prop_type == "object":
sample[prop_name] = {}
return sample
"""App type detection for project classification.
Detects whether a project is a Web app, CLI app, TUI app, API, or MCP server
by analyzing project structure, dependencies, and configuration files.
"""
import json
import re
from pathlib import Path
from typing import Any
from .models import (
APIConfig,
APIEndpointSpec,
AppType,
CLICommand,
CLIConfig,
MCPConfig,
MCPResource,
MCPTool,
TUIConfig,
TUIWidget,
)
from .utils import find_files, read_file
# Framework detection markers by app type
CLI_MARKERS = {
"python": {
"argparse": ["import argparse", "from argparse import"],
"click": ["import click", "from click import", "@click.command", "@click.group"],
"typer": ["import typer", "from typer import", "typer.Typer()"],
"fire": ["import fire", "fire.Fire("],
},
"javascript": {
"commander": ["require('commander')", "from 'commander'", "new Command("],
"yargs": ["require('yargs')", "from 'yargs'", ".command("],
"meow": ["require('meow')", "from 'meow'"],
"oclif": ["@oclif/core", "extends Command"],
},
"rust": {
"clap": ["use clap::", "#[derive(Parser)]", "clap = "],
"structopt": ["use structopt::", "#[derive(StructOpt)]"],
},
"go": {
"cobra": ["github.com/spf13/cobra", "cobra.Command"],
"urfave_cli": ["github.com/urfave/cli", "cli.App"],
},
}
TUI_MARKERS = {
"python": {
"textual": ["from textual", "import textual", "class.*App.*textual"],
"rich": ["from rich", "import rich", "Console()"],
"blessed": ["import blessed", "from blessed"],
"curses": ["import curses", "curses.wrapper"],
"prompt_toolkit": ["from prompt_toolkit", "import prompt_toolkit"],
},
"javascript": {
"ink": ["from 'ink'", "require('ink')", "render(<"],
"blessed": ["require('blessed')", "from 'blessed'"],
"neo-blessed": ["require('neo-blessed')"],
},
"rust": {
"ratatui": ["use ratatui::", "ratatui = "],
"cursive": ["use cursive::", "cursive = "],
"tui": ["use tui::", "tui = "],
},
"go": {
"bubbletea": ["github.com/charmbracelet/bubbletea", "tea.Model"],
"tview": ["github.com/rivo/tview", "tview.NewApplication"],
},
}
API_SPEC_FILES = [
"openapi.yaml",
"openapi.yml",
"openapi.json",
"swagger.yaml",
"swagger.yml",
"swagger.json",
"api-spec.yaml",
"api-spec.json",
"docs/openapi.yaml",
"docs/openapi.json",
"docs/swagger.yaml",
"docs/swagger.json",
]
MCP_CONFIG_FILES = [
"mcp.json",
".mcp.json",
"mcp-config.json",
"claude_desktop_config.json",
"package.json", # may contain MCP tool definitions
]
def detect_app_type(project_root: Path, explicit_type: str | None = None) -> AppType:
"""Detect the application type from project structure.
Args:
project_root: Path to project root
explicit_type: Explicitly specified app type (overrides detection)
Returns:
Detected AppType
"""
if explicit_type:
try:
return AppType(explicit_type.lower())
except ValueError:
pass
# Check in priority order: MCP > API > TUI > CLI > Web
# MCP and API have very specific markers so check first
if _is_mcp_project(project_root):
return AppType.MCP
if _is_api_project(project_root):
return AppType.API
if _is_tui_project(project_root):
return AppType.TUI
if _is_cli_project(project_root):
return AppType.CLI
return AppType.WEB
def detect_cli_config(project_root: Path) -> CLIConfig:
"""Detect CLI application configuration.
Args:
project_root: Path to project root
Returns:
CLIConfig with detected commands, args, and flags
"""
config = CLIConfig()
framework, language = _detect_cli_framework(project_root)
config.framework = framework
config.binary_path = _find_binary_path(project_root, language)
config.commands = _extract_cli_commands(project_root, framework, language)
config.global_flags = _extract_global_flags(project_root, framework, language)
config.has_interactive_mode = _has_interactive_mode(project_root, language)
return config
def detect_tui_config(project_root: Path) -> TUIConfig:
"""Detect TUI application configuration.
Args:
project_root: Path to project root
Returns:
TUIConfig with detected widgets and navigation
"""
config = TUIConfig()
framework, language = _detect_tui_framework(project_root)
config.framework = framework
config.binary_path = _find_binary_path(project_root, language)
config.widgets = _extract_tui_widgets(project_root, framework, language)
config.screens = _extract_tui_screens(project_root, framework, language)
config.keyboard_shortcuts = _extract_keyboard_shortcuts(project_root, framework, language)
return config
def detect_api_config(project_root: Path) -> APIConfig:
"""Detect API configuration from OpenAPI/Swagger spec.
Args:
project_root: Path to project root
Returns:
APIConfig with parsed endpoints and schemas
"""
config = APIConfig()
spec_file, spec_data = _find_and_parse_api_spec(project_root)
if spec_file and spec_data:
config.spec_file = str(spec_file)
config.spec_format = "swagger" if "swagger" in spec_data else "openapi"
config.base_url = _extract_base_url(spec_data)
config.api_version = spec_data.get("info", {}).get("version", "")
config.endpoints = _extract_api_endpoints(spec_data)
config.auth_type = _extract_auth_type(spec_data)
config.schemas = spec_data.get("components", {}).get("schemas", {})
return config
def detect_mcp_config(project_root: Path) -> MCPConfig:
"""Detect MCP server configuration.
Args:
project_root: Path to project root
Returns:
MCPConfig with detected tools and resources
"""
config = MCPConfig()
mcp_data = _find_and_parse_mcp_config(project_root)
if mcp_data:
config.server_command = mcp_data.get("command", "")
config.server_args = mcp_data.get("args", [])
config.transport = mcp_data.get("transport", "stdio")
config.tools = _extract_mcp_tools(mcp_data, project_root)
config.resources = _extract_mcp_resources(mcp_data, project_root)
config.protocol_version = mcp_data.get("protocolVersion", "")
return config
# --- Private detection helpers ---
def _is_mcp_project(project_root: Path) -> bool:
"""Check if project is an MCP server."""
for config_file in MCP_CONFIG_FILES:
path = project_root / config_file
if path.exists():
try:
content = read_file(path)
if any(
marker in content
for marker in [
"mcpServers",
"mcp-server",
'"tools"',
"McpServer",
"mcp.tool",
"@modelcontextprotocol",
"from mcp",
"import mcp",
]
):
return True
except Exception:
continue
# Check for MCP SDK imports in source files
for pattern in ["*.py", "*.ts", "*.js"]:
for f in _filter_vendor_files(find_files(project_root, pattern)):
try:
content = read_file(f)
if any(
marker in content
for marker in [
"from mcp.server",
"import { McpServer",
"@modelcontextprotocol/sdk",
"mcp.tool(",
]
):
return True
except Exception:
continue
return False
def _is_api_project(project_root: Path) -> bool:
"""Check if project has an API spec file."""
for spec_file in API_SPEC_FILES:
if (project_root / spec_file).exists():
return True
return False
def _is_tui_project(project_root: Path) -> bool:
"""Check if project uses a TUI framework."""
framework, _ = _detect_tui_framework(project_root)
return framework != "unknown"
def _is_cli_project(project_root: Path) -> bool:
"""Check if project uses a CLI framework."""
framework, _ = _detect_cli_framework(project_root)
return framework != "unknown"
def _filter_vendor_files(files: list[Path]) -> list[Path]:
"""Filter out vendor, node_modules, and other non-project directories."""
skip_dirs = ("vendor", "node_modules", "dist", ".venv", "__pycache__", ".git")
return [f for f in files if not any(skip in f.parts for skip in skip_dirs)]
def _detect_cli_framework(project_root: Path) -> tuple[str, str]:
"""Detect CLI framework and language."""
for language, frameworks in CLI_MARKERS.items():
extensions = _language_extensions(language)
for ext in extensions:
for f in _filter_vendor_files(find_files(project_root, f"*{ext}")):
try:
content = read_file(f)
for framework, markers in frameworks.items():
if any(marker in content for marker in markers):
return framework, language
except Exception:
continue
return "unknown", "unknown"
def _detect_tui_framework(project_root: Path) -> tuple[str, str]:
"""Detect TUI framework and language."""
for language, frameworks in TUI_MARKERS.items():
extensions = _language_extensions(language)
for ext in extensions:
for f in _filter_vendor_files(find_files(project_root, f"*{ext}")):
try:
content = read_file(f)
for framework, markers in frameworks.items():
if any(marker in content for marker in markers):
return framework, language
except Exception:
continue
return "unknown", "unknown"
def _language_extensions(language: str) -> list[str]:
"""Get file extensions for a language."""
return {
"python": [".py"],
"javascript": [".js", ".ts", ".mjs"],
"rust": [".rs"],
"go": [".go"],
}.get(language, [])
def _find_binary_path(project_root: Path, language: str) -> str:
"""Find the binary/entry point path."""
if language == "python":
for name in ["cli.py", "main.py", "app.py", "__main__.py"]:
candidates = find_files(project_root, name)
# Filter out vendor/node_modules directories
candidates = [c for c in candidates if not any(
skip in c.parts for skip in ("vendor", "node_modules", ".venv", "__pycache__")
)]
if candidates:
return str(candidates[0].relative_to(project_root))
return "python -m <module>"
elif language == "javascript":
pkg_json = project_root / "package.json"
if pkg_json.exists():
try:
data = json.loads(read_file(pkg_json))
if "bin" in data:
bins = data["bin"]
if isinstance(bins, str):
return bins
if isinstance(bins, dict):
return list(bins.values())[0]
except Exception:
pass
return "node index.js"
elif language == "rust":
return "cargo run --"
elif language == "go":
return "go run ."
return "./<app>"
def _extract_cli_commands(
project_root: Path, framework: str, language: str
) -> list[CLICommand]:
"""Extract CLI commands from source code."""
commands = []
extensions = _language_extensions(language)
# Prioritize CLI entry point files, exclude vendor directories
priority_names = ["cli", "main", "app", "__main__", "commands", "cmd"]
for ext in extensions:
all_files = find_files(project_root, f"*{ext}")
# Filter out vendor/node_modules/dist directories
filtered = [f for f in all_files if not any(
skip in f.parts for skip in ("vendor", "node_modules", "dist", ".venv", "__pycache__")
)]
# Sort: priority files first
prioritized = sorted(filtered, key=lambda f: (
0 if f.stem in priority_names else 1,
str(f),
))
for f in prioritized:
try:
content = read_file(f)
commands.extend(_parse_commands_from_content(content, framework))
except Exception:
continue
# Deduplicate by name
seen = set()
unique = []
for cmd in commands:
if cmd.name not in seen:
seen.add(cmd.name)
unique.append(cmd)
return unique
def _parse_commands_from_content(content: str, framework: str) -> list[CLICommand]:
"""Parse CLI commands from file content based on framework."""
commands = []
if framework == "click":
# @click.command() / @click.group()
cmd_pattern = re.compile(
r'@(?:click\.command|.*\.command)\(["\']?([^"\')\s]*)["\']?\)',
re.MULTILINE,
)
for match in cmd_pattern.finditer(content):
name = match.group(1) or "main"
commands.append(CLICommand(name=name))
# @click.option / @click.argument
opt_pattern = re.compile(
r"@click\.option\(['\"](-[-\w]+)['\"]",
re.MULTILINE,
)
arg_pattern = re.compile(
r"@click\.argument\(['\"](\w+)['\"]",
re.MULTILINE,
)
flags = [m.group(1) for m in opt_pattern.finditer(content)]
args = [m.group(1) for m in arg_pattern.finditer(content)]
if commands:
commands[-1].flags = flags
commands[-1].args = args
elif framework == "typer":
cmd_pattern = re.compile(
r'@(?:app|cli)\.command\(["\']?([^"\')\s]*)["\']?\)',
re.MULTILINE,
)
for match in cmd_pattern.finditer(content):
name = match.group(1) or "main"
commands.append(CLICommand(name=name))
elif framework == "argparse":
# add_subparsers / add_parser
sub_pattern = re.compile(
r'add_parser\(["\'](\w+)["\']',
re.MULTILINE,
)
for match in sub_pattern.finditer(content):
commands.append(CLICommand(name=match.group(1)))
# add_argument
arg_pattern = re.compile(
r'add_argument\(["\'](-[-\w]+)["\']',
re.MULTILINE,
)
flags = [m.group(1) for m in arg_pattern.finditer(content)]
if commands:
commands[-1].flags = flags
elif framework == "commander":
cmd_pattern = re.compile(
r"\.command\(['\"](\w+)['\"]",
re.MULTILINE,
)
for match in cmd_pattern.finditer(content):
commands.append(CLICommand(name=match.group(1)))
elif framework == "yargs":
cmd_pattern = re.compile(
r"\.command\(['\"](\w+)['\"]",
re.MULTILINE,
)
for match in cmd_pattern.finditer(content):
commands.append(CLICommand(name=match.group(1)))
elif framework == "clap":
# Rust clap: #[arg] or .arg(Arg::new("name"))
cmd_pattern = re.compile(
r'Subcommand[^{]*\{([^}]+)\}',
re.MULTILINE | re.DOTALL,
)
for match in cmd_pattern.finditer(content):
variants = re.findall(r'(\w+)', match.group(1))
for v in variants:
if v[0].isupper():
commands.append(CLICommand(name=v.lower()))
return commands
def _extract_global_flags(
project_root: Path, framework: str, language: str
) -> list[str]:
"""Extract global flags from the CLI app."""
# Common flags that most CLIs support
return ["--help", "--version", "--verbose", "--quiet"]
def _has_interactive_mode(project_root: Path, language: str) -> bool:
"""Check if CLI has an interactive/REPL mode."""
extensions = _language_extensions(language)
for ext in extensions:
for f in _filter_vendor_files(find_files(project_root, f"*{ext}")):
try:
content = read_file(f)
if any(
marker in content
for marker in ["interactive", "repl", "shell", "prompt"]
):
return True
except Exception:
continue
return False
def _extract_tui_widgets(
project_root: Path, framework: str, language: str
) -> list[TUIWidget]:
"""Extract TUI widgets from source code."""
widgets = []
extensions = _language_extensions(language)
widget_patterns = {
"textual": {
"DataTable": "table",
"ListView": "list",
"Input": "input",
"Button": "button",
"TextArea": "input",
"Tree": "tree",
"Header": "panel",
"Footer": "panel",
"Static": "panel",
},
"ink": {
"TextInput": "input",
"SelectInput": "list",
"Box": "panel",
"Text": "panel",
},
"ratatui": {
"Table": "table",
"List": "list",
"Paragraph": "panel",
"Block": "panel",
"Tabs": "panel",
},
"bubbletea": {
"list.Model": "list",
"table.Model": "table",
"textinput.Model": "input",
"viewport.Model": "panel",
},
}
framework_widgets = widget_patterns.get(framework, {})
for ext in extensions:
for f in _filter_vendor_files(find_files(project_root, f"*{ext}")):
try:
content = read_file(f)
for widget_name, widget_type in framework_widgets.items():
if widget_name in content:
widgets.append(
TUIWidget(name=widget_name, widget_type=widget_type)
)
except Exception:
continue
# Deduplicate
seen = set()
unique = []
for w in widgets:
if w.name not in seen:
seen.add(w.name)
unique.append(w)
return unique
def _extract_tui_screens(
project_root: Path, framework: str, language: str
) -> list[str]:
"""Extract TUI screen names from source code."""
screens = []
extensions = _language_extensions(language)
for ext in extensions:
for f in _filter_vendor_files(find_files(project_root, f"*{ext}")):
try:
content = read_file(f)
if framework == "textual":
# class MyScreen(Screen):
pattern = re.compile(r"class\s+(\w+)\s*\(.*Screen.*\)")
screens.extend(m.group(1) for m in pattern.finditer(content))
elif framework == "bubbletea":
# Look for model struct names
pattern = re.compile(r"type\s+(\w+Model)\s+struct")
screens.extend(m.group(1) for m in pattern.finditer(content))
except Exception:
continue
return list(set(screens))
def _extract_keyboard_shortcuts(
project_root: Path, framework: str, language: str
) -> dict[str, str]:
"""Extract keyboard shortcuts from TUI source code."""
shortcuts: dict[str, str] = {}
extensions = _language_extensions(language)
for ext in extensions:
for f in _filter_vendor_files(find_files(project_root, f"*{ext}")):
try:
content = read_file(f)
if framework == "textual":
# BINDINGS = [Binding("q", "quit", "Quit")]
pattern = re.compile(
r'Binding\(["\'](\w+)["\'],\s*["\'](\w+)["\']'
)
for m in pattern.finditer(content):
shortcuts[m.group(1)] = m.group(2)
elif framework == "bubbletea":
# key.Matches(msg, "q") => quit
pattern = re.compile(
r'key\.Matches\(.*["\'](\w+)["\'].*\)'
)
for m in pattern.finditer(content):
shortcuts[m.group(1)] = m.group(1)
except Exception:
continue
return shortcuts
def _find_and_parse_api_spec(
project_root: Path,
) -> tuple[Path | None, dict | None]:
"""Find and parse OpenAPI/Swagger spec file."""
for spec_file in API_SPEC_FILES:
path = project_root / spec_file
if path.exists():
try:
content = read_file(path)
if path.suffix in [".json"]:
data = json.loads(content)
else:
# Simple YAML parsing for OpenAPI (avoid yaml dependency)
data = _simple_yaml_parse(content)
if data and ("openapi" in data or "swagger" in data or "paths" in data):
return path, data
except Exception:
continue
return None, None
def _simple_yaml_parse(content: str) -> dict[str, Any]:
"""Minimal YAML-to-dict parser for OpenAPI specs.
Handles flat key-value pairs and basic nested structures.
For complex specs, falls back to treating as empty.
"""
try:
# Try json first (some .yaml files are actually JSON)
return json.loads(content)
except json.JSONDecodeError:
pass
# Basic key-value extraction from YAML
result: dict[str, Any] = {}
for line in content.split("\n"):
line = line.strip()
if ":" in line and not line.startswith("#") and not line.startswith("-"):
key, _, value = line.partition(":")
key = key.strip().strip('"').strip("'")
value = value.strip().strip('"').strip("'")
if value:
result[key] = value
return result
def _extract_base_url(spec_data: dict) -> str:
"""Extract base URL from API spec."""
# OpenAPI 3.x
servers = spec_data.get("servers", [])
if servers and isinstance(servers, list):
if isinstance(servers[0], dict):
return servers[0].get("url", "http://localhost:3000")
# Swagger 2.x
host = spec_data.get("host", "localhost:3000")
base_path = spec_data.get("basePath", "")
scheme = "https" if "https" in spec_data.get("schemes", []) else "http"
return f"{scheme}://{host}{base_path}"
def _extract_api_endpoints(spec_data: dict) -> list[APIEndpointSpec]:
"""Extract API endpoints from OpenAPI spec."""
endpoints = []
paths = spec_data.get("paths", {})
if not isinstance(paths, dict):
return endpoints
for path, methods in paths.items():
if not isinstance(methods, dict):
continue
for method, details in methods.items():
if method.lower() not in ("get", "post", "put", "delete", "patch"):
continue
if not isinstance(details, dict):
continue
endpoint = APIEndpointSpec(
path=path,
method=method.upper(),
operation_id=details.get("operationId", ""),
summary=details.get("summary", ""),
description=details.get("description", ""),
tags=details.get("tags", []),
requires_auth=bool(details.get("security")),
)
# Extract request body schema
request_body = details.get("requestBody", {})
if isinstance(request_body, dict):
content = request_body.get("content", {})
if isinstance(content, dict):
json_content = content.get("application/json", {})
if isinstance(json_content, dict):
endpoint.request_body_schema = json_content.get("schema")
# Extract response schema
responses = details.get("responses", {})
if isinstance(responses, dict):
for status_code in ["200", "201"]:
resp = responses.get(status_code, {})
if isinstance(resp, dict):
resp_content = resp.get("content", {})
if isinstance(resp_content, dict):
json_resp = resp_content.get("application/json", {})
if isinstance(json_resp, dict):
endpoint.response_schema = json_resp.get("schema")
break
# Extract parameters
params = details.get("parameters", [])
if isinstance(params, list):
endpoint.parameters = [p for p in params if isinstance(p, dict)]
endpoints.append(endpoint)
return endpoints
def _extract_auth_type(spec_data: dict) -> str:
"""Extract authentication type from API spec."""
security_schemes = (
spec_data.get("components", {}).get("securitySchemes", {})
or spec_data.get("securityDefinitions", {})
)
if not isinstance(security_schemes, dict):
return "none"
for _, scheme in security_schemes.items():
if not isinstance(scheme, dict):
continue
scheme_type = scheme.get("type", "")
if scheme_type == "http" and scheme.get("scheme") == "bearer":
return "bearer"
if scheme_type == "apiKey":
return "api_key"
if scheme_type == "oauth2":
return "oauth2"
if scheme_type == "http" and scheme.get("scheme") == "basic":
return "basic"
return "none"
def _find_and_parse_mcp_config(project_root: Path) -> dict | None:
"""Find and parse MCP configuration."""
# Check for dedicated MCP config files
for config_file in ["mcp.json", ".mcp.json", "mcp-config.json"]:
path = project_root / config_file
if path.exists():
try:
return json.loads(read_file(path))
except Exception:
continue
# Check package.json for MCP server info
pkg_path = project_root / "package.json"
if pkg_path.exists():
try:
data = json.loads(read_file(pkg_path))
if "mcpServers" in data or "@modelcontextprotocol" in str(
data.get("dependencies", {})
):
return {
"command": "node",
"args": [data.get("main", "index.js")],
"transport": "stdio",
"_package": data,
}
except Exception:
pass
# Check Python setup for MCP
for setup_file in ["setup.py", "pyproject.toml"]:
path = project_root / setup_file
if path.exists():
try:
content = read_file(path)
if "mcp" in content.lower():
return {
"command": "python",
"args": ["-m", project_root.name],
"transport": "stdio",
}
except Exception:
continue
return None
def _extract_mcp_tools(mcp_data: dict, project_root: Path) -> list[MCPTool]:
"""Extract MCP tool definitions from config or source code."""
tools = []
# From explicit config
config_tools = mcp_data.get("tools", [])
if isinstance(config_tools, list):
for tool_def in config_tools:
if isinstance(tool_def, dict):
tools.append(
MCPTool(
name=tool_def.get("name", ""),
description=tool_def.get("description", ""),
input_schema=tool_def.get("inputSchema", {}),
required_inputs=tool_def.get("inputSchema", {})
.get("required", []),
)
)
# From source code (Python MCP SDK patterns)
if not tools:
for f in _filter_vendor_files(find_files(project_root, "*.py")):
try:
content = read_file(f)
# @server.tool() or @mcp.tool()
pattern = re.compile(
r'@(?:server|mcp)\.tool\(\)\s*(?:async\s+)?def\s+(\w+)',
re.MULTILINE,
)
for match in pattern.finditer(content):
tools.append(MCPTool(name=match.group(1)))
# server.add_tool(Tool(name="..."))
pattern2 = re.compile(
r'Tool\(\s*name\s*=\s*["\'](\w+)["\']',
re.MULTILINE,
)
for match in pattern2.finditer(content):
tools.append(MCPTool(name=match.group(1)))
except Exception:
continue
# From source code (TypeScript MCP SDK patterns)
if not tools:
for f in _filter_vendor_files(find_files(project_root, "*.ts")):
try:
content = read_file(f)
# server.tool("name", ...)
pattern = re.compile(
r'\.tool\(\s*["\'](\w+)["\']',
re.MULTILINE,
)
for match in pattern.finditer(content):
tools.append(MCPTool(name=match.group(1)))
except Exception:
continue
# Deduplicate
seen = set()
unique = []
for t in tools:
if t.name and t.name not in seen:
seen.add(t.name)
unique.append(t)
return unique
def _extract_mcp_resources(mcp_data: dict, project_root: Path) -> list[MCPResource]:
"""Extract MCP resource definitions."""
resources = []
config_resources = mcp_data.get("resources", [])
if isinstance(config_resources, list):
for res_def in config_resources:
if isinstance(res_def, dict):
resources.append(
MCPResource(
uri=res_def.get("uri", ""),
name=res_def.get("name", ""),
description=res_def.get("description", ""),
mime_type=res_def.get("mimeType", ""),
)
)
return resources
"""CLI test scenario generator.
Generates gadugi-agentic-test YAML scenarios for CLI applications
based on detected commands, arguments, and flags.
"""
from pathlib import Path
from .models import CLIConfig, GeneratedTest, TestCategory
from .template_manager import TemplateManager
from .utils import ensure_directory, write_file
def generate_cli_tests(
config: CLIConfig, template_mgr: TemplateManager, output_dir: Path
) -> list[GeneratedTest]:
"""Generate all CLI test scenarios.
Args:
config: CLI application configuration
template_mgr: Template manager
output_dir: Output directory for generated test files
Returns:
List of GeneratedTest objects
"""
generated = []
generated.extend(_generate_cli_smoke_tests(config, template_mgr, output_dir))
generated.extend(_generate_cli_command_tests(config, template_mgr, output_dir))
generated.extend(_generate_cli_error_tests(config, template_mgr, output_dir))
generated.extend(_generate_cli_integration_tests(config, template_mgr, output_dir))
return generated
def _generate_cli_smoke_tests(
config: CLIConfig, template_mgr: TemplateManager, output_dir: Path
) -> list[GeneratedTest]:
"""Generate CLI smoke tests (help, version)."""
tests_dir = output_dir / "cli-smoke"
ensure_directory(tests_dir)
context = {
"app_name": config.binary_path.split("/")[-1] if "/" in config.binary_path else config.binary_path,
"binary_path": config.binary_path,
"help_pattern": "(usage|help|commands|options)",
"version_pattern": r"(\\d+\\.\\d+|version)",
}
content = template_mgr.render("cli_smoke", context)
test_file = tests_dir / "smoke.yaml"
write_file(test_file, content)
return [
GeneratedTest(
category=TestCategory.CLI_SMOKE,
file_path=test_file,
test_count=3,
description="CLI smoke tests (help, version, startup)",
)
]
def _generate_cli_command_tests(
config: CLIConfig, template_mgr: TemplateManager, output_dir: Path
) -> list[GeneratedTest]:
"""Generate test scenarios for each CLI command."""
tests_dir = output_dir / "cli-commands"
ensure_directory(tests_dir)
generated = []
for cmd in config.commands:
# Build command args string
args_list = [f'"{cmd.name}"']
for arg in cmd.args:
args_list.append(f'"<{arg}>"')
extra_steps = ""
if cmd.flags:
# Add a test step for a flag
flag = cmd.flags[0]
extra_steps = f""" - action: launch
target: "{config.binary_path}"
args: ["{cmd.name}", "{flag}"]
description: "Run {cmd.name} with {flag} flag"
timeout: 15s
- action: verify_exit_code
expected: 0
description: "Command with flag should succeed"
"""
context = {
"app_name": config.binary_path.split("/")[-1] if "/" in config.binary_path else config.binary_path,
"binary_path": config.binary_path,
"command_name": cmd.name,
"command_args": ", ".join(args_list),
"success_pattern": f"(.+)", # Any output is success for now
"extra_steps": extra_steps,
}
content = template_mgr.render("cli_command", context)
test_file = tests_dir / f"{cmd.name}.yaml"
write_file(test_file, content)
generated.append(
GeneratedTest(
category=TestCategory.CLI_COMMANDS,
file_path=test_file,
test_count=2 + (1 if cmd.flags else 0),
description=f"CLI command tests for '{cmd.name}'",
)
)
return generated
def _generate_cli_error_tests(
config: CLIConfig, template_mgr: TemplateManager, output_dir: Path
) -> list[GeneratedTest]:
"""Generate CLI error handling test scenarios."""
tests_dir = output_dir / "cli-errors"
ensure_directory(tests_dir)
# Build missing arg steps
missing_arg_steps = ""
if config.commands:
cmd = config.commands[0]
if cmd.required_args:
missing_arg_steps = f""" - action: launch
target: "{config.binary_path}"
args: ["{cmd.name}"]
description: "Run {cmd.name} without required arguments"
timeout: 10s
- action: verify_output
matches: "(error|missing|required)"
case_sensitive: false
timeout: 5s
description: "Should indicate missing arguments"
- action: verify_exit_code
expected: 1
description: "Should exit with error code"
"""
# Build invalid arg steps
invalid_arg_steps = ""
if config.commands:
cmd = config.commands[0]
invalid_arg_steps = f""" - action: launch
target: "{config.binary_path}"
args: ["{cmd.name}", "--invalid-flag-xyz"]
description: "Run with invalid flag"
timeout: 10s
- action: verify_output
matches: "(error|unknown|unrecognized|invalid)"
case_sensitive: false
timeout: 5s
description: "Should reject unknown flag"
"""
context = {
"app_name": config.binary_path.split("/")[-1] if "/" in config.binary_path else config.binary_path,
"binary_path": config.binary_path,
"missing_arg_steps": missing_arg_steps,
"invalid_arg_steps": invalid_arg_steps,
}
content = template_mgr.render("cli_error_handling", context)
test_file = tests_dir / "error-handling.yaml"
write_file(test_file, content)
return [
GeneratedTest(
category=TestCategory.CLI_ERROR_HANDLING,
file_path=test_file,
test_count=3,
description="CLI error handling tests",
)
]
def _generate_cli_integration_tests(
config: CLIConfig, template_mgr: TemplateManager, output_dir: Path
) -> list[GeneratedTest]:
"""Generate CLI integration test scenarios (command pipelines)."""
tests_dir = output_dir / "cli-integration"
ensure_directory(tests_dir)
generated = []
if len(config.commands) >= 2:
# Create a multi-command workflow test
steps = []
for i, cmd in enumerate(config.commands):
args = [f'"{cmd.name}"'] + [f'"test-arg-{j}"' for j in range(min(len(cmd.args), 2))]
steps.append(f""" - action: launch
target: "{config.binary_path}"
args: [{", ".join(args)}]
description: "Step {i+1}: Run {cmd.name}"
timeout: 15s
- action: verify_exit_code
expected: 0
description: "{cmd.name} should succeed"
""")
content = f"""# CLI Integration Test - Multi-Command Workflow
# Auto-generated outside-in test scenario
scenario:
name: "CLI Integration - Multi-Command Workflow"
description: |
Verifies that multiple CLI commands can be executed in sequence
as part of a typical workflow.
type: cli
level: 2
tags: [cli, integration, workflow, auto-generated]
prerequisites:
- "{config.binary_path} binary exists and is executable"
steps:
{"".join(steps)}
cleanup:
- action: stop_application
force: true
description: "Ensure process is terminated"
"""
test_file = tests_dir / "multi-command-workflow.yaml"
write_file(test_file, content)
generated.append(
GeneratedTest(
category=TestCategory.CLI_INTEGRATION,
file_path=test_file,
test_count=len(config.commands) * 2,
description="CLI multi-command integration test",
)
)
return generated
"""Coverage analysis and reporting.
Verifies all 7 categories present, ≥40 total tests, and generates recommendations.
"""
from .models import (
Bug,
BugSeverity,
CoverageAuditError,
CoverageReport,
GeneratedTest,
StackConfig,
TestCategory,
)
from .utils import calculate_coverage_percent
def audit_coverage(
stack: StackConfig,
generated_tests: list[GeneratedTest],
test_results: None = None, # Kept for API compatibility, not used
) -> CoverageReport:
"""Analyze test coverage and generate recommendations.
Args:
stack: Stack configuration
generated_tests: List of generated tests
test_results: Unused (kept for API compatibility)
Returns:
CoverageReport with coverage analysis
Raises:
CoverageAuditError: If audit fails
"""
try:
# Calculate total tests
total_tests = sum(t.test_count for t in generated_tests)
# Category breakdown
category_breakdown = calculate_category_breakdown(generated_tests)
# Route coverage
route_coverage = calculate_route_coverage(stack, generated_tests)
route_coverage_percent = calculate_coverage_percent(
sum(1 for covered in route_coverage.values() if covered), len(route_coverage)
)
# Endpoint coverage
endpoint_coverage = calculate_endpoint_coverage(stack, generated_tests)
endpoint_coverage_percent = calculate_coverage_percent(
sum(1 for covered in endpoint_coverage.values() if covered), len(endpoint_coverage)
)
# No bugs detected without test execution
bugs: list[Bug] = []
# Generate recommendations
recommendations = generate_recommendations(
total_tests,
category_breakdown,
route_coverage_percent,
endpoint_coverage_percent,
bugs,
)
return CoverageReport(
total_tests=total_tests,
category_breakdown=category_breakdown,
route_coverage=route_coverage,
endpoint_coverage=endpoint_coverage,
bugs_found=bugs,
recommendations=recommendations,
route_coverage_percent=route_coverage_percent,
endpoint_coverage_percent=endpoint_coverage_percent,
)
except Exception as e:
raise CoverageAuditError(f"Coverage audit failed: {e}")
def calculate_category_breakdown(generated_tests: list[GeneratedTest]) -> dict[str, int]:
"""Calculate test count per category.
Args:
generated_tests: List of generated tests
Returns:
Dict of category name -> test count
"""
breakdown = {category.value: 0 for category in TestCategory}
for test in generated_tests:
breakdown[test.category.value] += test.test_count
return breakdown
def calculate_route_coverage(
stack: StackConfig, generated_tests: list[GeneratedTest]
) -> dict[str, bool]:
"""Map routes to test coverage.
Args:
stack: Stack configuration
generated_tests: List of generated tests
Returns:
Dict of route path -> covered (bool)
"""
coverage = {route.path: False for route in stack.routes}
# Check which routes are covered by tests
for test in generated_tests:
# Simple heuristic: if test file mentions route, it's covered
route_slug = str(test.file_path.stem)
for route in stack.routes:
route_pattern = route.path.replace("/", "-").strip("-") or "home"
if route_pattern in route_slug:
coverage[route.path] = True
return coverage
def calculate_endpoint_coverage(
stack: StackConfig, generated_tests: list[GeneratedTest]
) -> dict[str, bool]:
"""Map API endpoints to test coverage.
Args:
stack: Stack configuration
generated_tests: List of generated tests
Returns:
Dict of endpoint key -> covered (bool)
"""
coverage = {}
for endpoint in stack.api_endpoints:
key = f"{endpoint.method} {endpoint.path}"
coverage[key] = False
# Check if any test covers this endpoint
# In real implementation, would parse test files
# For now, mark as not covered
coverage[key] = False
return coverage
def generate_recommendations(
total_tests: int,
category_breakdown: dict[str, int],
route_coverage_percent: float,
endpoint_coverage_percent: float,
bugs: list[Bug],
) -> list[str]:
"""Generate actionable recommendations.
Args:
total_tests: Total number of tests
category_breakdown: Tests per category
route_coverage_percent: Route coverage percentage
endpoint_coverage_percent: Endpoint coverage percentage
bugs: List of bugs found
Returns:
List of recommendations
"""
recommendations = []
# Check minimum test count
if total_tests < 40:
recommendations.append(f"Add {40 - total_tests} more tests to reach minimum of 40 tests")
# Check category coverage
for category, count in category_breakdown.items():
if count == 0:
recommendations.append(f"Add tests for {category} category (currently 0)")
elif count < 3:
recommendations.append(f"Consider adding more {category} tests (currently {count})")
# Check route coverage
if route_coverage_percent < 80.0:
recommendations.append(
f"Increase route coverage from {route_coverage_percent:.1f}% to at least 80%"
)
# Check endpoint coverage
if endpoint_coverage_percent < 70.0:
recommendations.append(
f"Increase API endpoint coverage from {endpoint_coverage_percent:.1f}% to at least 70%"
)
# Check bugs
critical_bugs = [b for b in bugs if b.severity == BugSeverity.CRITICAL]
if critical_bugs:
recommendations.append(f"Fix {len(critical_bugs)} CRITICAL bugs immediately")
high_bugs = [b for b in bugs if b.severity == BugSeverity.HIGH]
if high_bugs:
recommendations.append(f"Address {len(high_bugs)} HIGH severity bugs")
# If everything looks good
if not recommendations:
recommendations.append(
"Coverage looks good! All categories present, minimum test count met."
)
return recommendations
"""Infrastructure setup for E2E testing.
Generates Playwright config, test helpers, and seed data.
ENFORCES workers=1 in all generated configs.
"""
from pathlib import Path
from .models import InfrastructureSetupError, StackConfig
from .utils import ensure_directory, write_file, write_json_file
def setup_infrastructure(stack: StackConfig, output_dir: Path) -> None:
"""Create complete testing infrastructure.
Args:
stack: Detected stack configuration
output_dir: Directory to create infrastructure in (e2e/)
Raises:
InfrastructureSetupError: If setup fails
"""
try:
ensure_directory(output_dir)
# Generate playwright.config.ts
config_content = create_playwright_config(stack)
write_file(output_dir.parent / "playwright.config.ts", config_content)
# Generate test helpers
helpers_dir = output_dir / "test-helpers"
ensure_directory(helpers_dir)
helpers = create_test_helpers(stack)
for filename, content in helpers.items():
write_file(helpers_dir / filename, content)
# Generate seed data
fixtures_dir = output_dir / "fixtures"
ensure_directory(fixtures_dir)
seed_data = create_seed_data(stack)
for filename, data in seed_data.items():
write_json_file(fixtures_dir / filename, data)
except Exception as e:
raise InfrastructureSetupError(f"Infrastructure setup failed: {e}")
def create_playwright_config(stack: StackConfig) -> str:
"""Generate playwright.config.ts with workers=1.
Args:
stack: Stack configuration
Returns:
Playwright config file content
"""
config = """import {{ defineConfig, devices }} from '@playwright/test';
export default defineConfig({{
testDir: './e2e',
fullyParallel: false,
forbidOnly: !!process.env.CI,
retries: process.env.CI ? 2 : 0,
workers: 1, // MANDATORY: Must be 1 for deterministic execution
reporter: 'html',
use: {{
baseURL: '{base_url}',
trace: 'on-first-retry',
}},
projects: [
{{
name: 'chromium',
use: {{ ...devices['Desktop Chrome'] }},
}},
],
webServer: {{
command: '{dev_command}',
url: '{base_url}',
reuseExistingServer: !process.env.CI,
}},
}});
"""
# Determine dev command based on framework
dev_command = "npm run dev"
if stack.frontend_framework == "nextjs" or stack.frontend_framework in ["react", "vue"]:
dev_command = "npm run dev"
# Use api_base_url from stack
base_url = stack.api_base_url
return config.format(base_url=base_url, dev_command=dev_command)
def create_test_helpers(stack: StackConfig) -> dict[str, str]:
"""Generate helper functions for tests.
Args:
stack: Stack configuration
Returns:
Dict of filename -> content
"""
helpers = {}
# Authentication helper
auth_helper = """import {{ Page }} from '@playwright/test';
export async function login(page: Page, email: string, password: string) {{
await page.goto('/login');
await page.getByRole('textbox', {{ name: /email/i }}).fill(email);
await page.getByRole('textbox', {{ name: /password/i }}).fill(password);
await page.getByRole('button', {{ name: /sign in/i }}).click();
await page.waitForURL('/dashboard');
}}
export async function logout(page: Page) {{
await page.getByRole('button', {{ name: /logout/i }}).click();
await page.waitForURL('/login');
}}
"""
helpers["auth.ts"] = auth_helper
# Navigation helper
nav_helper = """import {{ Page }} from '@playwright/test';
export async function navigateTo(page: Page, route: string) {{
await page.goto(route);
await page.waitForLoadState('networkidle');
}}
export async function clickLink(page: Page, linkText: string) {{
await page.getByRole('link', {{ name: new RegExp(linkText, 'i') }}).click();
}}
"""
helpers["navigation.ts"] = nav_helper
# Assertions helper
assertions_helper = """import {{ Page, expect }} from '@playwright/test';
export async function assertPageTitle(page: Page, title: string) {{
await expect(page).toHaveTitle(new RegExp(title, 'i'));
}}
export async function assertElementVisible(page: Page, role: string, name: string) {{
await expect(page.getByRole(role as any, {{ name: new RegExp(name, 'i') }})).toBeVisible();
}}
export async function assertNoConsoleErrors(page: Page) {{
const errors: string[] = [];
page.on('console', msg => {{
if (msg.type() === 'error') {{
errors.push(msg.text());
}}
}});
return errors;
}}
"""
helpers["assertions.ts"] = assertions_helper
# Data setup helper
data_setup_helper = """import {{ Page }} from '@playwright/test';
import * as fs from 'fs';
import * as path from 'path';
export async function loadFixture(name: string): Promise<any> {{
const fixturePath = path.join(__dirname, '../fixtures', `${{name}}.json`);
const data = fs.readFileSync(fixturePath, 'utf-8');
return JSON.parse(data);
}}
export async function seedDatabase(page: Page, fixture: string) {{
const data = await loadFixture(fixture);
// POST to /api/test/seed endpoint
await page.request.post('/api/test/seed', {{ data }});
}}
export async function clearDatabase(page: Page) {{
await page.request.post('/api/test/clear');
}}
"""
helpers["data-setup.ts"] = data_setup_helper
return helpers
def create_seed_data(stack: StackConfig) -> dict[str, dict]:
"""Generate small deterministic seed datasets.
Creates 10-20 records max per fixture.
Args:
stack: Stack configuration
Returns:
Dict of filename -> data
"""
seed_data = {}
# Users fixture (10 users)
users = {
"users": [
{"id": i, "email": f"user{i}@example.com", "name": f"User {i}", "role": "user"}
for i in range(1, 11)
]
}
seed_data["users.json"] = users
# Products fixture (15 products)
products = {
"products": [
{
"id": i,
"name": f"Product {i}",
"price": 10.00 + i,
"category": ["Electronics", "Clothing", "Books"][i % 3],
"inStock": i % 2 == 0,
}
for i in range(1, 16)
]
}
seed_data["products.json"] = products
# Orders fixture (20 orders)
orders = {
"orders": [
{
"id": i,
"userId": (i % 10) + 1,
"productId": (i % 15) + 1,
"quantity": i % 5 + 1,
"status": ["pending", "shipped", "delivered"][i % 3],
"createdAt": f"2024-01-{(i % 28) + 1:02d}T00:00:00Z",
}
for i in range(1, 21)
]
}
seed_data["orders.json"] = orders
return seed_data
"""MCP test scenario generator.
Generates gadugi-agentic-test YAML scenarios for MCP servers
based on tool definitions, input schemas, and resource definitions.
"""
import json
from pathlib import Path
from .models import GeneratedTest, MCPConfig, MCPTool, TestCategory
from .template_manager import TemplateManager
from .utils import ensure_directory, write_file
def generate_mcp_tests(
config: MCPConfig, template_mgr: TemplateManager, output_dir: Path
) -> list[GeneratedTest]:
"""Generate all MCP test scenarios.
Args:
config: MCP server configuration
template_mgr: Template manager
output_dir: Output directory for generated test files
Returns:
List of GeneratedTest objects
"""
generated = []
generated.extend(_generate_mcp_smoke_tests(config, template_mgr, output_dir))
generated.extend(_generate_mcp_validation_tests(config, template_mgr, output_dir))
generated.extend(_generate_mcp_error_tests(config, template_mgr, output_dir))
generated.extend(_generate_mcp_workflow_tests(config, template_mgr, output_dir))
return generated
def _generate_mcp_smoke_tests(
config: MCPConfig, template_mgr: TemplateManager, output_dir: Path
) -> list[GeneratedTest]:
"""Generate MCP tool smoke tests (one per tool)."""
tests_dir = output_dir / "mcp-smoke"
ensure_directory(tests_dir)
generated = []
server_args = ", ".join(f'"{a}"' for a in config.server_args)
for tool in config.tools:
# Generate sample valid input from schema
valid_input = _generate_sample_input(tool)
valid_input_json = json.dumps(valid_input, indent=6)
# Build validation steps for required inputs
validation_steps = ""
if tool.required_inputs:
# Test with missing required field
if valid_input:
first_required = tool.required_inputs[0]
incomplete_input = {k: v for k, v in valid_input.items() if k != first_required}
validation_steps = f""" - action: mcp_call_tool
tool: "{tool.name}"
input: {json.dumps(incomplete_input)}
description: "Call without required field '{first_required}'"
timeout: 15s
- action: verify_mcp_error
matches: "(required|missing|{first_required})"
description: "Should report missing required field"
"""
# Build error steps
error_steps = f""" - action: mcp_call_tool
tool: "{tool.name}"
input: {{"__invalid__": true}}
description: "Call with invalid input schema"
timeout: 15s
- action: verify_mcp_error
matches: "(invalid|error|unexpected)"
description: "Should handle invalid input gracefully"
"""
context = {
"tool_name": tool.name,
"tool_description": tool.description or f"MCP tool: {tool.name}",
"transport": config.transport,
"server_command": config.server_command,
"server_args": server_args,
"valid_input": valid_input_json,
"response_pattern": ".*", # Any response is ok for smoke
"validation_steps": validation_steps,
"error_steps": error_steps,
}
content = template_mgr.render("mcp_tool", context)
test_file = tests_dir / f"{tool.name}.yaml"
write_file(test_file, content)
test_count = 2 # basic call + invalid input
if validation_steps:
test_count += 1
generated.append(
GeneratedTest(
category=TestCategory.MCP_TOOL_SMOKE,
file_path=test_file,
test_count=test_count,
description=f"MCP smoke test for tool '{tool.name}'",
)
)
return generated
def _generate_mcp_validation_tests(
config: MCPConfig, template_mgr: TemplateManager, output_dir: Path
) -> list[GeneratedTest]:
"""Generate MCP input validation test scenarios."""
tests_dir = output_dir / "mcp-validation"
ensure_directory(tests_dir)
generated = []
server_args = ", ".join(f'"{a}"' for a in config.server_args)
# Find tools with defined input schemas
schema_tools = [t for t in config.tools if t.input_schema]
for tool in schema_tools:
properties = tool.input_schema.get("properties", {})
if not isinstance(properties, dict):
continue
# Generate type mismatch tests
steps = ""
for prop_name, prop_def in properties.items():
if not isinstance(prop_def, dict):
continue
prop_type = prop_def.get("type", "string")
# Create input with wrong type
wrong_value = _wrong_type_value(prop_type)
wrong_input = {prop_name: wrong_value}
steps += f""" - action: mcp_call_tool
tool: "{tool.name}"
input: {json.dumps(wrong_input)}
description: "Send wrong type for '{prop_name}' (expected {prop_type})"
timeout: 15s
- action: verify_mcp_error
matches: "(type|invalid|error|{prop_name})"
description: "Should reject wrong type for {prop_name}"
"""
if steps:
content = f"""# MCP Input Validation Test - {tool.name}
# Auto-generated outside-in test scenario
scenario:
name: "MCP Validation - {tool.name} Input Types"
description: |
Verifies that the '{tool.name}' tool properly validates input types
and returns meaningful error messages for type mismatches.
type: mcp
level: 2
tags: [mcp, validation, {tool.name}, auto-generated]
prerequisites:
- "MCP server is available via {config.transport} transport"
steps:
- action: mcp_connect
command: "{config.server_command}"
args: [{server_args}]
transport: "{config.transport}"
description: "Connect to MCP server"
timeout: 15s
{steps}
- action: mcp_disconnect
description: "Disconnect from MCP server"
cleanup:
- action: mcp_disconnect
force: true
"""
test_file = tests_dir / f"validate-{tool.name}.yaml"
write_file(test_file, content)
prop_count = min(len(properties), 3)
generated.append(
GeneratedTest(
category=TestCategory.MCP_TOOL_VALIDATION,
file_path=test_file,
test_count=prop_count,
description=f"MCP validation test for '{tool.name}'",
)
)
return generated
def _generate_mcp_error_tests(
config: MCPConfig, template_mgr: TemplateManager, output_dir: Path
) -> list[GeneratedTest]:
"""Generate MCP error handling test scenarios."""
tests_dir = output_dir / "mcp-errors"
ensure_directory(tests_dir)
server_args = ", ".join(f'"{a}"' for a in config.server_args)
content = f"""# MCP Error Handling Test
# Auto-generated outside-in test scenario
scenario:
name: "MCP Error Handling - Server Resilience"
description: |
Verifies that the MCP server handles error conditions gracefully
including unknown tools, malformed input, and edge cases.
type: mcp
level: 2
tags: [mcp, error-handling, resilience, auto-generated]
prerequisites:
- "MCP server is available via {config.transport} transport"
steps:
- action: mcp_connect
command: "{config.server_command}"
args: [{server_args}]
transport: "{config.transport}"
description: "Connect to MCP server"
timeout: 15s
- action: mcp_call_tool
tool: "nonexistent_tool_xyz"
input: {{}}
description: "Call non-existent tool"
timeout: 10s
- action: verify_mcp_error
matches: "(not found|unknown|does not exist)"
description: "Should report unknown tool"
- action: mcp_call_tool
tool: "{config.tools[0].name if config.tools else 'test'}"
input: null
description: "Call with null input"
timeout: 10s
- action: verify_mcp_error
matches: "(invalid|null|error)"
description: "Should handle null input"
- action: mcp_disconnect
description: "Disconnect from MCP server"
cleanup:
- action: mcp_disconnect
force: true
"""
test_file = tests_dir / "error-handling.yaml"
write_file(test_file, content)
return [
GeneratedTest(
category=TestCategory.MCP_TOOL_ERROR,
file_path=test_file,
test_count=3,
description="MCP error handling tests",
)
]
def _generate_mcp_workflow_tests(
config: MCPConfig, template_mgr: TemplateManager, output_dir: Path
) -> list[GeneratedTest]:
"""Generate MCP multi-tool workflow test scenarios."""
if len(config.tools) < 2:
return []
tests_dir = output_dir / "mcp-workflows"
ensure_directory(tests_dir)
server_args = ", ".join(f'"{a}"' for a in config.server_args)
# Build workflow steps calling tools in sequence
steps = ""
for i, tool in enumerate(config.tools):
valid_input = _generate_sample_input(tool)
steps += f""" - action: mcp_call_tool
tool: "{tool.name}"
input: {json.dumps(valid_input)}
description: "Step {i+1}: Call {tool.name}"
timeout: 30s
- action: verify_mcp_response
matches: ".*"
description: "Step {i+1}: {tool.name} should respond"
"""
context = {
"workflow_name": "Multi-Tool Sequence",
"transport": config.transport,
"server_command": config.server_command,
"server_args": server_args,
"workflow_steps": steps,
}
content = template_mgr.render("mcp_workflow", context)
test_file = tests_dir / "multi-tool-workflow.yaml"
write_file(test_file, content)
return [
GeneratedTest(
category=TestCategory.MCP_WORKFLOW,
file_path=test_file,
test_count=min(len(config.tools), 4) * 2,
description="MCP multi-tool workflow test",
)
]
def _generate_sample_input(tool: MCPTool) -> dict:
"""Generate sample valid input for an MCP tool."""
if not tool.input_schema:
return {}
sample: dict = {}
properties = tool.input_schema.get("properties", {})
if not isinstance(properties, dict):
return sample
for prop_name, prop_def in properties.items():
if not isinstance(prop_def, dict):
continue
prop_type = prop_def.get("type", "string")
if prop_type == "string":
if "path" in prop_name.lower() or "file" in prop_name.lower():
sample[prop_name] = "/tmp/test-file.txt"
elif "url" in prop_name.lower():
sample[prop_name] = "https://example.com"
elif "query" in prop_name.lower():
sample[prop_name] = "test query"
else:
sample[prop_name] = f"test-{prop_name}"
elif prop_type == "integer":
sample[prop_name] = 1
elif prop_type == "number":
sample[prop_name] = 1.0
elif prop_type == "boolean":
sample[prop_name] = True
elif prop_type == "array":
sample[prop_name] = ["item1"]
elif prop_type == "object":
sample[prop_name] = {}
return sample
def _wrong_type_value(expected_type: str):
"""Return a value of the wrong type for testing."""
wrong_values = {
"string": 12345,
"integer": "not-a-number",
"number": "not-a-number",
"boolean": "not-a-bool",
"array": "not-an-array",
"object": "not-an-object",
}
return wrong_values.get(expected_type, None)
"""Security utilities for safe file and path operations."""
import json
from pathlib import Path
from typing import Any
class SecurityError(Exception):
"""Base exception for security violations."""
def validate_project_root(path: Path, allowed_root: Path | None = None) -> Path:
"""Validate path is within project boundaries.
Prevents path traversal attacks.
Args:
path: Path to validate
allowed_root: Root directory (default: current working directory)
Returns:
Resolved path if valid
Raises:
SecurityError: If path is outside allowed root
"""
if allowed_root is None:
allowed_root = Path.cwd()
resolved = path.resolve()
allowed_resolved = allowed_root.resolve()
# Check if path is within allowed root
try:
resolved.relative_to(allowed_resolved)
except ValueError:
raise SecurityError(f"Path traversal detected: {path} is outside {allowed_root}")
return resolved
def sanitize_path(path: str) -> str:
"""Remove shell metacharacters from path strings.
Prevents command injection via malicious paths.
Args:
path: Path string to sanitize
Returns:
Sanitized path string
Raises:
SecurityError: If path contains forbidden characters
"""
forbidden_chars = [";", "|", "&", "`", "$", "(", ")", "<", ">", "\n", "\r"]
for char in forbidden_chars:
if char in path:
raise SecurityError(f"Path contains forbidden character '{char}': {path}")
return path
def read_json_safe(path: Path, max_size_mb: int = 10) -> dict[str, Any]:
"""Read JSON file with DoS protection.
Prevents JSON bomb attacks via size limits.
Args:
path: Path to JSON file
max_size_mb: Maximum file size in megabytes (default: 10MB)
Returns:
Parsed JSON data
Raises:
SecurityError: If file is too large
ValueError: If JSON is invalid
"""
# Check file size before reading
file_size = path.stat().st_size
max_bytes = max_size_mb * 1024 * 1024
if file_size > max_bytes:
raise SecurityError(
f"JSON file too large: {file_size / 1024 / 1024:.1f}MB (max: {max_size_mb}MB)"
)
# Read and parse
with open(path, encoding="utf-8") as f:
try:
return json.load(f)
except json.JSONDecodeError as e:
raise ValueError(f"Invalid JSON in {path}: {e}")
# API Endpoint Test - {method} {path}
# Auto-generated outside-in test scenario
scenario:
name: "API {method} {path} - {summary}"
description: |
Verifies the {method} {path} endpoint responds correctly with valid input
and returns appropriate error responses for invalid input.
type: api
level: 1
tags: [api, {method_lower}, {tag}, auto-generated]
prerequisites:
- "API server is running at {base_url}"
variables:
base_url: "{base_url}"
steps:
- action: http_request
method: "{method}"
url: "{base_url}{path}"
{request_body}
{auth_header}
description: "Send valid {method} request to {path}"
timeout: 10s
- action: verify_status_code
expected: {expected_status}
description: "Should return {expected_status} status"
- action: verify_response
matches: "{response_pattern}"
description: "Response should match expected schema"
{validation_steps}
cleanup:
- action: log_response
save_as: "{method_lower}-{path_slug}-response.json"
description: "Save response for debugging"
# API Workflow Test - {workflow_name}
# Auto-generated outside-in test scenario
scenario:
name: "API Workflow - {workflow_name}"
description: |
Verifies the end-to-end {workflow_name} workflow by executing a sequence
of API calls and validating the business logic.
type: api
level: 2
tags: [api, workflow, integration, auto-generated]
prerequisites:
- "API server is running at {base_url}"
variables:
base_url: "{base_url}"
steps:
{workflow_steps}
cleanup:
- action: log_response
save_as: "workflow-{workflow_slug}-result.json"
description: "Save workflow results"
# MCP Multi-Tool Workflow Test
# Auto-generated outside-in test scenario
scenario:
name: "MCP Workflow - {workflow_name}"
description: |
Verifies a multi-tool MCP workflow by calling tools in sequence
and validating the combined output.
type: mcp
level: 2
tags: [mcp, workflow, multi-tool, auto-generated]
prerequisites:
- "MCP server is available via {transport} transport"
environment:
variables:
MCP_TRANSPORT: "{transport}"
steps:
- action: mcp_connect
command: "{server_command}"
args: [{server_args}]
transport: "{transport}"
description: "Connect to MCP server"
timeout: 15s
{workflow_steps}
- action: mcp_disconnect
description: "Disconnect from MCP server"
cleanup:
- action: mcp_disconnect
force: true
description: "Ensure MCP connection is closed"