-
Notifications
You must be signed in to change notification settings - Fork 25
/
fakeclient.py
127 lines (109 loc) · 4.02 KB
/
fakeclient.py
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
from __future__ import absolute_import
import mock
import koji
from koji.xmlrpcplus import Fault
class BaseFakeClientSession(koji.ClientSession):
def __init__(self, *a, **kw):
super(BaseFakeClientSession, self).__init__(*a, **kw)
def multiCall(self, strict=False):
if not self.multicall:
raise Exception("not in multicall")
ret = []
self.multicall = False
calls = self._calls
self._calls = []
for call in calls:
method = call['methodName']
args, kwargs = koji.decode_args(call['params'])
try:
result = self._callMethod(method, args, kwargs)
ret.append(result)
except Fault as fault:
if strict:
raise
else:
ret.append({'faultCode': fault.faultCode,
'faultString': fault.faultString})
return ret
class FakeClientSession(BaseFakeClientSession):
def __init__(self, *a, **kw):
super(FakeClientSession, self).__init__(*a, **kw)
self._calldata = {}
self._offsets = {}
def load_calls(self, data):
"""Load call data
Data should be a list of dictionaries with keys:
- method
- args
- kwargs
- result (for successful calls)
- fault (for errors)
That represent call data, e.g. as generated by RecordingClientSession
"""
for call in data:
key = self._munge([call['method'], call['args'], call['kwargs']])
self._calldata.setdefault(key, []).append(call)
def _callMethod(self, name, args, kwargs=None, retry=True):
if self.multicall:
return super(FakeClientSession, self)._callMethod(name, args,
kwargs, retry)
key = self._munge([name, args, kwargs])
# we may have a series of calls for each key
calls = self._calldata.get(key)
ofs = self._offsets.get(key, 0)
call = calls[ofs]
ofs += 1
if ofs < len(calls):
# don't go past the end
self._offsets[key] = ofs
if call:
if 'fault' in call:
fault = Fault(call['fault']['faultCode'],
call['fault']['faultString'])
raise koji.convertFault(fault)
else:
return call['result']
else:
return mock.MagicMock()
def _munge(self, data):
def callback(value):
if isinstance(value, list):
return tuple(value)
elif isinstance(value, dict):
keys = sorted(value.keys())
return tuple([(k, value[k]) for k in keys])
else:
return value
walker = koji.util.DataWalker(data, callback)
return walker.walk()
class RecordingClientSession(BaseFakeClientSession):
def __init__(self, *a, **kw):
super(RecordingClientSession, self).__init__(*a, **kw)
self._calldata = []
def get_calls(self):
return self._calldata
def _callMethod(self, name, args, kwargs=None, retry=True):
if self.multicall:
return super(RecordingClientSession, self)._callMethod(name, args,
kwargs, retry)
call = {
'method': name,
'args': args,
'kwargs': kwargs,
}
self._calldata.append(call)
try:
result = super(RecordingClientSession, self)._callMethod(name, args,
kwargs, retry)
call['result'] = result
return result
except Fault as fault:
err = {'faultCode': fault.faultCode,
'faultString': fault.faultString}
call['fault'] = err
raise
except koji.GenericError as e:
err = {'faultCode': e.faultCode,
'faultString': str(e)}
call['fault'] = err
raise