fixture.py

"""Fresh data and service process for every case; never imports implementation."""

import json
import sqlite3
import subprocess
import sys
import tempfile
import time
from pathlib import Path

from .adapter import PaymentAdapter


class PaymentEnvironment:
    def __init__(self, implementation, fault):
        self.temp = tempfile.TemporaryDirectory(prefix="quality-workflow-")
        self.root = Path(self.temp.name)
        self.database = self.root / "payments.sqlite"
        with sqlite3.connect(self.database) as connection:
            connection.executescript("""
                CREATE TABLE orders (id TEXT PRIMARY KEY, status TEXT NOT NULL);
                CREATE TABLE payments (id INTEGER PRIMARY KEY, order_id TEXT NOT NULL);
                INSERT INTO orders VALUES ('paid-order', 'PAID'), ('pending-order', 'PENDING');
                INSERT INTO payments(order_id) VALUES ('paid-order');
            """)
        ready_file = self.root / "ready.json"
        self.process = subprocess.Popen([
            sys.executable, str(implementation), "--db", str(self.database),
            "--ready-file", str(ready_file), "--fault", fault,
        ], stdout=subprocess.DEVNULL, stderr=subprocess.PIPE)
        try:
            deadline = time.monotonic() + 5
            while True:
                if self.process.poll() is not None:
                    raise RuntimeError(self.process.stderr.read().decode("utf-8", errors="replace"))
                try:
                    ready = json.loads(ready_file.read_text(encoding="utf-8"))
                    break
                except (FileNotFoundError, json.JSONDecodeError):
                    if time.monotonic() >= deadline:
                        raise TimeoutError("HTTP service did not start; not a business assertion failure")
                    time.sleep(0.01)
            self.adapter = PaymentAdapter(f"http://{ready['host']}:{ready['port']}", self.database)
        except Exception:
            self.close()
            raise

    def close(self):
        if self.process.poll() is None:
            self.process.terminate()
            try:
                self.process.wait(timeout=3)
            except subprocess.TimeoutExpired:
                self.process.kill()
                self.process.wait(timeout=3)
        self.process.stderr.close()
        self.temp.cleanup()