How to use barman - 10 common examples

def test_init_bad_command(self, exit_mock, caplog):
        # preparation
        writer = self._mock_writer()
        del writer.init_bad_command

        output.init('bad_command')

        # logging test
        for record in caplog.records:
            assert record.levelname == 'ERROR'
        assert 'bad_command' in caplog.text
        assert 'Traceback' in caplog.text

        # writer test
        writer.error_occurred.assert_called_once_with()
        assert writer.exception.call_count == 1

        # exit with error
        assert exit_mock.called
        assert exit_mock.call_count == 1
        assert exit_mock.call_args[0] != 0

def test_info_with_args(self, caplog):
        # See all logs
        caplog.set_level(0)

        # preparation
        writer = self._mock_writer()

        msg = 'test format %02d %s'
        args = (1, '2nd')
        output.info(msg, *args)

        # logging test
        for record in caplog.records:
            assert record.levelname == 'INFO'
            assert record.name == __name__
        assert msg % args in caplog.text

        # writer test
        assert not writer.error_occurred.called
        writer.info.assert_called_once_with(msg, *args)

        # global status test
        assert not output.error_occurred

class TestU:
            if sys.version_info[0] >= 3:
                def __str__(self):
                    return accented

                def __bytes__(self):
                    return b'Wrong'
            else:
                def __str__(self):
                    return b'Wrong'

                def __unicode__(self):
                    return accented

        assert (barman.utils.force_str(Test()) == accented)
        assert (barman.utils.force_str(TestU()) == accented)
        assert (barman.utils.force_str(Exception(Test())) == accented)
        assert (barman.utils.force_str(Exception(TestU())) == accented)
        assert (barman.utils.force_str(1) == '1')
        assert (barman.utils.force_str('foo') == 'foo')
        assert (barman.utils.force_str(('foo', 'bar')) == "('foo', 'bar')")

def test_result_check_ok_color(self, capsys, monkeypatch):
        monkeypatch.setattr(output, 'ansi_colors_enabled', True)
        writer = output.ConsoleOutputWriter()
        output.error_occurred = False

        server = 'test'
        check = 'test check'

        writer.result_check(server, check, True)
        (out, err) = capsys.readouterr()
        assert out == '\t%s: %sOK%s\n' % (check, GREEN, RESET)
        assert err == ''
        assert not output.error_occurred

def test_single_result_check_error(self, capsys):
        writer = output.NagiosOutputWriter()
        output.error_occurred = False

        # one server with one error
        writer.result_check('a', 'test', False, None)

        writer.close()
        (out, err) = capsys.readouterr()
        assert out == 'BARMAN CRITICAL - server a has issues * ' \
                      'a FAILED: test\na.test: FAILED\n'
        assert err == ''
        assert output.error_occurred
        assert output.error_exit_code == 2

def test_result_check_ok_hint_color(self, capsys, monkeypatch):
        monkeypatch.setattr(output, 'ansi_colors_enabled', True)
        writer = output.ConsoleOutputWriter()
        output.error_occurred = False

        server = 'test'
        check = 'test check'
        hint = 'do something'

        writer.result_check(server, check, True, hint)
        (out, err) = capsys.readouterr()
        assert out == '\t%s: %sOK%s (%s)\n' % (check, GREEN, RESET, hint)
        assert err == ''
        assert not output.error_occurred

def teardown_module(module):
    """
    Set the output API to a functional state, after testing it
    """
    output.set_output_writer(output.DEFAULT_WRITER)

begin_time=None,
        begin_wal='000000010000000000000002',
        begin_xlog='0/2000028',
        config_file='/pgdata/location/postgresql.conf',
        end_offset=184,
        end_time=None,
        end_wal='000000010000000000000002',
        end_xlog='0/20000B8',
        error=None,
        hba_file='/pgdata/location/pg_hba.conf',
        ident_file='/pgdata/location/pg_ident.conf',
        mode='default',
        pgdata='/pgdata/location',
        server_name='test_server',
        size=12345,
        status=BackupInfo.DONE,
        included_files=None,
        tablespaces=(
            ('tbs1', 16387, '/fake/location'),
            ('tbs2', 16405, '/another/location'),
        ),
        timeline=1,
        version=90302,
        server=None,
        copy_stats=None):
    """
    Create an 'Ad Hoc' BackupInfo object for testing purposes.

    A BackupInfo object is the barman representation of a physical backup,
    for testing purposes is necessary to build a BackupInfo avoiding the usage
    of Mock/MagicMock classes as much as possible.

def test_result(self):
        # preparation
        writer = self._mock_writer()

        args = ('1', 'two')
        kwargs = dict(three=3, four=5)
        output.result('command', *args, **kwargs)
        output.result('another_command')

        # writer test
        writer.result_command.assert_called_once_with(*args, **kwargs)
        writer.result_another_command.assert_called_once_with()

'pg_receivexlog_installed': True,
                'pg_receivexlog_compatible': True,
                'pg_receivexlog_synchronous': False,
                'pg_receivexlog_path': 'fake/path'
            }
            archiver.receive_wal()
        # Test: general failure executing pg_receivexlog
        with pytest.raises(ArchiverFailure):
            remote_mock.return_value = {
                'pg_receivexlog_installed': True,
                'pg_receivexlog_compatible': True,
                'pg_receivexlog_synchronous': False,
                'pg_receivexlog_path': 'fake/path'
            }
            receivexlog_mock.return_value.execute.side_effect = \
                CommandFailedException
            archiver.receive_wal()