forked from sigmavirus24/github3.py
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathutils.py
More file actions
127 lines (93 loc) · 3.14 KB
/
Copy pathutils.py
File metadata and controls
127 lines (93 loc) · 3.14 KB
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
import json
import os
import sys
if sys.version_info < (3, 0):
from unittest2 import TestCase
else:
from unittest import TestCase
import requests
import github3
try:
from unittest import mock
except ImportError:
import mock
from io import BytesIO
from requests.structures import CaseInsensitiveDict
is_py3 = sys.version_info > (3, 0)
def load(name):
with path(name) as f:
j = json.load(f)
return j
def path(name, mode='r'):
return open('tests/json/{0}'.format(name), mode)
class BaseCase(TestCase):
github_url = 'https://api.github.com/'
def setUp(self):
self.g = github3.GitHub()
self.session = self.g._session
if os.environ.get('GH_AUTH'):
self.g.login(token=os.environ['GH_AUTH'])
self.args = ()
self.conf = {'allow_redirects': True}
self.mock = mock.patch.object(requests.sessions.Session, 'request')
self.request = self.mock.start()
def tearDown(self):
self.mock.stop()
def login(self):
self.g.login('user', 'password')
def mock_assertions(self):
assert self.request.called is True
conf = self.conf.copy()
args, kwargs = self.request.call_args
assert self.args == args
if 'data' in self.conf:
if isinstance(self.conf['data'], dict):
for k, v in list(self.conf['data'].items()):
s = json.dumps({k: v})[1:-1]
assert s in kwargs['data']
else:
assert self.conf['data'] == kwargs['data']
del self.conf['data']
for k in self.conf:
assert k in kwargs
assert self.conf[k] == kwargs[k]
self.request.reset_mock()
self.conf = conf
def response(self, path_name, status_code=200, enc='utf-8',
_iter=False, **headers):
r = requests.Response()
r.status_code = status_code
r.encoding = enc
if path_name:
with path(path_name) as f:
content = f.read().strip()
if _iter:
content = '[{0}]'.format(content)
r.raw = RequestsBytesIO(content.encode())
elif is_py3:
r.raw = RequestsBytesIO(content.encode())
else:
r.raw = RequestsBytesIO(content)
else:
r.raw = RequestsBytesIO()
if headers:
r.headers = CaseInsensitiveDict(headers)
self.request.return_value = r
def delete(self, url):
self.args = ('DELETE', url)
self.conf = {}
def get(self, url):
self.args = ('GET', url)
def patch(self, url):
self.args = ('PATCH', url)
def post(self, url):
self.args = ('POST', url)
def put(self, url):
self.args = ('PUT', url)
def not_called(self):
assert self.request.called is False
def assertGitHubErrorRaised(self, func, *args, **kwargs):
return self.assertRaises(github3.GitHubError, func(*args, **kwargs))
class RequestsBytesIO(BytesIO):
def read(self, chunk_size, *args, **kwargs):
return super(RequestsBytesIO, self).read(chunk_size)