diff --git a/examples/like_dbg_confs/echo_module_arm64.ini b/examples/like_dbg_confs/echo_module_arm64.ini new file mode 100644 index 0000000..cd172da --- /dev/null +++ b/examples/like_dbg_confs/echo_module_arm64.ini @@ -0,0 +1,9 @@ +[general] +arch = arm64 + +[kernel_builder] +## Provide a path to a parent directory that houses custom kernel modules (see the example) +custom_modules = examples/c_kmod/ + +[kernel_dl] +tag = 5.15 diff --git a/src/debuggee.py b/src/debuggee.py index f540dc6..827adef 100644 --- a/src/debuggee.py +++ b/src/debuggee.py @@ -49,11 +49,9 @@ def infer_panic_behavior(self) -> int: return 0 elif "wait" in self.panic: try: - ret = self.panic.split(" ")[1] - if not ret: - return 15 + ret = int(self.panic.split(" ")[1]) return ret - except IndexError: + except (IndexError, ValueError): return 15 else: logger.error("Unknown requested panic behavior...") @@ -72,7 +70,6 @@ def run_container(self): self.cmd = f"qemu-system-{self.qemu_arch} -m {self.memory} -smp {self.smp} -kernel {kernel}" if self.qemu_arch == "aarch64": self.cmd += " -cpu cortex-a72" - # self._add_smep_smap() self.cmd += ' -machine type=virt -append "console=ttyAMA0 root=/dev/vda' elif self.qemu_arch == "x86_64": self.cmd += " -cpu qemu64" diff --git a/src/debugger.py b/src/debugger.py index 1feee0c..64ee76e 100644 --- a/src/debugger.py +++ b/src/debugger.py @@ -20,32 +20,39 @@ def __init__(self, **kwargs): super().__init__(**kwargs) user_cfg = kwargs.get("user_cfg", "") cfg_setter(self, ["general", "debugger"], user_cfg, exclude_keys=["kernel_root"]) - ctf_ctx = kwargs.get("ctf_ctx", False) - if ctf_ctx: - self.ctf_kernel = Path(kwargs.get("ctf_kernel", "")) - self.project_dir = Path.cwd() / self.ctf_dir - vmlinux = Path(self.project_dir) / "vmlinux" - if not vmlinux.exists() or b"ELF" not in sp.run(f"file {vmlinux}", shell=True, capture_output=True).stdout: - self._extract_vmlinux() + if kwargs.get("ctf_ctx", False): + self.ctf = True + self._set_ctf_ctx(kwargs) else: + self.ctf = False self.project_dir = Path.cwd() / self.kernel_root - self.ctf = 1 if ctf_ctx else 0 self.custom_gdb_script = Path("/home/") / self.user / Path(self.gdb_script).name self.script_logging = "set -e" if kwargs.get("log_level", "INFO") == "INFO" else "set -eux" self.skip_prompts = kwargs.get("skip_prompts", False) - def _extract_vmlinux(self) -> None: + def _set_ctf_ctx(self, kwargs) -> None: + self.ctf_kernel = Path(kwargs.get("ctf_kernel", "")) + self.project_dir = Path(self.ctf_dir).resolve().absolute() + vmlinux = Path(self.project_dir) / "vmlinux" + if not vmlinux.exists() or b"ELF" not in sp.run(f"file {vmlinux}", shell=True, capture_output=True).stdout: + if self._extract_vmlinux(): + exit(-1) + + def _extract_vmlinux(self) -> int: + vml_ext = Path(glob("**/extract-vmlinux.sh", recursive=True)[0]).resolve().absolute() + pkernel = self.ctf_kernel.resolve().absolute() with new_context(self.ctf_dir): - vml_ext = glob("**/extract*", recursive=True)[0] - ret = sp.run(f"./{vml_ext} {Path(*self.ctf_kernel.parts[1:])} > vmlinux", shell=True, capture_output=True) + cmd = f"{vml_ext} {pkernel} > vmlinux" + ret = sp.run(f"{cmd}", shell=True, capture_output=True) if ret.returncode == 0: logger.info("Successfully extracted 'vmlinux' from compressed kernel") + return 0 else: logger.error("Failed to extract 'vmlinux'") - exit(-1) + return 1 def run_container(self) -> None: - entrypoint = f'/bin/bash -c "{self.script_logging}; . /home/{self.user}/debugger.sh -a {self.arch} -p {self.docker_mnt} -c {self.ctf} -g {self.custom_gdb_script}"' + entrypoint = f'/bin/bash -c "{self.script_logging}; . /home/{self.user}/debugger.sh -a {self.arch} -p {self.docker_mnt} -c {int(self.ctf)} -g {self.custom_gdb_script}"' runner = f'docker run -it --rm --security-opt seccomp=unconfined --cap-add=SYS_PTRACE -v {self.project_dir}:/io --net="host" {self.tag} {entrypoint}' tmux("selectp -t 2") tmux_shell(runner) diff --git a/src/rootfs_builder.py b/src/rootfs_builder.py index 18d74ac..5151765 100644 --- a/src/rootfs_builder.py +++ b/src/rootfs_builder.py @@ -15,7 +15,7 @@ class RootFSBuilder(DockerRunner): def __init__(self, partial_run: bool = False, **kwargs) -> None: super().__init__(**kwargs) user_cfg = kwargs.get("user_cfg", "") - cfg_setter(self, ["rootfs_general", "rootfs_builder", "general"], user_cfg) + cfg_setter(self, ["general", "rootfs_general", "rootfs_builder"], user_cfg) self.partial = partial_run self.fs_name = self.rootfs_base + self.arch + self.rootfs_ftype self.rootfs_path = self.rootfs_dir + self.fs_name diff --git a/src/tests/files/testKernel_packed b/src/tests/files/testKernel_packed new file mode 100644 index 0000000..29c347b Binary files /dev/null and b/src/tests/files/testKernel_packed differ diff --git a/src/tests/test_debuggee.py b/src/tests/test_debuggee.py index e69de29..c07fdee 100644 --- a/src/tests/test_debuggee.py +++ b/src/tests/test_debuggee.py @@ -0,0 +1,211 @@ +from pathlib import Path + +from src.debuggee import Debuggee +from unittest.mock import patch, MagicMock +import pytest + + +@patch("subprocess.run") +def test_infer_qemu_fs_mount_cpio(sp_mock) -> None: + d = Debuggee(**{"kroot": "foo"}) + mock = MagicMock() + mock.configure_mock(**{"stdout": b"A cpio archive dummy archive"}) + sp_mock.return_value = mock + assert d.infer_qemu_fs_mount() == f" -initrd {d.rootfs}" + + +@patch("subprocess.run") +def test_infer_qemu_fs_mount_filesystem(sp_mock) -> None: + d = Debuggee(**{"kroot": "foo"}) + mock = MagicMock() + mock.configure_mock(**{"stdout": b"Some filesystem data..."}) + sp_mock.return_value = mock + assert d.infer_qemu_fs_mount() == f" -drive file={d.rootfs},format=raw" + + +@patch("subprocess.run") +def test_infer_qemu_fs_mount_error(sp_mock) -> None: + d = Debuggee(**{"kroot": "foo"}) + mock = MagicMock() + mock.configure_mock(**{"stdout": b"foo bar baz"}) + sp_mock.return_value = mock + with pytest.raises(SystemExit) as ext: + d.infer_qemu_fs_mount() + assert ext.type == SystemExit + assert ext.value.code == -1 + + +@patch("subprocess.run") +def test_infer_qemu_fs_mount_cpio_ctf(sp_mock) -> None: + d = Debuggee(**{"kroot": "foo", "ctf_ctx": True}) + mock = MagicMock() + mock.configure_mock(**{"stdout": b"A cpio archive dummy archive"}) + sp_mock.return_value = mock + assert d.infer_qemu_fs_mount() == f" -initrd {d.rootfs.name}" + + +def test_assert_ctf_ctx_mode() -> None: + d = Debuggee(**{"kroot": "foo", "ctf_ctx": True, "ctf_mount": "/foo", "ctf_kernel": "/a/path", "ctf_fs": "/another/path"}) + assert d.ctf is True + assert d.ctf_mount == "/foo" + assert d.kernel == Path("/a/path") + assert d.rootfs == Path("/another/path") + + +@patch("src.misc.cfg_setter", return_value=None) +def test_assert_normal_mode(self) -> None: + d = Debuggee(**{"kroot": "foo", "ctf_ctx": False}) + d.kernel_root = "/foo" + d.arch = "x86_64" + d.rootfs_ftype = "" + assert d.ctf is False + assert d.kernel == Path(f"{d.docker_mnt}/{d.kernel_root}/arch/{d.arch}/boot/Image") + assert d.rootfs == Path(f"{d.docker_mnt}/{d.rootfs_dir}/{d.rootfs_base + d.arch + d.rootfs_ftype}") + + +def test_infer_panic_behavior_panic() -> None: + d = Debuggee(**{"kroot": "foo", "ctf_ctx": False}) + d.panic = "reboot" + assert d.infer_panic_behavior() == -1 + + +def test_infer_panic_behavior_halt() -> None: + d = Debuggee(**{"kroot": "foo", "ctf_ctx": False}) + d.panic = "halt" + assert d.infer_panic_behavior() == 0 + + +def test_infer_panic_behavior_wait_90() -> None: + d = Debuggee(**{"kroot": "foo", "ctf_ctx": False}) + d.panic = "wait 90" + assert d.infer_panic_behavior() == 90 + + +def test_infer_panic_behavior_wait_split_fail() -> None: + d = Debuggee(**{"kroot": "foo", "ctf_ctx": False}) + d.panic = "wait" + assert d.infer_panic_behavior() == 15 + + +def test_infer_panic_behavior_wait_conversion_fail() -> None: + d = Debuggee(**{"kroot": "foo", "ctf_ctx": False}) + d.panic = "wait a" + assert d.infer_panic_behavior() == 15 + + +def test_infer_panic_behavior_wait_unknown() -> None: + d = Debuggee(**{"kroot": "foo", "ctf_ctx": False}) + d.panic = "foo" + with pytest.raises(SystemExit) as ext: + d.infer_panic_behavior() + assert ext.type == SystemExit + assert ext.value.code == -1 + + +def test_add_smep() -> None: + d = Debuggee(**{"kroot": "foo", "ctf_ctx": False}) + d.smep = True + d.smap = False + tmp = "some cmd -cpu foo" + d.cmd = tmp + d._add_smep_smap() + assert d.cmd == f"{tmp},+smep" + + +def test_add_smap() -> None: + d = Debuggee(**{"kroot": "foo", "ctf_ctx": False}) + d.smep = False + d.smap = True + tmp = "some cmd -cpu foo" + d.cmd = tmp + d._add_smep_smap() + assert d.cmd == f"{tmp},+smap" + + +def test_add_smep_smap() -> None: + d = Debuggee(**{"kroot": "foo", "ctf_ctx": False}) + d.smep = True + d.smap = True + tmp = "some cmd -cpu foo" + d.cmd = tmp + d._add_smep_smap() + assert d.cmd == f"{tmp},+smep,+smap" + + +@patch("src.misc.tmux") +@patch("src.misc.tmux_shell") +@patch("termios.tcflush", return_value=True) +@patch("builtins.input", lambda *args: "y") +@patch.object(Debuggee, "infer_qemu_fs_mount", return_value=" -initrd /foo/bar.cpio") +def test_run_x86_all_mitigations_kvm_gdb(tmock, tsmock, infer_mock, flush_mock) -> None: + d = Debuggee(**{"kroot": "foo", "ctf_ctx": False}) + d.kaslr = True + d.smep = True + d.smap = True + d.kpti = True + d.kvm = True + d.arch = "x86_64" + d.kernel = "/some/kernel/Image" + d.rootfs = Path("/foo/rootfs") + d.run() + assert ( + d.cmd + == 'qemu-system-x86_64 -m 1024 -smp 1 -kernel /some/kernel/Image -cpu qemu64,+smep,+smap -append "console=ttyS0 root=/dev/sda earlyprintk=serial net.ifnames=0 kaslr pti=on oops=panic panic=0" -initrd /foo/bar.cpio -net user,host=10.0.2.10,hostfwd=tcp:127.0.0.1:10021-:22 -net nic,model=e1000 -nographic -pidfile vm.pid -enable-kvm -S -s' + ) + + +@patch("src.misc.tmux") +@patch("src.misc.tmux_shell") +@patch("termios.tcflush", return_value=True) +@patch("builtins.input", lambda *args: "y") +@patch.object(Debuggee, "infer_qemu_fs_mount", return_value=" -initrd /foo/bar.cpio") +def test_run_x86_no_mitigations_kvm_gdb(tmock, tsmock, infer_mock, flush_mock) -> None: + d = Debuggee(**{"kroot": "foo", "ctf_ctx": False}) + d.kaslr = False + d.smep = False + d.smap = False + d.kpti = False + d.kvm = False + d.gdb = False + d.arch = "x86_64" + d.kernel = "/some/kernel/Image" + d.rootfs = Path("/foo/rootfs") + d.run() + assert ( + d.cmd + == 'qemu-system-x86_64 -m 1024 -smp 1 -kernel /some/kernel/Image -cpu qemu64 -append "console=ttyS0 root=/dev/sda earlyprintk=serial net.ifnames=0 nokaslr nosmep nosmap nopti oops=panic panic=0" -initrd /foo/bar.cpio -net user,host=10.0.2.10,hostfwd=tcp:127.0.0.1:10021-:22 -net nic,model=e1000 -nographic -pidfile vm.pid' + ) + + +@patch("src.misc.tmux") +@patch("src.misc.tmux_shell") +@patch("termios.tcflush", return_value=True) +@patch("builtins.input", lambda *args: "y") +@patch.object(Debuggee, "infer_qemu_fs_mount", return_value=" -initrd /foo/bar.cpio") +def test_run_arm_no_mitigations_kvm_on(tmock, tsmock, infer_mock, flush_mock) -> None: + d = Debuggee(**{"kroot": "foo", "ctf_ctx": False}) + d.kaslr = False + d.smep = False + d.smap = False + d.kpti = False + d.kvm = True + d.gdb = False + d.qemu_arch = "aarch64" + d.kernel = "/some/kernel/Image" + d.rootfs = Path("/foo/rootfs") + d.run() + assert ( + d.cmd + == 'qemu-system-aarch64 -m 1024 -smp 1 -kernel /some/kernel/Image -cpu cortex-a72 -machine type=virt -append "console=ttyAMA0 root=/dev/vda earlyprintk=serial net.ifnames=0 nokaslr nosmep nosmap nopti oops=panic panic=0" -initrd /foo/bar.cpio -net user,host=10.0.2.10,hostfwd=tcp:127.0.0.1:10021-:22 -net nic,model=e1000 -nographic -pidfile vm.pid' + ) + + +@patch("termios.tcflush", return_value=True) +@patch("builtins.input", lambda *args: "y") +def test_run_unknown_arch(flush_mock) -> None: + d = Debuggee(**{"kroot": "foo", "ctf_ctx": False}) + d.qemu_arch = "foobar" + with pytest.raises(SystemExit) as ext: + d.run() + assert ext.type == SystemExit + assert ext.value.code == -1 diff --git a/src/tests/test_debugger.py b/src/tests/test_debugger.py index e69de29..67d634a 100644 --- a/src/tests/test_debugger.py +++ b/src/tests/test_debugger.py @@ -0,0 +1,92 @@ +from pathlib import Path +from src.debugger import GDB_SCRIPT_HIST, Debugger +from unittest.mock import patch +import hashlib + +TPATH = Path("/tmp/.hist") +TPATH_NEW = Path("/tmp/.hist_new") +PACKED_KERNEL = Path("src/tests/files/testKernel_packed") + + +def test_is_gdb_script_success() -> None: + d = Debugger(**{"kroot": "foo"}) + GDB_SCRIPT_HIST.touch() + assert d._is_gdb_script_hist() is True + + +@patch("src.debugger.GDB_SCRIPT_HIST", Path("/tmp/.fake_file")) +def test_is_gdb_script_fail() -> None: + d = Debugger(**{"kroot": "foo"}) + assert d._is_gdb_script_hist() is False + + +@patch("src.debugger.GDB_SCRIPT_HIST", TPATH) +def test_handle_gdb_change_update_existing() -> None: + d = Debugger(**{"kroot": "foo"}) + d.force_rebuild = False + TPATH.touch() + d.gdb_script = Path("/tmp/.hist2") + d.gdb_script.touch() + cntn = "start\nbreak *0xdeadbeef\ncontinue" + d.gdb_script.write_text(cntn) + d._handle_gdb_change() + assert d.force_rebuild is True + assert TPATH.read_text() == hashlib.sha256(cntn.encode()).hexdigest() + TPATH.unlink() + d.gdb_script.unlink() + + +@patch("src.debugger.GDB_SCRIPT_HIST", TPATH_NEW) +def test_handle_gdb_change_new() -> None: + d = Debugger(**{"kroot": "foo"}) + d.force_rebuild = False + d.gdb_script = Path("/tmp/.hist2") + d.gdb_script.touch() + cntn = "start\nbreak *0xdeadbeef\ncontinue" + d.gdb_script.write_text(cntn) + d._handle_gdb_change() + assert d.force_rebuild is False + assert TPATH_NEW.read_text() == hashlib.sha256(cntn.encode()).hexdigest() + d.gdb_script.unlink() + TPATH_NEW.unlink() + + +def test_set_ctf_ctx_in_init(tmp_path) -> None: + d = Debugger(**{"kroot": "foo", "ctf_ctx": True, "ctf_dir": tmp_path, "ctf_kernel": PACKED_KERNEL}) + assert d.ctf is True + + +def test_set_ctf_ctx(tmp_path) -> None: + d = Debugger(**{"kroot": "foo"}) + d.ctf_dir = tmp_path + d._set_ctf_ctx({"ctf_kernel": PACKED_KERNEL}) + assert d.ctf is False + + +def test_extract_vmlinux_success(tmp_path) -> None: + d = Debugger(**{"kroot": "foo"}) + d.ctf_kernel = PACKED_KERNEL + d.ctf_dir = tmp_path + assert d._extract_vmlinux() == 0 + + +def test_extract_vmlinux_fail(tmp_path) -> None: + d = Debugger(**{"kroot": "foo"}) + p = Path(tmp_path / "fake_kernel") + p.touch() + d.ctf_kernel = p + d.ctf_dir = tmp_path + assert d._extract_vmlinux() == 1 + + +@patch("src.misc.tmux", return_value=None) +@patch("src.misc.tmux_shell") +@patch("termios.tcflush", return_value=True) +@patch("builtins.input", lambda *args: "y") +def test_run_container(tflush, tsmock, tmock) -> None: + d = Debugger(**{"kroot": "foo"}) + d.project_dir = "/some/project_dir" + d.tag = "tag" + d.run_container() + expected = f"send-keys 'docker run -it --rm --security-opt seccomp=unconfined --cap-add=SYS_PTRACE -v {d.project_dir}:{d.docker_mnt} --net=\"host\" {d.tag} /bin/bash -c \"set -e; . /home/user/debugger.sh -a {d.arch} -p /io -c 0 -g /home/user/gdb_script\"' 'C-m'" + tmock.assert_called_with(expected) diff --git a/src/tests/test_docker_runner.py b/src/tests/test_docker_runner.py index b271778..2ad63f9 100644 --- a/src/tests/test_docker_runner.py +++ b/src/tests/test_docker_runner.py @@ -1,7 +1,8 @@ from docker.models.containers import Container +from docker import DockerClient from ..docker_runner import DockerRunner from pathlib import Path -from unittest.mock import patch +from unittest.mock import patch, Mock import pytest import uuid import shutil @@ -201,3 +202,26 @@ def test_no_kroot() -> None: _ = DockerRunner(**GENERIC_ARGS) assert ext.type == SystemExit assert ext.value.code == -1 + + +@patch("time.sleep", return_value=None) +@patch("fabric.Connection", return_value=Mock) +def test_init_ssh_fail_missing_port(fab, slp) -> None: + dr = DockerRunner(**GENERIC_ARGS | MOCK_UNPACKER_RES) + dr.client = DockerClient() + with pytest.raises((Exception, SystemExit)) as err: + dr.init_ssh() + assert err.type == SystemExit + assert err.value.code == -1 + + +@patch("fabric.Connection", return_value=Mock) +@patch("time.sleep", return_value=None) +def test_init_ssh_fail_sucess(fab, slp) -> None: + dr = DockerRunner(**GENERIC_ARGS | MOCK_UNPACKER_RES) + dr.client = DockerClient() + dr.ssh_fwd_port = 1337 + dr.user = "user" + dr.init_ssh() + assert dr.ssh_conn.host == "localhost" + assert dr.ssh_conn.port == 1337 diff --git a/src/tests/test_kernel_downloader.py b/src/tests/test_kernel_downloader.py index b8b8e58..bcc69b8 100644 --- a/src/tests/test_kernel_downloader.py +++ b/src/tests/test_kernel_downloader.py @@ -65,4 +65,4 @@ def test_run(tmp_path): lkdl.archive = Path(lkdl.kernel_dl_path) / f"linux-{lkdl._set_choice()}.tar.gz" lkdl.dl_uri = "https://file-examples.com/wp-content/uploads/2017/02/zip_2MB.zip" lkdl.run() - assert hashlib.md5(lkdl.archive.read_bytes()).hexdigest() == "3bcc73cf17293dae9ae3003bc82bd11f" + assert hashlib.md5(lkdl.archive.read_bytes()).hexdigest() == "805b6db3298f838d1998466e54471490" diff --git a/src/tests/test_rootfs_builder.py b/src/tests/test_rootfs_builder.py index c68cef8..e2c0a35 100644 --- a/src/tests/test_rootfs_builder.py +++ b/src/tests/test_rootfs_builder.py @@ -1,9 +1,13 @@ from pathlib import Path + +from src.docker_runner import DockerRunner from src.rootfs_builder import RootFSBuilder +from unittest.mock import patch, MagicMock def test_is_exist_fail() -> None: rfsb = RootFSBuilder(False, **{"kroot": "/tmp"}) + rfsb.rootfs_path = Path("/foobar") assert rfsb.is_exist() is False @@ -12,3 +16,125 @@ def test_is_exist_success(tmp_path) -> None: rfsb = RootFSBuilder(False, **{"kroot": tmp_path}) rfsb.rootfs_path = tmp_path / "foobar" assert rfsb.is_exist() is True + + +def test_run_container_full_cfg() -> None: + rfsb = RootFSBuilder(False, **{"kroot": "/tmp"}) + rfsb.client = MagicMock() + rfsb.run_container() + assert ( + rfsb.client.containers.run.assert_called_with( + None, + volumes={f"{Path.cwd() / 'io'}": {"bind": f"{rfsb.docker_mnt}", "mode": "rw"}}, + detach=True, + privileged=True, + remove=True, + command=f"/bin/bash -c 'set -e; . /home/{rfsb.user}/rootfs.sh -n {rfsb.fs_name} -a x86_64 -d {rfsb.distribution} -p {rfsb.packages} -u {rfsb.user} -h {rfsb.hostname.strip()}'", + ) + is None + ) + + +def test_run_container_no_hostname() -> None: + rfsb = RootFSBuilder(False, **{"kroot": "/tmp"}) + rfsb.hostname = "" + rfsb.client = MagicMock() + rfsb.run_container() + assert ( + rfsb.client.containers.run.assert_called_with( + None, + volumes={f"{Path.cwd() / 'io'}": {"bind": f"{rfsb.docker_mnt}", "mode": "rw"}}, + detach=True, + privileged=True, + remove=True, + command=f"/bin/bash -c 'set -e; . /home/{rfsb.user}/rootfs.sh -n {rfsb.fs_name} -a x86_64 -d {rfsb.distribution} -p {rfsb.packages} -u {rfsb.user}'", + ) + is None + ) + + +@patch.object(DockerRunner, "get_image", return_value="foobar") +def test__run(mock) -> None: + rfsb = RootFSBuilder(False, **{"kroot": "/tmp"}) + rfsb.client = MagicMock() + rfsb._run() + assert rfsb.image == "foobar" + + +@patch.object(DockerRunner, "get_image", return_value="foobar") +@patch("termios.tcflush", return_value=True) +@patch("builtins.input", lambda *args: "y") +def test_run_is_reuse(tflush, gimg) -> None: + rfsb = RootFSBuilder(False, **{"kroot": "/tmp"}) + rfsb.client = MagicMock() + assert rfsb.run() is None + + +@patch.object(DockerRunner, "get_image", return_value="foobar") +@patch.object(DockerRunner, "build_image", return_value=None) +def test_run_update(bimg, gimg) -> None: + rfsb = RootFSBuilder(False, **{"kroot": "/tmp"}) + rfsb.hostname = "" + rfsb.update_containers = True + assert rfsb.run() is None + + +@patch.object(DockerRunner, "get_image", return_value="foobar") +@patch.object(DockerRunner, "build_image", return_value=None) +def test_run_frebuild(bimg, gimg) -> None: + rfsb = RootFSBuilder(False, **{"kroot": "/tmp"}) + rfsb.client = MagicMock() + rfsb.hostname = "" + rfsb.update_containers = False + rfsb.force_rebuild = True + assert rfsb.run() is None + assert rfsb.image == "foobar" + + +@patch.object(DockerRunner, "get_image", return_value="foobar") +@patch.object(DockerRunner, "build_image", return_value=None) +def test_run_partial(bimg, gimg) -> None: + rfsb = RootFSBuilder(False, **{"kroot": "/tmp"}) + rfsb.client = MagicMock() + rfsb.hostname = "" + rfsb.update_containers = False + rfsb.force_rebuild = False + rfsb.partial = True + assert rfsb.run() is None + assert rfsb.image == "foobar" + + +@patch.object(DockerRunner, "get_image", return_value="foobar") +@patch.object(DockerRunner, "build_image", return_value=None) +@patch("termios.tcflush", return_value=True) +@patch("builtins.input", lambda *args: "n") +def test_run_skip_prompts(tflush, bimg, gimg, tmp_path) -> None: + rfsb = RootFSBuilder(False, **{"kroot": "/tmp"}) + rfsb.client = MagicMock() + p = tmp_path / "rootfs" + p.touch() + rfsb.rootfs_path = p + rfsb.hostname = "" + rfsb.update_containers = False + rfsb.force_rebuild = False + rfsb.partial = False + rfsb.skip_prompts = True + assert rfsb.run() is None + + +@patch.object(DockerRunner, "get_image", return_value="foobar") +@patch.object(DockerRunner, "build_image", return_value=None) +@patch("termios.tcflush", return_value=True) +@patch("builtins.input", lambda *args: "n") +def test_run_else(tflush, bimg, gimg, tmp_path) -> None: + rfsb = RootFSBuilder(False, **{"kroot": "/tmp"}) + rfsb.client = MagicMock() + p = tmp_path / "rootfs" + p.touch() + rfsb.rootfs_path = p + rfsb.hostname = "" + rfsb.update_containers = False + rfsb.force_rebuild = False + rfsb.partial = False + rfsb.skip_prompts = False + assert rfsb.run() is None