- rewrote "def commit"
[opensuse:osc.git] / tests / common.py
1 import unittest
2 import urllib2
3 import osc.core
4 import StringIO
5 import shutil
6 import tempfile
7 import os
8 import sys
9 from xml.etree import cElementTree as ET
10 EXPECTED_REQUESTS = []
11
12 class RequestWrongOrder(Exception):
13     """raised if an unexpected request is issued to urllib2"""
14     def __init__(self, url, exp_url, method, exp_method):
15         Exception.__init__(self)
16         self.url = url
17         self.exp_url = exp_url
18         self.method = method
19         self.exp_method = exp_method
20
21     def __str__(self):
22         return '%s, %s, %s, %s' % (self.url, self.exp_url, self.method, self.exp_method)
23
24 class RequestDataMismatch(Exception):
25     """raised if POSTed or PUTed data doesn't match with the expected data"""
26     def __init__(self, url, got, exp):
27         self.url = url
28         self.got = got
29         self.exp = exp
30
31     def __str__(self):
32         return '%s, %s, %s' % (self.url, self.got, self.exp)
33
34 class MyHTTPHandler(urllib2.HTTPHandler):
35     def __init__(self, exp_requests, fixtures_dir):
36         urllib2.HTTPHandler.__init__(self)
37         self.__exp_requests = exp_requests
38         self.__fixtures_dir = fixtures_dir
39
40     def http_open(self, req):
41         r = self.__exp_requests.pop(0)
42         if req.get_full_url() != r[1] or req.get_method() != r[0]:
43             raise RequestWrongOrder(req.get_full_url(), r[1], req.get_method(), r[0])
44         if req.get_method() in ('GET', 'DELETE'):
45             return self.__mock_GET(r[1], **r[2])
46         elif req.get_method() in ('PUT', 'POST'):
47             return self.__mock_PUT(req, **r[2])
48
49     def __mock_GET(self, fullurl, **kwargs):
50         return self.__get_response(fullurl, **kwargs)
51
52     def __mock_PUT(self, req, **kwargs):
53         exp = kwargs.get('exp', None)
54         if exp is not None and kwargs.has_key('expfile'):
55             raise RuntimeError('either specify exp or expfile')
56         elif kwargs.has_key('expfile'):
57             exp = open(os.path.join(self.__fixtures_dir, kwargs['expfile']), 'r').read()
58         elif exp is None:
59             raise RuntimeError('exp or expfile required')
60         if exp is not None:
61             if req.get_data() != exp:
62                 raise RequestDataMismatch(req.get_full_url(), repr(req.get_data()), repr(exp))
63         return self.__get_response(req.get_full_url(), **kwargs)
64
65     def __get_response(self, url, **kwargs):
66         f = None
67         if kwargs.has_key('exception'):
68             raise kwargs['exception']
69         if not kwargs.has_key('text') and kwargs.has_key('file'):
70             f = StringIO.StringIO(open(os.path.join(self.__fixtures_dir, kwargs['file']), 'r').read())
71         elif kwargs.has_key('text') and not kwargs.has_key('file'):
72             f = StringIO.StringIO(kwargs['text'])
73         else:
74             raise RuntimeError('either specify text or file')
75         resp = urllib2.addinfourl(f, '', url)
76         resp.code = 200
77         resp.msg = ''
78         return resp
79
80 def urldecorator(method, fullurl, **kwargs):
81     def decorate(test_method):
82         def wrapped_test_method(*args):
83             addExpectedRequest(method, fullurl, **kwargs)
84             test_method(*args)
85         # "rename" method otherwise we cannot specify a TestCaseClass.testName
86         # cmdline arg when using unittest.main()
87         wrapped_test_method.__name__ = test_method.__name__
88         return wrapped_test_method
89     return decorate
90
91 def GET(fullurl, **kwargs):
92     return urldecorator('GET', fullurl, **kwargs)
93
94 def PUT(fullurl, **kwargs):
95     return urldecorator('PUT', fullurl, **kwargs)
96
97 def POST(fullurl, **kwargs):
98     return urldecorator('POST', fullurl, **kwargs)
99
100 def DELETE(fullurl, **kwargs):
101     return urldecorator('DELETE', fullurl, **kwargs)
102
103 def addExpectedRequest(method, url, **kwargs):
104     global EXPECTED_REQUESTS
105     EXPECTED_REQUESTS.append((method, url, kwargs))
106
107 class OscTestCase(unittest.TestCase):
108     def setUp(self):
109         osc.core.conf.get_config(override_conffile=os.path.join(self._get_fixtures_dir(), 'oscrc'))
110         self.tmpdir = tempfile.mkdtemp(prefix='osc_test')
111         shutil.copytree(os.path.join(self._get_fixtures_dir(), 'osctest'), os.path.join(self.tmpdir, 'osctest'))
112         global EXPECTED_REQUESTS
113         EXPECTED_REQUESTS = []
114         osc.core.conf._build_opener = lambda u: urllib2.build_opener(MyHTTPHandler(EXPECTED_REQUESTS, self._get_fixtures_dir()))
115         self.stdout = sys.stdout
116         sys.stdout = StringIO.StringIO()
117
118     def tearDown(self):
119         self.assertTrue(len(EXPECTED_REQUESTS) == 0)
120         sys.stdout = self.stdout
121         try:
122             shutil.rmtree(self.tmpdir)
123         except:
124             pass
125
126     def _get_fixtures_dir(self):
127         raise NotImplementedError('subclasses should implement this method')
128
129     def _change_to_pkg(self, name):
130         os.chdir(os.path.join(self.tmpdir, 'osctest', name))
131
132     def _check_list(self, fname, exp):
133         fname = os.path.join('.osc', fname)
134         self.assertTrue(os.path.exists(fname))
135         self.assertEqual(open(fname, 'r').read(), exp)
136
137     def _check_addlist(self, exp):
138         self._check_list('_to_be_added', exp)
139
140     def _check_deletelist(self, exp):
141         self._check_list('_to_be_deleted', exp)
142
143     def _check_conflictlist(self, exp):
144         self._check_list('_in_conflict', exp)
145
146     def _check_status(self, p, fname, exp):
147         self.assertEqual(p.status(fname), exp)
148
149     def _check_digests(self, fname, *skipfiles):
150         fname = os.path.join(self._get_fixtures_dir(), fname)
151         self.assertEqual(open(os.path.join('.osc', '_files'), 'r').read(), open(fname, 'r').read())
152         root = ET.parse(fname).getroot()
153         for i in root.findall('entry'):
154             if i.get('name') in skipfiles:
155                 continue
156             self.assertTrue(os.path.exists(os.path.join('.osc', i.get('name'))))
157             self.assertEqual(osc.core.dgst(os.path.join('.osc', i.get('name'))), i.get('md5'))