testapp: moved login/csrf session methods into TestApp itself.

- this allows easier control over which app is responsible for session handling.
This comes in handy in CE vs EE test where EE get's a custom app.

rhodecode/tests/utils.py

@@ -1,411 +1,429 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2017 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import threading
 import time
 import logging
 import os.path
 import subprocess32
 import tempfile
 import urllib2
 from lxml.html import fromstring, tostring
 from lxml.cssselect import CSSSelector
 from urlparse import urlparse, parse_qsl
 from urllib import unquote_plus
 import webob
 
 from webtest.app import TestResponse, TestApp, string_types
 from webtest.compat import print_stderr
 
 import pytest
 import rc_testdata
 
 from rhodecode.model.db import User, Repository
 from rhodecode.model.meta import Session
 from rhodecode.model.scm import ScmModel
 from rhodecode.lib.vcs.backends.svn.repository import SubversionRepository
 from rhodecode.lib.vcs.backends.base import EmptyCommit
-
+from rhodecode.tests import login_user_session
 
 log = logging.getLogger(__name__)
 
 
 class CustomTestResponse(TestResponse):
     def _save_output(self, out):
         f = tempfile.NamedTemporaryFile(
             delete=False, prefix='rc-test-', suffix='.html')
         f.write(out)
         return f.name
 
     def mustcontain(self, *strings, **kw):
         """
         Assert that the response contains all of the strings passed
         in as arguments.
 
         Equivalent to::
 
             assert string in res
         """
         if 'no' in kw:
             no = kw['no']
             del kw['no']
             if isinstance(no, string_types):
                 no = [no]
         else:
             no = []
         if kw:
             raise TypeError(
-                "The only keyword argument allowed is 'no'")
+                "The only keyword argument allowed is 'no' got %s" % kw)
 
         f = self._save_output(str(self))
 
         for s in strings:
             if not s in self:
                 print_stderr("Actual response (no %r):" % s)
                 print_stderr(str(self))
                 raise IndexError(
                     "Body does not contain string %r, output saved as %s" % (
                     s, f))
 
         for no_s in no:
             if no_s in self:
                 print_stderr("Actual response (has %r)" % no_s)
                 print_stderr(str(self))
                 raise IndexError(
                     "Body contains bad string %r, output saved as %s" % (
                     no_s, f))
 
     def assert_response(self):
         return AssertResponse(self)
 
     def get_session_from_response(self):
         """
         This returns the session from a response object.
         """
 
         from pyramid_beaker import session_factory_from_settings
         session = session_factory_from_settings(
             self.test_app.app.config.get_settings())
         return session(self.request)
 
 
 class TestRequest(webob.BaseRequest):
 
     # for py.test
     disabled = True
     ResponseClass = CustomTestResponse
 
     def add_response_callback(self, callback):
         pass
 
 
 class CustomTestApp(TestApp):
     """
-    Custom app to make mustcontain more usefull
+    Custom app to make mustcontain more usefull, and extract special methods
     """
     RequestClass = TestRequest
