borgmatic/tests/unit/borg/test_check.py

import logging

import pytest
from flexmock import flexmock

from borgmatic.borg import check as module

from ..test_verbosity import insert_logging_mock


def insert_execute_command_mock(command, borg_exit_codes=None):
    flexmock(module.environment).should_receive('make_environment')
    flexmock(module).should_receive('execute_command').with_args(
        command,
        extra_environment=None,
        borg_local_path=command[0],
        borg_exit_codes=borg_exit_codes,
    ).once()


def insert_execute_command_never():
    flexmock(module).should_receive('execute_command').never()


def test_make_archive_filter_flags_with_default_checks_and_prefix_returns_default_flags():
    flexmock(module.feature).should_receive('available').and_return(True)
    flexmock(module.flags).should_receive('make_match_archives_flags').and_return(())

    flags = module.make_archive_filter_flags(
        '1.2.3',
        {'prefix': 'foo'},
        ('repository', 'archives'),
        check_arguments=flexmock(match_archives=None),
    )

    assert flags == ('--match-archives', 'sh:foo*')


def test_make_archive_filter_flags_with_all_checks_and_prefix_returns_default_flags():
    flexmock(module.feature).should_receive('available').and_return(True)
    flexmock(module.flags).should_receive('make_match_archives_flags').and_return(())

    flags = module.make_archive_filter_flags(
        '1.2.3',
        {'prefix': 'foo'},
        ('repository', 'archives', 'extract'),
        check_arguments=flexmock(match_archives=None),
    )

    assert flags == ('--match-archives', 'sh:foo*')


def test_make_archive_filter_flags_with_all_checks_and_prefix_without_borg_features_returns_glob_archives_flags():
    flexmock(module.feature).should_receive('available').and_return(False)
    flexmock(module.flags).should_receive('make_match_archives_flags').and_return(())

    flags = module.make_archive_filter_flags(
        '1.2.3',
        {'prefix': 'foo'},
        ('repository', 'archives', 'extract'),
        check_arguments=flexmock(match_archives=None),
    )

    assert flags == ('--glob-archives', 'foo*')


def test_make_archive_filter_flags_with_archives_check_and_last_includes_last_flag():
    flexmock(module.feature).should_receive('available').and_return(True)
    flexmock(module.flags).should_receive('make_match_archives_flags').and_return(())

    flags = module.make_archive_filter_flags(
        '1.2.3',
        {'check_last': 3},
        ('archives',),
        check_arguments=flexmock(match_archives=None),
    )

    assert flags == ('--last', '3')


def test_make_archive_filter_flags_with_data_check_and_last_includes_last_flag():
    flexmock(module.feature).should_receive('available').and_return(True)
    flexmock(module.flags).should_receive('make_match_archives_flags').and_return(())

    flags = module.make_archive_filter_flags(
        '1.2.3',
        {'check_last': 3},
        ('data',),
        check_arguments=flexmock(match_archives=None),
    )

    assert flags == ('--last', '3')


def test_make_archive_filter_flags_with_repository_check_and_last_omits_last_flag():
    flexmock(module.feature).should_receive('available').and_return(True)
    flexmock(module.flags).should_receive('make_match_archives_flags').and_return(())

    flags = module.make_archive_filter_flags(
        '1.2.3',
        {'check_last': 3},
        ('repository',),
        check_arguments=flexmock(match_archives=None),
    )

    assert flags == ()


def test_make_archive_filter_flags_with_default_checks_and_last_includes_last_flag():
    flexmock(module.feature).should_receive('available').and_return(True)
    flexmock(module.flags).should_receive('make_match_archives_flags').and_return(())

    flags = module.make_archive_filter_flags(
        '1.2.3',
        {'check_last': 3},
        ('repository', 'archives'),
        check_arguments=flexmock(match_archives=None),
    )

    assert flags == ('--last', '3')


def test_make_archive_filter_flags_with_archives_check_and_prefix_includes_match_archives_flag():
    flexmock(module.feature).should_receive('available').and_return(True)
    flexmock(module.flags).should_receive('make_match_archives_flags').and_return(())

    flags = module.make_archive_filter_flags(
        '1.2.3',
        {'prefix': 'foo-'},
        ('archives',),
        check_arguments=flexmock(match_archives=None),
    )

    assert flags == ('--match-archives', 'sh:foo-*')


