mm/lazy_serve
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

handle signals and add tests

Browse Source
This commit is contained in:
Michael Pilosov, PhD 2024-01-25 16:39:48 +00:00
parent 6693911c00
commit cf8eaaaf83
3 changed files with 54 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
lazy_serve.py
View File

@ -1,7 +1,7 @@
import subprocess
import threading
import os
import signal
import subprocess
import threading
class ServerThread(threading.Thread):
@ -30,7 +30,7 @@ class ServerThread(threading.Thread):
self.process.wait()
def serve(servers):
def serve(servers, handle_signals=True):
threads = []
for port, directory in servers:
thread = ServerThread(port, directory)
@ -42,7 +42,9 @@ def serve(servers):
for thread in threads:
thread.stop()
if handle_signals:
signal.signal(signal.SIGINT, signal_handler)
signal.signal(signal.SIGTERM, signal_handler)
# Wait for all threads to complete
for thread in threads:

4
setup.py
View File

@ -1,8 +1,8 @@
from setuptools import setup, find_packages
from setuptools import find_packages, setup
setup(
name="lazy_serve",
version="0.1",
version="0.2",
description="A simple package for effortlessly starting HTTP servers.",
long_description=open("README.md").read(),
long_description_content_type="text/markdown",

46
test_lazy_serve.py Normal file
View File

@ -0,0 +1,46 @@
import os
import subprocess
import tempfile
import time
from multiprocessing import Process
import pytest
from lazy_serve import serve # Replace with the name of your script
@pytest.fixture(scope="module")
def server_directories():
temp_dirs = []
for _ in range(2):
temp_dir = tempfile.TemporaryDirectory()
index_path = os.path.join(temp_dir.name, "index.html")
with open(index_path, "w") as file:
file.write("<html><body>Hello World</body></html>")
temp_dirs.append((9090 + len(temp_dirs), temp_dir))
yield [(port, dir.name) for port, dir in temp_dirs]
for _, temp_dir in temp_dirs:
temp_dir.cleanup()
@pytest.mark.parametrize("handle_signals", (True, False))
def test_serve(server_directories, handle_signals):
args = (server_directories, handle_signals)
server_process = Process(target=serve, args=args)
server_process.start()
time.sleep(3) # Allow servers to start
server_process.terminate()
server_process.join()
for port, _ in server_directories:
with pytest.raises(subprocess.CalledProcessError):
subprocess.check_call(
["python", "-m", "http.server", str(port)],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
if __name__ == "__main__":
test_serve(server_directories())