pbs recipe: added test
[slapos.git] / slapos / test / recipe / test_pbs.py
1
2 import os
3 import shutil
4 import tempfile
5 import unittest
6
7 from slapos.recipe import pbs
8
9
10 class PBSTest(unittest.TestCase):
11
12 def new_recipe(self):
13 buildout = {
14 'buildout': {
15 'bin-directory': '',
16 'find-links': '',
17 'allow-hosts': '',
18 'develop-eggs-directory': '',
19 'eggs-directory': '',
20 },
21 'slap-connection': {
22 'computer-id': '',
23 'partition-id': '',
24 'server-url': '',
25 'software-release-url': '',
26 }
27 }
28
29 options = {
30 'rdiffbackup-binary': ''
31 }
32
33 return pbs.Recipe(buildout=buildout, name='pbs', options=options)
34
35 def test_push(self):
36 recipe = self.new_recipe()
37
38 with tempfile.NamedTemporaryFile() as rdiff_wrapper:
39 recipe.wrapper_push(remote_schema='TEST_REMOTE_SCHEMA',
40 local_dir='TEST_LOCAL_DIR',
41 remote_dir='TEST_REMOTE_DIR',
42 rdiff_wrapper_path=rdiff_wrapper.name)
43 content = rdiff_wrapper.read()
44 self.assertIn('--remote-schema TEST_REMOTE_SCHEMA', content)
45 self.assertIn('TEST_LOCAL_DIR', content)
46 self.assertIn('TEST_REMOTE_DIR', content)
47
48 def test_pull(self):
49 recipe = self.new_recipe()
50
51 with tempfile.NamedTemporaryFile() as rdiff_wrapper:
52 recipe.wrapper_pull(remote_schema='TEST_REMOTE_SCHEMA',
53 local_dir='TEST_LOCAL_DIR',
54 remote_dir='TEST_REMOTE_DIR',
55 rdiff_wrapper_path=rdiff_wrapper.name,
56 remove_backup_older_than='TEST_OLDER')
57 content = rdiff_wrapper.read()
58 self.assertIn('--remote-schema TEST_REMOTE_SCHEMA', content)
59 self.assertIn('TEST_LOCAL_DIR', content)
60 self.assertIn('TEST_REMOTE_DIR', content)
61 self.assertIn('--remove-older-than TEST_OLDER', content)
62
63 def test_invalid_type(self):
64 recipe = self.new_recipe()
65
66 entry = {
67 'url': 'http://url.to.something/',
68 'type': 'invalid'
69 }
70
71 self.assertRaisesRegexp(ValueError,
72 'type parameter must be either pull or push',
73 recipe.add_slave, entry=entry, known_hosts_file={})
74
75 def test_install(self):
76 recipe = self.new_recipe()
77
78 promises_directory = tempfile.mkdtemp()
79 wrappers_directory = tempfile.mkdtemp()
80 directory = tempfile.mkdtemp()
81 feeds_directory = tempfile.mkdtemp()
82 run_directory = tempfile.mkdtemp()
83 cron_directory = tempfile.mkdtemp()
84
85 recipe.options.update({
86 'promises-directory': promises_directory,
87 'wrappers-directory': wrappers_directory,
88 'sshclient-binary': 'TEST_SSH_CLIENT',
89 'directory': directory,
90 'notifier-binary': 'TEST_NOTIFIER',
91 'feeds': feeds_directory,
92 'notifier-url': 'http://url.to.notifier/',
93 'run-directory': run_directory,
94 'cron-entries': cron_directory,
95 'known-hosts': 'TEST_KNOWN_HOSTS',
96 'slave-instance-list': '''[
97 {
98 "url": "http://url.to.pull/",
99 "type": "pull",
100 "notification-id": "pulltest",
101 "server-key": "TEST_SERVER_KEY",
102 "name": "TEST_NAME",
103 "notify": "http://url.to.notify/",
104 "frequency": "TEST_FREQUENCY"
105 }, {
106 "url": "http://url.to.push/",
107 "type": "push",
108 "notification-id": "pushtest",
109 "server-key": "TEST_SERVER_KEY",
110 "name": "TEST_NAME",
111 "notify": "http://url.to.notify/",
112 "frequency": "TEST_FREQUENCY"
113 }
114 ]
115 '''
116 })
117
118 recipe._install()
119
120 self.assertItemsEqual(os.listdir(promises_directory),
121 ['pulltest', 'pushtest'])
122
123 self.assertItemsEqual(os.listdir(wrappers_directory),
124 ['pulltest_raw', 'pulltest', 'pushtest_raw', 'pushtest'])
125
126 self.assertItemsEqual(os.listdir(directory),
127 ['TEST_NAME'])
128
129 self.assertItemsEqual(os.listdir(feeds_directory),
130 ['pulltest', 'pushtest'])
131
132 self.assertItemsEqual(os.listdir(run_directory),
133 [])
134
135 self.assertItemsEqual(os.listdir(cron_directory),
136 ['pulltest', 'pushtest'])
137
138 shutil.rmtree(promises_directory)
139 shutil.rmtree(wrappers_directory)
140 shutil.rmtree(directory)
141 shutil.rmtree(feeds_directory)
142 shutil.rmtree(run_directory)
143 shutil.rmtree(cron_directory)
144
145