def test_make_archive_filter_flags_with_data_check_and_prefix_includes_match_archives_flag():
    flexmock(module.feature).should_receive('available').and_return(True)
    flexmock(module.flags).should_receive('make_match_archives_flags').and_return(())

    flags = module.make_archive_filter_flags(
        '1.2.3',
        {'prefix': 'foo-'},
        ('data',),
        check_arguments=flexmock(match_archives=None),
    )

    assert flags == ('--match-archives', 'sh:foo-*')


def test_make_archive_filter_flags_prefers_check_arguments_match_archives_to_config_match_archives():
    flexmock(module.feature).should_receive('available').and_return(True)
    flexmock(module.flags).should_receive('make_match_archives_flags').with_args(
        'baz-*', None, '1.2.3'
    ).and_return(('--match-archives', 'sh:baz-*'))

    flags = module.make_archive_filter_flags(
        '1.2.3',
        {'match_archives': 'bar-{now}', 'prefix': ''},  # noqa: FS003
        ('archives',),
        check_arguments=flexmock(match_archives='baz-*'),
    )

    assert flags == ('--match-archives', 'sh:baz-*')


def test_make_archive_filter_flags_with_archives_check_and_empty_prefix_uses_archive_name_format_instead():
    flexmock(module.feature).should_receive('available').and_return(True)
    flexmock(module.flags).should_receive('make_match_archives_flags').with_args(
        None, 'bar-{now}', '1.2.3'  # noqa: FS003
    ).and_return(('--match-archives', 'sh:bar-*'))

    flags = module.make_archive_filter_flags(
        '1.2.3',
        {'archive_name_format': 'bar-{now}', 'prefix': ''},  # noqa: FS003
        ('archives',),
        check_arguments=flexmock(match_archives=None),
    )

    assert flags == ('--match-archives', 'sh:bar-*')


def test_make_archive_filter_flags_with_archives_check_and_none_prefix_omits_match_archives_flag():
    flexmock(module.feature).should_receive('available').and_return(True)
    flexmock(module.flags).should_receive('make_match_archives_flags').and_return(())

    flags = module.make_archive_filter_flags(
        '1.2.3',
        {},
        ('archives',),
        check_arguments=flexmock(match_archives=None),
    )

    assert flags == ()


def test_make_archive_filter_flags_with_repository_check_and_prefix_omits_match_archives_flag():
    flexmock(module.feature).should_receive('available').and_return(True)
    flexmock(module.flags).should_receive('make_match_archives_flags').and_return(())

    flags = module.make_archive_filter_flags(
        '1.2.3',
        {'prefix': 'foo-'},
        ('repository',),
        check_arguments=flexmock(match_archives=None),
    )

    assert flags == ()


def test_make_archive_filter_flags_with_default_checks_and_prefix_includes_match_archives_flag():
    flexmock(module.feature).should_receive('available').and_return(True)
    flexmock(module.flags).should_receive('make_match_archives_flags').and_return(())

    flags = module.make_archive_filter_flags(
        '1.2.3',
        {'prefix': 'foo-'},
        ('repository', 'archives'),
        check_arguments=flexmock(match_archives=None),
    )

    assert flags == ('--match-archives', 'sh:foo-*')


def test_make_check_flags_with_repository_check_returns_flag():
    flags = module.make_check_flags(('repository',), ())

    assert flags == ('--repository-only',)


def test_make_check_flags_with_archives_check_returns_flag():
    flags = module.make_check_flags(('archives',), ())

    assert flags == ('--archives-only',)


def test_make_check_flags_with_archives_check_and_archive_filter_flags_includes_those_flags():
    flags = module.make_check_flags(('archives',), ('--match-archives', 'sh:foo-*'))

    assert flags == ('--archives-only', '--match-archives', 'sh:foo-*')


def test_make_check_flags_without_archives_check_and_with_archive_filter_flags_includes_those_flags():
    flags = module.make_check_flags(('repository',), ('--match-archives', 'sh:foo-*'))

    assert flags == ('--repository-only',)


