tests: Adapt tests to recent changes.

rhodecode/tests/lib/middleware/utils/test_scm_app_http.py

@@ -1,130 +1,130 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2016-2016  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 mock
 import Pyro4
 import pytest
 import webtest
 
 from rhodecode.lib.middleware.utils import scm_app_http, scm_app
 from rhodecode.lib.vcs.conf import settings
 
 
 def vcs_http_app(vcsserver_http_echo_app):
     """
     VcsHttpProxy wrapped in WebTest.
     """
     git_url = vcsserver_http_echo_app.http_url + 'stream/git/'
     vcs_http_proxy = scm_app_http.VcsHttpProxy(
-        git_url, 'stub_path', 'stub_name', None)
+        git_url, 'stub_path', 'stub_name', None, 'stub_backend')
     app = webtest.TestApp(vcs_http_proxy)
     return app
 
 
 @pytest.fixture(scope='module')
 def vcsserver_http_echo_app(request, vcsserver_factory):
     """
     A running VCSServer with the EchoApp activated via HTTP.
     """
     vcsserver = vcsserver_factory(
         request=request,
         use_http=True,
         overrides=[{'app:main': {'dev.use_echo_app': 'true'}}])
     return vcsserver
 
 
 @pytest.fixture(scope='session')
 def data():
     one_kb = "x" * 1024
     return one_kb * 1024 * 10
 
 
 def test_reuse_app_no_data(repeat, vcsserver_http_echo_app):
     app = vcs_http_app(vcsserver_http_echo_app)
     for x in xrange(repeat / 10):
         response = app.post('/')
         assert response.status_code == 200
 
 
 def test_reuse_app_with_data(data, repeat, vcsserver_http_echo_app):
     app = vcs_http_app(vcsserver_http_echo_app)
     for x in xrange(repeat / 10):
         response = app.post('/', params=data)
         assert response.status_code == 200
 
 
 def test_create_app_per_request_no_data(repeat, vcsserver_http_echo_app):
     for x in xrange(repeat / 10):
         app = vcs_http_app(vcsserver_http_echo_app)
         response = app.post('/')
         assert response.status_code == 200
 
 
 def test_create_app_per_request_with_data(
         data, repeat, vcsserver_http_echo_app):
     for x in xrange(repeat / 10):
         app = vcs_http_app(vcsserver_http_echo_app)
         response = app.post('/', params=data)
         assert response.status_code == 200
 
 
 @pytest.fixture(scope='module')
 def vcsserver_pyro_echo_app(request, vcsserver_factory):
     """
     A running VCSServer with the EchoApp activated via Pyro4.
     """
     vcsserver = vcsserver_factory(
         request=request,
         use_http=False,
         overrides=[{'DEFAULT': {'dev.use_echo_app': 'true'}}])
     return vcsserver
 
 
 def vcs_pyro4_app(vcsserver_pyro_echo_app):
     """
     Pyro4 based Vcs proxy wrapped in WebTest
     """
     stub_config = {
         'git_update_server_info': 'stub',
     }
     server_and_port = vcsserver_pyro_echo_app.server_and_port
     GIT_REMOTE_WSGI = Pyro4.Proxy(
         settings.pyro_remote(
             settings.PYRO_GIT_REMOTE_WSGI, server_and_port))
     with mock.patch('rhodecode.lib.middleware.utils.scm_app.GIT_REMOTE_WSGI',
                     GIT_REMOTE_WSGI):
         pyro4_app = scm_app.create_git_wsgi_app(
             'stub_path', 'stub_name', stub_config)
     app = webtest.TestApp(pyro4_app)
     return app
 
 
 def test_pyro4_no_data(repeat, pylonsapp, vcsserver_pyro_echo_app):
     for x in xrange(repeat / 10):
         app = vcs_pyro4_app(vcsserver_pyro_echo_app)
         response = app.post('/')
         assert response.status_code == 200
 
 
 def test_pyro4_with_data(repeat, pylonsapp, vcsserver_pyro_echo_app, data):
     for x in xrange(repeat / 10):
         app = vcs_pyro4_app(vcsserver_pyro_echo_app)
         response = app.post('/', params=data)
         assert response.status_code == 200

