diff --git a/tuf/tests/test_push.py b/tuf/tests/test_push.py new file mode 100644 index 00000000..ef5d7bfc --- /dev/null +++ b/tuf/tests/test_push.py @@ -0,0 +1,164 @@ +""" + + test_push.py + + + Konstantin Andrianov + + + April 2013. + + + See LICENSE for licensing information. + + + Test push.py. + +""" + +import os +import getpass +import tempfile +import unittest +import ConfigParser + +import tuf +import tuf.pushtools.push as push +import tuf.pushtools.transfer.scp as scp +import tuf.pushtools.pushtoolslib as pushtoolslib +import tuf.tests.system_tests.util_test_tools as util_test_tools + + + +class TestPush(unittest.TestCase): + src_push_dict = {} + print scp.transfer + + ORIGINAL_PUSH_CONFIG = pushtoolslib.PUSH_CONFIG + + + + @staticmethod + def _mock_scp_transfer(config_dict): + pass + + + + @staticmethod + def write_config_file(config_filename, config_dictionary): + """Create a configuration file by writing supplied configuration + dictionary ('config_dictionary') into the file ('config_filename').""" + + config = ConfigParser.RawConfigParser() + + for section, values_dict in config_dictionary.iteritems(): + config.add_section(section) + for key in values_dict: + config.set(section, key, values_dict[key]) + + # Writing our configuration file to 'config_filename'. + with open(config_filename, 'wb') as configfile: + config.write(configfile) + + + + def setUp(self): + # Create push configuration file, general temporary dir 'root_repo'. + cwd = os.getcwd() + self.push_config = tempfile.mkstemp(prefix='tmp_push_conf_', dir=cwd)[1] + self.root_repo = tempfile.mkdtemp(prefix='tmp_tuf_repo_', dir=cwd) + + # Drop target files into the working project directory 'reg_repo'. + # When targets metadata updates, the target files in the 'reg_repo' + # will be copied into the tuf's targets directory. + self.reg_repo = os.path.join(self.root_repo, 'reg_repo') + os.mkdir(self.reg_repo) + + # Add a file to the 'reg_repo'. + util_test_tools.add_file_to_repository(self.reg_repo, data='Test String') + + # Create TUF repository. + util_test_tools.init_tuf(self.root_repo) + + # Update the tuf targets metadata. + util_test_tools.make_targets_meta(self.root_repo) + + # Populate 'src_push_dict'. + targets_dir = os.path.join(self.root_repo, 'tuf_repo', 'targets') + metadate_path = os.path.join(self.root_repo, 'tuf_repo', 'metadata', + 'targets.txt') + remote_dir = tempfile.mkdtemp(prefix='tmp_tuf_repo_', dir=self.root_repo) + + self.src_push_dict = {'general':{'transfer_module':'scp', + 'metadata_path':metadate_path, + 'targets_directory':targets_dir}, + 'scp':{'host':'localhost', + 'identity_file':None, + 'user':getpass.getuser(), + 'remote_directory':remote_dir}} + + # Write the config dictionary into the push configuration file. + self.write_config_file(self.push_config, self.src_push_dict) + + # Patch 'pushtoolslib.PUSH_CONFIG'. + pushtoolslib.PUSH_CONFIG = os.path.basename(self.push_config) + + + + def tearDown(self): + # Remove tempfile. + os.remove(self.push_config) + + # Remove TUF repository. + util_test_tools.cleanup(self.root_repo) + + # Clear 'src_push_dict'. + self.src_push_dict.clear() + + # # Reassign 'pushtoolslib.PUSH_CONFIG'. + pushtoolslib.PUSH_CONFIG = self.ORIGINAL_PUSH_CONFIG + + + + def test_expected_behaviour_of_push_without_scp(self): + # Patch 'scp.transfer' function. + ORIGINAL_SCP_TRANSFER_MODULE = scp.transfer + scp.transfer = self._mock_scp_transfer + + push.push(self.push_config) + + # Restore 'scp.transfer' function. + scp.transfer = ORIGINAL_SCP_TRANSFER_MODULE + + + + def test_exceptions_handeling_of_push_without_scp(self): + # Patch 'scp.transfer' function. + ORIGINAL_SCP_TRANSFER_MODULE = scp.transfer + scp.transfer = self._mock_scp_transfer + + self.assertRaises(tuf.Error, push.push, self.root_repo) + self.assertRaises(tuf.FormatError, push.push, None) + self.assertRaises(tuf.FormatError, push.push, 12345) + self.assertRaises(tuf.FormatError, push.push, ['test']) + self.assertRaises(tuf.FormatError, push.push, {'test':'test'}) + + # Restore 'scp.transfer' function. + scp.transfer = ORIGINAL_SCP_TRANSFER_MODULE + + + + # Unit test bellow executes secure copy 'scp' command. Hence user's + # password is requered. Comment-out or remove the line bellow to + # un-skip the unit test. + @unittest.skip("Requires user's password!") + def test_expected_behaviour_of_push_with_scp(self): + push.push(self.push_config) + + + + + +# Run the unittests +suite = unittest.TestLoader().loadTestsFromTestCase(TestPush) +unittest.TextTestRunner(verbosity=2).run(suite) \ No newline at end of file diff --git a/tuf/tests/test_pushtoolslib.py b/tuf/tests/test_pushtoolslib.py new file mode 100644 index 00000000..a1938cd6 --- /dev/null +++ b/tuf/tests/test_pushtoolslib.py @@ -0,0 +1,169 @@ +""" + + test_pushtoolslib.py + + + Konstantin Andrianov + + + April 2013. + + + See LICENSE for licensing information. + + + Test pushtoolslib.py. + +""" + +import os +import tempfile +import unittest +import ConfigParser + +import tuf.formats +import tuf.pushtools.pushtoolslib as pushtoolslib + + + +class TestPushtoolslib(unittest.TestCase): + src_push_dict = {} + src_receive_dict = {} + ORIGINAL_PUSH_CONFIG = pushtoolslib.PUSH_CONFIG + ORIGINAL_RECEIVE_CONFIG = pushtoolslib.RECEIVE_CONFIG + + + + def setUp(self): + # Create config tempfiles. + cwd = os.getcwd() + self.push_config_file = tempfile.mkstemp(prefix='tmp_push_conf_', dir=cwd)[1] + self.receive_config_file = tempfile.mkstemp(prefix='tmp_receive_conf_', dir=cwd)[1] + + # Populate 'src_push_dict' and 'src_receive_dict'. + self.src_push_dict = {'general':{'transfer_module':'scp', + 'metadata_path':'some/path', + 'targets_directory':'some/path'}, + 'scp':{'host':'localhost', + 'user':'user', + 'identity_file':None, + 'remote_directory':'~/pushes'}} + + self.src_receive_dict = {'general':{'pushroots':['some/path'], + 'repository_directory':'some/path', + 'metadata_directory':'some/path', + 'targets_directory':'some/path', + 'backup_directory':'some/path'}} + + # Patch the 'pushtoolslib.PUSH_CONFIG' and 'pushtoolslib.RECEIVE_CONFIG' + pushtoolslib.PUSH_CONFIG = os.path.basename(self.push_config_file) + pushtoolslib.RECEIVE_CONFIG = os.path.basename(self.receive_config_file) + + + + def tearDown(self): + # Remove tempfile. + os.remove(self.push_config_file) + os.remove(self.receive_config_file) + + # Clear dictionaries. + self.src_push_dict.clear() + self.src_receive_dict.clear() + + # Reassign 'pushtoolslib.PUSH_CONFIG' and 'pushtoolslib.RECEIVE_CONFIG' + # to original values. + pushtoolslib.PUSH_CONFIG = self.ORIGINAL_PUSH_CONFIG + pushtoolslib.RECEIVE_CONFIG = self.ORIGINAL_RECEIVE_CONFIG + + + + @staticmethod + def write_config_file(config_filename, config_dictionary): + """Create a configuration file by writing supplied configuration + dictionary ('config_dictionary') into the file ('config_filename').""" + + config = ConfigParser.RawConfigParser() + + for section, values_dict in config_dictionary.iteritems(): + config.add_section(section) + for key in values_dict: + config.set(section, key, values_dict[key]) + + # Writing our configuration file to 'config_filename'. + with open(config_filename, 'wb') as configfile: + config.write(configfile) + + + + def test_expected_behavior_of_read_config_file(self): + # Test 'push' configuration type. + self.write_config_file(self.push_config_file, self.src_push_dict) + config_dict = pushtoolslib.read_config_file(self.push_config_file, 'push') + tuf.formats.SCPCONFIG_SCHEMA.check_match(config_dict) + + # Test 'receive' configuration type. + self.write_config_file(self.receive_config_file, self.src_receive_dict) + config_dict = pushtoolslib.read_config_file(self.receive_config_file, 'receive') + tuf.formats.RECEIVECONFIG_SCHEMA.check_match(config_dict) + + + + def test_exceptions_handeling_of_read_config_file(self): + # Test an incorrect configuration file. + with open(self.push_config_file, 'wb') as configfile: + configfile.write('test') + self.assertRaises(tuf.Error, pushtoolslib.read_config_file, + self.push_config_file, 'push') + + self.write_config_file(self.push_config_file, {}) + self.assertRaises(tuf.Error, pushtoolslib.read_config_file, + self.push_config_file, 'push') + + self.write_config_file(self.receive_config_file, self.src_receive_dict) + self.assertRaises(tuf.Error, pushtoolslib.read_config_file, + self.receive_config_file, 'push') + + self.write_config_file(self.push_config_file, self.src_push_dict) + self.assertRaises(tuf.Error, pushtoolslib.read_config_file, + self.push_config_file, 'receive') + + # Test incorrect configuration type. + self.write_config_file(self.push_config_file, self.src_push_dict) + self.assertRaises(tuf.Error, pushtoolslib.read_config_file, + self.push_config_file, 'junk') + + # Test 'push' type configuration with 'transfer_module' absent from + # config_dict['general']. + saved_transfer_module = self.src_push_dict['general']['transfer_module'] + del self.src_push_dict['general']['transfer_module'] + self.write_config_file(self.push_config_file, self.src_push_dict) + self.assertRaises(tuf.Error, pushtoolslib.read_config_file, + self.push_config_file, 'push') + + # Test 'push' type configuration with + # config_dict['general']['transfer_module'] != 'scp'. + self.src_push_dict['general']['transfer_module'] = 'test' + self.write_config_file(self.push_config_file, self.src_push_dict) + self.assertRaises(tuf.Error, pushtoolslib.read_config_file, + self.push_config_file, 'push') + + # Test 'push' type configuration with one of the keys missing. + self.src_push_dict['general']['transfer_module'] = 'scp' + del self.src_push_dict['general']['metadata_path'] + self.write_config_file(self.push_config_file, self.src_push_dict) + self.assertRaises(tuf.FormatError, pushtoolslib.read_config_file, + self.push_config_file, 'push') + + # Test 'receive' type configuration with one of the keys missing. + del self.src_receive_dict['general']['repository_directory'] + self.write_config_file(self.receive_config_file, self.src_receive_dict) + self.assertRaises(tuf.FormatError, pushtoolslib.read_config_file, + self.receive_config_file, 'receive') + + + + + +# Run the unittests +suite = unittest.TestLoader().loadTestsFromTestCase(TestPushtoolslib) +unittest.TextTestRunner(verbosity=2).run(suite) \ No newline at end of file