summaryrefslogtreecommitdiff
path: root/systest
blob: 7142d98a10ad9fe17d3c5b26b5c609fe2b6904a1 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
#!/usr/bin/python

import cliapp
import logging
import re
import subprocess
import unittest


class AssertionFailure(cliapp.AppException):

    def __init__(self, msg):
        self._str = msg
        
    def __str__(self):
        return self._str


class TestCase(unittest.TestCase):

    '''Base class for system tests.
    
    This extends ``unittest.TestCase`` with some things that are 
    relevant to system tests.
    
    '''

    def runcmd(self, argv, stdin='', *args, **kwargs):
        p = subprocess.Popen(argv, 
                             stdin=subprocess.PIPE, 
                             stdout=subprocess.PIPE, 
                             stderr=subprocess.PIPE)
        out, err = p.communicate(stdin)
        return p.returncode, out, err

    def hostcmd(self, argv, *args, **kwargs):
        '''Run external command on test host.'''
        returncode, out, err = self.runcmd(argv, *args, **kwargs)
        if returncode:
            msg = 'host command failed: %s\n%s' % (' '.join(argv), err)
            logging.error(msg)
            raise cliapp.AppException(msg)
        return out

    def targetcmd(self, argv, *args, **kwargs):
        '''Run command on test target.'''
        full_argv = (['ssh', '-l', self.settings['user'], 
                      self.settings['target']] +
                     argv)
        returncode, out, err = self.runcmd(full_argv, *args, **kwargs)
        if returncode:
            msg = 'target command failed: %s\n%s' % (' '.join(argv), err)
            logging.error(msg)
            raise cliapp.AppException(msg)
        return out

    def assertMatches(self, pat, text, msg=None):
        self.assert_(re.match(pat, text),
                     msg=('pattern %s does not match %s' % (pat, text)) or msg)


class DebianBaseTests(TestCase):

    def test_only_ssh_port(self):
        out = self.hostcmd(['nmap', self.settings['target']])
        ports = [line.split()[0]
                 for line in out.splitlines()
                 if ' open ' in line]
        self.assertEqual(ports, ['22/tcp'])

    def test_ssh_login(self):
        user = self.settings['user']
        out = self.hostcmd(['ssh', '-l', user, self.settings['target'], 'id'])
        self.assertMatches(r'^uid=1000\(%s\)' % user, out)

    def test_simple_dns_lookup(self):
        out = self.targetcmd(['host', 'www.debian.org'])
        self.assert_('www.debian.org' in out)
        
    def test_ping_localhost(self):
        self.targetcmd(['ping', '-c1', 'localhost'])
        
    def test_ping6_localhost(self):
        self.targetcmd(['ping6', '-c1', 'ip6-localhost'])
        
    def test_cat(self):
        out = self.targetcmd(['cat'], stdin='foo')
        self.assertEqual(out, 'foo')

#    def test_sudo(self):
#        out = self.targetcmd(['sudo', 'id'], 
#                             stdin=self.settings['user-password'])
#        self.assertMatches(r'^uid=0\(root\)', out)



class SystemTest(cliapp.Application):

    def add_settings(self):
        self.settings.boolean(['verbose', 'v'], 
                              'print names of tests when run')
        self.settings.string(['target'], 'target domain name or IP address')
        self.settings.string(['user'], 'user on target')
        self.settings.string(['user-password'], 'password for target user')

    def process_args(self, args):
        loader = unittest.defaultTestLoader
        loader.suiteClass = self.create_suite
        suite = loader.loadTestsFromTestCase(DebianBaseTests)
        unittest.TextTestRunner().run(suite)

    def create_suite(self, tests):
        for test in tests:
            test.settings = self.settings
        suite = unittest.TestSuite(tests)
        return suite

    def mangle(self, testname):
        return 'test_' + testname.replace('-', '_')
    
    def unmangle(self, methodname):
        assert methodname.startswith('test_')
        methodname = methodname[len('test_'):]
        return methodname.replace('_', '-')


if __name__ == '__main__':
    SystemTest().run()