def test_make_check_flags_with_data_check_returns_flag_and_implies_archives():
    flexmock(module.feature).should_receive('available').and_return(True)
    flexmock(module.flags).should_receive('make_match_archives_flags').and_return(())

    flags = module.make_check_flags(('data',), ())

    assert flags == (
        '--archives-only',
        '--verify-data',
    )


def test_make_check_flags_with_extract_omits_extract_flag():
    flexmock(module.feature).should_receive('available').and_return(True)
    flexmock(module.flags).should_receive('make_match_archives_flags').and_return(())

    flags = module.make_check_flags(('extract',), ())

    assert flags == ()


def test_make_check_flags_with_repository_and_data_checks_does_not_return_repository_only():
    flexmock(module.feature).should_receive('available').and_return(True)
    flexmock(module.flags).should_receive('make_match_archives_flags').and_return(())

    flags = module.make_check_flags(
        (
            'repository',
            'data',
        ),
        (),
    )

    assert flags == ('--verify-data',)


def test_get_repository_id_with_valid_json_does_not_raise():
    config = {}
    flexmock(module.rinfo).should_receive('display_repository_info').and_return(
        '{"repository": {"id": "repo"}}'
    )

    assert module.get_repository_id(
        repository_path='repo',
        config=config,
        local_borg_version='1.2.3',
        global_arguments=flexmock(log_json=False),
        local_path='borg',
        remote_path=None,
    )


def test_get_repository_id_with_json_error_raises():
    config = {}
    flexmock(module.rinfo).should_receive('display_repository_info').and_return(
        '{"unexpected": {"id": "repo"}}'
    )

    with pytest.raises(ValueError):
        module.get_repository_id(
            repository_path='repo',
            config=config,
            local_borg_version='1.2.3',
            global_arguments=flexmock(log_json=False),
            local_path='borg',
            remote_path=None,
        )


def test_get_repository_id_with_missing_json_keys_raises():
    config = {}
    flexmock(module.rinfo).should_receive('display_repository_info').and_return('{invalid JSON')

    with pytest.raises(ValueError):
        module.get_repository_id(
            repository_path='repo',
            config=config,
            local_borg_version='1.2.3',
            global_arguments=flexmock(log_json=False),
            local_path='borg',
            remote_path=None,
        )


def test_check_archives_with_progress_passes_through_to_borg():
    config = {}
    flexmock(module).should_receive('make_check_flags').and_return(())
    flexmock(module).should_receive('execute_command').never()
    flexmock(module.flags).should_receive('make_repository_flags').and_return(('repo',))
    flexmock(module.environment).should_receive('make_environment')
    flexmock(module).should_receive('execute_command').with_args(
        ('borg', 'check', '--progress', 'repo'),
        output_file=module.DO_NOT_CAPTURE,
        extra_environment=None,
        borg_local_path='borg',
        borg_exit_codes=None,
    ).once()

    module.check_archives(
        repository_path='repo',
        config=config,
        local_borg_version='1.2.3',
        check_arguments=flexmock(
            progress=True, repair=None, only_checks=None, force=None, match_archives=None
        ),
        global_arguments=flexmock(log_json=False),
        checks={'repository'},
        archive_filter_flags=(),
    )


def test_check_archives_with_repair_passes_through_to_borg():
    config = {}
    flexmock(module).should_receive('make_check_flags').and_return(())
    flexmock(module).should_receive('execute_command').never()
    flexmock(module.flags).should_receive('make_repository_flags').and_return(('repo',))
    flexmock(module.environment).should_receive('make_environment')
    flexmock(module).should_receive('execute_command').with_args(
        ('borg', 'check', '--repair', 'repo'),
        output_file=module.DO_NOT_CAPTURE,
        extra_environment=None,
        borg_local_path='borg',
        borg_exit_codes=None,
    ).once()

    module.check_archives(
        repository_path='repo',
        config=config,
        local_borg_version='1.2.3',
        check_arguments=flexmock(
            progress=None, repair=True, only_checks=None, force=None, match_archives=None
        ),
        global_arguments=flexmock(log_json=False),
        checks={'repository'},
        archive_filter_flags=(),
    )


