Python snipets

background_runner.py

import asyncio
import subprocess
import threading
import time

class BackgroundRunner:
    def __init__(self):
        self.proc = None
        self.is_running = False
        self.count = 0
        self.search_string = None

    def start(self, command: list[str], search_string: str = ''):
        """
        Parameters
        ----------
        command : list[str]
            Command and its arguments as supplied to subprocess.Popen
        search_string : str
            Search string to search in command output that determines
            that program is up and running (e.g. 'listening on port 8000')
        """
        self.search_string = search_string
        thread = threading.Thread(target=self._run, args=[command])
        thread.daemon = True
        thread.start()
        if (search_string == ''):
            return
        while (not self.is_running):
            time.sleep(0.1)

    def _run(self, command: list[str]):
        self.proc = subprocess.Popen(
            command,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True
        )
        asyncio.run(self.print_stream())

    async def print_stream(self):
        if (self.proc):
            while (self.proc.poll() is None):
                if (self.proc.stderr):
                    for str in self.proc.stderr:
                        self.handle_str(str)
                if (self.proc.stdout):
                    for str in self.proc.stdout:
                        self.handle_str(str)
                await asyncio.sleep(0.1)

    def handle_str(self, str: str):
        print(str, end="")
        if (self.search_string is None):
            return
        if (self.search_string in str):
            self.is_running = True

    def stop(self):
        if (self.proc):
            self.proc.terminate()

mock_server.py

from http.server import BaseHTTPRequestHandler, HTTPServer
import json
import threading
from typing import Tuple


class SimplifiedRequest:
    def __init__(self, method: str, url: str, headers, body=''):
        self.method = method
        self.url = url
        self.headers = headers
        self.body = body

    def __repr__(self):
        return f'{self.method}: {self.url}\n{self.headers}\n{self.body}'


def make_handler(return_data: dict, code: int):
    req_queue = []

    class MockHandler(BaseHTTPRequestHandler):
        def do_POST(self):
            self.send_response(code)
            self.send_header('Content-type', 'application/json')
            self.end_headers()
            content_length = int(self.headers['Content-Length'])
            req_queue.append(
                SimplifiedRequest(
                    'POST',
                    self.path,
                    self.headers,
                    self.rfile.read(content_length).decode('utf-8')
                ))
            self.wfile.write(json.dumps(return_data).encode('utf-8'))
        def do_GET(self):
            self.send_response(code)
            self.send_header('Content-type', 'application/json')
            self.end_headers()
            req_queue.append(
                SimplifiedRequest(
                    'GET',
                    self.path,
                    self.headers,
                ))
            self.wfile.write(json.dumps(return_data).encode('utf-8'))
    return MockHandler, req_queue


def mock_http_server(port: int, return_data: dict, code=200) -> Tuple[HTTPServer, list[SimplifiedRequest]]:
    MockHandler, req_queue = make_handler(return_data, code)
    httpd = HTTPServer(('localhost', port), MockHandler)
    thread = threading.Thread(target=httpd.serve_forever)
    thread.daemon = True
    thread.start()
    return httpd, req_queue

Looks like for mocking http servers there's a good option: https://httpretty.readthedocs.io/en/latest/