1 from unittest2 import TestCase
3 from StringIO import StringIO
4 from htsworkflow.automation import copier
5 from htsworkflow.automation.solexa import is_runfolder
7 class testCopier(TestCase):
8 def test_empty_config(self):
9 cfg = StringIO("""[fake]
12 bot = copier.CopierBot('fake', configfile=cfg)
13 self.failUnlessRaises(RuntimeError, bot.read_config)
15 def test_full_config(self):
16 cfg = StringIO("""[copier]
17 jid: copier@example.fake
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
27 c = copier.CopierBot("copier", configfile=cfg)
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.
46 c.validate_url('rsync://localhost/tmp/sequencer_source'),
47 'rsync://localhost/tmp/sequencer_source')
49 c.validate_url('rsync://localhost/tmp/sequencer_source/'),
50 'rsync://localhost/tmp/sequencer_source/')
52 c.validate_url('rsync://localhost/tmp/sequencer_source/bleem'),
53 'rsync://localhost/tmp/sequencer_source/bleem')
55 c.validate_url('rsync://user@server:1234/other_sequencer'),
56 'rsync://user@server:1234/other_sequencer')
59 def test_dirlist_filter(self):
61 test our dir listing parser
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('/', '/', '/')
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',
73 result = r.list_filter(listing)
74 self.failUnlessEqual(len(result), 1)
75 self.failUnlessEqual(result[0][-1], '4')
79 from unittest2 import TestSuite, defaultTestLoader
81 suite.addTests(defaultTestLoader.loadTestsFromTestCase(testCopier))
85 if __name__ == "__main__":
86 from unittest2 import main
87 main(defaultTest="suite")