@pytest.mark.parametrize(
    'checks',
    (
        ('repository',),
        ('archives',),
        ('repository', 'archives'),
        ('repository', 'archives', 'other'),
    ),
)
def test_check_archives_calls_borg_with_parameters(checks):
    config = {}
    flexmock(module).should_receive('make_check_flags').with_args(checks, ()).and_return(())
    flexmock(module.flags).should_receive('make_repository_flags').and_return(('repo',))
    insert_execute_command_mock(('borg', 'check', 'repo'))

    module.check_archives(
        repository_path='repo',
        config=config,
        local_borg_version='1.2.3',
        check_arguments=flexmock(
            progress=None, repair=None, only_checks=None, force=None, match_archives=None
        ),
        global_arguments=flexmock(log_json=False),
        checks=checks,
        archive_filter_flags=(),
    )


def test_check_archives_with_log_info_passes_through_to_borg():
    config = {}
    flexmock(module).should_receive('make_check_flags').and_return(())
    flexmock(module.flags).should_receive('make_repository_flags').and_return(('repo',))
    insert_logging_mock(logging.INFO)
    insert_execute_command_mock(('borg', 'check', '--info', 'repo'))

    module.check_archives(
        repository_path='repo',
        config=config,
        local_borg_version='1.2.3',
        check_arguments=flexmock(
            progress=None, repair=None, only_checks=None, force=None, match_archives=None
        ),
        global_arguments=flexmock(log_json=False),
        checks={'repository'},
        archive_filter_flags=(),
    )


def test_check_archives_with_log_debug_passes_through_to_borg():
    config = {}
    flexmock(module).should_receive('make_check_flags').and_return(())
    flexmock(module.flags).should_receive('make_repository_flags').and_return(('repo',))
    insert_logging_mock(logging.DEBUG)
    insert_execute_command_mock(('borg', 'check', '--debug', '--show-rc', 'repo'))

    module.check_archives(
        repository_path='repo',
        config=config,
        local_borg_version='1.2.3',
        check_arguments=flexmock(
            progress=None, repair=None, only_checks=None, force=None, match_archives=None
        ),
        global_arguments=flexmock(log_json=False),
        checks={'repository'},
        archive_filter_flags=(),
    )


def test_check_archives_with_local_path_calls_borg_via_local_path():
    checks = {'repository'}
    config = {}
    flexmock(module).should_receive('make_check_flags').with_args(checks, ()).and_return(())
    flexmock(module.flags).should_receive('make_repository_flags').and_return(('repo',))
    insert_execute_command_mock(('borg1', 'check', 'repo'))

    module.check_archives(
        repository_path='repo',
        config=config,
        local_borg_version='1.2.3',
        check_arguments=flexmock(
            progress=None, repair=None, only_checks=None, force=None, match_archives=None
        ),
        global_arguments=flexmock(log_json=False),
        checks=checks,
        archive_filter_flags=(),
        local_path='borg1',
    )


def test_check_archives_with_exit_codes_calls_borg_using_them():
    checks = {'repository'}
    borg_exit_codes = flexmock()
    config = {'borg_exit_codes': borg_exit_codes}
    flexmock(module).should_receive('make_check_flags').with_args(checks, ()).and_return(())
    flexmock(module.flags).should_receive('make_repository_flags').and_return(('repo',))
    insert_execute_command_mock(('borg', 'check', 'repo'), borg_exit_codes=borg_exit_codes)

    module.check_archives(
        repository_path='repo',
        config=config,
        local_borg_version='1.2.3',
        check_arguments=flexmock(
            progress=None, repair=None, only_checks=None, force=None, match_archives=None
        ),
        global_arguments=flexmock(log_json=False),
        checks=checks,
        archive_filter_flags=(),
    )


