addons/website/tests/test_requests.py

   1 # -*- coding: utf-8 -*-
   2 import urlparse
   3 import unittest2
   4 import urllib2
   5 import werkzeug.urls
   6
   7 import lxml.html
   8
   9 import openerp
  10 from openerp import tools
  11
  12 import cases
  13
  14 __all__ = ['load_tests', 'CrawlSuite']
  15
  16 class RedirectHandler(urllib2.HTTPRedirectHandler):
  17     """
  18     HTTPRedirectHandler is predicated upon HTTPErrorProcessor being used and
  19     works by intercepting 3xy "errors".
  20
  21     Inherit from it to handle 3xy non-error responses instead, as we're not
  22     using the error processor
  23     """
  24
  25     def http_response(self, request, response):
  26         code, msg, hdrs = response.code, response.msg, response.info()
  27
  28         if 300 <= code < 400:
  29             return self.parent.error(
  30                 'http', request, response, code, msg, hdrs)
  31
  32         return response
  33
  34     https_response = http_response
  35
  36 class CrawlSuite(unittest2.TestSuite):
  37     """ Test suite crawling an openerp CMS instance and checking that all
  38     internal links lead to a 200 response.
  39
  40     If a username and a password are provided, authenticates the user before
  41     starting the crawl
  42     """
  43
  44     def __init__(self, user=None, password=None):
  45         super(CrawlSuite, self).__init__()
  46
  47         registry = openerp.registry(tools.config['db_name'])
  48         try:
  49             # switch registry to test mode, so that requests can be made
  50             registry.enter_test_mode()
  51
  52             self.opener = urllib2.OpenerDirector()
  53             self.opener.add_handler(urllib2.UnknownHandler())
  54             self.opener.add_handler(urllib2.HTTPHandler())
  55             self.opener.add_handler(urllib2.HTTPSHandler())
  56             self.opener.add_handler(urllib2.HTTPCookieProcessor())
  57             self.opener.add_handler(RedirectHandler())
  58
  59             self._authenticate(user, password)
  60             self.user = user
  61
  62         finally:
  63             registry.leave_test_mode()
  64
  65     def _request(self, path):
  66         return self.opener.open(urlparse.urlunsplit([
  67             'http', 'localhost:%s' % tools.config['xmlrpc_port'],
  68             path, '', ''
  69         ]))
  70
  71     def _authenticate(self, user, password):
  72         # force tools.config['db_name'] in user session so opening `/` doesn't
  73         # blow up in multidb situations
  74         self.opener.open('http://localhost:{port}/web/?db={db}'.format(
  75             port=tools.config['xmlrpc_port'],
  76             db=werkzeug.urls.url_quote_plus(tools.config['db_name']),
  77         ))
  78         if user is not None:
  79             url = 'http://localhost:{port}/login?{query}'.format(
  80                 port=tools.config['xmlrpc_port'],
  81                 query=werkzeug.urls.url_encode({
  82                     'db': tools.config['db_name'],
  83                     'login': user,
  84                     'key': password,
  85                 })
  86             )
  87             auth = self.opener.open(url)
  88             assert auth.getcode() < 400, "Auth failure %d" % auth.getcode()
  89
  90     def _wrapped_run(self, result, debug=False):
  91         registry = openerp.registry(tools.config['db_name'])
  92         try:
  93             # switch registry to test mode, so that requests can be made
  94             registry.enter_test_mode()
  95
  96             paths = [URL('/'), URL('/sitemap')]
  97             seen = set(paths)
  98
  99             while paths:
 100                 url = paths.pop(0)
 101                 r = self._request(url.url)
 102                 url.to_case(self.user, r).run(result)
 103
 104                 if r.info().gettype() != 'text/html':
 105                     continue
 106
 107                 doc = lxml.html.fromstring(r.read())
 108                 for link in doc.xpath('//a[@href]'):
 109                     href = link.get('href')
 110
 111                     # avoid repeats, even for links we won't crawl no need to
 112                     # bother splitting them if we've already ignored them
 113                     # previously
 114                     if href in seen: continue
 115                     seen.add(href)
 116
 117                     parts = urlparse.urlsplit(href)
 118
 119                     if parts.netloc or \
 120                         not parts.path.startswith('/') or \
 121                         parts.path == '/web' or\
 122                         parts.path.startswith('/web/') or \
 123                         (parts.scheme and parts.scheme not in ('http', 'https')):
 124                         continue
 125
 126                     paths.append(URL(href, url.url))
 127
 128         finally:
 129             registry.leave_test_mode()
 130
 131 class URL(object):
 132     def __init__(self, url, source=None):
 133         self.url = url
 134         self.source = source
 135
 136     def to_case(self, user, result):
 137         return cases.URLCase(user, self.url, self.source, result)
 138
 139 def load_tests(loader, base, _):
 140     base.addTest(CrawlSuite())
 141     base.addTest(CrawlSuite('admin', 'admin'))
 142     base.addTest(CrawlSuite('demo', 'demo'))
 143     return base