# -*- coding: utf-8 -*- import contextlib import logging import pytest import threading import time import os from tests.utils import unique_name from tests.utils.project import Project from tests.utils.group import Group from tests.utils.runner import Runner from tests.utils.user import User from tests.utils import git from tests.utils import hg from tests.utils.heptapod import ( Heptapod, OmnibusHeptapod, DockerHeptapod, SourceHeptapod, GdkHeptapod, ) logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) def pytest_addoption(parser): """Add command-line options for Heptapod host.""" parser.addoption('--heptapod-source-install', action="store_true", help="Test a source install. This means that Heptapod " "is running on the same system, these tests will " "perform direct subprocess calls and filesystem " "manipulations.") parser.addoption('--heptapod-gdk', action="store_true", help="Test a Heptapod running " "within the GitLab Development Kit (GDK) on the " "same system as the tests." "The tests will perform direct service management " "(start/stop), subprocess calls " "and filesystem manipulations." "This mode needs rbenv or equivalent to be active." ) parser.addoption('--heptapod-omnibus', action="store_true", help="Test a Heptapod Omnibus by running the tests " "as root on the same system ") parser.addoption('--heptapod-remote', action='store_true', help="Test a remote server. This means that no " "direct command nor file system access is possible " "leading to skipping some tests and some teardown " "reliability actions.") parser.addoption('--heptapod-hg-native', action='store_true', help="Have all Mercurial projects created as 'native' " "(HGitaly backed)") parser.addoption('--heptapod-repositories-root', help="Root of the group/repository hierarchy. " "This is mandatory for source installs and ignored " "in Docker mode.") parser.addoption('--heptapod-gdk-root', help="Root of the installation for GDK Heptapod.") parser.addoption('--heptapod-url', default='http://heptapod:81') parser.addoption('--heptapod-ssh-port', type=int, default=2022) parser.addoption('--heptapod-ssh-user', default='git') parser.addoption('--heptapod-docker-container', default='heptapod') parser.addoption('--heptapod-root-password', default='5iveL!fe') parser.addoption('--heptapod-reverse-call-host', help="Network address that the Heptapod server " "can use to initiate network connections to " "the system running these tests.") parser.addoption('--heptapod-webdriver-remote-url', help="URL to a Selenium RemoteWebDriver server. " "For instance, for the standalone-chrome " "docker image, it would be http://container:4444/wd/hub " "Care must be taken that `heptapod-url` is " "appropriate from *both* the RemoteWebDriver server " "*and* the host running these tests") parser.addoption('--heptapod-wait-after-first-response', type=int, default=0, help="Time to wait after we got a first non error " "response from the server. This is especially useful " "after waiting while the instance was being " "created from scratch and in constrained environments " "(CI…). Without this, it can even happen that the " "first test passes and the second fails") parser.addoption('--heptapod-hg-client-executable', default='hg', help="Path to the hg executable on the client side, " "i.e., the one used for local repos that get pulled " "and pushed from the Heptapod server.") parser.addoption('--heptapod-git-client-executable', default='git', help="Path to the git executable on the client side, " "i.e., the one used for local repos that get pulled " "and pushed from the Heptapod server.") def pytest_configure(config): config.addinivalue_line("markers", "fs_access: mark test as needing file system " "access to Heptapod server (Docker or local).") config.addinivalue_line("markers", "reverse_call: mark test as involving network " "calls from the Heptapod server to the host " "running these tests") config.addinivalue_line("markers", "docker: mark test as needing to run against " "Docker Heptapod servers") config.addinivalue_line("markers", "services: mark test as needing services " "management " ) # if we regroup the tests with separate fixtures for the two # modes, these two will become useless: config.addinivalue_line("markers", "hg_git: mark test as running only if " "session is *not* in hg-native mode" ) config.addinivalue_line("markers", "hg_native: mark test as running only if " "session is in hg-native mode" ) def pytest_collection_modifyitems(config, items): skip_fs_needed = pytest.mark.skip(reason="needs filesystem access") skip_docker = pytest.mark.skip( reason="needs Heptapod as a Docker container") skip_reverse_call = pytest.mark.skip( reason="needs to be able to initiate network connections to the host " "running these tests") skip_services = pytest.mark.skip( reason="needs to manage Heptapod services") skip_hg_git = pytest.mark.skip(reason="needs non hg-native mode") skip_hg_native = pytest.mark.skip(reason="needs hg-native mode") no_reverse_call = not(config.getoption('heptapod_reverse_call_host')) remote = config.getoption('heptapod_remote') source_install = config.getoption('heptapod_source_install') gdk = config.getoption('heptapod_gdk') omnibus = config.getoption('heptapod_omnibus') docker = not (gdk or source_install or remote or omnibus) hg_native = config.getoption("heptapod_hg_native") for item in items: if remote and "fs_access" in item.keywords: item.add_marker(skip_fs_needed) if "docker" in item.keywords and not docker: item.add_marker(skip_docker) if "reverse_call" in item.keywords and no_reverse_call: item.add_marker(skip_reverse_call) if "services" in item.keywords and (remote or source_install): item.add_marker(skip_services) if "hg_git" in item.keywords and hg_native: item.add_marker(skip_hg_git) if "hg_native" in item.keywords and not hg_native: item.add_marker(skip_hg_native) heptapod_instance = None lock = threading.Lock() active_threads = 0 @pytest.fixture(scope="session") def heptapod(pytestconfig): global lock global active_threads global heptapod_instance lock.acquire() active_threads += 1 # Setting Git and Mercurial client-side executables. # This looks hacky as any monkey-patching does, but the alternative would # be to forward the pytestconfig fixture to pretty much all tests # or to store on the Heptapod object (all tests use it), which creates # confusion between the client and server sides. hg.HG_EXECUTABLE = pytestconfig.getoption('heptapod_hg_client_executable') git.GIT_EXECUTABLE = pytestconfig.getoption( 'heptapod_git_client_executable') url = pytestconfig.getoption('heptapod_url') common = dict( url=url, ssh_user=pytestconfig.getoption('heptapod_ssh_user'), ssh_port=pytestconfig.getoption('heptapod_ssh_port'), reverse_call_host=pytestconfig.getoption( 'heptapod_reverse_call_host'), webdriver_remote_url=pytestconfig.getoption( 'heptapod_webdriver_remote_url'), wait_after_first_response=pytestconfig.getoption( 'heptapod_wait_after_first_response'), hg_native=pytestconfig.getoption('heptapod_hg_native') ) # make tests fully independent of current user settings os.environ['HGRCPATH'] = '' try: if active_threads == 1: # we're the first if pytestconfig.getoption('heptapod_source_install'): repos_root = pytestconfig.getoption( 'heptapod_repositories_root') heptapod = SourceHeptapod(repositories_root=repos_root, **common) elif pytestconfig.getoption('heptapod_gdk'): heptapod = GdkHeptapod( gdk_root=pytestconfig.getoption('heptapod_gdk_root'), **common) elif pytestconfig.getoption('heptapod_remote'): heptapod = Heptapod(**common) elif pytestconfig.getoption('heptapod_omnibus'): heptapod = OmnibusHeptapod(**common) else: ct = pytestconfig.getoption('heptapod_docker_container') heptapod = DockerHeptapod(docker_container=ct, **common) heptapod.prepare(pytestconfig.getoption('heptapod_root_password')) heptapod_instance = heptapod finally: lock.release() yield heptapod_instance lock.acquire() active_threads -= 1 try: if active_threads == 0: heptapod_instance.close() finally: lock.release() @contextlib.contextmanager def project_fixture(heptapod, name_prefix, owner, group=None, **opts): name = '%s_%s' % (name_prefix, str(time.time()).replace('.', '_')) with Project.api_create(heptapod, owner, name, group=group, **opts) as proj: yield proj @contextlib.contextmanager def group_fixture(heptapod, path_prefix, creator_name='root', parent=None): with Group.api_create(heptapod, unique_name(path_prefix), user_name=creator_name, parent=parent) as group: yield group @pytest.fixture() def external_user(heptapod, accepts_concurrent): with User.create(heptapod, unique_name('external_user'), external=True) as user: yield user @pytest.fixture() def additional_user(heptapod, accepts_concurrent): with User.create(heptapod, unique_name('add_user')) as user: user.ensure_private_token() yield user @pytest.fixture() def test_project(heptapod, accepts_concurrent): with project_fixture(heptapod, 'test_project', 'root') as proj: yield proj @pytest.fixture() def test_project_with_runner(test_project): with Runner.api_register(test_project, unique_name('fixt')) as runner: yield test_project, runner @pytest.fixture() def test_group(heptapod, accepts_concurrent): with group_fixture(heptapod, 'test_group', 'root') as group: yield group @pytest.fixture() def project_breaks_concurrent(heptapod, breaks_concurrent): """Used for tests that break concurrent and need a testing project""" with project_fixture(heptapod, 'test_project', 'root') as proj: yield proj @pytest.fixture() def public_project(heptapod, accepts_concurrent): with project_fixture(heptapod, 'public_project', 'test_basic') as proj: resp = proj.api_edit(visibility='public') assert resp.status_code == 200 yield proj @pytest.fixture() def git_project(heptapod, accepts_concurrent): with project_fixture(heptapod, 'git_project', 'test_basic', vcs_type='git') as proj: yield proj @pytest.fixture() def git_project_with_runner(git_project): with Runner.api_register(git_project, unique_name('fixt')) as runner: yield git_project, runner @pytest.fixture() def public_project_breaks_concurrent(heptapod, breaks_concurrent): with project_fixture(heptapod, 'public_project', 'test_basic') as proj: resp = proj.api_edit(visibility='public') assert resp.status_code == 200 yield proj no_concurrency_lock = threading.Lock() """Lock to implement cases where concurrency is to be avoided. Tests that break concurrency will keep a hold on this lock. Other tests will release it right after startup. """ running_count_change = threading.Condition() """Guard changes in the number of running tests accepting concurrency. Tests that break concurrency will wait for this condition. Other tests will notify it. """ running_tests = 0 """Number of tests accepting concurrency that are currently running.""" def thread_name(): return threading.current_thread().name @pytest.fixture() def breaks_concurrent(): """Fixture for tests that cannot run concurrently with any other one. For instance, some tests will restart the Rails application, hence breaking anything that's concurrent. Use this fixture to declare them as such, and have pytest do the right thing, hence avoid launching concurrent tests while they are running. """ # make sure that we are in the only test breaking concurrency # that can proceed with no_concurrency_lock: # now that we have this lock, all concurrent tests should be based on # the `accepts_concurrent` fixture (barring forgotten declarations) logger.info("[%s] breaks_concurrency(): waiting " "for concurrent tests to finish", thread_name()) running_count_change.acquire() while running_tests > 0: logger.debug("[%s] breaks_concurrency(): " "there are %d running concurrent tests", thread_name(), running_tests) running_count_change.wait() running_count_change.release() logger.info("[%s] breaks_concurrency(): proceeding to the test", thread_name()) yield logger.debug("[%s] breaks_concurrency() done", thread_name()) @pytest.fixture() def accepts_concurrent(): """Fixture for tests that can run concurrently among each other. Of course they can't run concurrently with tests that use the :func:`breaks_concurrent` fixture """ global running_tests logger.debug("[%s] accepts_concurrent(): waiting for " "tests breaking concurrency to finish", thread_name()) # wait for tests breaking concurrency to finish. with no_concurrency_lock: # while no other thread can hold the no_concurrency_lock, # some other test might be tearing down, using the lock # behind running_count_changes with running_count_change: running_tests += 1 logger.debug("[%s] accepts_concurrent(): incremented counter to " "%d and proceeding to the test", thread_name(), running_tests) yield with running_count_change: running_tests -= 1 logger.debug("[%s] accepts_concurrent(): decremented to %d " "and notifying tests that break concurrency", thread_name(), running_tests) running_count_change.notify() @pytest.fixture() def app_settings(heptapod): """Fixture for tests that change application settings. This restores these settings in the teardown. The value yielded is the `Heptapod` instance. Depending on the existence of other tests that rely on a given value of the changed settings, these could break concurrency or not. """ heptapod.sync_application_settings() settings = heptapod.settings yield heptapod new_settings = heptapod.settings # stored settings (from API GET or PUT responses) typically have # much more than what can be set, and include problematic values. # Let's rollback only what has changed rollback = {name: value for name, value in settings.items() if new_settings.get(name) != value} if rollback: heptapod.set_application_settings(**rollback)