← Back to branch summary

~ricardokirkner/rnr-server/sentry

« back to all changes in this revision

Viewing changes to src/core/tests/doubles.py

  • Committer: Tarmac
  • Author(s): Ricardo Kirkner
  • Date: 2016-05-17 14:37:37 UTC
  • mfrom: (316.1.11 trunk)
  • Revision ID: tarmac-20160517143737-upuntzyo0kv78mzg
[r=fgallina] support submitting reviews using macaroon based authentication

Show diffs side-by-side

added added

removed removed

Lines of Context:
src/core/tests/doubles.py
 
1
from __future__ import absolute_import, unicode_literals
 
2
 
 
3
import json
 
4
 
 
5
import responses
 
6
from django.conf import settings
 
7
from requests import PreparedRequest
 
8
 
 
9
from core.utilities import WebServices
 
10
 
 
11
 
 
12
# XXX: RequestsDouble and RequestsDoubleTestCaseMixin come from
 
13
# appstore-common, this copypasta (with modifications) is an attempt
 
14
# to generalize it a bit and later on move the code into its own
 
15
# module and reuse it across projects.
 
16
class RequestsDouble(object):
 
17
 
 
18
    def start(self):
 
19
        self.reset()
 
20
        responses.start()
 
21
 
 
22
    def stop(self):
 
23
        # Stopping does not triggers a reset and it should because
 
24
        # responses dirty state can become a problem for other tests.
 
25
        try:
 
26
            self.reset()
 
27
            responses.stop()
 
28
        except RuntimeError:
 
29
            raise RuntimeError(
 
30
                'Calling stop on a RequestDouble that is not started.'
 
31
                'This may be because you are instantiating your own instead '
 
32
                'of reusing the RequestDouble from your TestCase')
 
33
 
 
34
    def reset(self):
 
35
        responses.reset()
 
36
 
 
37
    def reset_calls(self):
 
38
        responses.calls.reset()
 
39
 
 
40
    def make_body(self, data=None, files=None):
 
41
        req = PreparedRequest()
 
42
        # need to define headers before body
 
43
        req.prepare_headers({})
 
44
        req.prepare_body(data=data, files=files)
 
45
        return req.body
 
46
 
 
47
    def add_response(self, url, status_code=200, body='',
 
48
                     match_querystring=False, method='GET',
 
49
                     callback=None, reset=False):
 
50
        if reset:
 
51
            self.reset()
 
52
        if callback:
 
53
            responses.add_callback(method, url, callback=callback,
 
54
                                   match_querystring=match_querystring)
 
55
        else:
 
56
            responses.add(method, url, match_querystring=match_querystring,
 
57
                          status=status_code, body=body)
 
58
 
 
59
    def _call_has_headers(self, call, headers):
 
60
        has_headers = True
 
61
 
 
62
        if headers is not None:
 
63
            # make sure all given headers were included in call
 
64
            for key, value in headers.items():
 
65
                if call.request.headers.get(key) != value:
 
66
                    has_headers = False
 
67
                    break
 
68
 
 
69
        return has_headers
 
70
 
 
71
    def find_call(self, url, headers=None, body=None, method=None):
 
72
        """Find calls for url optionally matching headers, body and method.
 
73
 
 
74
        The given headers can be a subset of the actual call headers,
 
75
        body must be an exact match.
 
76
 
 
77
        """
 
78
        for call in responses.calls:
 
79
            test_body = call.request.body if body is None else body
 
80
            if (call.request.url == url
 
81
                    and self._call_has_headers(call, headers)
 
82
                    and test_body == call.request.body):
 
83
                if method is not None and call.request.method != method:
 
84
                    continue
 
85
                return (url, call.request.headers, call.request.body)
 
86
 
 
87
    def find_calls(self, url, method=None):
 
88
        calls = (c for c in responses.calls if c.request.url == url)
 
89
        if method:
 
90
            calls = (c for c in calls if c.request.method == method)
 
91
        return [(url, c.request.headers, c.request.body) for c in calls]
 
92
 
 
93
 
 
94
class SCADouble(WebServices):
 
95
 
 
96
    def __init__(self, requests_double, *args, **kwargs):
 
97
        WebServices.__init__(self, *args, **kwargs)
 
98
        self.requests_double = requests_double
 
99
 
 
100
        self.verify_acl_url = (
 
101
            settings.SCA_HOST_URL.strip('/') + '/dev/api/acl/verify/')
 
102
 
 
103
    def set_verify_acl_response(self, **kwargs):
 
104
        body = {
 
105
            'allowed': True,
 
106
            'refresh_required': False,
 
107
            'account': {
 
108
                'openid': 'oid1234',
 
109
                'email': 'user@example.com',
 
110
                'displayname': 'user1234',
 
111
                'verified': True,
 
112
            },
 
113
            'last_auth': '20160515T12:00:00.000',
 
114
            'permissions': ['package_access'],
 
115
        }
 
116
        body.update(**kwargs)
 
117
        self.requests_double.add_response(
 
118
            self.verify_acl_url, method='POST',
 
119
            body=json.dumps(body))