Isolated Test Execution Service

Build a production-ready isolated test execution service for CI/CD pipelines. This cookbook demonstrates how to create a test execution platform that runs tests in isolated sandboxes, supports parallel execution, and aggregates results.

Overview

Isolated test execution services run tests in completely isolated environments, ensuring test reliability and preventing interference between test runs. This pattern is essential for CI/CD pipelines, test automation platforms, and quality assurance systems. HopX provides the secure isolation needed for this use case.

Prerequisites

HopX API key (Get one here)
Python 3.8+ or Node.js 16+
Understanding of test frameworks (pytest, jest, etc.)
Basic knowledge of CI/CD concepts

Architecture

┌──────────────┐
│  CI/CD       │ Test execution requests
│   Pipeline   │
└──────┬───────┘
       │
       ▼
┌─────────────────┐
│  Test Runner    │ Parallel execution
└──────┬──────────┘
       │
       ▼
┌─────────────────┐
│  HopX Sandboxes │ Isolated execution
│  (One per test) │
└──────┬──────────┘
       │
       ▼
┌─────────────────┐
│  Result         │ Aggregate & report
│  Aggregator     │
└─────────────────┘

Implementation

Step 1: Basic Isolated Test Execution

Execute tests in isolated sandboxes:

from hopx_ai import Sandbox
import os
import json
from typing import Dict, List, Any
from concurrent.futures import ThreadPoolExecutor, as_completed

class IsolatedTestExecutor:
    def __init__(self, api_key: str):
        self.api_key = api_key
    
    def execute_test(self, test_code: str, test_framework: str = "pytest", timeout: int = 60) -> Dict[str, Any]:
        """Execute a single test in isolated sandbox"""
        sandbox = None
        try:
            # Create fresh sandbox for each test
            sandbox = Sandbox.create(
                template="code-interpreter",
                api_key=self.api_key,
                timeout_seconds=timeout + 10
            )
            
            # Write test file
            test_file = "/workspace/test_file.py"
            sandbox.files.write(test_file, test_code)
            
            # Install test framework if needed
            if test_framework == "pytest":
                sandbox.commands.run("pip install pytest", timeout=30)
                # Execute test
                result = sandbox.commands.run(f"pytest {test_file} -v", timeout=timeout)
            elif test_framework == "unittest":
                result = sandbox.commands.run(f"python -m unittest {test_file}", timeout=timeout)
            else:
                result = sandbox.run_code(test_code, timeout=timeout)
            
            # Parse test results
            test_results = self._parse_test_results(result.stdout, test_framework)
            
            return {
                "success": result.exit_code == 0,
                "exit_code": result.exit_code,
                "stdout": result.stdout,
                "stderr": result.stderr,
                "test_results": test_results,
                "execution_time": getattr(result, 'execution_time', 0)
            }
            
        except Exception as e:
            return {
                "success": False,
                "error": str(e),
                "stderr": str(e)
            }
        finally:
            if sandbox:
                sandbox.kill()
    
    def _parse_test_results(self, output: str, framework: str) -> Dict[str, Any]:
        """Parse test framework output"""
        if framework == "pytest":
            # Parse pytest output
            passed = output.count("PASSED")
            failed = output.count("FAILED")
            return {
                "passed": passed,
                "failed": failed,
                "total": passed + failed
            }
        else:
            return {
                "passed": 0,
                "failed": 0,
                "total": 0
            }

# Usage
executor = IsolatedTestExecutor(api_key=os.getenv("HOPX_API_KEY"))

test_code = """
def test_addition():
    assert 1 + 1 == 2

def test_subtraction():
    assert 5 - 3 == 2
"""

result = executor.execute_test(test_code, test_framework="pytest")
print(json.dumps(result, indent=2))

import { Sandbox } from '@hopx-ai/sdk';

class IsolatedTestExecutor {
    constructor(apiKey) {
        this.apiKey = apiKey;
    }
    
