77a61da760
The libdir is passed down for sources but it is never used in
any of our sources. As this is confusing and we want to eventually
support multiple libdirs remove this code.
It looks like the libdir for soruces was added a long time ago in 8423da3
but there is no indication if/how it is/was supposed to get used and
AFACT from going over the git history it was very used.
SourceService:dispatch() never sends "libdir" to the actual sources,
so it is not an even technically an API break.
135 lines
4 KiB
Python
135 lines
4 KiB
Python
#
|
|
# Runtime Tests for Source Modules
|
|
#
|
|
|
|
import contextlib
|
|
import ctypes
|
|
import http.server
|
|
import json
|
|
import os
|
|
import socketserver
|
|
import subprocess
|
|
import threading
|
|
|
|
import pytest
|
|
|
|
import osbuild.meta
|
|
import osbuild.objectstore
|
|
import osbuild.sources
|
|
from osbuild import host
|
|
|
|
from .. import test
|
|
|
|
|
|
def errcheck(ret, _func, _args):
|
|
if ret == -1:
|
|
e = ctypes.get_errno()
|
|
raise OSError(e, os.strerror(e))
|
|
|
|
|
|
CLONE_NEWNET = 0x40000000
|
|
libc = ctypes.CDLL('libc.so.6', use_errno=True)
|
|
libc.setns.errcheck = errcheck
|
|
|
|
|
|
@contextlib.contextmanager
|
|
def netns():
|
|
# Grab a reference to the current namespace.
|
|
with open("/proc/self/ns/net", encoding="utf8") as oldnet:
|
|
# Create a new namespace and enter it.
|
|
libc.unshare(CLONE_NEWNET)
|
|
try:
|
|
# Up the loopback device in the new namespace.
|
|
subprocess.run(["ip", "link", "set", "up", "dev", "lo"], check=True)
|
|
yield
|
|
finally:
|
|
# Revert to the old namespace, dropping our
|
|
# reference to the new one.
|
|
libc.setns(oldnet.fileno(), CLONE_NEWNET)
|
|
|
|
|
|
@contextlib.contextmanager
|
|
def fileServer(directory):
|
|
with netns():
|
|
# This is leaked until the program exits, but inaccessible after the with
|
|
# due to the network namespace.
|
|
barrier = threading.Barrier(2)
|
|
thread = threading.Thread(target=runFileServer, args=(barrier, directory))
|
|
thread.daemon = True
|
|
thread.start()
|
|
barrier.wait()
|
|
yield
|
|
|
|
|
|
def can_setup_netns() -> bool:
|
|
try:
|
|
with netns():
|
|
return True
|
|
except BaseException: # pylint: disable=broad-exception-caught
|
|
return False
|
|
|
|
|
|
def runFileServer(barrier, directory):
|
|
class Handler(http.server.SimpleHTTPRequestHandler):
|
|
def __init__(self, request, client_address, server):
|
|
super().__init__(request, client_address, server)
|
|
|
|
def translate_path(self, path: str) -> str:
|
|
translated_path = super().translate_path(path)
|
|
common = os.path.commonpath([translated_path, directory])
|
|
translated_path = os.path.join(directory, os.path.relpath(translated_path, common))
|
|
return translated_path
|
|
|
|
def guess_type(self, path):
|
|
try:
|
|
with open(path + ".mimetype", "r", encoding="utf8") as f:
|
|
return f.read().strip()
|
|
except FileNotFoundError:
|
|
pass
|
|
return super().guess_type(path)
|
|
|
|
httpd = socketserver.TCPServer(('', 80), Handler)
|
|
barrier.wait()
|
|
httpd.serve_forever()
|
|
|
|
|
|
def make_test_cases():
|
|
sources = os.path.join(test.TestBase.locate_test_data(), "sources")
|
|
if os.path.exists(sources):
|
|
for source in os.listdir(sources):
|
|
for case in os.listdir(f"{sources}/{source}/cases"):
|
|
yield source, case
|
|
|
|
|
|
def check_case(source, case_options, store):
|
|
with host.ServiceManager() as mgr:
|
|
expects = case_options["expects"]
|
|
if expects == "error":
|
|
with pytest.raises(host.RemoteError):
|
|
source.download(mgr, store)
|
|
elif expects == "success":
|
|
source.download(mgr, store)
|
|
else:
|
|
raise ValueError(f"invalid expectation: {expects}")
|
|
|
|
|
|
@pytest.mark.skipif(not can_setup_netns(), reason="network namespace setup failed")
|
|
@pytest.mark.parametrize("source,case", make_test_cases())
|
|
def test_sources(source, case, tmp_path):
|
|
index = osbuild.meta.Index(os.curdir)
|
|
sources = os.path.join(test.TestBase.locate_test_data(), "sources")
|
|
|
|
with open(f"{sources}/{source}/cases/{case}", encoding="utf8") as f:
|
|
case_options = json.load(f)
|
|
|
|
info = index.get_module_info("Source", source)
|
|
desc = case_options[source]
|
|
items = desc.get("items", {})
|
|
options = desc.get("options", {})
|
|
|
|
src = osbuild.sources.Source(info, items, options)
|
|
|
|
with osbuild.objectstore.ObjectStore(tmp_path) as store, \
|
|
fileServer(test.TestBase.locate_test_data()):
|
|
check_case(src, case_options, store)
|
|
check_case(src, case_options, store)
|