test/test_copier.py

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