    async executeTest(testCode, testFramework = 'jest', timeout = 60) {
        let sandbox = null;
        try {
            // Create fresh sandbox for each test
            sandbox = await Sandbox.create({
                template: 'code-interpreter',
                apiKey: this.apiKey,
                timeoutSeconds: timeout + 10
            });
            
            // Write test file
            const testFile = '/workspace/test_file.js';
            await sandbox.files.write(testFile, testCode);
            
            // Install test framework if needed
            if (testFramework === 'jest') {
                await sandbox.commands.run('npm install jest', { timeout: 30 });
                // Execute test
                const result = await sandbox.commands.run(`npx jest ${testFile}`, { timeout });
                
                // Parse test results
                const testResults = this.parseTestResults(result.stdout, testFramework);
                
                return {
                    success: result.exit_code === 0,
                    exitCode: result.exit_code,
                    stdout: result.stdout,
                    stderr: result.stderr,
                    testResults,
                    executionTime: result.execution_time || 0
                };
            } else {
                const result = await sandbox.runCode(testCode, { timeout });
                return {
                    success: result.success,
                    stdout: result.stdout,
                    stderr: result.stderr
                };
            }
            
        } catch (error) {
            return {
                success: false,
                error: error.message,
                stderr: error.toString()
            };
        } finally {
            if (sandbox) {
                await sandbox.kill();
            }
        }
    }
    
    parseTestResults(output, framework) {
        if (framework === 'jest') {
            // Parse jest output
            const passed = (output.match(/PASS/g) || []).length;
            const failed = (output.match(/FAIL/g) || []).length;
            return {
                passed,
                failed,
                total: passed + failed
            };
        } else {
            return {
                passed: 0,
                failed: 0,
                total: 0
            };
        }
    }
}

// Usage
const executor = new IsolatedTestExecutor(process.env.HOPX_API_KEY);

const testCode = `
test('addition', () => {
    expect(1 + 1).toBe(2);
});

test('subtraction', () => {
    expect(5 - 3).toBe(2);
});
`;

const result = await executor.executeTest(testCode, 'jest');
console.log(JSON.stringify(result, null, 2));

Step 2: Parallel Test Execution

Execute multiple tests in parallel:

class ParallelTestExecutor(IsolatedTestExecutor):
    def execute_test_suite(self, test_suite: List[Dict], max_workers: int = 5) -> Dict[str, Any]:
        """Execute multiple tests in parallel"""
        results = []
        
        with ThreadPoolExecutor(max_workers=max_workers) as executor:
            # Submit all tests
            future_to_test = {
                executor.submit(
                    self.execute_test,
                    test["code"],
                    test.get("framework", "pytest"),
                    test.get("timeout", 60)
                ): test
                for test in test_suite
            }
            
            # Collect results as they complete
            for future in as_completed(future_to_test):
                test = future_to_test[future]
                try:
                    result = future.result()
                    results.append({
                        "test_name": test.get("name", "unknown"),
                        "result": result
                    })
                except Exception as e:
                    results.append({
                        "test_name": test.get("name", "unknown"),
                        "result": {
                            "success": False,
                            "error": str(e)
                        }
                    })
        
        # Aggregate results
        total_tests = len(results)
        passed_tests = sum(1 for r in results if r["result"].get("success", False))
        failed_tests = total_tests - passed_tests
        
        return {
            "total_tests": total_tests,
            "passed": passed_tests,
            "failed": failed_tests,
            "success_rate": (passed_tests / total_tests * 100) if total_tests > 0 else 0,
            "results": results
        }

# Usage
executor = ParallelTestExecutor(api_key=os.getenv("HOPX_API_KEY"))

test_suite = [
    {
        "name": "test_math",
        "code": """
def test_addition():
    assert 1 + 1 == 2
""",
        "framework": "pytest",
        "timeout": 30
    },
    {
        "name": "test_strings",
        "code": """
def test_concatenation():
    assert "hello" + " " + "world" == "hello world"
""",
        "framework": "pytest",
        "timeout": 30
    }
]

results = executor.execute_test_suite(test_suite, max_workers=2)
print(json.dumps(results, indent=2))

