test/test_copier.py

   1 from unittest import TestCase
   2
   3 from io import StringIO
   4 from htsworkflow.automation import copier
   5 from htsworkflow.automation.solexa import is_runfolder
   6
   7 class testCopier(TestCase):
   8     def test_empty_config(self):
   9         cfg = StringIO("""[fake]
  10 something: unrelated
  11 """)
  12         bot = copier.CopierBot('fake', configfile=cfg)
  13         self.failUnlessRaises(RuntimeError, bot.read_config)
  14
  15     def test_full_config(self):
  16         cfg = StringIO("""[copier]
  17 jid: copier@example.fake
  18 password: badpassword
  19 authorized_users: user1@example.fake user2@example.fake
  20 rsync_password_file: ~/.sequencer
  21 rsync_sources: rsync://localhost/tmp/sequencer_source rsync://user@server:1234/other_sequencer
  22 rsync_destination: /tmp/sequencer_destination
  23 notify_users: user3@example.fake
  24 # who to run to
  25 #runner:
  26 """)
  27         c = copier.CopierBot("copier", configfile=cfg)
  28         c.read_config()
  29         c._init_rsync()
  30         self.failUnlessEqual(c.jid, 'copier@example.fake')
  31         self.failUnlessEqual(c.cfg['password'], 'badpassword')
  32         self.failUnlessEqual(len(c.authorized_users), 2)
  33         self.failUnlessEqual(c.authorized_users[0], 'user1@example.fake')
  34         self.failUnlessEqual(c.authorized_users[1], 'user2@example.fake')
  35         self.failUnlessEqual(c.rsync.source_base_list[0],
  36                              'rsync://localhost/tmp/sequencer_source/')
  37         self.failUnlessEqual(c.rsync.dest_base, '/tmp/sequencer_destination')
  38         self.failUnlessEqual(len(c.notify_users), 1)
  39         self.failUnlessEqual(c.notify_users[0], 'user3@example.fake')
  40         self.failUnlessEqual(c.validate_url('rsync://other/tmp'), None)
  41         self.failUnlessEqual(c.validate_url('http://localhost/tmp'), None)
  42         # In the rsync process the URL gets a trailing '/' added to it
  43         # But in the bot config its still slash-less.
  44         # It is debatable when to add the trailing slash.
  45         self.failUnlessEqual(
  46           c.validate_url('rsync://localhost/tmp/sequencer_source'),
  47           'rsync://localhost/tmp/sequencer_source')
  48         self.failUnlessEqual(
  49           c.validate_url('rsync://localhost/tmp/sequencer_source/'),
  50           'rsync://localhost/tmp/sequencer_source/')
  51         self.failUnlessEqual(
  52           c.validate_url('rsync://localhost/tmp/sequencer_source/bleem'),
  53           'rsync://localhost/tmp/sequencer_source/bleem')
  54         self.failUnlessEqual(
  55           c.validate_url('rsync://user@server:1234/other_sequencer'),
  56           'rsync://user@server:1234/other_sequencer')
  57
  58
  59     def test_dirlist_filter(self):
  60        """
  61        test our dir listing parser
  62        """
  63        # everyone should have a root dir, and since we're not
  64        # currently writing files... it should all be good
  65        r = copier.rsync('/', '/', '/')
  66
  67        listing = [
  68          'drwxrwxr-x           0 2007/12/29 12:34:56 071229_USI-EAS229_001_FC1234\n',
  69          '-rwxrw-r--      123268 2007/12/29 17:39:31 2038EAAXX.rtf\n',
  70          '-rwxrw-r--           6 2007/12/29 15:10:29 New Text Document.txt\n',
  71        ]
  72
  73        result = r.list_filter(listing)
  74        self.failUnlessEqual(len(result), 1)
  75        self.failUnlessEqual(result[0][-1], '4')
  76
  77
  78 def suite():
  79     from unittest import TestSuite, defaultTestLoader
  80     suite = TestSuite()
  81     suite.addTests(defaultTestLoader.loadTestsFromTestCase(testCopier))
  82     return suite
  83
  84
  85 if __name__ == "__main__":
  86     from unittest import main
  87     main(defaultTest="suite")