File: utilities.py

package info (click to toggle)
python-moto 5.1.18-3
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid
  • size: 116,520 kB
  • sloc: python: 636,725; javascript: 181; makefile: 39; sh: 3
file content (35 lines) | stat: -rw-r--r-- 1,108 bytes parent folder | download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
from http.server import BaseHTTPRequestHandler, HTTPServer
from threading import Event, Thread
from typing import Optional


class SimpleServer:
    def __init__(self, handler: type[BaseHTTPRequestHandler]) -> None:
        self._port = 0
        self._handler = handler

        self._thread: Optional[Thread] = None
        self._ip_address = "0.0.0.0"
        self._server: Optional[HTTPServer] = None
        self._server_ready_event = Event()

    def _server_entry(self) -> None:
        self._server = HTTPServer(("0.0.0.0", 0), self._handler)
        self._server_ready_event.set()
        self._server.serve_forever()

    def start(self) -> None:
        self._thread = Thread(target=self._server_entry, daemon=True)
        self._thread.start()
        self._server_ready_event.wait()

    def get_host_and_port(self) -> tuple[str, int]:
        assert self._server
        host, port = self._server.server_address[:2]
        return str(host), port

    def stop(self) -> None:
        if self._server:
            self._server.shutdown()
        if self._thread:
            self._thread.join()