class ParallelTestExecutor extends IsolatedTestExecutor {
    async executeTestSuite(testSuite, maxWorkers = 5) {
        const results = [];
        
        // Execute tests in parallel (limited by maxWorkers)
        const chunks = [];
        for (let i = 0; i < testSuite.length; i += maxWorkers) {
            chunks.push(testSuite.slice(i, i + maxWorkers));
        }
        
        for (const chunk of chunks) {
            const chunkResults = await Promise.all(
                chunk.map(async (test) => {
                    try {
                        const result = await this.executeTest(
                            test.code,
                            test.framework || 'jest',
                            test.timeout || 60
                        );
                        return {
                            testName: test.name || 'unknown',
                            result
                        };
                    } catch (error) {
                        return {
                            testName: test.name || 'unknown',
                            result: {
                                success: false,
                                error: error.message
                            }
                        };
                    }
                })
            );
            results.push(...chunkResults);
        }
        
        // Aggregate results
        const totalTests = results.length;
        const passedTests = results.filter(r => r.result.success).length;
        const failedTests = totalTests - passedTests;
        
        return {
            totalTests,
            passed: passedTests,
            failed: failedTests,
            successRate: totalTests > 0 ? (passedTests / totalTests * 100) : 0,
            results
        };
    }
}

// Usage
const executor = new ParallelTestExecutor(process.env.HOPX_API_KEY);

const testSuite = [
    {
        name: 'test_math',
        code: `
test('addition', () => {
    expect(1 + 1).toBe(2);
});
`,
        framework: 'jest',
        timeout: 30
    },
    {
        name: 'test_strings',
        code: `
test('concatenation', () => {
    expect('hello' + ' ' + 'world').toBe('hello world');
});
`,
        framework: 'jest',
        timeout: 30
    }
];

const results = await executor.executeTestSuite(testSuite, 2);
console.log(JSON.stringify(results, null, 2));

Step 3: Test Isolation Strategies

Ensure complete test isolation:

class IsolatedTestRunner:
    def __init__(self, api_key: str):
        self.api_key = api_key
    
    def run_isolated_test(self, test_config: Dict) -> Dict[str, Any]:
        """Run test with complete isolation"""
        sandbox = None
        try:
            # Create sandbox with clean environment
            sandbox = Sandbox.create(
                template="code-interpreter",
                api_key=self.api_key,
                timeout_seconds=test_config.get("timeout", 60) + 10,
                env_vars=test_config.get("env_vars", {})
            )
            
            # Set up test environment
            setup_code = test_config.get("setup", "")
            if setup_code:
                setup_result = sandbox.run_code(setup_code, timeout=30)
                if not setup_result.success:
                    return {
                        "success": False,
                        "error": f"Setup failed: {setup_result.stderr}",
                        "stage": "setup"
                    }
            
            # Write test files
            for file_config in test_config.get("files", []):
                sandbox.files.write(
                    file_config["path"],
                    file_config["content"]
                )
            
            # Execute test
            test_result = sandbox.run_code(
                test_config["test_code"],
                language=test_config.get("language", "python"),
                timeout=test_config.get("timeout", 60)
            )
            
            # Cleanup (teardown)
            teardown_code = test_config.get("teardown", "")
            if teardown_code:
                sandbox.run_code(teardown_code, timeout=30)
            
            return {
                "success": test_result.success,
                "stdout": test_result.stdout,
                "stderr": test_result.stderr,
                "exit_code": test_result.exit_code,
                "execution_time": test_result.execution_time
            }
            
        except Exception as e:
            return {
                "success": False,
                "error": str(e)
            }
        finally:
            # Always cleanup - fresh sandbox for next test
            if sandbox:
                sandbox.kill()

# Usage
runner = IsolatedTestRunner(api_key=os.getenv("HOPX_API_KEY"))

test_config = {
    "test_code": """
import unittest

class TestMath(unittest.TestCase):
    def test_addition(self):
        self.assertEqual(1 + 1, 2)
    
    def test_multiplication(self):
        self.assertEqual(2 * 3, 6)

if __name__ == '__main__':
    unittest.main()
""",
    "setup": "import unittest",
    "timeout": 30,
    "env_vars": {
        "TEST_MODE": "true"
    },
    "files": [
        {
            "path": "/workspace/test_utils.py",
            "content": "def helper(): return 'helper'"
        }
    ]
}

result = runner.run_isolated_test(test_config)
print(json.dumps(result, indent=2))

class IsolatedTestRunner {
    constructor(apiKey) {
        this.apiKey = apiKey;
    }
    
