source: web/old/remctl-2.14/python/test_remctl.py @ f6f3e91

web
Last change on this file since f6f3e91 was f6f3e91, checked in by Jessica B. Hamrick <jhamrick@…>, 15 years ago

Preserve directory hierarchy (not sure what happened to it)

  • Property mode set to 100644
File size: 8.0 KB
Line 
1# test_remctl.py -- Test suite for remctl Python bindings
2#
3# Written by Russ Allbery <rra@stanford.edu>
4# Copyright 2008 Board of Trustees, Leland Stanford Jr. University
5#
6# See LICENSE for licensing terms.
7
8import remctl
9import errno, os, re, signal, time, unittest
10
11def needs_kerberos(func):
12    """unittest test method decorator to skip tests requiring Kerberos
13
14    Used to annotate test methods that require a Kerberos configuration.
15    Ignores failures in the annotated test method.
16    """
17    def wrapper(*args, **kw):
18        if not os.path.isfile('data/test.principal'):
19            return True
20        else:
21            return func(*args, **kw)
22    wrapper.__name__ = func.__name__
23    wrapper.__doc__ = func.__doc__
24    return wrapper
25
26class TestRemctl(unittest.TestCase):
27    def get_principal(self):
28        file = open('data/test.principal', 'r')
29        principal = file.read().rstrip()
30        file.close()
31        return principal
32
33    @needs_kerberos
34    def start_remctld(self):
35        try:
36            os.remove('data/pid')
37        except OSError, (error, strerror):
38            if error != errno.ENOENT:
39                raise
40        principal = self.get_principal()
41        child = os.fork()
42        if child == 0:
43            output = open('data/test.output', 'w')
44            os.dup2(output.fileno(), 1)
45            os.chdir('/afs/sipb.mit.edu/project/sipb-www/sipbmp3web/remctl-2.14/tests')
46            os.execl('/afs/sipb.mit.edu/project/sipb-www/sipbmp3web/remctl-2.14/server/remctld', 'remctld', '-m',
47                     '-p', '14373', '-s', principal, '-f', 'data/conf-simple',
48                     '-P', '/afs/sipb.mit.edu/project/sipb-www/sipbmp3web/remctl-2.14/tests/data/pid', '-d', '-S',
49                     '-F', '-k', '/afs/sipb.mit.edu/project/sipb-www/sipbmp3web/remctl-2.14/tests/data/test.keytab')
50        if not os.path.isfile('data/pid'):
51            time.sleep(1)
52
53    def stop_remctld(self):
54        try:
55            file = open('data/pid', 'r')
56            pid = file.read().rstrip()
57            file.close()
58            os.kill(int(pid), signal.SIGTERM)
59            child, status = os.waitpid(int(pid), 0)
60        except IOError, (error, strerror):
61            if error != errno.ENOENT:
62                raise
63
64    @needs_kerberos
65    def run_kinit(self):
66        os.environ['KRB5CCNAME'] = 'data/test.cache'
67        self.principal = self.get_principal()
68        commands = ('kinit -k -t data/test.keytab ' + self.principal,
69                    'kinit -t data/test.keytab ' + self.principal,
70                    'kinit -k -K data/test.keytab ' + self.principal)
71        for command in commands:
72            if os.system(command + ' >/dev/null </dev/null') == 0:
73                return True
74        if not os.path.isfile('data/pid'):
75            time.sleep(1)
76        stop_remctld()
77        return False
78
79    def setUp(self):
80        os.chdir('/afs/sipb.mit.edu/project/sipb-www/sipbmp3web/remctl-2.14/tests')
81        self.start_remctld()
82        assert(self.run_kinit())
83
84    @needs_kerberos
85    def tearDown(self):
86        self.stop_remctld()
87        os.remove('data/test.output')
88        try:
89            os.remove('data/test.cache')
90        except OSError, (error, strerror):
91            if error != errno.ENOENT:
92                raise
93
94class TestRemctlSimple(TestRemctl):
95    @needs_kerberos
96    def test_simple_success(self):
97        command = ('test', 'test')
98        result = remctl.remctl('localhost', 14373, self.principal, command)
99        self.assertEqual(result.stdout, "hello world\n")
100        self.assertEqual(result.stderr, None)
101        self.assertEqual(result.status, 0)
102
103    @needs_kerberos
104    def test_simple_status(self):
105        command = [ 'test', 'status', '2' ]
106        result = remctl.remctl(host = 'localhost', command = command,
107                               port = '14373', principal = self.principal)
108        self.assertEqual(result.stdout, None)
109        self.assertEqual(result.stderr, None)
110        self.assertEqual(result.status, 2)
111
112    @needs_kerberos
113    def test_simple_failure(self):
114        command = ('test', 'bad-command')
115        try:
116            result = remctl.remctl('localhost', 14373, self.principal, command)
117        except remctl.RemctlProtocolError, error:
118            self.assertEqual(str(error), 'Unknown command')
119
120    @needs_kerberos
121    def test_simple_errors(self):
122        try:
123            remctl.remctl()
124        except TypeError:
125            pass
126        try:
127            remctl.remctl('localhost')
128        except ValueError, error:
129            self.assertEqual(str(error), 'command must not be empty')
130        try:
131            remctl.remctl(host = 'localhost', command = 'foo')
132        except TypeError, error:
133            self.assertEqual(str(error),
134                             'command must be a sequence or iterator')
135        try:
136            remctl.remctl('localhost', "foo", self.principal, [])
137        except TypeError, error:
138            self.assertEqual(str(error), "port must be a number: 'foo'")
139        try:
140            remctl.remctl('localhost', -1, self.principal, [])
141        except ValueError, error:
142            self.assertEqual(str(error), 'invalid port number: -1')
143        try:
144            remctl.remctl('localhost', 14373, self.principal, [])
145        except ValueError, error:
146            self.assertEqual(str(error), 'command must not be empty')
147        try:
148            remctl.remctl('localhost', 14373, self.principal, 'test')
149        except TypeError, error:
150            self.assertEqual(str(error),
151                             'command must be a sequence or iterator')
152
153class TestRemctlFull(TestRemctl):
154    @needs_kerberos
155    def test_full_success(self):
156        r = remctl.Remctl()
157        r.open('localhost', 14373, self.principal)
158        r.command(['test', 'test'])
159        type, data, stream, status, error = r.output()
160        self.assertEqual(type, "output")
161        self.assertEqual(data, "hello world\n")
162        self.assertEqual(stream, 1)
163        type, data, stream, status, error = r.output()
164        self.assertEqual(type, "status")
165        self.assertEqual(status, 0)
166        type, data, stream, status, error = r.output()
167        self.assertEqual(type, "done")
168        r.close()
169
170    @needs_kerberos
171    def test_full_failure(self):
172        r = remctl.Remctl('localhost', 14373, self.principal)
173        r.command(['test', 'bad-command'])
174        type, data, stream, status, error = r.output()
175        self.assertEqual(type, "error")
176        self.assertEqual(data, 'Unknown command')
177        self.assertEqual(error, 5)
178
179    @needs_kerberos
180    def test_full_errors(self):
181        r = remctl.Remctl()
182        try:
183            r.open()
184        except TypeError:
185            pass
186        try:
187            r.open('localhost', 'foo')
188        except TypeError, error:
189            self.assertEqual(str(error), "port must be a number: 'foo'")
190        try:
191            r.open('localhost', -1)
192        except ValueError, error:
193            self.assertEqual(str(error), 'invalid port number: -1')
194        pattern = 'cannot connect to localhost \(port 14444\): .*'
195        try:
196            r.open('localhost', 14444)
197        except remctl.RemctlError, error:
198            self.assert_(re.compile(pattern).match(str(error)))
199        self.assert_(re.compile(pattern).match(r.error()))
200        try:
201            r.command(['test', 'test'])
202        except remctl.RemctlNotOpenedError, error:
203            self.assertEqual(str(error), 'no currently open connection')
204        r.open('localhost', 14373, self.principal)
205        try:
206            r.command('test')
207        except TypeError, error:
208            self.assertEqual(str(error),
209                             'command must be a sequence or iterator')
210        try:
211            r.command([])
212        except ValueError, error:
213            self.assertEqual(str(error), 'command must not be empty')
214        r.close()
215        try:
216            r.output()
217        except remctl.RemctlNotOpenedError, error:
218            self.assertEqual(str(error), 'no currently open connection')
219        self.assertEqual(r.error(), 'no currently open connection')
220
221if __name__ == '__main__':
222    unittest.main()
Note: See TracBrowser for help on using the repository browser.