def test_check_archives_with_remote_path_passes_through_to_borg():
    checks = {'repository'}
    config = {}
    flexmock(module).should_receive('make_check_flags').with_args(checks, ()).and_return(())
    flexmock(module.flags).should_receive('make_repository_flags').and_return(('repo',))
    insert_execute_command_mock(('borg', 'check', '--remote-path', 'borg1', 'repo'))

    module.check_archives(
        repository_path='repo',
        config=config,
        local_borg_version='1.2.3',
        check_arguments=flexmock(
            progress=None, repair=None, only_checks=None, force=None, match_archives=None
        ),
        global_arguments=flexmock(log_json=False),
        checks=checks,
        archive_filter_flags=(),
        remote_path='borg1',
    )


def test_check_archives_with_log_json_passes_through_to_borg():
    checks = {'repository'}
    config = {}
    flexmock(module).should_receive('make_check_flags').with_args(checks, ()).and_return(())
    flexmock(module.flags).should_receive('make_repository_flags').and_return(('repo',))
    insert_execute_command_mock(('borg', 'check', '--log-json', 'repo'))

    module.check_archives(
        repository_path='repo',
        config=config,
        local_borg_version='1.2.3',
        check_arguments=flexmock(
            progress=None, repair=None, only_checks=None, force=None, match_archives=None
        ),
        global_arguments=flexmock(log_json=True),
        checks=checks,
        archive_filter_flags=(),
    )


def test_check_archives_with_lock_wait_passes_through_to_borg():
    checks = {'repository'}
    config = {'lock_wait': 5}
    flexmock(module).should_receive('make_check_flags').with_args(checks, ()).and_return(())
    flexmock(module.flags).should_receive('make_repository_flags').and_return(('repo',))
    insert_execute_command_mock(('borg', 'check', '--lock-wait', '5', 'repo'))

    module.check_archives(
        repository_path='repo',
        config=config,
        local_borg_version='1.2.3',
        check_arguments=flexmock(
            progress=None, repair=None, only_checks=None, force=None, match_archives=None
        ),
        global_arguments=flexmock(log_json=False),
        checks=checks,
        archive_filter_flags=(),
    )


def test_check_archives_with_retention_prefix():
    checks = {'repository'}
    prefix = 'foo-'
    config = {'prefix': prefix}
    flexmock(module).should_receive('make_check_flags').with_args(checks, ()).and_return(())
    flexmock(module.flags).should_receive('make_repository_flags').and_return(('repo',))
    insert_execute_command_mock(('borg', 'check', 'repo'))

    module.check_archives(
        repository_path='repo',
        config=config,
        local_borg_version='1.2.3',
        check_arguments=flexmock(
            progress=None, repair=None, only_checks=None, force=None, match_archives=None
        ),
        global_arguments=flexmock(log_json=False),
        checks=checks,
        archive_filter_flags=(),
    )


def test_check_archives_with_extra_borg_options_passes_through_to_borg():
    config = {'extra_borg_options': {'check': '--extra --options'}}
    flexmock(module).should_receive('make_check_flags').and_return(())
    flexmock(module.flags).should_receive('make_repository_flags').and_return(('repo',))
    insert_execute_command_mock(('borg', 'check', '--extra', '--options', 'repo'))

    module.check_archives(
        repository_path='repo',
        config=config,
        local_borg_version='1.2.3',
        check_arguments=flexmock(
            progress=None, repair=None, only_checks=None, force=None, match_archives=None
        ),
        global_arguments=flexmock(log_json=False),
        checks={'repository'},
        archive_filter_flags=(),
    )


def test_check_archives_with_match_archives_passes_through_to_borg():
    config = {}
    flexmock(module).should_receive('make_check_flags').and_return(('--match-archives', 'foo-*'))
    flexmock(module.flags).should_receive('make_repository_flags').and_return(('repo',))
    flexmock(module.environment).should_receive('make_environment')
    flexmock(module).should_receive('execute_command').with_args(
        ('borg', 'check', '--match-archives', 'foo-*', 'repo'),
        extra_environment=None,
        borg_local_path='borg',
        borg_exit_codes=None,
    ).once()

    module.check_archives(
        repository_path='repo',
        config=config,
        local_borg_version='1.2.3',
        check_arguments=flexmock(
            progress=None, repair=None, only_checks=None, force=None, match_archives='foo-*'
        ),
        global_arguments=flexmock(log_json=False),
        checks={'archives'},
        archive_filter_flags=('--match-archives', 'foo-*'),
    )