Added test_pushtoolslib.py and test_push.py.

tuf/tests/test_push.py

@@ -0,0 +1,164 @@
+"""
+<Program Name>
+  test_push.py
+
+<Author>
+  Konstantin Andrianov
+
+<Started>
+  April 2013.
+
+<Copyright>
+  See LICENSE for licensing information.
+
+<Purpose>
+  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)

tuf/tests/test_pushtoolslib.py

@@ -0,0 +1,169 @@
+"""
+<Program Name>
+  test_pushtoolslib.py
+
+<Author>
+  Konstantin Andrianov
+
+<Started>
+  April 2013.
+
+<Copyright>
+  See LICENSE for licensing information.
+
+<Purpose>
+  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)