    async runIsolatedTest(testConfig) {
        let sandbox = null;
        try {
            // Create sandbox with clean environment
            sandbox = await Sandbox.create({
                template: 'code-interpreter',
                apiKey: this.apiKey,
                timeoutSeconds: (testConfig.timeout || 60) + 10,
                envVars: testConfig.envVars || {}
            });
            
            // Set up test environment
            const setupCode = testConfig.setup || '';
            if (setupCode) {
                const setupResult = await sandbox.runCode(setupCode, { timeout: 30 });
                if (!setupResult.success) {
                    return {
                        success: false,
                        error: `Setup failed: ${setupResult.stderr}`,
                        stage: 'setup'
                    };
                }
            }
            
            // Write test files
            for (const fileConfig of testConfig.files || []) {
                await sandbox.files.write(
                    fileConfig.path,
                    fileConfig.content
                );
            }
            
            // Execute test
            const testResult = await sandbox.runCode(
                testConfig.testCode,
                {
                    language: testConfig.language || 'python',
                    timeout: testConfig.timeout || 60
                }
            );
            
            // Cleanup (teardown)
            const teardownCode = testConfig.teardown || '';
            if (teardownCode) {
                await sandbox.runCode(teardownCode, { timeout: 30 });
            }
            
            return {
                success: testResult.success,
                stdout: testResult.stdout,
                stderr: testResult.stderr,
                exitCode: testResult.exit_code,
                executionTime: testResult.execution_time
            };
            
        } catch (error) {
            return {
                success: false,
                error: error.message
            };
        } finally {
            // Always cleanup - fresh sandbox for next test
            if (sandbox) {
                await sandbox.kill();
            }
        }
    }
}

// Usage
const runner = new IsolatedTestRunner(process.env.HOPX_API_KEY);

const testConfig = {
    testCode: `
const assert = require('assert');

describe('Math Tests', () => {
    it('should add numbers', () => {
        assert.strictEqual(1 + 1, 2);
    });
    
    it('should multiply numbers', () => {
        assert.strictEqual(2 * 3, 6);
    });
});
`,
    setup: 'const assert = require("assert");',
    timeout: 30,
    envVars: {
        TEST_MODE: 'true'
    },
    files: [
        {
            path: '/workspace/test_utils.js',
            content: "function helper() { return 'helper'; }"
        }
    ]
};

const result = await runner.runIsolatedTest(testConfig);
console.log(JSON.stringify(result, null, 2));

Step 4: Result Aggregation and Reporting

Aggregate test results and generate reports:

class TestResultAggregator:
    def __init__(self):
        self.results = []
    
    def add_result(self, test_name: str, result: Dict[str, Any]):
        """Add test result to aggregation"""
        self.results.append({
            "test_name": test_name,
            "result": result,
            "timestamp": self._get_timestamp()
        })
    
    def generate_report(self) -> Dict[str, Any]:
        """Generate aggregated test report"""
        total = len(self.results)
        passed = sum(1 for r in self.results if r["result"].get("success", False))
        failed = total - passed
        
        # Calculate execution times
        execution_times = [
            r["result"].get("execution_time", 0)
            for r in self.results
            if "execution_time" in r["result"]
        ]
        avg_time = sum(execution_times) / len(execution_times) if execution_times else 0
        total_time = sum(execution_times)
        
        # Group by status
        passed_tests = [r for r in self.results if r["result"].get("success", False)]
        failed_tests = [r for r in self.results if not r["result"].get("success", False)]
        
        return {
            "summary": {
                "total": total,
                "passed": passed,
                "failed": failed,
                "success_rate": (passed / total * 100) if total > 0 else 0,
                "average_execution_time": avg_time,
                "total_execution_time": total_time
            },
            "passed_tests": [
                {
                    "name": r["test_name"],
                    "execution_time": r["result"].get("execution_time", 0)
                }
                for r in passed_tests
            ],
            "failed_tests": [
                {
                    "name": r["test_name"],
                    "error": r["result"].get("error") or r["result"].get("stderr", ""),
                    "execution_time": r["result"].get("execution_time", 0)
                }
                for r in failed_tests
            ],
            "all_results": self.results
        }
    
    def _get_timestamp(self) -> str:
        from datetime import datetime
        return datetime.now().isoformat()

# Usage
aggregator = TestResultAggregator()

# Add results
aggregator.add_result("test_1", {"success": True, "execution_time": 1.5})
aggregator.add_result("test_2", {"success": False, "error": "Assertion failed"})
aggregator.add_result("test_3", {"success": True, "execution_time": 2.1})

# Generate report
report = aggregator.generate_report()
print(json.dumps(report, indent=2))

class TestResultAggregator {
    constructor() {
        this.results = [];
    }
    
    addResult(testName, result) {
        this.results.push({
            testName,
            result,
            timestamp: new Date().toISOString()
        });
    }
    
    generateReport() {
        const total = this.results.length;
        const passed = this.results.filter(r => r.result.success).length;
        const failed = total - passed;
        
        // Calculate execution times
        const executionTimes = this.results
            .map(r => r.result.executionTime || 0)
            .filter(t => t > 0);
        const avgTime = executionTimes.length > 0
            ? executionTimes.reduce((a, b) => a + b, 0) / executionTimes.length
            : 0;
        const totalTime = executionTimes.reduce((a, b) => a + b, 0);
        
        // Group by status
        const passedTests = this.results.filter(r => r.result.success);
        const failedTests = this.results.filter(r => !r.result.success);
        
        return {
            summary: {
                total,
                passed,
                failed,
                successRate: total > 0 ? (passed / total * 100) : 0,
                averageExecutionTime: avgTime,
                totalExecutionTime: totalTime
            },
            passedTests: passedTests.map(r => ({
                name: r.testName,
                executionTime: r.result.executionTime || 0
            })),
            failedTests: failedTests.map(r => ({
                name: r.testName,
                error: r.result.error || r.result.stderr || '',
                executionTime: r.result.executionTime || 0
            })),
            allResults: this.results
        };
    }
}

// Usage
const aggregator = new TestResultAggregator();

// Add results
aggregator.addResult('test_1', { success: true, executionTime: 1.5 });
aggregator.addResult('test_2', { success: false, error: 'Assertion failed' });
aggregator.addResult('test_3', { success: true, executionTime: 2.1 });

// Generate report
const report = aggregator.generateReport();
console.log(JSON.stringify(report, null, 2));

Best Practices

Isolation

Always create a fresh sandbox for each test to ensure complete isolation. Never reuse sandboxes between tests.

Fresh Sandboxes: One sandbox per test execution
Clean Environment: Start with clean environment variables
No Shared State: Tests should not depend on each other
Resource Cleanup: Always cleanup sandboxes after tests

Performance

Use parallel execution for independent tests, but ensure each test runs in its own isolated sandbox.

Parallel Execution: Run independent tests in parallel
Resource Limits: Set appropriate timeouts per test
Result Caching: Cache test results when possible
Efficient Cleanup: Clean up sandboxes efficiently

Reliability

Error Handling: Handle all error cases gracefully
Timeout Management: Set appropriate timeouts
Retry Logic: Implement retry for transient failures
Logging: Log all test executions for debugging

Real-World Examples

This pattern is used by:

GitHub Actions: CI/CD test execution
Jenkins: Continuous integration platform
CircleCI: Cloud-based CI/CD
Travis CI: Continuous integration service

API Testing Platform - API testing automation

Next Steps

Integrate with CI/CD platforms (GitHub Actions, GitLab CI, etc.)
Add test result persistence and history
Implement test result notifications
Create test execution dashboard
Add support for more test frameworks

AI & LLM Integration

Educational Platforms

Development Tools

Data Science & Analytics

Testing & CI/CD

Automation & Workflows

Serverless & Edge

Marketplace & Plugins

Enterprise & SaaS

Overview

Prerequisites

Architecture

Implementation

Step 1: Basic Isolated Test Execution

Step 2: Parallel Test Execution

Step 3: Test Isolation Strategies

Step 4: Result Aggregation and Reporting

Best Practices

Isolation

Performance

Reliability

Real-World Examples

Next Steps

​Overview

​Prerequisites

​Architecture

​Implementation

​Step 1: Basic Isolated Test Execution

​Step 2: Parallel Test Execution

​Step 3: Test Isolation Strategies

​Step 4: Result Aggregation and Reporting

​Best Practices

​Isolation

​Performance

​Reliability

​Real-World Examples

​Related Cookbooks

​Next Steps

Overview

Prerequisites

Architecture

Implementation

Step 1: Basic Isolated Test Execution

Step 2: Parallel Test Execution

Step 3: Test Isolation Strategies

Step 4: Result Aggregation and Reporting

Best Practices

Isolation

Performance

Reliability

Real-World Examples

Related Cookbooks

Next Steps