File: __init__.py

package info (click to toggle)
python-pypump 0.7-6
  • links: PTS, VCS
  • area: main
  • in suites: sid, trixie
  • size: 560 kB
  • sloc: python: 3,153; makefile: 134
file content (189 lines) | stat: -rw-r--r-- 5,271 bytes parent folder | download | duplicates (4)
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
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
from __future__ import absolute_import

import json
import os
import unittest

import six

from pypump import WebPump, PyPump, Client, AbstractStore


class Response(object):

    def __init__(self, url, data, params=None, status_code=200):
        self.url = url
        self.data = data
        self.status_code = status_code
        self.params = params or {}

    def __getitem__(self, key):
        return self.json()[key]

    def json(self):
        if isinstance(self.data, six.string_types):
            return json.loads(self.data)
        return self.data

    @property
    def content(self):
        return self.data


class Bucket(object):
    """ Container for useful test data """

    def __init__(self, **data):
        for key, value in data.items():
            setattr(self, key, value)


class TestStore(AbstractStore):
    def save(self):
        pass

    @classmethod
    def load(cls, webfinger, pump):
        store = cls()
        store.prefix = webfinger

        # Set testing data
        store["client-key"] = "ClientToken"
        store["client-secret"] = "ClientSecret"
        store["client-expirey"] = 0

        store["oauth-request-token"] = "RequestToken"
        store["oauth-request-secret"] = "RequestSecret"

        store["oauth-access-token"] = "AccessToken"
        store["oauth-access-secret"] = "AccessSecret"

        store["verifier"] = "AVerifier"  # not strictly needed.

        return store


class TestMixin(object):

    _response = None
    _unit_testing = True

    store_class = TestStore

    def __init__(self, *args, **kwargs):
        self.__oauth_testing = {
            'request': {
                'token': 'RequestToken',
                'token_secret': 'RequestTokenSecret',
            },
            'access': {
                'token': 'AccessToken',
                'token_secret': 'AccessTokenSecret',
            },
            'verifier': 'AVerifier',
        }

        # most of the time we don't want to go through oauth
        self.client = Client(
            webfinger="Test@example.com",
            key="AKey",
            secret="ASecret",
            name="PumpTest",
            type="native"
        )

        new_kwargs = dict(
            client=self.client,
        )

        self._response = kwargs.pop("response")
        self._testcase = kwargs.pop("testcase")

        new_kwargs.update(kwargs)

        super(TestMixin, self).__init__(*args, **new_kwargs)

    def get_access(self, *args, **kwargs):
        """ Get verifier """
        return self.store["verifier"]

    def request_token(self, *args, **kwargs):
        """ Gets request token and token secret """
        return {
            "token": self.store["oauth-request-token"],
            "token_secret": self.store["oauth-request-secret"]
        }

    def request_access(self, *args, **kwargs):
        """ Gets access token and token secret """
        return {
            "token": self.store["oauth-access-token"],
            "token_secret": self.store["oauth-access-secret"]
        }

    def set_status_code(self, status_code):
        if self._response is None:
            raise Exception("Response must be given before status code is specified")
        self._response.status_code = int(status_code)

    def get_status_code(self):
        return self._response.status_code

    def _requester(self, *args, **kwargs):
        """ Instead of requesting to a pump server we'll return the data we've been given """
        self._testcase.requests.append(Response(
            url=kwargs.get("endpoint", None),
            data=kwargs.get("data", None),
            params=kwargs.get("params", None)
        ))
        return self._response

    def construct_oauth_url(self):
        return "https://{server}/oauth/authorize?oauth_token=Atoken".format(server=self.client.server)


class TestWebPump(TestMixin, WebPump):
    pass


class TestPump(TestMixin, PyPump):

    def __init__(self, *args, **kwargs):
        kwargs.setdefault('verifier_callback', self._callback)
        return super(TestPump, self).__init__(*args, **kwargs)

    def _callback(self, url):
        return 'a verifier'


class PyPumpTest(unittest.TestCase):
    """
        This is the base test class for PyPump.

        This will provide a testing PyPump class which allows you to easily
        test code in PyPump. It easily allows you to specify pump server return
        values.

    """
    def setUp(self):
        """ This will setup everything needed to test PyPump """
        # response from server, any string will be treated as a json string
        self.response = Response(url=None, data={})

        # These will be set when a request is made
        self.requests = []

        # Setup the bucket
        test_directory = os.path.abspath(os.path.dirname(__file__))
        self.bucket = Bucket(
            path_to_png=os.path.join(test_directory, "bucket", "test_image.png")
        )

        # make the pump object for testing.
        self.pump = TestPump(response=self.response, testcase=self)
        self.webpump = TestWebPump(response=self.response, testcase=self)

    @property
    def request(self):
        """ Returns the last (cronologically) request made """
        return self.requests[-1]