rhodecode/tests/lib/middleware/utils/test_scm_app_http_chunking.py

@@ -1,136 +1,136 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2016-2016  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/
 
 """
 Checking the chunked data transfer via HTTP
 """
 
 import os
 import time
 import subprocess
 
 import pytest
 import requests
 
 from rhodecode.lib.middleware.utils import scm_app_http
 from rhodecode.tests.utils import wait_for_url
 
 
 def test_does_chunked_end_to_end_transfer(scm_app):
     response = requests.post(scm_app, data='', stream=True)
     assert response.headers['Transfer-Encoding'] == 'chunked'
     times = [time.time() for chunk in response.raw.read_chunked()]
     assert times[1] - times[0] > 0.1, "Chunks arrived at the same time"
 
 
 @pytest.fixture
 def echo_app_chunking(request, available_port_factory):
     """
     Run the EchoApp via Waitress in a subprocess.
 
     Return the URL endpoint to reach the app.
     """
     port = available_port_factory()
     command = (
         'waitress-serve --send-bytes 1 --port {port} --call '
         'rhodecode.tests.lib.middleware.utils.test_scm_app_http_chunking'
         ':create_echo_app')
     command = command.format(port=port)
     proc = subprocess.Popen(command.split(' '), bufsize=0)
     echo_app_url = 'http://localhost:' + str(port)
 
     @request.addfinalizer
     def stop_echo_app():
         proc.kill()
 
     return echo_app_url
 
 
 @pytest.fixture
 def scm_app(request, available_port_factory, echo_app_chunking):
     """
     Run the scm_app in Waitress.
 
     Returns the URL endpoint where this app can be reached.
     """
     port = available_port_factory()
     command = (
         'waitress-serve --send-bytes 1 --port {port} --call '
         'rhodecode.tests.lib.middleware.utils.test_scm_app_http_chunking'
         ':create_scm_app')
     command = command.format(port=port)
     env = os.environ.copy()
     env["RC_ECHO_URL"] = echo_app_chunking
     proc = subprocess.Popen(command.split(' '), bufsize=0, env=env)
     scm_app_url = 'http://localhost:' + str(port)
     wait_for_url(scm_app_url)
 
     @request.addfinalizer
     def stop_echo_app():
         proc.kill()
 
     return scm_app_url
 
 
 class EchoApp(object):
     """
     Stub WSGI application which returns a chunked response to every request.
     """
 
     def __init__(self, repo_path, repo_name, config):
         self._repo_path = repo_path
 
     def __call__(self, environ, start_response):
         environ['wsgi.input'].read()
         status = '200 OK'
         headers = []
         start_response(status, headers)
         return result_generator()
 
 
 def result_generator():
     """
     Simulate chunked results.
 
     The intended usage is to simulate a chunked response as we would get it
     out of a vcs operation during a call to "hg clone".
     """
     yield 'waiting 2 seconds'
     # Wait long enough so that the first chunk can go out
     time.sleep(2)
     yield 'final chunk'
     # Another small wait, otherwise they go together
     time.sleep(0.1)
 
 
 def create_echo_app():
     """
     Create EchoApp filled with stub data.
     """
     return EchoApp('stub_path', 'repo_name', {})
 
 
 def create_scm_app():
     """
     Create a scm_app hooked up to speak to EchoApp.
     """
     echo_app_url = os.environ["RC_ECHO_URL"]
     return scm_app_http.VcsHttpProxy(
-        echo_app_url, 'stub_path', 'stub_name', None)
+        echo_app_url, 'stub_path', 'stub_name', None, 'stub_backend')