20565403496

Committed 29 Dec 2025 05:11AM UTC coverage: 84.103% (-2.8%) from 86.921%

Build # 20565403496

Build Type

Pull #13567

github

Committed by

web-flow

Commit Message

Merge 4816837a5 into 2417384aa

Pull Request Pull Request #13567: Update ASF APIs

Run Details

67166 of 79862 relevant lines covered (84.1%)

0.84 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

80.85

/localstack-core/localstack/utils/serving.py

import abc
import logging
import threading

from localstack.utils.net import is_port_open
from localstack.utils.sync import poll_condition
from localstack.utils.threads import FuncThread, start_thread

LOG = logging.getLogger(__name__)


class StopServer(Exception):
    pass


class Server(abc.ABC):
    """
    A Server implements the lifecycle of a server running in a thread.
    """

    def __init__(self, port: int, host: str = "localhost") -> None:
        super().__init__()
        self._thread: FuncThread | None = None

        self._lifecycle_lock = threading.RLock()
        self._stopped = threading.Event()
        self._started = threading.Event()

        self._host = host
        self._port = port

    @property
    def host(self):
        return self._host

    @property
    def port(self):
        return self._port

    @property
    def protocol(self):
        return "http"

    @property
    def url(self):
        return f"{self.protocol}://{self.host}:{self.port}"

    def get_error(self) -> Exception | None:
        """
        If the thread running the server returned with an Exception, then this function will return that exception.
        """
        if not self._started.is_set():
            return None

        future = self._thread.result_future
        if future.done():
            return future.exception()
        return None

    def wait_is_up(self, timeout: float = None) -> bool:
        """
        Waits until the server is started and is_up returns true.

        :param timeout: the time in seconds to wait before returning false. If timeout is None, then wait indefinitely.
        :returns: true if the server is up, false if not or the timeout was reached while waiting.
        """
        # first wait until the started event was called
        self._started.wait(timeout=timeout)
        # then poll the health check
        return poll_condition(self.is_up, timeout=timeout)

    def is_running(self) -> bool:
        """
        Checks whether the thread holding the server is running. The server may be running but not healthy (
        is_running == True, is_up == False).

        :returns: true if the server thread is running
        """
        if not self._started.is_set():
            return False
        if self._stopped.is_set():
            return False
        return self._thread.running

    def is_up(self) -> bool:
        """
        Checks whether the server is up by executing the health check function.

        :returns: false if the server has not been started or if the health check failed, true otherwise
        """
        if not self._started.is_set():
            return False

        try:
            return True if self.health() else False
        except Exception:
            return False

    def shutdown(self) -> None:
        """
        Attempts to shut down the server by calling the internal do_shutdown method. It only does this if the server
        has been started. Repeated calls to this function have no effect.

        :raises RuntimeError: shutdown was called before start
        """
        with self._lifecycle_lock:
            if not self._started.is_set():
                raise RuntimeError("cannot shutdown server before it is started")
            if self._stopped.is_set():
                return

            self._thread.stop()
            self._stopped.set()
            self.do_shutdown()

    def start(self) -> bool:
        """
        Starts the server by calling the internal do_run method in a new thread, and then returns True. Repeated
        calls to this function have no effect but return False.

        :return: True if the server was started in this call, False if the server was already started previously
        """
        with self._lifecycle_lock:
            if self._started.is_set():
                return False

            self._thread = self.do_start_thread()
            self._started.set()
            return True

    def join(self, timeout=None):
        """
        Waits for the given amount of time until the thread running the server returns. If the server hasn't started
        yet, it first waits for the server to start.

        :params: the time in seconds to wait. If None then wait indefinitely.
        :raises TimeoutError: If the server didn't shut down before the given timeout.
        """
        if not self._started.is_set():
            raise RuntimeError("cannot join server before it is started")

        if not self._started.wait(timeout):
            raise TimeoutError

        try:
            self._thread.result_future.result(timeout)
        except TimeoutError:
            raise
        except Exception:
            # Future.result() will re-raise the exception that was raised in the thread
            return

    def health(self):
        """
        Runs a health check on the server. The default implementation performs is_port_open on the server URL.
        """
        return is_port_open(self.url)

    def do_start_thread(self) -> FuncThread:
        """
        Creates and starts the thread running the server. By default, it calls the do_run method in a FuncThread, but
        can be overridden to if the subclass wants to return its own thread.
        """

        def _run(*_):
            try:
                return self.do_run()
            except StopServer:
                LOG.debug("stopping server %s", self.url)
            finally:
                self._stopped.set()

        return start_thread(_run, name=f"server-{self.__class__.__name__}")

    def do_run(self):
        """
        Runs the server (blocking method). (Needs to be overridden by subclasses of do_start_thread is not overridden).

        :raises StopServer: can be raised by the subclass to indicate the server should be stopped.
        """
        pass

    def do_shutdown(self):
        """
        Called when shutdown() is performed. (Should be overridden by subclasses).
        """
        pass

