> ## Documentation Index
> Fetch the complete documentation index at: https://docs.hopx.ai/llms.txt
> Use this file to discover all available pages before exploring further.

# Isolated Test Execution Service

> Build a CI/CD test execution platform with parallel test execution, test isolation strategies, and result aggregation

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](https://console.hopx.dev/api-keys))
* 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:

<CodeGroup>
  ```python Python theme={null}
  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))
  ```

  ```javascript JavaScript theme={null}
  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));
  ```
</CodeGroup>

### Step 2: Parallel Test Execution

Execute multiple tests in parallel:

<CodeGroup>
  ```python Python theme={null}
  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))
  ```

  ```javascript JavaScript theme={null}
  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));
  ```
</CodeGroup>

### Step 3: Test Isolation Strategies

Ensure complete test isolation:

<CodeGroup>
  ```python Python theme={null}
  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))
  ```

  ```javascript JavaScript theme={null}
  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));
  ```
</CodeGroup>

### Step 4: Result Aggregation and Reporting

Aggregate test results and generate reports:

<CodeGroup>
  ```python Python theme={null}
  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))
  ```

  ```javascript JavaScript theme={null}
  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));
  ```
</CodeGroup>

## Best Practices

### Isolation

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

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

### Performance

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

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

### Reliability

1. **Error Handling**: Handle all error cases gracefully
2. **Timeout Management**: Set appropriate timeouts
3. **Retry Logic**: Implement retry for transient failures
4. **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

## Related Cookbooks

* [API Testing Platform](/cookbooks/automation/api-testing-platform) - API testing automation

## Next Steps

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


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.