+    rc_login_data = {}
+    rc_current_session = None
+
+    def login(self, username=None, password=None):
+        from rhodecode.lib import auth
+
+        if username and password:
+            session = login_user_session(self, username, password)
+        else:
+            session = login_user_session(self)
+
+        self.rc_login_data['csrf_token'] = auth.get_csrf_token(session)
+        self.rc_current_session = session
+        return session['rhodecode_user']
+
+    @property
+    def csrf_token(self):
+        return self.rc_login_data['csrf_token']
 
 
 def set_anonymous_access(enabled):
     """(Dis)allows anonymous access depending on parameter `enabled`"""
     user = User.get_default_user()
     user.active = enabled
     Session().add(user)
     Session().commit()
     time.sleep(1.5)  # must sleep for cache (1s to expire)
     log.info('anonymous access is now: %s', enabled)
     assert enabled == User.get_default_user().active, (
         'Cannot set anonymous access')
 
 
 def check_xfail_backends(node, backend_alias):
     # Using "xfail_backends" here intentionally, since this marks work
     # which is "to be done" soon.
     skip_marker = node.get_marker('xfail_backends')
     if skip_marker and backend_alias in skip_marker.args:
         msg = "Support for backend %s to be developed." % (backend_alias, )
         msg = skip_marker.kwargs.get('reason', msg)
         pytest.xfail(msg)
 
 
 def check_skip_backends(node, backend_alias):
     # Using "skip_backends" here intentionally, since this marks work which is
     # not supported.
     skip_marker = node.get_marker('skip_backends')
     if skip_marker and backend_alias in skip_marker.args:
         msg = "Feature not supported for backend %s." % (backend_alias, )
         msg = skip_marker.kwargs.get('reason', msg)
         pytest.skip(msg)
 
 
 def extract_git_repo_from_dump(dump_name, repo_name):
     """Create git repo `repo_name` from dump `dump_name`."""
     repos_path = ScmModel().repos_path
     target_path = os.path.join(repos_path, repo_name)
     rc_testdata.extract_git_dump(dump_name, target_path)
     return target_path
 
 
 def extract_hg_repo_from_dump(dump_name, repo_name):
     """Create hg repo `repo_name` from dump `dump_name`."""
     repos_path = ScmModel().repos_path
     target_path = os.path.join(repos_path, repo_name)
     rc_testdata.extract_hg_dump(dump_name, target_path)
     return target_path
 
 
 def extract_svn_repo_from_dump(dump_name, repo_name):
     """Create a svn repo `repo_name` from dump `dump_name`."""
     repos_path = ScmModel().repos_path
     target_path = os.path.join(repos_path, repo_name)
     SubversionRepository(target_path, create=True)
     _load_svn_dump_into_repo(dump_name, target_path)
     return target_path
 
 
 def assert_message_in_log(log_records, message, levelno, module):
     messages = [
         r.message for r in log_records
         if r.module == module and r.levelno == levelno
     ]
     assert message in messages
 
 
 def _load_svn_dump_into_repo(dump_name, repo_path):
     """
     Utility to populate a svn repository with a named dump
 
     Currently the dumps are in rc_testdata. They might later on be
     integrated with the main repository once they stabilize more.
     """
     dump = rc_testdata.load_svn_dump(dump_name)
     load_dump = subprocess32.Popen(
         ['svnadmin', 'load', repo_path],
         stdin=subprocess32.PIPE, stdout=subprocess32.PIPE,
         stderr=subprocess32.PIPE)
     out, err = load_dump.communicate(dump)
     if load_dump.returncode != 0:
         log.error("Output of load_dump command: %s", out)
         log.error("Error output of load_dump command: %s", err)
         raise Exception(
             'Failed to load dump "%s" into repository at path "%s".'
             % (dump_name, repo_path))
 
 
 class AssertResponse(object):
     """
     Utility that helps to assert things about a given HTML response.
     """
 
     def __init__(self, response):
         self.response = response
 
     def get_imports(self):
         return fromstring, tostring, CSSSelector
 
     def one_element_exists(self, css_selector):
         self.get_element(css_selector)
 
     def no_element_exists(self, css_selector):
         assert not self._get_elements(css_selector)
 
     def element_equals_to(self, css_selector, expected_content):
         element = self.get_element(css_selector)
         element_text = self._element_to_string(element)
         assert expected_content in element_text
 
     def element_contains(self, css_selector, expected_content):
         element = self.get_element(css_selector)
         assert expected_content in element.text_content()
 
     def element_value_contains(self, css_selector, expected_content):
         element = self.get_element(css_selector)
         assert expected_content in element.value
 
     def contains_one_link(self, link_text, href):
         fromstring, tostring, CSSSelector = self.get_imports()
         doc = fromstring(self.response.body)
         sel = CSSSelector('a[href]')
         elements = [
             e for e in sel(doc) if e.text_content().strip() == link_text]
         assert len(elements) == 1, "Did not find link or found multiple links"
         self._ensure_url_equal(elements[0].attrib.get('href'), href)
 
     def contains_one_anchor(self, anchor_id):
         fromstring, tostring, CSSSelector = self.get_imports()
         doc = fromstring(self.response.body)
         sel = CSSSelector('#' + anchor_id)
         elements = sel(doc)
         assert len(elements) == 1, 'cannot find 1 element {}'.format(anchor_id)
 
     def _ensure_url_equal(self, found, expected):
         assert _Url(found) == _Url(expected)
 
     def get_element(self, css_selector):
         elements = self._get_elements(css_selector)
         assert len(elements) == 1, 'cannot find 1 element {}'.format(css_selector)
         return elements[0]
 
     def get_elements(self, css_selector):
         return self._get_elements(css_selector)
 
     def _get_elements(self, css_selector):
         fromstring, tostring, CSSSelector = self.get_imports()
         doc = fromstring(self.response.body)
         sel = CSSSelector(css_selector)
         elements = sel(doc)
         return elements
 
     def _element_to_string(self, element):
         fromstring, tostring, CSSSelector = self.get_imports()
         return tostring(element)
 
 
 class _Url(object):
     """
     A url object that can be compared with other url orbjects
     without regard to the vagaries of encoding, escaping, and ordering
     of parameters in query strings.
 
     Inspired by
     http://stackoverflow.com/questions/5371992/comparing-two-urls-in-python
     """
 
     def __init__(self, url):
         parts = urlparse(url)
         _query = frozenset(parse_qsl(parts.query))
         _path = unquote_plus(parts.path)
         parts = parts._replace(query=_query, path=_path)
         self.parts = parts
 
     def __eq__(self, other):
         return self.parts == other.parts
 
     def __hash__(self):
         return hash(self.parts)
 
 
 def run_test_concurrently(times, raise_catched_exc=True):
     """
     Add this decorator to small pieces of code that you want to test
     concurrently
 
     ex:
 
     @test_concurrently(25)
     def my_test_function():
         ...
     """
     def test_concurrently_decorator(test_func):
         def wrapper(*args, **kwargs):
             exceptions = []
 
             def call_test_func():
                 try:
                     test_func(*args, **kwargs)
                 except Exception as e:
                     exceptions.append(e)
                     if raise_catched_exc:
                         raise
             threads = []
             for i in range(times):
                 threads.append(threading.Thread(target=call_test_func))
             for t in threads:
                 t.start()
             for t in threads:
                 t.join()
             if exceptions:
                 raise Exception(
                     'test_concurrently intercepted %s exceptions: %s' % (
                         len(exceptions), exceptions))
         return wrapper
     return test_concurrently_decorator
 
 
 def wait_for_url(url, timeout=10):
     """
     Wait until URL becomes reachable.
 
     It polls the URL until the timeout is reached or it became reachable.
     If will call to `py.test.fail` in case the URL is not reachable.
     """
     timeout = time.time() + timeout
     last = 0
     wait = 0.1
 
     while timeout > last:
         last = time.time()
         if is_url_reachable(url):
             break
         elif (last + wait) > time.time():
             # Go to sleep because not enough time has passed since last check.
             time.sleep(wait)
     else:
         pytest.fail("Timeout while waiting for URL {}".format(url))
 
 
 def is_url_reachable(url):
     try:
         urllib2.urlopen(url)
     except urllib2.URLError:
         return False
     return True
 
 
 def repo_on_filesystem(repo_name):
     from rhodecode.lib import vcs
     from rhodecode.tests import TESTS_TMP_PATH
     repo = vcs.get_vcs_instance(
         os.path.join(TESTS_TMP_PATH, repo_name), create=False)
     return repo is not None
 
 
 def commit_change(
         repo, filename, content, message, vcs_type, parent=None, newfile=False):
     from rhodecode.tests import TEST_USER_ADMIN_LOGIN
 
     repo = Repository.get_by_repo_name(repo)
     _commit = parent
     if not parent:
         _commit = EmptyCommit(alias=vcs_type)
 
     if newfile:
         nodes = {
             filename: {
                 'content': content
             }
         }
         commit = ScmModel().create_nodes(
             user=TEST_USER_ADMIN_LOGIN, repo=repo,
             message=message,
             nodes=nodes,
             parent_commit=_commit,
             author=TEST_USER_ADMIN_LOGIN,
         )
     else:
         commit = ScmModel().commit_change(
             repo=repo.scm_instance(), repo_name=repo.repo_name,
             commit=parent, user=TEST_USER_ADMIN_LOGIN,
             author=TEST_USER_ADMIN_LOGIN,
             message=message,
             content=content,
             f_path=filename
         )
     return commit