psw/borgmatic
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Configuration support for additional attic prune flags: keep_within, keep_hourly, keep_yearly, and prefix.

Browse source
This commit is contained in:
Dan Helfman 2014-12-06 18:35:20 -08:00
parent b94c106a36
commit 056ed7184b
9 changed files with 341 additions and 124 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
.hgignore
View file

@ -1,3 +1,4 @@
syntax: glob
*.pyc
*.egg-info
*.pyc
*.swp

8
NEWS Normal file
View file

@ -0,0 +1,8 @@
0.0.2
* Configuration support for additional attic prune flags: keep_within, keep_hourly, keep_yearly,
and prefix.
0.0.1
* Initial release.

39
atticmatic/attic.py
View file

@ -5,6 +5,10 @@ import subprocess
def create_archive(excludes_filename, verbose, source_directories, repository):
'''
Given an excludes filename, a vebosity flag, a space-separated list of source directories, and
a local or remote repository path, create an attic archive.
'''
sources = tuple(source_directories.split(' '))
command = (
@ -22,13 +26,40 @@ def create_archive(excludes_filename, verbose, source_directories, repository):
subprocess.check_call(command)
def prune_archives(repository, verbose, keep_daily, keep_weekly, keep_monthly):
def make_prune_flags(retention_config):
'''
Given a retention config dict mapping from option name to value, tranform it into an iterable of
command-line name-value flag pairs.
For example, given a retention config of:
{'keep_weekly': 4, 'keep_monthly': 6}
This will be returned as an iterable of:
(
('--keep-weekly', '4'),
('--keep-monthly', '6'),
)
'''
return (
('--' + option_name.replace('_', '-'), str(retention_config[option_name]))
for option_name, value in retention_config.items()
)
def prune_archives(verbose, repository, retention_config):
'''
Given a verbosity flag, a local or remote repository path, and a retention config dict, prune
attic archives according the the retention policy specified in that configuration.
'''
command = (
'attic', 'prune',
repository,
'--keep-daily', str(keep_daily),
'--keep-weekly', str(keep_weekly),
'--keep-monthly', str(keep_monthly),
) + tuple(
element
for pair in make_prune_flags(retention_config)
for element in pair
) + (('--verbose',) if verbose else ())
subprocess.check_call(command)

4
atticmatic/command.py
View file

@ -38,8 +38,8 @@ def main():
args = parse_arguments()
location_config, retention_config = parse_configuration(args.config_filename)
create_archive(args.excludes_filename, args.verbose, *location_config)
prune_archives(location_config.repository, args.verbose, *retention_config)
create_archive(args.excludes_filename, args.verbose, **location_config)
prune_archives(args.verbose, location_config['repository'], retention_config)
except (ValueError, IOError, CalledProcessError) as error:
print(error, file=sys.stderr)
sys.exit(1)

145
atticmatic/config.py
View file

@ -1,4 +1,4 @@
from collections import namedtuple
from collections import OrderedDict, namedtuple
try:
# Python 2
@ -8,58 +8,121 @@ except ImportError:
from configparser import ConfigParser
CONFIG_SECTION_LOCATION = 'location'
CONFIG_SECTION_RETENTION = 'retention'
Section_format = namedtuple('Section_format', ('name', 'options'))
Config_option = namedtuple('Config_option', ('name', 'value_type', 'required'))
CONFIG_FORMAT = {
CONFIG_SECTION_LOCATION: ('source_directories', 'repository'),
CONFIG_SECTION_RETENTION: ('keep_daily', 'keep_weekly', 'keep_monthly'),
def option(name, value_type=str, required=True):
'''
Given a config file option name, an expected type for its value, and whether it's required,
return a Config_option capturing that information.
'''
return Config_option(name, value_type, required)
CONFIG_FORMAT = (
Section_format(
'location',
(
option('source_directories'),
option('repository'),
),
),
Section_format(
'retention',
(
option('keep_within', required=False),
option('keep_hourly', int, required=False),
option('keep_daily', int, required=False),
option('keep_weekly', int, required=False),
option('keep_monthly', int, required=False),
option('keep_yearly', int, required=False),
option('prefix', required=False),
),
)
)
def validate_configuration_format(parser, config_format):
'''
Given an open ConfigParser and an expected config file format, validate that the parsed
configuration file has the expected sections, that any required options are present in those
sections, and that there aren't any unexpected options.
Raise ValueError if anything is awry.
'''
section_names = parser.sections()
required_section_names = tuple(section.name for section in config_format)
if set(section_names) != set(required_section_names):
raise ValueError(
'Expected config sections {} but found sections: {}'.format(
', '.join(required_section_names),
', '.join(section_names)
)
)
for section_format in config_format:
option_names = parser.options(section_format.name)
expected_options = section_format.options
unexpected_option_names = set(option_names) - set(option.name for option in expected_options)
if unexpected_option_names:
raise ValueError(
'Unexpected options found in config section {}: {}'.format(
section_format.name,
', '.join(sorted(unexpected_option_names)),
)
)
missing_option_names = tuple(
option.name for option in expected_options if option.required
if option.name not in option_names
)
if missing_option_names:
raise ValueError(
'Required options missing from config section {}: {}'.format(
section_format.name,
', '.join(missing_option_names)
)
)
def parse_section_options(parser, section_format):
'''
Given an open ConfigParser and an expected section format, return the option values from that
section as a dict mapping from option name to value. Omit those options that are not present in
the parsed options.
Raise ValueError if any option values cannot be coerced to the expected Python data type.
'''
type_getter = {
str: parser.get,
int: parser.getint,
}
LocationConfig = namedtuple('LocationConfig', CONFIG_FORMAT[CONFIG_SECTION_LOCATION])
RetentionConfig = namedtuple('RetentionConfig', CONFIG_FORMAT[CONFIG_SECTION_RETENTION])
return OrderedDict(
(option.name, type_getter[option.value_type](section_format.name, option.name))
for option in section_format.options
if parser.has_option(section_format.name, option.name)
)
def parse_configuration(config_filename):
'''
Given a config filename of the expected format, return the parse configuration as a tuple of
(LocationConfig, RetentionConfig).
Given a config filename of the expected format, return the parsed configuration as a tuple of
(location config, retention config) where each config is a dict of that section's options.
Raise IOError if the file cannot be read, or ValueError if the format is not as expected.
'''
parser = ConfigParser()
parser.readfp(open(config_filename))
section_names = parser.sections()
expected_section_names = CONFIG_FORMAT.keys()
if set(section_names) != set(expected_section_names):
raise ValueError(
'Expected config sections {} but found sections: {}'.format(
', '.join(expected_section_names),
', '.join(section_names)
)
)
validate_configuration_format(parser, CONFIG_FORMAT)
for section_name in section_names:
option_names = parser.options(section_name)
expected_option_names = CONFIG_FORMAT[section_name]
if set(option_names) != set(expected_option_names):
raise ValueError(
'Expected options {} in config section {} but found options: {}'.format(
', '.join(expected_option_names),
section_name,
', '.join(option_names)
)
)
return (
LocationConfig(*(
parser.get(CONFIG_SECTION_LOCATION, option_name)
for option_name in CONFIG_FORMAT[CONFIG_SECTION_LOCATION]
)),
RetentionConfig(*(
parser.getint(CONFIG_SECTION_RETENTION, option_name)
for option_name in CONFIG_FORMAT[CONFIG_SECTION_RETENTION]
))
return tuple(
parse_section_options(parser, section_format)
for section_format in CONFIG_FORMAT
)

41
atticmatic/tests/unit/test_attic.py
View file

@ -1,3 +1,5 @@
from collections import OrderedDict
from flexmock import flexmock
from atticmatic import attic as module
@ -52,7 +54,32 @@ def test_create_archive_with_verbose_should_call_attic_with_verbose_parameters()
)
BASE_PRUNE_FLAGS = (
('--keep-daily', '1'),
('--keep-weekly', '2'),
('--keep-monthly', '3'),
)
def test_make_prune_flags_should_return_flags_from_config():
retention_config = OrderedDict(
(
('keep_daily', 1),
('keep_weekly', 2),
('keep_monthly', 3),
)
)
result = module.make_prune_flags(retention_config)
assert tuple(result) == BASE_PRUNE_FLAGS
def test_prune_archives_should_call_attic_with_parameters():
retention_config = flexmock()
flexmock(module).should_receive('make_prune_flags').with_args(retention_config).and_return(
BASE_PRUNE_FLAGS,
)
insert_subprocess_mock(
(
'attic', 'prune', 'repo', '--keep-daily', '1', '--keep-weekly', '2', '--keep-monthly',
@ -61,15 +88,17 @@ def test_prune_archives_should_call_attic_with_parameters():
)
module.prune_archives(
repository='repo',
verbose=False,
keep_daily=1,
keep_weekly=2,
keep_monthly=3
repository='repo',
retention_config=retention_config,
)
def test_prune_archives_with_verbose_should_call_attic_with_verbose_parameters():
retention_config = flexmock()
flexmock(module).should_receive('make_prune_flags').with_args(retention_config).and_return(
BASE_PRUNE_FLAGS,
)
insert_subprocess_mock(
(
'attic', 'prune', 'repo', '--keep-daily', '1', '--keep-weekly', '2', '--keep-monthly',
@ -80,7 +109,5 @@ def test_prune_archives_with_verbose_should_call_attic_with_verbose_parameters()
module.prune_archives(
repository='repo',
verbose=True,
keep_daily=1,
keep_weekly=2,
keep_monthly=3
retention_config=retention_config,
)

216
atticmatic/tests/unit/test_config.py
View file

@ -1,84 +1,166 @@
from collections import OrderedDict
from flexmock import flexmock
from nose.tools import assert_raises
from atticmatic import config as module
def insert_mock_parser(section_names):
def test_option_should_create_config_option():
option = module.option('name', bool, required=False)
assert option == module.Config_option('name', bool, False)
def test_option_should_create_config_option_with_defaults():
option = module.option('name')
assert option == module.Config_option('name', str, True)
def test_validate_configuration_format_with_valid_config_should_not_raise():
parser = flexmock()
parser.should_receive('sections').and_return(('section', 'other'))
parser.should_receive('options').with_args('section').and_return(('stuff',))
parser.should_receive('options').with_args('other').and_return(('such',))
config_format = (
module.Section_format(
'section',
options=(
module.Config_option('stuff', str, required=True),
),
),
module.Section_format(
'other',
options=(
module.Config_option('such', str, required=True),
),
),
)
module.validate_configuration_format(parser, config_format)
def test_validate_configuration_format_with_missing_section_should_raise():
parser = flexmock()
parser.should_receive('sections').and_return(('section',))
config_format = (
module.Section_format('section', options=()),
module.Section_format('missing', options=()),
)
with assert_raises(ValueError):
module.validate_configuration_format(parser, config_format)
def test_validate_configuration_format_with_extra_section_should_raise():
parser = flexmock()
parser.should_receive('sections').and_return(('section', 'extra'))
config_format = (
module.Section_format('section', options=()),
)
with assert_raises(ValueError):
module.validate_configuration_format(parser, config_format)
def test_validate_configuration_format_with_missing_required_option_should_raise():
parser = flexmock()
parser.should_receive('sections').and_return(('section',))
parser.should_receive('options').with_args('section').and_return(('option',))
config_format = (
module.Section_format(
'section',
options=(
module.Config_option('option', str, required=True),
module.Config_option('missing', str, required=True),
),
),
)
with assert_raises(ValueError):
module.validate_configuration_format(parser, config_format)
def test_validate_configuration_format_with_missing_optional_option_should_not_raise():
parser = flexmock()
parser.should_receive('sections').and_return(('section',))
parser.should_receive('options').with_args('section').and_return(('option',))
config_format = (
module.Section_format(
'section',
options=(
module.Config_option('option', str, required=True),
module.Config_option('missing', str, required=False),
),
),
)
module.validate_configuration_format(parser, config_format)
def test_validate_configuration_format_with_extra_option_should_raise():
parser = flexmock()
parser.should_receive('sections').and_return(('section',))
parser.should_receive('options').with_args('section').and_return(('option', 'extra'))
config_format = (
module.Section_format(
'section',
options=(module.Config_option('option', str, required=True),),
),
)
with assert_raises(ValueError):
module.validate_configuration_format(parser, config_format)
def test_parse_section_options_should_return_section_options():
parser = flexmock()
parser.should_receive('get').with_args('section', 'foo').and_return('value')
parser.should_receive('getint').with_args('section', 'bar').and_return(1)
parser.should_receive('has_option').with_args('section', 'foo').and_return(True)
parser.should_receive('has_option').with_args('section', 'bar').and_return(True)
section_format = module.Section_format(
'section',
(
module.Config_option('foo', str, required=True),
module.Config_option('bar', int, required=True),
),
)
config = module.parse_section_options(parser, section_format)
assert config == OrderedDict(
(
('foo', 'value'),
('bar', 1),
)
)
def insert_mock_parser():
parser = flexmock()
parser.should_receive('readfp')
parser.should_receive('sections').and_return(section_names)
flexmock(module).open = lambda filename: None
flexmock(module).ConfigParser = parser
return parser
def test_parse_configuration_should_return_config_data():
section_names = (module.CONFIG_SECTION_LOCATION, module.CONFIG_SECTION_RETENTION)
parser = insert_mock_parser(section_names)
def test_parse_configuration_should_return_section_configs():
parser = insert_mock_parser()
mock_module = flexmock(module)
mock_module.should_receive('validate_configuration_format').with_args(
parser, module.CONFIG_FORMAT,
).once()
mock_section_configs = (flexmock(), flexmock())
for section_name in section_names:
parser.should_receive('options').with_args(section_name).and_return(
module.CONFIG_FORMAT[section_name],
)
for section_format, section_config in zip(module.CONFIG_FORMAT, mock_section_configs):
mock_module.should_receive('parse_section_options').with_args(
parser, section_format,
).and_return(section_config).once()
expected_config = (
module.LocationConfig(flexmock(), flexmock()),
module.RetentionConfig(flexmock(), flexmock(), flexmock()),
)
sections = (
(module.CONFIG_SECTION_LOCATION, expected_config[0], 'get'),
(module.CONFIG_SECTION_RETENTION, expected_config[1], 'getint'),
)
section_configs = module.parse_configuration('filename')
for section_name, section_config, method_name in sections:
for index, option_name in enumerate(module.CONFIG_FORMAT[section_name]):
(
parser.should_receive(method_name).with_args(section_name, option_name)
.and_return(section_config[index])
)
config = module.parse_configuration(flexmock())
assert config == expected_config
def test_parse_configuration_with_missing_section_should_raise():
insert_mock_parser((module.CONFIG_SECTION_LOCATION,))
with assert_raises(ValueError):
module.parse_configuration(flexmock())
def test_parse_configuration_with_extra_section_should_raise():
insert_mock_parser((module.CONFIG_SECTION_LOCATION, module.CONFIG_SECTION_RETENTION, 'extra'))
with assert_raises(ValueError):
module.parse_configuration(flexmock())
def test_parse_configuration_with_missing_option_should_raise():
section_names = (module.CONFIG_SECTION_LOCATION, module.CONFIG_SECTION_RETENTION)
parser = insert_mock_parser(section_names)
for section_name in section_names:
parser.should_receive('options').with_args(section_name).and_return(
module.CONFIG_FORMAT[section_name][:-1],
)
with assert_raises(ValueError):
module.parse_configuration(flexmock())
def test_parse_configuration_with_extra_option_should_raise():
section_names = (module.CONFIG_SECTION_LOCATION, module.CONFIG_SECTION_RETENTION)
parser = insert_mock_parser(section_names)
for section_name in section_names:
parser.should_receive('options').with_args(section_name).and_return(
module.CONFIG_FORMAT[section_name] + ('extra',),
)
with assert_raises(ValueError):
module.parse_configuration(flexmock())
assert section_configs == mock_section_configs

5
sample/config
View file

@ -7,6 +7,11 @@ repository: user@backupserver:sourcehostname.attic
[retention]
# Retention policy for how many backups to keep in each category.
# See https://attic-backup.org/usage.html#attic-prune for details.
#keep_within: 3h
#keep_hourly: 24
keep_daily: 7
keep_weekly: 4
keep_monthly: 6
keep_yearly: 1
#prefix: sourcehostname

2
setup.py
View file

@ -2,7 +2,7 @@ from setuptools import setup, find_packages
setup(
name='atticmatic',
version='0.0.1',
version='0.0.2',
description='A wrapper script for Attic backup software that creates and prunes backups',
author='Dan Helfman',
author_email='witten@torsion.org',