Registry.mock_send()   F
last analyzed

Complexity

Conditions 11

Size

Total Lines 36

Duplication

Lines 0
Ratio 0 %

Importance

Changes 0
Metric Value
cc 11
dl 0
loc 36
rs 3.1764
c 0
b 0
f 0

How to fix   Complexity   

Complexity

Complex classes like Registry.mock_send() often do a lot of different things. To break such a class down, we need to identify a cohesive component within that class. A common approach to find such a component is to look for fields/methods that share the same prefixes, or suffixes.

Once you have determined the fields that belong together, you can apply the Extract Class refactoring. If the component makes sense as a sub-class, Extract Subclass is also a candidate, and is often faster.

1
'''
2
Created on Feb 22, 2014
3
4
@author: sean
5
'''
6
from __future__ import unicode_literals
7
import requests
8
from functools import wraps
9
import json
10
11
try:
12
    unicode
13
except NameError:
14
    unicode = str
15
16
from collections import namedtuple
17
18
rule = namedtuple('rule', ('url', 'path', 'method', 'status', 'content', 'side_effect', 'res', 'headers', 'expected_headers'))
19
20
def filter_request(m, prepared_request):
21
    if m.url and  m.url != prepared_request.url:
22
        return False
23
24
    if m.path and  m.path != prepared_request.path_url:
25
        return False
26
27
    if m.method and m.method != prepared_request.method:
28
        return False
29
30
    return True
31
32
class Responses(object):
33
    def __init__(self):
34
        self._resps = []
35
36
    def append(self, res):
37
        self._resps.append(res)
38
39
    @property
40
    def called(self):
41
        return (len(self._resps))
42
43
    @property
44
    def req(self):
45
        if self._resps:
46
            return self._resps[0][1]
47
48
    def assertCalled(self, url=''):
49
        assert self.called, "The url %s was not called" % url
50
51
    def assertNotCalled(self, url=''):
52
        assert not self.called, "The url %s was called" % url
53
54
class Registry(object):
55
    def __init__(self):
56
        self._map = []
57
58
    def __enter__(self):
59
        self.real_send = requests.Session.send
60
        requests.Session.send = self.mock_send
61
62
        return self
63
64
    def __exit__(self, *exec_info):
65
        requests.Session.send = self.real_send
66
        return
67
68
    def mock_send(self, prepared_request, *args, **kwargs):
69
70
        rule = next((m for m in self._map[::-1] if filter_request(m, prepared_request)), None)
71
72
        if rule is None:
73
            raise Exception('No matching rule found for url [%s] %s' % (prepared_request.method,
74
                                                                          prepared_request.url,
75
                                                                          ))
76
77
        if rule.expected_headers:
78
            for header, value in rule.expected_headers.items():
79
                if header not in prepared_request.headers:
80
                    raise Exception("{}: header {} expected in {}".format(prepared_request.url, header, prepared_request.headers))
81
82
                if prepared_request.headers[header] != value:
83
                    raise Exception("{}: header {} has unexpected value {} was expecting {}".format(prepared_request.url, header, prepared_request.headers[header], value))
84
85
        content = rule.content
86
        if isinstance(content, dict):
87
            content = json.dumps(content)
88
        if isinstance(content, unicode):
89
            content = content.encode()
90
91
        res = requests.models.Response()
92
        res.status_code = rule.status
93
        res._content_consumed = True
94
        res._content = content
95
        res.encoding = 'utf-8'
96
        res.request = prepared_request
97
        res.headers.update(rule.headers or {})
98
        rule.res.append((res, prepared_request))
99
100
        if rule.side_effect:
101
            rule.side_effect()
102
103
        return res
104
105
    def register(self, url=None, path=None, method='GET', status=200, content=b'', side_effect=None, headers=None, expected_headers=None):
106
        res = Responses()
107
        self._map.append(rule(url, path, method, status, content, side_effect, res, headers, expected_headers))
108
        return res
109
110
    def unregister(self, res):
111
        for item in list(self._map):
112
            if res == item.res:
113
                self._map.remove(item)
114
                return
115
116
    def assertAllCalled(self):
117
        for item in self._map:
118
            res = item.res
119
            res.assertCalled('[%s] %s%s' % (item.method or 'any', item.url or 'http://<any>', item.path))
120
121
def urlpatch(func):
122
    @wraps(func)
123
    def inner(self, *args, **kwargs):
124
        with Registry() as r:
125
            return func(self, r, *args, **kwargs)
126
    return inner
127
128