def test_add_product(self):
""" POST request to a product page augments the CartItem instance count """
quantity = 2
product_url = self.product.get_absolute_url()
response = self.client.get(product_url)
self.assertEqual(response.status_code, httplib.OK )
# store count in cart_count variable
cart_item_count = self.get_cart_item_count()
# assert that the cart item count is zero
self.failUnlessEqual(cart_item_count, 0)
# perform the post of adding to the cart
cookie = self.client.cookies[settings.SESSION_COOKIE_NAME]
csrf_token = csrf.middleware._make_token(cookie.value)
#self.failUnlessEqual(csrf_token, None)
postdata = {'product_slug': self.product.slug,
'quantity': quantity,
'csrfmiddlewaretoken': csrf_token }
response = self.client.post(product_url, postdata )
# assert redirected to cart page - 302 then 200?
cart_url = urlresolvers.reverse('show_cart')
self.assertRedirects(response, cart_url, status_code=httplib.FOUND, target_status_code=httplib.OK)
# assert cart item count is incremented by one
self.assertEqual(self.get_cart_item_count(), cart_item_count + 1)
cart_id = self.get_cart_id()
last_item = CartItem.objects.filter(cart_id=cart_id).latest('date_added')
# assert the latest cart item has a quantity of two
self.failUnlessEqual(last_item.quantity, quantity)
# assert the latest cart item is the correct product
self.failUnlessEqual(last_item.product, self.product)
python类FOUND的实例源码
def test_user_login_redirect(self):
""" if URL contains 'next' parameter, customer is redirected after successfully logging in """
my_account_url = urlresolvers.reverse('my_account')
redirect_url = self.login_url + '?next=' + my_account_url
response = self.client.get(my_account_url)
self.assertRedirects(response, redirect_url,
status_code=httplib.FOUND, target_status_code=httplib.OK)
response = self.client.post(redirect_url, self.login_data)
self.assertRedirects(response, my_account_url,
status_code=httplib.FOUND, target_status_code=httplib.OK)
def _redirect_302_path_info(self, updated_path_info, environ, start_response):
"""Redirect to an updated path.
Respond to the current request with a 302 Found status with an updated path
but preserving the rest of the request.
Notes:
- WSGI does not make the fragment available so we are not able to preserve
it. Luckily prod does not preserve the fragment so it works out.
Args:
updated_path_info: the new HTTP path to redirect to.
environ: WSGI environ object.
start_response: WSGI start response callable.
Returns:
WSGI-compatible iterable object representing the body of the response.
"""
correct_url = urlparse.urlunsplit(
(environ['wsgi.url_scheme'],
environ['HTTP_HOST'],
urllib.quote(updated_path_info),
self._quote_querystring(environ['QUERY_STRING']),
None))
content_type = 'text/html; charset=utf-8'
output = _REDIRECT_HTML % {
'content-type': content_type,
'status': httplib.FOUND,
'correct-url': correct_url
}
start_response('%d %s' % (httplib.FOUND, httplib.responses[httplib.FOUND]),
[('Content-Type', content_type),
('Location', correct_url),
('Content-Length', str(len(output)))])
return output
def ReadHttpResponse(conn, expect_status=200, ignore_404=True):
"""Reads an http response from a connection into a string buffer.
Args:
conn: An HTTPSConnection or HTTPConnection created by CreateHttpConn, above.
expect_status: Success is indicated by this status in the response.
ignore_404: For many requests, gerrit-on-borg will return 404 if the request
doesn't match the database contents. In most such cases, we
want the API to return None rather than raise an Exception.
Returns: A string buffer containing the connection's reply.
"""
sleep_time = 0.5
for idx in range(TRY_LIMIT):
response = conn.getresponse()
# Check if this is an authentication issue.
www_authenticate = response.getheader('www-authenticate')
if (response.status in (httplib.UNAUTHORIZED, httplib.FOUND) and
www_authenticate):
auth_match = re.search('realm="([^"]+)"', www_authenticate, re.I)
host = auth_match.group(1) if auth_match else conn.req_host
reason = ('Authentication failed. Please make sure your .netrc file '
'has credentials for %s' % host)
raise GerritAuthenticationError(response.status, reason)
# If response.status < 500 then the result is final; break retry loop.
if response.status < 500:
break
# A status >=500 is assumed to be a possible transient error; retry.
http_version = 'HTTP/%s' % ('1.1' if response.version == 11 else '1.0')
msg = (
'A transient error occurred while querying %s:\n'
'%s %s %s\n'
'%s %d %s' % (
conn.host, conn.req_params['method'], conn.req_params['url'],
http_version, http_version, response.status, response.reason))
if TRY_LIMIT - idx > 1:
msg += '\n... will retry %d more times.' % (TRY_LIMIT - idx - 1)
time.sleep(sleep_time)
sleep_time = sleep_time * 2
req_host = conn.req_host
req_params = conn.req_params
conn = GetConnectionClass()(req_host)
conn.req_host = req_host
conn.req_params = req_params
conn.request(**req_params)
LOGGER.warn(msg)
if ignore_404 and response.status == 404:
return StringIO()
if response.status != expect_status:
reason = '%s: %s' % (response.reason, response.read())
raise GerritError(response.status, reason)
return StringIO(response.read())
def ReadHttpResponse(conn, expect_status=200, ignore_404=True):
"""Reads an http response from a connection into a string buffer.
Args:
conn: An HTTPSConnection or HTTPConnection created by CreateHttpConn, above.
expect_status: Success is indicated by this status in the response.
ignore_404: For many requests, gerrit-on-borg will return 404 if the request
doesn't match the database contents. In most such cases, we
want the API to return None rather than raise an Exception.
Returns: A string buffer containing the connection's reply.
"""
sleep_time = 0.5
for idx in range(TRY_LIMIT):
response = conn.getresponse()
# Check if this is an authentication issue.
www_authenticate = response.getheader('www-authenticate')
if (response.status in (httplib.UNAUTHORIZED, httplib.FOUND) and
www_authenticate):
auth_match = re.search('realm="([^"]+)"', www_authenticate, re.I)
host = auth_match.group(1) if auth_match else conn.req_host
reason = ('Authentication failed. Please make sure your .netrc file '
'has credentials for %s' % host)
raise GerritAuthenticationError(response.status, reason)
# If response.status < 500 then the result is final; break retry loop.
if response.status < 500:
LOGGER.debug('got response %d for %s %s', response.status,
conn.req_params['method'], conn.req_params['url'])
break
# A status >=500 is assumed to be a possible transient error; retry.
http_version = 'HTTP/%s' % ('1.1' if response.version == 11 else '1.0')
LOGGER.warn('A transient error occurred while querying %s:\n'
'%s %s %s\n'
'%s %d %s',
conn.host, conn.req_params['method'], conn.req_params['url'],
http_version, http_version, response.status, response.reason)
if TRY_LIMIT - idx > 1:
LOGGER.warn('... will retry %d more times.', TRY_LIMIT - idx - 1)
time.sleep(sleep_time)
sleep_time = sleep_time * 2
req_host = conn.req_host
req_params = conn.req_params
conn = GetConnectionClass()(req_host)
conn.req_host = req_host
conn.req_params = req_params
conn.request(**req_params)
if ignore_404 and response.status == 404:
return StringIO()
if response.status != expect_status:
reason = '%s: %s' % (response.reason, response.read())
raise GerritError(response.status, reason)
return StringIO(response.read())
