From aae4abe6c1da5f484de531f2c41afea2a751ff7e Mon Sep 17 00:00:00 2001 From: Thiago Kenji Okada Date: Mon, 16 Dec 2024 12:53:08 +0000 Subject: [PATCH] nixos-rebuild-ng: convert side_effects mocks to function --- .../nixos-rebuild-ng/src/tests/test_main.py | 135 ++++++++++-------- .../ni/nixos-rebuild-ng/src/tests/test_nix.py | 10 +- 2 files changed, 80 insertions(+), 65 deletions(-) diff --git a/pkgs/by-name/ni/nixos-rebuild-ng/src/tests/test_main.py b/pkgs/by-name/ni/nixos-rebuild-ng/src/tests/test_main.py index 39a50ecc405a..b372a3e8f9a1 100644 --- a/pkgs/by-name/ni/nixos-rebuild-ng/src/tests/test_main.py +++ b/pkgs/by-name/ni/nixos-rebuild-ng/src/tests/test_main.py @@ -82,18 +82,20 @@ def test_execute_nix_boot(mock_run: Any, tmp_path: Path) -> None: nixpkgs_path.mkdir() config_path = tmp_path / "test" config_path.touch() - mock_run.side_effect = [ - # update_nixpkgs_rev - CompletedProcess([], 0, str(nixpkgs_path)), - CompletedProcess([], 0, "nixpkgs-rev"), - CompletedProcess([], 0), - # nixos_build - CompletedProcess([], 0, str(config_path)), - # set_profile - CompletedProcess([], 0), - # switch_to_configuration - CompletedProcess([], 0), - ] + + def run_wrapper_side_effect( + args: list[str], **kwargs: Any + ) -> CompletedProcess[str]: + if args[0] == "nix-instantiate": + return CompletedProcess([], 0, str(nixpkgs_path)) + elif args[0] == "git" and "rev-parse" in args: + return CompletedProcess([], 0, "nixpkgs-rev") + elif args[0] == "nix-build": + return CompletedProcess([], 0, str(config_path)) + else: + return CompletedProcess([], 0) + + mock_run.side_effect = run_wrapper_side_effect nr.execute(["nixos-rebuild", "boot", "--no-flake", "-vvv", "--fast"]) @@ -155,14 +157,16 @@ def test_execute_nix_boot(mock_run: Any, tmp_path: Path) -> None: def test_execute_nix_switch_flake(mock_run: Any, tmp_path: Path) -> None: config_path = tmp_path / "test" config_path.touch() - mock_run.side_effect = [ - # nixos_build_flake - CompletedProcess([], 0, str(config_path)), - # set_profile - CompletedProcess([], 0), - # switch_to_configuration - CompletedProcess([], 0), - ] + + def run_wrapper_side_effect( + args: list[str], **kwargs: Any + ) -> CompletedProcess[str]: + if args[0] == "nix": + return CompletedProcess([], 0, str(config_path)) + else: + return CompletedProcess([], 0) + + mock_run.side_effect = run_wrapper_side_effect nr.execute( [ @@ -226,16 +230,16 @@ def test_execute_nix_switch_flake_target_host( ) -> None: config_path = tmp_path / "test" config_path.touch() - mock_run.side_effect = [ - # nixos_build_flake - CompletedProcess([], 0, str(config_path)), - # set_profile - CompletedProcess([], 0), - # copy_closure - CompletedProcess([], 0), - # switch_to_configuration - CompletedProcess([], 0), - ] + + def run_wrapper_side_effect( + args: list[str], **kwargs: Any + ) -> CompletedProcess[str]: + if args[0] == "nix": + return CompletedProcess([], 0, str(config_path)) + else: + return CompletedProcess([], 0) + + mock_run.side_effect = run_wrapper_side_effect nr.execute( [ @@ -317,18 +321,18 @@ def test_execute_nix_switch_flake_build_host( ) -> None: config_path = tmp_path / "test" config_path.touch() - mock_run.side_effect = [ - # nixos_build_flake - CompletedProcess([], 0, str(config_path)), - CompletedProcess([], 0), - CompletedProcess([], 0, str(config_path)), - # set_profile - CompletedProcess([], 0), - # copy_closure - CompletedProcess([], 0), - # switch_to_configuration - CompletedProcess([], 0), - ] + + def run_wrapper_side_effect( + args: list[str], **kwargs: Any + ) -> CompletedProcess[str]: + if args[0] == "nix" and "eval" in args: + return CompletedProcess([], 0, str(config_path)) + if args[0] == "ssh" and "nix" in args: + return CompletedProcess([], 0, str(config_path)) + else: + return CompletedProcess([], 0) + + mock_run.side_effect = run_wrapper_side_effect nr.execute( [ @@ -478,12 +482,16 @@ def test_execute_build(mock_run: Any, tmp_path: Path) -> None: def test_execute_test_flake(mock_run: Any, tmp_path: Path) -> None: config_path = tmp_path / "test" config_path.touch() - mock_run.side_effect = [ - # nixos_build_flake - CompletedProcess([], 0, str(config_path)), - # switch_to_configuration - CompletedProcess([], 0), - ] + + def run_wrapper_side_effect( + args: list[str], **kwargs: Any + ) -> CompletedProcess[str]: + if args[0] == "nix": + return CompletedProcess([], 0, str(config_path)) + else: + return CompletedProcess([], 0) + + mock_run.side_effect = run_wrapper_side_effect nr.execute( ["nixos-rebuild", "test", "--flake", "github:user/repo#hostname", "--fast"] @@ -522,20 +530,23 @@ def test_execute_test_rollback( mock_path_exists: Any, mock_run: Any, ) -> None: - mock_run.side_effect = [ - # rollback_temporary_profile - CompletedProcess( - [], - 0, - stdout=textwrap.dedent("""\ - 2082 2024-11-07 22:58:56 - 2083 2024-11-07 22:59:41 - 2084 2024-11-07 23:54:17 (current) - """), - ), - # switch_to_configuration - CompletedProcess([], 0), - ] + def run_wrapper_side_effect( + args: list[str], **kwargs: Any + ) -> CompletedProcess[str]: + if args[0] == "nix-env": + return CompletedProcess( + [], + 0, + stdout=textwrap.dedent("""\ + 2082 2024-11-07 22:58:56 + 2083 2024-11-07 22:59:41 + 2084 2024-11-07 23:54:17 (current) + """), + ) + else: + return CompletedProcess([], 0) + + mock_run.side_effect = run_wrapper_side_effect nr.execute( ["nixos-rebuild", "test", "--rollback", "--profile-name", "foo", "--fast"] diff --git a/pkgs/by-name/ni/nixos-rebuild-ng/src/tests/test_nix.py b/pkgs/by-name/ni/nixos-rebuild-ng/src/tests/test_nix.py index 6245aa5cd196..b2345d191a95 100644 --- a/pkgs/by-name/ni/nixos-rebuild-ng/src/tests/test_nix.py +++ b/pkgs/by-name/ni/nixos-rebuild-ng/src/tests/test_nix.py @@ -1,11 +1,11 @@ import textwrap +import uuid from pathlib import Path from subprocess import PIPE, CompletedProcess from typing import Any from unittest.mock import ANY, call, patch import pytest -import uuid import nixos_rebuild.models as m import nixos_rebuild.nix as n @@ -80,13 +80,17 @@ def test_remote_build(mock_uuid4: Any, mock_run: Any, monkeypatch: Any) -> None: build_host = m.Remote("user@host", [], None) monkeypatch.setenv("NIX_SSHOPTS", "--ssh opts") - def run_wrapper_side_effect(args, **kwargs): # type: ignore + def run_wrapper_side_effect( + args: list[str], **kwargs: Any + ) -> CompletedProcess[str]: if args[0] == "nix-instantiate": return CompletedProcess([], 0, stdout=" \n/path/to/file\n ") elif args[0] == "mktemp": return CompletedProcess([], 0, stdout=" \n/tmp/tmpdir\n ") elif args[0] == "nix-store": - return CompletedProcess([], 0, stdout=" \n/tmp/tmpdir/00000000000000000000000000000000\n ") + return CompletedProcess( + [], 0, stdout=" \n/tmp/tmpdir/00000000000000000000000000000000\n " + ) elif args[0] == "readlink": return CompletedProcess([], 0, stdout=" \n/path/to/config\n ") else: