Repository: osbuild/osbuild
Pull request #1942 (2024-11-25 09:58:23)

sources: fix ostree_sources test to work without `/var/empty`
Not all distros ship `/var/empty` so just create an empty dir on demand as needed.

This also tweaks `test_ostree_source_exists()` into calling `make_repo()` instead of duplicating that code.


Patch:
diff --git a/stages/org.osbuild.users b/stages/org.osbuild.users
index 08fbf287..40392eeb 100755
--- a/stages/org.osbuild.users
+++ b/stages/org.osbuild.users
@@ -1,9 +1,9 @@
 #!/usr/bin/python3
 import os
-import subprocess
 import sys
 
 import osbuild.api
+from osbuild.util.chroot import Chroot
 
 
 def getpwnam(root, name):
@@ -48,7 +48,8 @@ def useradd(
     if expiredate is not None:
         arguments += ["--expiredate", str(expiredate)]
 
-    subprocess.run(["chroot", root, "useradd", *arguments, name], check=True)
+    with Chroot(root) as chroot:
+        chroot.run(["useradd", *arguments, name], check=True)
 
 
 def usermod(root, name, gid=None, groups=None, description=None, home=None, shell=None, password=None, expiredate=None):
@@ -69,7 +70,8 @@ def usermod(root, name, gid=None, groups=None, description=None, home=None, shel
         arguments += ["--expiredate", str(expiredate)]
 
     if arguments:
-        subprocess.run(["chroot", root, "usermod", *arguments, name], check=True)
+        with Chroot(root) as chroot:
+            chroot.run(["usermod", *arguments, name], check=True)
 
 
 def add_ssh_keys(root, user, keys):
@@ -95,7 +97,8 @@ def ensure_homedir(root, name, home):
     if os.path.exists(os.path.join(root, home.lstrip("/"))):
         return
 
-    subprocess.run(["chroot", root, "mkhomedir_helper", name], check=True)
+    with Chroot(root) as chroot:
+        chroot.run(["mkhomedir_helper", name], check=True)
 
 
 def main(tree, options):
@@ -127,7 +130,8 @@ def main(tree, options):
             useradd(tree, name, uid, gid, groups, description, home, shell, password, expiredate)
 
         if force_password_reset:
-            subprocess.run(["chroot", tree, "passwd", "--expire", name], check=True)
+            with Chroot(tree) as chroot:
+                chroot.run(["passwd", "--expire", name], check=True)
 
         # following maintains backwards compatibility for handling a single ssh key
         key = user_options.get("key")   # Public SSH key


Test patch:
diff --git a/sources/test/test_ostree_source.py b/sources/test/test_ostree_source.py
index cfa990c0..6909390d 100644
--- a/sources/test/test_ostree_source.py
+++ b/sources/test/test_ostree_source.py
@@ -1,6 +1,7 @@
 #!/usr/bin/python3
 
 import pathlib
+import tempfile
 
 from osbuild.testutil.net import http_serve_directory, https_serve_directory
 from osbuild.util import ostree
@@ -16,9 +17,9 @@ def test_ostree_source_not_exists(tmp_path, sources_service):
 
 def test_ostree_source_exists(tmp_path, sources_service):
     sources_service.setup({"cache": tmp_path, "options": {}})
-    repo = tmp_path / "org.osbuild.ostree" / "repo"
-    commit = ostree.cli("commit", f"--repo={repo}", "--orphan", "/var/empty")
-    assert sources_service.exists("sha256:" + commit.stdout, None)
+    root = tmp_path / "org.osbuild.ostree" / "repo"
+    commit = make_repo(root)
+    assert sources_service.exists("sha256:" + commit, None)
 
 
 def make_test_sources(proto, port, fake_commit, **secrets):
@@ -35,8 +36,9 @@ def make_test_sources(proto, port, fake_commit, **secrets):
 
 
 def make_repo(root):
-    ostree.cli("init", f"--repo={root}")
-    return ostree.cli("commit", f"--repo={root}", "--orphan", "/var/empty").stdout.rstrip()
+    with tempfile.TemporaryDirectory() as empty_tmpdir:
+        ostree.cli("init", f"--repo={root}")
+        return ostree.cli("commit", f"--repo={root}", "--orphan", empty_tmpdir).stdout.rstrip()
 
 
 def test_ostree_pull_plain(tmp_path, sources_service):
diff --git a/stages/test/test_users.py b/stages/test/test_users.py
index 58b681e5..7915696b 100644
--- a/stages/test/test_users.py
+++ b/stages/test/test_users.py
@@ -7,7 +7,6 @@ import pytest
 from osbuild.testutil import (
     assert_jsonschema_error_contains,
     make_fake_tree,
-    mock_command,
 )
 
 STAGE_NAME = "org.osbuild.users"
@@ -57,95 +56,111 @@ def test_users_happy(mocked_run, tmp_path, stage_module, user_opts, expected_arg
     }
     options["users"]["foo"].update(user_opts)
 
-    stage_module.main(tmp_path, options)
+    stage_module.main(tmp_path.as_posix(), options)
 
-    assert len(mocked_run.call_args_list) == 1
-    args, kwargs = mocked_run.call_args_list[0]
-    assert args[0] == ["chroot", tmp_path, "useradd"] + expected_args + ["foo"]
+    # We expect 7 calls to run(): 3 mount + chroot + 3 umount
+    assert len(mocked_run.call_args_list) == 7
+    args, kwargs = mocked_run.call_args_list[3]  # chroot is the 4th call
+    assert args[0] == ["chroot", tmp_path.as_posix(), "useradd"] + expected_args + ["foo"]
     assert kwargs.get("check")
 
 
+@patch("subprocess.run")
 @pytest.mark.parametrize("user_opts,expected_args", TEST_CASES)
-def test_users_mock_bin(tmp_path, stage_module, user_opts, expected_args):
-    with mock_command("chroot", "") as mocked_chroot:
-        make_fake_tree(tmp_path, {
-            "/etc/passwd": "",
-        })
-
-        options = {
-            "users": {
-                "foo": {},
-            }
+def test_users_mock_bin(mocked_run, tmp_path, stage_module, user_opts, expected_args):
+    make_fake_tree(tmp_path, {
+        "/etc/passwd": "",
+    })
+
+    options = {
+        "users": {
+            "foo": {},
         }
-        options["users"]["foo"].update(user_opts)
+    }
+    options["users"]["foo"].update(user_opts)
 
-        stage_module.main(tmp_path, options)
-        assert len(mocked_chroot.call_args_list) == 1
-        assert mocked_chroot.call_args_list[0][2:] == expected_args + ["foo"]
+    stage_module.main(tmp_path.as_posix(), options)
+    # We expect 7 calls to run(): 3 mount + chroot + 3 umount
+    assert len(mocked_run.call_args_list) == 7
+    args, kwargs = mocked_run.call_args_list[3]  # chroot is the 4th call
+    assert args[0] == ["chroot", tmp_path.as_posix(), "useradd"] + expected_args + ["foo"]
+    assert kwargs.get("check")
 
 
 # separate test right now as it results in two binaries being called
 # (adduser,passwd) which our parameter tests cannot do yet
 
+@patch("subprocess.run")
+def test_users_with_password_reset_none(mocked_run, tmp_path, stage_module):
+    make_fake_tree(tmp_path, {
+        "/etc/passwd": "",
+    })
 
-def test_users_with_password_reset_none(tmp_path, stage_module):
-    with mock_command("chroot", "") as mocked_chroot:
-        make_fake_tree(tmp_path, {
-            "/etc/passwd": "",
-        })
-
-        options = {
-            "users": {
-                "foo": {
-                },
-            }
+    options = {
+        "users": {
+            "foo": {
+            },
         }
+    }
 
-        stage_module.main(tmp_path, options)
-        assert len(mocked_chroot.call_args_list) == 1
-        assert mocked_chroot.call_args_list[0][1:] == ["useradd", "foo"]
+    stage_module.main(tmp_path.as_posix(), options)
+    # We expect 7 calls to run(): 3 mount + chroot + 3 umount
+    assert len(mocked_run.call_args_list) == 7
+    args, kwargs = mocked_run.call_args_list[3]  # chroot is the 4th call
+    assert args[0] == ["chroot", tmp_path.as_posix(), "useradd", "foo"]
+    assert kwargs.get("check")
 
 # separate test right now as it results in two binaries being called
 # (adduser,passwd) which our parameter tests cannot do yet
 
 
-def test_users_with_password_reset_false(tmp_path, stage_module):
-    with mock_command("chroot", "") as mocked_chroot:
-        make_fake_tree(tmp_path, {
-            "/etc/passwd": "",
-        })
+@patch("subprocess.run")
+def test_users_with_password_reset_false(mocked_run, tmp_path, stage_module):
+    make_fake_tree(tmp_path, {
+        "/etc/passwd": "",
+    })
 
-        options = {
-            "users": {
-                "foo": {
-                    "force_password_reset": False,
-                },
-            }
+    options = {
+        "users": {
+            "foo": {
+                "force_password_reset": False,
+            },
         }
+    }
 
-        stage_module.main(tmp_path, options)
-        assert len(mocked_chroot.call_args_list) == 1
-        assert mocked_chroot.call_args_list[0][1:] == ["useradd", "foo"]
+    stage_module.main(tmp_path.as_posix(), options)
+    # We expect 7 calls to run(): 3 mount + chroot + 3 umount
+    assert len(mocked_run.call_args_list) == 7
+    args, kwargs = mocked_run.call_args_list[3]  # chroot is the 4th call
+    assert args[0] == ["chroot", tmp_path.as_posix(), "useradd", "foo"]
+    assert kwargs.get("check")
 
 # separate test right now as it results in two binaries being called
 # (adduser,passwd) which our parameter tests cannot do yet
 
 
-def test_users_with_password_reset_true(tmp_path, stage_module):
-    with mock_command("chroot", "") as mocked_chroot:
-        make_fake_tree(tmp_path, {
-            "/etc/passwd": "",
-        })
+@patch("subprocess.run")
+def test_users_with_password_reset_true(mocked_run, tmp_path, stage_module):
+    make_fake_tree(tmp_path, {
+        "/etc/passwd": "",
+    })
 
-        options = {
-            "users": {
-                "foo": {
-                    "force_password_reset": True,
-                },
-            }
+    options = {
+        "users": {
+            "foo": {
+                "force_password_reset": True,
+            },
         }
+    }
 
-        stage_module.main(tmp_path, options)
-        assert len(mocked_chroot.call_args_list) == 2
-        assert mocked_chroot.call_args_list[0][1:] == ["useradd", "foo"]
-        assert mocked_chroot.call_args_list[1][1:] == ["passwd", "--expire", "foo"]
+    stage_module.main(tmp_path.as_posix(), options)
+    # We expect 14 calls to run(): 2x (3 mount + chroot + 3 umount)
+    assert len(mocked_run.call_args_list) == 14
+
+    args, kwargs = mocked_run.call_args_list[3]  # chroot is the 4th call
+    assert args[0] == ["chroot", tmp_path.as_posix(), "useradd", "foo"]
+    assert kwargs.get("check")
+
+    args, kwargs = mocked_run.call_args_list[10]  # chroot is the 11th call
+    assert args[0] == ["chroot", tmp_path.as_posix(), "passwd", "--expire", "foo"]
+    assert kwargs.get("check")