1	import abc	1✔
2	import logging	1✔
3	import threading	1✔
4
5	from localstack.utils.net import is_port_open	1✔
6	from localstack.utils.sync import poll_condition	1✔
7	from localstack.utils.threads import FuncThread, start_thread	1✔
8
9	LOG = logging.getLogger(__name__)	1✔
10
11
12	class StopServer(Exception):	1✔
13	pass	1✔
14
15
16	class Server(abc.ABC):	1✔
17	"""
18	A Server implements the lifecycle of a server running in a thread.
19	"""
20
21	def __init__(self, port: int, host: str = "localhost") -> None:	1✔
22	super().__init__()	1✔
23	self._thread: FuncThread \| None = None	1✔
24
25	self._lifecycle_lock = threading.RLock()	1✔
26	self._stopped = threading.Event()	1✔
27	self._started = threading.Event()	1✔
28
29	self._host = host	1✔
30	self._port = port	1✔
31
32	@property	1✔
33	def host(self):	1✔
34	return self._host	1✔
35
36	@property	1✔
37	def port(self):	1✔
38	return self._port	1✔
39
40	@property	1✔
41	def protocol(self):	1✔
42	return "http"	1✔
43
44	@property	1✔
45	def url(self):	1✔
46	return f"{self.protocol}://{self.host}:{self.port}"	1✔
47
48	def get_error(self) -> Exception \| None:	1✔
49	"""
50	If the thread running the server returned with an Exception, then this function will return that exception.
51	"""
52	if not self._started.is_set():	×
53	return None	×
54
55	future = self._thread.result_future	×
56	if future.done():	×
57	return future.exception()	×
58	return None	×
59
60	def wait_is_up(self, timeout: float = None) -> bool:	1✔
61	"""
62	Waits until the server is started and is_up returns true.
63
64	:param timeout: the time in seconds to wait before returning false. If timeout is None, then wait indefinitely.
65	:returns: true if the server is up, false if not or the timeout was reached while waiting.
66	"""
67	# first wait until the started event was called
68	self._started.wait(timeout=timeout)	1✔
69	# then poll the health check
70	return poll_condition(self.is_up, timeout=timeout)	1✔
71
72	def is_running(self) -> bool:	1✔
73	"""
74	Checks whether the thread holding the server is running. The server may be running but not healthy (
75	is_running == True, is_up == False).
76
77	:returns: true if the server thread is running
78	"""
79	if not self._started.is_set():	1✔
80	return False	1✔
81	if self._stopped.is_set():	×
82	return False	×
83	return self._thread.running	×
84
85	def is_up(self) -> bool:	1✔
86	"""
87	Checks whether the server is up by executing the health check function.
88
89	:returns: false if the server has not been started or if the health check failed, true otherwise
90	"""
91	if not self._started.is_set():	1✔
92	return False	1✔
93
94	try:	1✔
95	return True if self.health() else False	1✔
96	except Exception:	1✔
97	return False	1✔
98
99	def shutdown(self) -> None:	1✔
100	"""
101	Attempts to shut down the server by calling the internal do_shutdown method. It only does this if the server
102	has been started. Repeated calls to this function have no effect.
103
104	:raises RuntimeError: shutdown was called before start
105	"""
106	with self._lifecycle_lock:	1✔
107	if not self._started.is_set():	1✔
108	raise RuntimeError("cannot shutdown server before it is started")	×
109	if self._stopped.is_set():	1✔
110	return	1✔
111
112	self._thread.stop()	1✔
113	self._stopped.set()	1✔
114	self.do_shutdown()	1✔
115
116	def start(self) -> bool:	1✔
117	"""
118	Starts the server by calling the internal do_run method in a new thread, and then returns True. Repeated
119	calls to this function have no effect but return False.
120
121	:return: True if the server was started in this call, False if the server was already started previously
122	"""
123	with self._lifecycle_lock:	1✔
124	if self._started.is_set():	1✔
125	return False	1✔
126
127	self._thread = self.do_start_thread()	1✔
128	self._started.set()	1✔
129	return True	1✔
130
131	def join(self, timeout=None):	1✔
132	"""
133	Waits for the given amount of time until the thread running the server returns. If the server hasn't started
134	yet, it first waits for the server to start.
135
136	:params: the time in seconds to wait. If None then wait indefinitely.
137	:raises TimeoutError: If the server didn't shut down before the given timeout.
138	"""
139	if not self._started.is_set():	1✔
140	raise RuntimeError("cannot join server before it is started")	×
141
142	if not self._started.wait(timeout):	1✔
143	raise TimeoutError	×
144
145	try:	1✔
146	self._thread.result_future.result(timeout)	1✔
147	except TimeoutError:	×
148	raise	×
149	except Exception:	×
150	# Future.result() will re-raise the exception that was raised in the thread
151	return	×
152
153	def health(self):	1✔
154	"""
155	Runs a health check on the server. The default implementation performs is_port_open on the server URL.
156	"""
157	return is_port_open(self.url)	1✔
158
159	def do_start_thread(self) -> FuncThread:	1✔
160	"""
161	Creates and starts the thread running the server. By default, it calls the do_run method in a FuncThread, but
162	can be overridden to if the subclass wants to return its own thread.
163	"""
164
165	def _run(*_):	1✔
166	try:	1✔
167	return self.do_run()	1✔
168	except StopServer:	1✔
169	LOG.debug("stopping server %s", self.url)	×
170	finally:
171	self._stopped.set()	1✔
172
173	return start_thread(_run, name=f"server-{self.__class__.__name__}")	1✔
174
175	def do_run(self):	1✔
176	"""
177	Runs the server (blocking method). (Needs to be overridden by subclasses of do_start_thread is not overridden).
178
179	:raises StopServer: can be raised by the subclass to indicate the server should be stopped.
180	"""
181	pass	×
182
183	def do_shutdown(self):	1✔
184	"""
185	Called when shutdown() is performed. (Should be overridden by subclasses).
186	"""
187	pass	1✔

localstack / localstack / 20565403496

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous