Test Environment Management — Strategy, Infrastructure & Best Practices

# docker-compose.test.yml
version: '3.8'

services:
  app:
    build:
      context: .
      dockerfile: Dockerfile.test
    environment:
      - DB_HOST=db
      - DB_NAME=testdb
      - REDIS_HOST=redis
      - API_KEY=test-api-key-123
    depends_on:
      db:
        condition: service_healthy
      redis:
        condition: service_healthy
    ports:
      - "3000"

  db:
    image: postgres:16-alpine
    environment:
      POSTGRES_DB: testdb
      POSTGRES_USER: test
      POSTGRES_PASSWORD: testpass
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U test -d testdb"]
      interval: 5s
      timeout: 5s
      retries: 5
    tmpfs: /var/lib/postgresql/data

  redis:
    image: redis:7-alpine
    healthcheck:
      test: ["CMD", "redis-cli", "ping"]
      interval: 5s
      timeout: 3s
      retries: 5

# environment-manager.py
class TestEnvironmentManager:
    def __init__(self, env_type):
        self.env_type = env_type
        self.services = {}
        self.health_checks = {}
        self.data = {}

    def add_service(self, name, image, port, health_check_cmd):
        self.services[name] = {
            "image": image,
            "port": port,
            "health_check": health_check_cmd,
        }

    def provision(self):
        print(f"===== Provisioning {self.env_type} Environment =====")
        print(f"Services to start: {len(self.services)}")

        for name, svc in self.services.items():
            print(f"  Starting {name} ({svc['image']})...")
            import time
            time.sleep(0.5)
            print(f"    Health check: {svc['health_check']}")
            self.health_checks[name] = True
            print(f"    Status: READY on port {svc['port']}")

        ready = sum(1 for v in self.health_checks.values() if v)
        print(f"\nEnvironment: {self.env_type}")
        print(f"Services ready: {ready}/{len(self.services)}")
        print(f"Status: {'ALL HEALTHY' if ready == len(self.services) else 'DEGRADED'}")

    def destroy(self):
        print(f"\n===== Destroying {self.env_type} Environment =====")
        for name in self.services:
            print(f"  Stopping {name}...")
            self.health_checks.pop(name, None)
        print("Environment fully decommissioned.")

env = TestEnvironmentManager("ephemeral-pr-42")
env.add_service("api", "myapp:latest", 3000, "curl -f http://localhost:3000/health")
env.add_service("postgres", "postgres:16-alpine", 5432, "pg_isready")
env.add_service("redis", "redis:7-alpine", 6379, "redis-cli ping")
env.provision()
env.destroy()

// ephemeral-lifecycle.js
class EphemeralEnvironment {
  constructor(branch, prNumber) {
    this.branch = branch;
    this.prNumber = prNumber;
    this.created = new Date();
    this.timeout = 24 * 60 * 60 * 1000;
    this.status = 'provisioning';
    this.endpoint = null;
  }

  async deploy() {
    console.log(`[${this.branch}] Deploying ephemeral environment...`);

    // Build, deploy, wait for health
    await new Promise(r => setTimeout(r, 500));
    this.endpoint = `https://pr-${this.prNumber}.dev.dodatech.com`;
    this.status = 'running';

    console.log(`  Environment: ${this.endpoint}`);
    console.log(`  Status: ${this.status}`);
    console.log(`  TTL: ${this.timeout / 60000} minutes`);
  }

  async runTests() {
    console.log(`\n[${this.branch}] Running tests...`);
    const tests = [
      { name: 'Unit Tests', status: 'passed', duration: 45 },
      { name: 'Integration Tests', status: 'passed', duration: 120 },
      { name: 'E2E Smoke Tests', status: 'passed', duration: 180 },
      { name: 'Contract Tests', status: 'passed', duration: 60 },
    ];

    let passed = 0;
    tests.forEach(t => {
      console.log(`  ${t.name}: ${t.status} (${t.duration}s)`);
      if (t.status === 'passed') passed++;
    });

    console.log(`\n  Results: ${passed}/${tests.length} passed`);
    return passed === tests.length;
  }

