Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_origin_db_head(self):
prev_data = json.dumps({'account': 'acc', 'container': 'cont',
'ttl': 1234, 'logs_enabled': 'true', 'cdn_enabled': 'false'})
self.test_origin.app = FakeApp(iter([
('204 No Content', {}, prev_data)])) # call to _get_cdn_data
req = Request.blank('http://origin_db.com:8080/v1/acc/cont',
environ={'REQUEST_METHOD': 'HEAD'})
resp = req.get_response(self.test_origin)
self.assertEquals(resp.status_int, 204)
self.assertEquals(resp.headers['x-ttl'], 1234)
self.sent += 1
return ' '
return ''
req = Request.blank('/a/c/o',
environ={'REQUEST_METHOD': 'PUT', 'wsgi.input': SlowBody()},
headers={'Content-Length': '4', 'Content-Type': 'text/plain'})
self.app.update_request(req)
controller = proxy_server.ObjectController(self.app, 'account',
'container', 'object')
proxy_server.http_connect = \
fake_http_connect(200, 200, 201, 201, 201)
# acct cont obj obj obj
resp = controller.PUT(req)
self.assertEquals(resp.status_int, 201)
self.app.client_timeout = 0.1
req = Request.blank('/a/c/o',
environ={'REQUEST_METHOD': 'PUT', 'wsgi.input': SlowBody()},
headers={'Content-Length': '4', 'Content-Type': 'text/plain'})
self.app.update_request(req)
proxy_server.http_connect = \
fake_http_connect(201, 201, 201)
# obj obj obj
resp = controller.PUT(req)
self.assertEquals(resp.status_int, 408)
def test_ignore_robotstxt():
"""Ignore the robots.txt resource on root resources."""
class Dummy(wsgiservice.Resource):
_path = '/{id}'
def GET(self, id):
return id
req = webob.Request.blank('/robots.txt')
res = webob.Response()
usr = Dummy(request=req, response=res, path_params={})
res = usr()
print(res)
assert res.status_int == 404
def test_auth_bad_path(self):
res = self.controller.handle_auth(
Request.blank('', environ={'REQUEST_METHOD': 'GET'}))
self.assertEquals(res.status_int, 400)
res = self.controller.handle_auth(Request.blank('/bad',
environ={'REQUEST_METHOD': 'GET'}))
self.assertEquals(res.status_int, 400)
def test_method(self):
r = route(POST, 'news', 'target')
req = Request.blank('/news', {'REQUEST_METHOD': 'POST'})
tr = r(req)
self.assertEqual(
(tr.args, tr.kwargs, tr.target),
((), {}, 'target'))
self.assertRaises(
MethodNotAllowed,
r, Request.blank('/news', {'REQUEST_METHOD': 'DELETE'}))
def test_get_action_success(self, m_repository):
contents = random_string()
cfg = {'return_value.read.return_value': contents}
m_repository.return_value.get_file.configure_mock(**cfg)
filename = random_string()
url = '/actions/%s' % filename
request = Request.blank(url)
resp = request.get_response(ztpserver.controller.Router())
m_repository.return_value.get_file.assert_called_with(url[1:])
self.assertEqual(resp.status_code, constants.HTTP_STATUS_OK)
self.assertEqual(resp.content_type, constants.CONTENT_TYPE_PYTHON)
self.assertEqual(resp.body, contents)
if HostProxy is not no_default:
app = HostProxy(path_info)
path_info = '/'
else:
raise ImportError('restkit is not installed')
environ = kwargs.pop('environ').copy()
environ.update(kwargs)
# unsuported (came from Deliverance)
for key in ['HTTP_ACCEPT_ENCODING', 'HTTP_IF_MATCH',
'HTTP_IF_UNMODIFIED_SINCE', 'HTTP_RANGE', 'HTTP_IF_RANGE']:
if key in environ:
del environ[key]
req = Request.blank(path_info)
req.environ.update(environ)
resp = req.get_response(app)
status = resp.status.split()
ctype = resp.content_type.split(';')[0]
if status[0] not in '45' and ctype == 'text/html':
body = resp.body
else:
body = []
result = self.__class__(body,
parent=self._parent,
app=self.app, # always return self.app
response=resp)
return result
def take_over(self, bad_node, other_nodes, backups=0, root=None):
"""Called when the node should take over from `bad_node`"""
req = Request.blank(
self.url + '/take-over', json={'other': other_nodes, 'bad': bad_node, 'name': self.url, 'backups': backups})
body = forward(req, root=root).body.strip()
if body:
print body
The method returns a webob.Request object; the default
implementation returns a blank Request with only the header
``x-deliverance-theme-subrequest`` set. Subclasses can
override this behavior, e.g. to preserve certain headers from
the original request into subrequests.
``url``:
The URL of the resource to be fetched
``orig_req``:
The original request received by Deliverance
``log``:
The logging object
"""
subreq = Request.blank(url)
subreq.headers['x-deliverance-theme-subrequest'] = "1"
return subreq
def create_container(self, account, container):
"""
Create container.
:param account: account name to put the container in
:param container: container name to create
:returns: True if successful, otherwise False
"""
req = webob.Request.blank('/v1/%s/%s' % (account, container),
environ={'REQUEST_METHOD': 'PUT'})
req.account = account
resp = self.upload_app.handle_request(
self.upload_app.update_request(req))
tries = 1
while (resp.status_int < 200 or resp.status_int > 299) \
and tries <= self.retries:
resp = self.upload_app.handle_request(
self.upload_app.update_request(req))
tries += 1
return 200 <= resp.status_int < 300