Merge "Extract and test inventory and backup I/O"

This commit is contained in:
Jenkins 2016-06-09 22:42:58 +00:00 committed by Gerrit Code Review
commit 8110716870
3 changed files with 67 additions and 22 deletions

@ -1051,6 +1051,36 @@ def load_user_configuration(config_path):
return user_defined_config
def make_backup(config_path, inventory_file_path):
# Create a backup of all previous inventory files as a tar archive
inventory_backup_file = os.path.join(
config_path,
'backup_openstack_inventory.tar'
)
with tarfile.open(inventory_backup_file, 'a') as tar:
basename = os.path.basename(inventory_file_path)
backup_name = get_backup_name(basename)
tar.add(inventory_file_path, arcname=backup_name)
def get_backup_name(basename):
utctime = datetime.datetime.utcnow()
utctime = utctime.strftime("%Y%m%d_%H%M%S")
return '%s-%s.json' % (basename, utctime)
def get_inventory(config_path, inventory_file_path):
if os.path.isfile(inventory_file_path):
with open(inventory_file_path, 'rb') as f:
dynamic_inventory = json.loads(f.read())
make_backup(config_path, inventory_file_path)
else:
dynamic_inventory = copy.deepcopy(INVENTORY_SKEL)
return dynamic_inventory
def main(all_args):
"""Run the main application."""
# Get the path to the user configuration files
@ -1066,24 +1096,8 @@ def main(all_args):
dynamic_inventory_file = os.path.join(
config_path, 'openstack_inventory.json'
)
if os.path.isfile(dynamic_inventory_file):
with open(dynamic_inventory_file, 'rb') as f:
dynamic_inventory = json.loads(f.read())
# Create a backup of all previous inventory files as a tar archive
inventory_backup_file = os.path.join(
config_path,
'backup_openstack_inventory.tar'
)
with tarfile.open(inventory_backup_file, 'a') as tar:
basename = os.path.basename(dynamic_inventory_file)
# Time stamp the inventory file in UTC
utctime = datetime.datetime.utcnow()
utctime = utctime.strftime("%Y%m%d_%H%M%S")
backup_name = '%s-%s.json' % (basename, utctime)
tar.add(dynamic_inventory_file, arcname=backup_name)
else:
dynamic_inventory = copy.deepcopy(INVENTORY_SKEL)
dynamic_inventory = get_inventory(config_path, dynamic_inventory_file)
# Save the users container cidr as a group variable
cidr_networks = user_defined_config.get('cidr_networks')

@ -5,6 +5,7 @@ coverage<=4.0.3 # Apache-2.0
flake8==2.2.4
hacking>=0.10.0,<0.11
mccabe==0.2.1 # capped for flake8
mock == 2.0.0
pep8==1.5.7
pyflakes==0.8.1
virtualenv>=14.0.0

@ -3,6 +3,7 @@
import collections
import copy
import json
import mock
import os
from os import path
import Queue
@ -644,15 +645,30 @@ class TestNetAddressSearch(unittest.TestCase):
class TestMultipleRuns(unittest.TestCase):
def test_creating_backup_file(self):
# Generate the initial inventory files
get_inventory(clean=False)
inventory_file_path = os.path.join(TARGET_DIR,
'openstack_inventory.json')
get_backup_name_path = 'dynamic_inventory.get_backup_name'
backup_name = 'openstack_inventory.json-20160531_171804.json'
# run again to force creation of the backup files
get_inventory(clean=False)
tar_file = mock.MagicMock()
tar_file.__enter__.return_value = tar_file
# run make backup with faked tarfiles and date
with mock.patch('dynamic_inventory.tarfile.open') as tar_open:
tar_open.return_value = tar_file
with mock.patch(get_backup_name_path) as backup_mock:
backup_mock.return_value = backup_name
di.make_backup(TARGET_DIR, inventory_file_path)
backup_path = path.join(TARGET_DIR, 'backup_openstack_inventory.tar')
self.assertTrue(os.path.exists(backup_path))
tar_open.assert_called_with(backup_path, 'a')
# This chain is present because of how tarfile.open is called to
# make a context manager inside the make_backup function.
tar_file.add.assert_called_with(inventory_file_path,
arcname=backup_name)
def test_recreating_files(self):
# Deleting the files after the first run should cause the files to be
@ -665,6 +681,20 @@ class TestMultipleRuns(unittest.TestCase):
self.assertFalse(os.path.exists(backup_path))
def test_rereading_files(self):
# Generate the initial inventory files
get_inventory(clean=False)
inventory_file_path = os.path.join(TARGET_DIR,
'openstack_inventory.json')
inv = di.get_inventory(TARGET_DIR, inventory_file_path)
self.assertIsInstance(inv, dict)
self.assertIn('_meta', inv)
# This test is basically just making sure we get more than
# INVENTORY_SKEL populated, so we're not going to do deep testing
self.assertIn('log_hosts', inv)
def tearDown(self):
# Clean up here since get_inventory will not do it by design in
# this test.