From cee0615f4d5a10f973c212fd31e64d971bac0c98 Mon Sep 17 00:00:00 2001
From: Michael Vogt <michael.vogt@gmail.com>
Date: Wed, 3 Apr 2024 18:09:16 +0200
Subject: [PATCH] testutil: add http_serve_director() test helper

To test the curl sources it is very useful to have a small httpd
server that can serve an arbitrary directory. This helper will
ensure that via:
```python
with with osbuild.testutil.net.http_serve_directory(fake_httpd_root) as httpd:
    port = httpd.server_port
    # download from http://localhost:{port}/<any-path-under-httpd-root>
```
---
 osbuild/testutil/net.py | 53 +++++++++++++++++++++++++++++++++++++++++
 1 file changed, 53 insertions(+)
 create mode 100644 osbuild/testutil/net.py

diff --git a/osbuild/testutil/net.py b/osbuild/testutil/net.py
new file mode 100644
index 00000000..02be47b4
--- /dev/null
+++ b/osbuild/testutil/net.py
@@ -0,0 +1,53 @@
+#!/usr/bin/python3
+"""
+network related utilities
+"""
+import contextlib
+import http.server
+import socket
+import threading
+
+
+def _get_free_port():
+    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    s.bind(("localhost", 0))
+    return s.getsockname()[1]
+
+
+class SilentHTTPRequestHandler(http.server.SimpleHTTPRequestHandler):
+    def log_message(self, *args, **kwargs):
+        pass
+
+
+class DirHTTPServer(http.server.ThreadingHTTPServer):
+    def __init__(self, *args, directory=None, simulate_failures=0, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.directory = directory
+        self.simulate_failures = simulate_failures
+        self.reqs = 0
+
+    def finish_request(self, request, client_address):
+        self.reqs += 1  # racy on non GIL systems
+        if self.simulate_failures > 0:
+            self.simulate_failures -= 1  # racy on non GIL systems
+            SilentHTTPRequestHandler(
+                request, client_address, self, directory="does-not-exists")
+            return
+        SilentHTTPRequestHandler(
+            request, client_address, self, directory=self.directory)
+
+
+@contextlib.contextmanager
+def http_serve_directory(rootdir, simulate_failures=0):
+    port = _get_free_port()
+    httpd = DirHTTPServer(
+        ("localhost", port),
+        http.server.SimpleHTTPRequestHandler,
+        directory=rootdir,
+        simulate_failures=simulate_failures,
+    )
+    threading.Thread(target=httpd.serve_forever).start()
+    try:
+        yield httpd
+    finally:
+        httpd.shutdown()