particle-os/debian-forge
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

test: add/use new testutil.make_fake_service_fd()

Browse source 
All inputs/sources tests need a fake service fd to instanciate
their services. Consolidate the creation in a single helper.
This commit is contained in:
Michael Vogt 2024-02-27 20:33:19 +01:00 • committed by Achilleas Koutsou
parent cccdf8b784
commit 5f31ccf9f2
4 changed files with 33 additions and 19 deletions
Download patch file Download diff file Expand all files Collapse all files

6
inputs/test/test_containers.py
View file

@ -2,12 +2,12 @@
import os
import pathlib
import socket
import subprocess
import tempfile
import pytest
from osbuild import testutil
from osbuild.testutil import has_executable, make_container
INPUTS_NAME = "org.osbuild.containers-storage"
@ -40,8 +40,8 @@ def test_containers_local_inputs_integration(tmp_path, inputs_module):
}
}
}
sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
cnt_inputs = inputs_module.ContainersStorageInput.from_args(["--service-fd", str(sock.fileno())])
fd = testutil.make_fake_service_fd()
cnt_inputs = inputs_module.ContainersStorageInput.from_args(["--service-fd", str(fd)])
store = FakeStoreClient(tmp_path / "fake-sources")
# not using "tmp_path" here as it will "rm -rf" on cleanup and
# that is dangerous as during the tests we bind mount the

13
osbuild/testutil/__init__.py
View file

@ -6,6 +6,7 @@ import os
import pathlib
import re
import shutil
import socket
import subprocess
import tempfile
import textwrap
@ -155,3 +156,15 @@ def pull_oci_archive_container(archive_path, image_name):
yield
finally:
subprocess.check_call(["skopeo", "delete", f"containers-storage:{image_name}"])
def make_fake_service_fd() -> int:
"""Create a file descriptor suitable as input for --service-fd for any
host.Service
Note that the service will take over the fd and take care of the
lifecycle so no need to close it.
"""
sock = socket.socket(socket.AF_UNIX, socket.SOCK_SEQPACKET)
fd = os.dup(sock.fileno())
return fd

14
sources/test/test_container_storage_source.py
View file

@ -1,11 +1,11 @@
#!/usr/bin/python3
import os
import socket
import subprocess
import pytest
from osbuild import testutil
from osbuild.testutil import has_executable, make_container
SOURCES_NAME = "org.osbuild.containers-storage"
@ -20,8 +20,8 @@ def test_containers_storage_integration(tmp_path, sources_module):
image_id = subprocess.check_output(["podman", "inspect", "-f", "{{ .Id }}", base_tag],
universal_newlines=True).strip()
checksum = f"sha256:{image_id}"
sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
cnt_storage = sources_module.ContainersStorageSource.from_args(["--service-fd", str(sock.fileno())])
fd = testutil.make_fake_service_fd()
cnt_storage = sources_module.ContainersStorageSource.from_args(["--service-fd", str(fd)])
assert cnt_storage.exists(checksum, None)
@ -29,8 +29,8 @@ def test_containers_storage_integration(tmp_path, sources_module):
@pytest.mark.skipif(os.getuid() != 0, reason="root only")
def test_containers_storage_integration_missing(sources_module):
checksum = "sha256:1234567890123456789012345678901234567890909b14ffb032aa20fa23d9ad6"
sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
cnt_storage = sources_module.ContainersStorageSource.from_args(["--service-fd", str(sock.fileno())])
fd = testutil.make_fake_service_fd()
cnt_storage = sources_module.ContainersStorageSource.from_args(["--service-fd", str(fd)])
assert not cnt_storage.exists(checksum, None)
@ -40,8 +40,8 @@ def test_containers_storage_integration_invalid(sources_module):
# put an invalid reference into the source to ensure skopeo errors with
# a different error than image not found
checksum = "sha256:["
sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
cnt_storage = sources_module.ContainersStorageSource.from_args(["--service-fd", str(sock.fileno())])
fd = testutil.make_fake_service_fd()
cnt_storage = sources_module.ContainersStorageSource.from_args(["--service-fd", str(fd)])
with pytest.raises(RuntimeError) as exc:
cnt_storage.exists(checksum, None)
assert "unknown skopeo error:" in str(exc)

19
sources/test/test_curl_source.py
View file

@ -3,17 +3,18 @@
import contextlib
import os
import pathlib
import socket
import tempfile
import pytest
from osbuild import testutil
SOURCES_NAME = "org.osbuild.curl"
def test_curl_source_not_exists(sources_module):
sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
curl_source = sources_module.CurlSource.from_args(["--service-fd", str(sock.fileno())])
fd = testutil.make_fake_service_fd()
curl_source = sources_module.CurlSource.from_args(["--service-fd", str(fd)])
tmpdir = tempfile.TemporaryDirectory()
curl_source.cache = tmpdir.name
desc = {
@ -24,8 +25,8 @@ def test_curl_source_not_exists(sources_module):
def test_curl_source_exists(sources_module):
sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
curl_source = sources_module.CurlSource.from_args(["--service-fd", str(sock.fileno())])
fd = testutil.make_fake_service_fd()
curl_source = sources_module.CurlSource.from_args(["--service-fd", str(fd)])
tmpdir = tempfile.TemporaryDirectory()
curl_source.cache = tmpdir.name
desc = {
@ -37,8 +38,8 @@ def test_curl_source_exists(sources_module):
def test_curl_source_amend_secrets(sources_module):
sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
curl_source = sources_module.CurlSource.from_args(["--service-fd", str(sock.fileno())])
fd = testutil.make_fake_service_fd()
curl_source = sources_module.CurlSource.from_args(["--service-fd", str(fd)])
tmpdir = tempfile.TemporaryDirectory()
curl_source.cache = tmpdir.name
desc = {
@ -65,8 +66,8 @@ def test_curl_source_amend_secrets(sources_module):
def test_curl_source_amend_secrets_fail(sources_module):
sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
curl_source = sources_module.CurlSource.from_args(["--service-fd", str(sock.fileno())])
fd = testutil.make_fake_service_fd()
curl_source = sources_module.CurlSource.from_args(["--service-fd", str(fd)])
tmpdir = tempfile.TemporaryDirectory()
curl_source.cache = tmpdir.name
desc = {