How to use watchmaker - 10 common examples

def __init__(self, arguments):
        self.log = logging.getLogger(
            '{0}.{1}'.format(__name__, self.__class__.__name__)
        )
        # Pop extra_arguments now so we can log it separately
        extra_arguments = arguments.pop('extra_arguments', [])

        header = ' WATCHMAKER RUN '
        header = header.rjust((40 + len(header) // 2), '#').ljust(80, '#')
        self.log.info(header)
        self.log.debug('Watchmaker Version: %s', __version__)
        self.log.debug('Parameters: %s', arguments)
        self.log.debug('Extra Parameters: %s', extra_arguments)

        # Pop remaining arguments used by watchmaker.Client itself
        self.default_config = os.path.join(static.__path__[0], 'config.yaml')
        self.no_reboot = arguments.pop('no_reboot', False)
        self.config_path = arguments.pop('config_path')
        self.log_dir = arguments.pop('log_dir')
        self.log_level = arguments.pop('log_level')

        # Get the system params
        self.system = platform.system().lower()
        self._set_system_params()

        self.log.debug('System Type: %s', self.system)
        self.log.debug('System Parameters: %s', self.system_params)

        # All remaining arguments are worker_args
        worker_args = arguments

        # Convert extra_arguments to a dict and merge it with worker_args.

def test_copytree_no_force(mock_copy, mock_rm, mock_exists):
    """Test that copytree results in correct calls without force option."""
    random_src = 'aba51e65-afd2-5020-8117-195f75e64258'
    random_dst = 'f74d03de-7c1d-596f-83f3-73748f2e238f'

    watchmaker.utils.copytree(random_src, random_dst)
    mock_copy.assert_called_with(random_src, random_dst)
    assert mock_rm.call_count == 0
    assert mock_exists.call_count == 0

    watchmaker.utils.copytree(random_src, random_dst, force=False)
    mock_copy.assert_called_with(random_src, random_dst)
    assert mock_rm.call_count == 0
    assert mock_exists.call_count == 0

def test_clean_none():
    """Check string 'None' conversion to None."""
    assert not watchmaker.utils.clean_none('None')
    assert not watchmaker.utils.clean_none('none')
    assert not watchmaker.utils.clean_none(None)
    assert watchmaker.utils.clean_none('not none') == 'not none'

def test_linux_computer_name_none():
    """Test that Pythonic None can be used without error rather than 'None'."""
    # setup ========================
    system_params = {}
    salt_config = {}
    system_params["prepdir"] = "662f1bdb-5992-5f8f-87d6-15c4de958b7b"
    system_params["logdir"] = "76b74ceb-e81d-5fac-b293-0d7d45901ef7"
    system_params["workingdir"] = "4e6a1827-1d3b-5612-a7fd-f6fed00b5a2f"

    # try "normal" first, with a value. try with none below.
    salt_config["computer_name"] = "5f0c8635-4a10-5802-8145-732052a0b44b"

    # execution ====================
    saltworker_lx = SaltLinux(system_params, **salt_config)

    saltworker_lx._set_grain = MagicMock(return_value=None)
    saltworker_lx.run_salt = MagicMock(
        return_value={"retcode": 0, "stdout": b"", "stderr": b""}
    )

    saltworker_lx.process_grains()

    # assertions ===================
    assert saltworker_lx._set_grain.call_count == 3
    saltworker_lx._set_grain.assert_called_with(
        'name-computer', {'computername': salt_config["computer_name"]})

    # tried "normal" first, with a value, above. now, trying with none.
    salt_config["computer_name"] = None

def test_linux_salt_content_path_none(
        mock_copysubdirs, mock_glob, mock_copytree, mock_yload,
        mock_ydump, mock_os, mock_codec):
    """Test that Pythonic None can be used without error rather than 'None'."""
    # setup ========================
    system_params = {}
    salt_config = {}
    system_params["prepdir"] = "4504257a-76d0-49bd-9d04-53c1459b7156"
    system_params["logdir"] = "045143d6-0e87-497f-a11a-5eebb1ec7edf"
    system_params["workingdir"] = "83f16e7b-c2cf-482b-93c8-32a558f6ded6"

    salt_config["salt_content"] = "33691f8e-e245-4be2-827b-2fa727600fb4.zip"
    salt_config["salt_content_path"] = None

    # execution ====================
    saltworker_lx = SaltLinux(system_params, **salt_config)
    saltworker_lx.working_dir = system_params["workingdir"]

    saltworker_lx.retrieve_file = MagicMock(return_value=None)
    saltworker_lx.extract_contents = MagicMock(return_value=None)

    saltworker_lx._build_salt_formula("e8d7398e-49fa-4eb9-8f8b-22c9d3fdb7f7")

    # assertions ===================
    assert saltworker_lx.retrieve_file.call_count == 1
    assert saltworker_lx.extract_contents.call_count == 1
    assert mock_copysubdirs.call_count == 1
    assert mock_codec.call_count == 1
    assert mock_os.call_count == 1
    assert mock_ydump.call_count == 1
    assert mock_yload.call_count == 1
    assert mock_copytree.call_count > 1

saltworker_lx.run_salt = MagicMock(
        return_value={"retcode": 0, "stdout": b"", "stderr": b""}
    )

    saltworker_lx.process_grains()

    # assertions ===================
    assert saltworker_lx._set_grain.call_count == 3
    saltworker_lx._set_grain.assert_called_with(
        'join-domain', {'oupath': salt_config["ou_path"]})

    # tried "normal" first, with a value, above. now, trying with none.
    salt_config["ou_path"] = None

    # execution ====================
    saltworker_lx = SaltLinux(system_params, **salt_config)

    saltworker_lx._set_grain = MagicMock(return_value=None)
    saltworker_lx.run_salt = MagicMock(
        return_value={"retcode": 0, "stdout": b"", "stderr": b""}
    )

    saltworker_lx.process_grains()

    # assertions ===================
    assert saltworker_lx._set_grain.call_count == 2

# try "normal" first, with a value. try with none below.
    salt_config["salt_debug_log"] = "211aba6a-72f3-5659-85ce-22fc3854da2e"

    # execution ====================
    saltworker_lx = SaltLinux(system_params, **salt_config)
    saltworker_lx.create_working_dir = MagicMock(return_value=None)
    saltworker_lx._prepare_for_install()

    # assertions ===================
    assert saltworker_lx.salt_debug_logfile == salt_config["salt_debug_log"]

    # tried "normal" first, with a value, above. now, trying with none.
    salt_config["salt_debug_log"] = None

    # execution ====================
    saltworker_lx = SaltLinux(system_params, **salt_config)
    saltworker_lx.create_working_dir = MagicMock(return_value=None)
    saltworker_lx._prepare_for_install()

    # assertions ===================
    assert saltworker_lx.salt_debug_logfile == os.sep.join(
        (system_params["logdir"], 'salt_call.debug.log'))

def test_linux_salt_content_path(
        mock_glob, mock_copytree, mock_yload,
        mock_ydump, mock_os, mock_codec):
    """Ensure that files from salt_content_path are retrieved correctly."""
    # setup ========================
    system_params = {}
    salt_config = {}
    system_params["prepdir"] = "96003f32-5808-4ef8-a573-763b7f47ba9d"
    system_params["logdir"] = "0585f9d7-ed0e-4a1b-ac0d-2b10a245a0eb"
    system_params["workingdir"] = "35f411db-355b-4953-ae31-d6f592753e58"

    salt_config["salt_content"] = "d002be6e-645d-4f58-97c9-8335df0ff5e4.zip"
    salt_config["salt_content_path"] = "05628e08-f1be-474d-8c12-5bb6517fc5f9"

    # execution ====================
    saltworker_lx = SaltLinux(system_params, **salt_config)

    saltworker_lx.retrieve_file = MagicMock(return_value=None)
    saltworker_lx.extract_contents = MagicMock(return_value=None)
    saltworker_lx.working_dir = system_params["workingdir"]
    mock_glob.return_value = ['05628e08-f1be-474d-8c12-5bb6517fc5f9/87a2324d']

    saltworker_lx._build_salt_formula("8822e968-deea-410f-9b6e-d25a36c512d1")

    # assertions ===================
    assert saltworker_lx.retrieve_file.call_count == 1
    assert saltworker_lx.extract_contents.call_count == 1
    assert mock_codec.call_count == 1
    assert mock_os.call_count == 3
    assert mock_ydump.call_count == 1
    assert mock_yload.call_count == 1
    assert mock_copytree.call_count > 1

def test_linux_salt_content_none(
        mock_copytree, mock_yload, mock_ydump, mock_os, mock_codec):
    """Test that Pythonic None can be used without error rather than 'None'."""
    # setup ========================
    system_params = {}
    salt_config = {}
    system_params["prepdir"] = "1868795c-7d47-5d4d-9021-941b449967af"
    system_params["logdir"] = "e88611fe-d444-5951-bb20-4d7508ed7a0c"
    system_params["workingdir"] = "f671935d-6669-50d1-a63c-157799cd13cb"

    salt_config["salt_content"] = None

    # execution ====================
    saltworker_lx = SaltLinux(system_params, **salt_config)

    saltworker_lx.retrieve_file = MagicMock(return_value=None)
    saltworker_lx.extract_contents = MagicMock(return_value=None)

    saltworker_lx._build_salt_formula("c6295b26-c043-5cd8-aa83-e3aab9df263c")

    # assertions ===================
    assert saltworker_lx.retrieve_file.call_count == 0
    assert saltworker_lx.extract_contents.call_count == 0
    assert mock_codec.call_count == 1
    assert mock_os.call_count == 2
    assert mock_ydump.call_count == 1
    assert mock_yload.call_count == 1
    assert mock_copytree.call_count > 1

def test_linux_admin_groups_none():
    """Test that Pythonic None can be used without error rather than 'None'."""
    # setup ========================
    system_params = {}
    salt_config = {}
    system_params["prepdir"] = "a27f71e5-4af3-5ad4-9d69-05fb8b7cbe4f"
    system_params["logdir"] = "9b504864-814c-5226-9a05-b7613dd96efa"
    system_params["workingdir"] = "d468d406-1e89-5041-9900-0a9189e541b0"

    # try "normal" first, with a value. try with none below.
    salt_config["admin_groups"] = "5a8915b6-8ceb-53c8-a9c6-064365dd10bd"

    # execution ====================
    saltworker_lx = SaltLinux(system_params, **salt_config)

    saltworker_lx._set_grain = MagicMock(return_value=None)
    saltworker_lx.run_salt = MagicMock(
        return_value={"retcode": 0, "stdout": b"", "stderr": b""}
    )

    saltworker_lx.process_grains()

    # assertions ===================
    assert saltworker_lx._set_grain.call_count == 3
    saltworker_lx._set_grain.assert_called_with(
        'join-domain', {'admin_groups': [salt_config["admin_groups"]]})

    # tried "normal" first, with a value, above. now, trying with none.
    salt_config["admin_groups"] = None