  async destroy() {
    await new Promise(r => setTimeout(r, 300));
    this.status = 'destroyed';
    console.log(`\n[${this.branch}] Environment destroyed.`);
    console.log(`  Lifetime: ${Math.round((Date.now() - this.created) / 1000)}s`);
  }
}

(async () => {
  const env = new EphemeralEnvironment('feature/payment-update', 1234);
  await env.deploy();
  const passed = await env.runTests();
  await env.destroy();
  console.log(`\nPR #1234 merge ${passed ? 'approved' : 'blocked'}`);
})();

class TestDataSeeder:
    def __init__(self, env_name):
        self.env_name = env_name
        self.data = {}

    def seed_core_data(self):
        self.data['users'] = [
            {"id": 1, "email": f"admin@{self.env_name}.com", "role": "admin"},
            {"id": 2, "email": f"user@{self.env_name}.com", "role": "user"},
        ]
        self.data['products'] = [
            {"id": 101, "name": "Widget A", "price": 19.99, "in_stock": True},
            {"id": 102, "name": "Widget B", "price": 29.99, "in_stock": True},
            {"id": 103, "name": "Widget C", "price": 9.99, "in_stock": False},
        ]
        self.data['orders'] = [
            {"id": 1001, "user_id": 2, "product_id": 101,
             "quantity": 2, "status": "completed"},
            {"id": 1002, "user_id": 2, "product_id": 102,
             "quantity": 1, "status": "pending"},
        ]
        print(f"[{self.env_name}] Core data seeded")
        return self

    def seed_specific_scenario(self, scenario):
        scenario_data = {
            "promo-code": {
                "promotions": [
                    {"code": "WELCOME10", "discount": 0.10, "max_uses": 100},
                ]
            },
            "subscription": {
                "plans": [
                    {"id": "free", "price": 0, "features": ["basic"]},
                    {"id": "pro", "price": 29, "features": ["basic", "advanced"]},
                ]
            }
        }
        if scenario in scenario_data:
            self.data.update(scenario_data[scenario])
            print(f"[{self.env_name}] Scenario '{scenario}' loaded")
        return self

    def print_summary(self):
        print(f"\n{self.env_name} Test Data Summary:")
        total_records = sum(len(v) for v in self.data.values())
        print(f"  Entities: {', '.join(self.data.keys())}")
        print(f"  Total records: {total_records}")

seeder = TestDataSeeder("staging")
seeder.seed_core_data().seed_specific_scenario("promo-code").print_summary()

class EnvironmentHealthChecker:
    def __init__(self, env_name):
        self.env_name = env_name
        self.checks = {}

    def add_check(self, name, check_fn, critical=True):
        self.checks[name] = {"fn": check_fn, "critical": critical}

    def run_all(self):
        print(f"=== Health Check: {self.env_name} ===\n")
        all_passed = True
        critical_failures = 0

        for name, check in self.checks.items():
            try:
                result = check["fn"]()
                status = "PASS" if result else "FAIL"
                if not result:
                    all_passed = False
                    if check["critical"]:
                        critical_failures += 1
                print(f"  [{status}] {name}")
            except Exception as e:
                print(f"  [ERROR] {name}: {e}")
                all_passed = False
                if check["critical"]:
                    critical_failures += 1

        print(f"\n  Critical failures: {critical_failures}")
        print(f"  Overall status: {'HEALTHY' if all_passed else 'UNHEALTHY'}")

        if critical_failures > 0:
            print(f"  Action: Stop tests, investigate critical services")

        return all_passed and critical_failures == 0

health = EnvironmentHealthChecker("staging")

def api_health(): return True
def db_connected(): return True
def redis_connected(): return True
def queue_backlog(): return False

health.add_check("API /health", api_health, critical=True)
health.add_check("PostgreSQL connection", db_connected, critical=True)
health.add_check("Redis connection", redis_connected, critical=True)
health.add_check("Queue backlog < 1000", queue_backlog, critical=False)

health.run_all()