# ================================================================= # # Authors: Ricardo Garcia Silva # # Copyright (c) 2016 Ricardo Garcia Silva # # Permission is hereby granted, free of charge, to any person # obtaining a copy of this software and associated documentation # files (the "Software"), to deal in the Software without # restriction, including without limitation the rights to use, # copy, modify, merge, publish, distribute, sublicense, and/or sell # copies of the Software, and to permit persons to whom the # Software is furnished to do so, subject to the following # conditions: # # The above copyright notice and this permission notice shall be # included in all copies or substantial portions of the Software. # # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, # EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES # OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND # NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT # HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, # WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING # FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR # OTHER DEALINGS IN THE SOFTWARE. # # ================================================================= """pytest configuration file for functional tests""" import codecs from collections import namedtuple import logging import os import re import configparser import apipkg import pytest from pycsw.core import admin from pycsw.core.config import StaticContext apipkg.initpkg("optionaldependencies", { "psycopg2": "psycopg2", }) from optionaldependencies import psycopg2 # NOQA: E402 TESTS_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) SuiteDirs = namedtuple("SuiteDirs", [ "get_tests_dir", "post_tests_dir", "data_tests_dir", "expected_results_dir", "export_tests_dir", ]) def pytest_generate_tests(metafunc): """Parametrize tests programmatically. This function scans the filesystem directories under ``tests/functionaltests/suites`` and automatically generates pytest tests based on the available test suites. Each suite directory has the following structure: * A mandatory ``default.cfg`` file specifying the configuration for the pycsw instance to use in the tests of the suite. * An optional ``get/`` subdirectory containing a ``requests.txt`` file with any HTTP GET requests for which to generate tests for. Each request is specified in a new line, with the following pattern: * , * An optional ``post/`` subdirectory containing files that are used as the payload for HTTP POST requests. The name of each file is used as the name of the test (without the file's extension); * An optional ``data/`` subdirectory. This directory, if present, indicates that the suite uses a custom database. The database is populated with any additional files that are contained inside this directory. If the ``data`` directory does not exist then the suite's tests will use the CITE database; * An ``expected/`` subdirectory containing a file for each of the expected test outcomes. The tests are autogenerated by parametrizing the ``tests/functionaltests/test_suites_functional::test_suites`` function Notes ----- Check pytest's documentation for information on autogenerating parametrized tests for further details on how the ``pytest_generate_tests`` function can be used: http://pytest.org/latest/parametrize.html#basic-pytest-generate-tests-example """ global TESTS_ROOT if metafunc.function.__name__ == "test_suites": suites_root_dir = os.path.join(TESTS_ROOT, "functionaltests", "suites") suite_names = os.listdir(suites_root_dir) arg_values = [] test_ids = [] logging.basicConfig(level=getattr( logging, metafunc.config.getoption("--pycsw-loglevel").upper())) if metafunc.config.getoption("--database-backend") == "postgresql": _recreate_postgresql_database(metafunc.config) for suite in suite_names: suite_dir = os.path.join(suites_root_dir, suite) config_path = os.path.join(suite_dir, "default.cfg") if not os.path.isfile(config_path): print("Directory {0!r} does not have a suite " "configuration file".format(suite_dir)) continue print("Generating tests for suite {0!r}...".format(suite)) normalize_ids = True if suite in ("harvesting", "manager") else False suite_dirs = _get_suite_dirs(suite) if suite_dirs.post_tests_dir is not None: post_argvalues, post_ids = _get_post_parameters( post_tests_dir=suite_dirs.post_tests_dir, expected_tests_dir=suite_dirs.expected_results_dir, config_path=config_path, suite_name=suite, normalize_ids=normalize_ids, ) arg_values.extend(post_argvalues) test_ids.extend(post_ids) if suite_dirs.get_tests_dir is not None: get_argvalues, get_ids = _get_get_parameters( get_tests_dir=suite_dirs.get_tests_dir, expected_tests_dir=suite_dirs.expected_results_dir, config_path=config_path, suite_name=suite, normalize_ids=normalize_ids, ) arg_values.extend(get_argvalues) test_ids.extend(get_ids) metafunc.parametrize( argnames=["configuration", "request_method", "request_data", "expected_result", "normalize_identifier_fields",], argvalues=arg_values, indirect=["configuration"], ids=test_ids, ) @pytest.fixture() def test_identifier(request): """Extract a meaningful identifier from the request's node.""" return re.search(r"[\w_]+\[(.*)\]", request.node.name).group(1) @pytest.fixture() def use_xml_canonicalisation(request): return not request.config.getoption("--functional-prefer-diffs") @pytest.fixture() def save_results_directory(request): return request.config.getoption("--functional-save-results-directory") @pytest.fixture() def configuration(request, tests_directory, log_level): """Configure a suite for execution in tests. This function is executed once for each individual test request, after tests have been collected. The configuration file for each test suite is read into memory. Some configuration parameters, like the repository's url and table name are adjusted. The suite's repository is also created, if needed. Parameters ---------- request: pytest.fixtures.FixtureRequest tests_directory: py.path.local Directory created by pytest where any test artifacts are to be saved log_level: str Log level for the pycsw server instance that will be created during tests. """ config_path = request.param config = configparser.ConfigParser() with codecs.open(config_path, encoding="utf-8") as fh: config.read_file(fh) suite_name = config_path.split(os.path.sep)[-2] suite_dirs = _get_suite_dirs(suite_name) data_dir = suite_dirs.data_tests_dir export_dir = suite_dirs.export_tests_dir if data_dir is not None: # suite has its own database repository_url = _get_repository_url(request.config, suite_name, tests_directory) elif suite_name == 'opensearcheo': repository_url = _get_repository_url(request.config, "apiso", tests_directory) else: # suite uses the CITE database data_dir, export_dir = _get_cite_suite_dirs() repository_url = _get_repository_url(request.config, "cite", tests_directory) table_name = _get_table_name(suite_name, config, repository_url) if not _repository_exists(repository_url, table_name): _initialize_database(repository_url=repository_url, table_name=table_name, data_dir=data_dir, test_dir=tests_directory, export_dir=export_dir) config.set("server", "loglevel", log_level) config.set("server", "logfile", "") config.set("repository", "database", repository_url) config.set("repository", "table", table_name) return config @pytest.fixture(scope="session", name="tests_directory") def fixture_tests_directory(tmpdir_factory): """Create a temporary directory for each test session. This directory is typically situated under ``/tmp`` and is used to create eventual sqlite databases for each suite. This functionality is mostly provided by pytest's built-in ``tmpdir_factory`` fixture. More information on this is available at: http://doc.pytest.org/en/2.9.0/tmpdir.html#the-tmpdir-factory-fixture """ tests_dir = tmpdir_factory.mktemp("functional_tests") return tests_dir def _get_cite_suite_dirs(): """Return the path to the data directory of the CITE test suite.""" global TESTS_ROOT suites_root_dir = os.path.join(TESTS_ROOT, "functionaltests", "suites") suite_dir = os.path.join(suites_root_dir, "cite") data_tests_dir = os.path.join(suite_dir, "data") export_tests_dir = os.path.join(suite_dir, "export") data_dir = data_tests_dir if os.path.isdir(data_tests_dir) else None export_dir = export_tests_dir if os.path.isdir(export_tests_dir) else None return data_dir, export_dir def _get_get_parameters(get_tests_dir, expected_tests_dir, config_path, suite_name, normalize_ids): """Return the parameters suitable for parametrizing HTTP GET tests.""" method = "GET" test_argvalues = [] test_ids = [] requests_file_path = os.path.join(get_tests_dir, "requests.txt") with open(requests_file_path) as fh: for line in fh: test_name, test_params = [i.strip() for i in line.partition(",")[::2]] expected_result_path = os.path.join( expected_tests_dir, "{method}_{name}.xml".format(method=method.lower(), name=test_name) ) test_argvalues.append( (config_path, method, test_params, expected_result_path, normalize_ids) ) test_ids.append( "{suite}_{http_method}_{name}".format( suite=suite_name, http_method=method.lower(), name=test_name) ) return test_argvalues, test_ids def _get_post_parameters(post_tests_dir, expected_tests_dir, config_path, suite_name, normalize_ids): """Return the parameters suitable for parametrizing HTTP POST tests.""" method = "POST" test_argvalues = [] test_ids = [] # we are sorting the directory contents because the # `harvesting` suite requires tests to be executed in alphabetical order directory_contents = sorted(os.listdir(post_tests_dir)) for request_file_name in directory_contents: request_path = os.path.join(post_tests_dir, request_file_name) expected_result_path = os.path.join( expected_tests_dir, "{method}_{filename}".format( method=method.lower(), filename=request_file_name ) ) test_argvalues.append( (config_path, method, request_path, expected_result_path, normalize_ids) ) test_ids.append( "{suite}_{http_method}_{file_name}".format( suite=suite_name, http_method=method.lower(), file_name=os.path.splitext( request_file_name)[0]) ) return test_argvalues, test_ids def _get_repository_url(conf, suite_name, test_dir): """Return the repository_url for the input parameters. Returns ------- repository_url: str SQLAlchemy URL for the repository in use. """ db_type = conf.getoption("--database-backend") if db_type == "sqlite": repository_url = "sqlite:///{test_dir}/{suite}.db".format( test_dir=test_dir, suite=suite_name) elif db_type == "postgresql": repository_url = ( "postgresql://{user}:{password}@{host}:{port}/{database}".format( user=conf.getoption("--database-user-postgresql"), password=conf.getoption("--database-password-postgresql"), host=conf.getoption("--database-host-postgresql"), port=conf.getoption("--database-port-postgresql"), database=conf.getoption("--database-name-postgresql")) ) else: raise NotImplementedError return repository_url def _get_suite_dirs(suite_name): """Get the paths to relevant suite directories. Parameters ---------- suite_name: str Name of the site Returns ------- SuiteDirs A four element named tuple with the input suite's relevant test directories. """ global TESTS_ROOT suites_root_dir = os.path.join(TESTS_ROOT, "functionaltests", "suites") suite_dir = os.path.join(suites_root_dir, suite_name) data_tests_dir = os.path.join(suite_dir, "data") post_tests_dir = os.path.join(suite_dir, "post") get_tests_dir = os.path.join(suite_dir, "get") export_tests_dir = os.path.join(suite_dir, "export") expected_results_dir = os.path.join(suite_dir, "expected") data_dir = data_tests_dir if os.path.isdir(data_tests_dir) else None posts_dir = post_tests_dir if os.path.isdir(post_tests_dir) else None gets_dir = get_tests_dir if os.path.isdir(get_tests_dir) else None expected_dir = (expected_results_dir if os.path.isdir( expected_results_dir) else None) export_dir = export_tests_dir if os.path.isdir(export_tests_dir) else None return SuiteDirs(get_tests_dir=gets_dir, post_tests_dir=posts_dir, data_tests_dir=data_dir, expected_results_dir=expected_dir, export_tests_dir=export_tests_dir) def _get_table_name(suite, config, repository_url): """Get the name of the table used to store records in the database. Parameters ---------- suite: str Name of the suite. config: ConfigParser Configuration for the suite. repository_url: str SQLAlchemy URL for the repository in use. Returns ------- str Name of the table to use in the database """ if repository_url.startswith("sqlite"): result = config.get("repository", "table") elif repository_url.startswith("postgresql"): result = "{suite}_records".format(suite=suite) else: raise NotImplementedError return result def _initialize_database(repository_url, table_name, data_dir, test_dir, export_dir): """Initialize database for tests. This function will create the database and load any test data that the suite may require. Parameters ---------- repository_url: str URL for the repository, as used by SQLAlchemy engines table_name: str Name of the table that is to be used to store pycsw records data_dir: str Path to a directory that contains sample data records to be loaded into the database test_dir: str Directory where the database is to be created, in case of sqlite. export_dir: str Diretory where the exported records are to be saved, if any """ print("Setting up {0!r} repository...".format(repository_url)) if repository_url.startswith("postgresql"): extra_kwargs = { "create_sfsql_tables": True, "create_plpythonu_functions": False } else: extra_kwargs = {} admin.setup_db(database=repository_url, table=table_name, home=test_dir, **extra_kwargs) if len(os.listdir(data_dir)) > 0: print("Loading database data...") loaded = admin.load_records( context=StaticContext(), database=repository_url, table=table_name, xml_dirpath=data_dir, recursive=True ) admin.optimize_db(context=StaticContext(), database=repository_url, table=table_name) if export_dir is not None: # Attempt to export files exported = admin.export_records( context=StaticContext(), database=repository_url, table=table_name, xml_dirpath=export_dir ) if len(loaded) != len(exported): raise ValueError( "Loaded records (%s) is different from exported records (%s)" % (len(loaded), len(exported)) ) # Remove the files that were exported since this was just a test for toremove in exported: os.remove(toremove) def _parse_postgresql_repository_url(repository_url): """Parse a SQLAlchemy engine URL describing a postgresql database. Parameters ---------- repository_url: str SQLAlchemy URL for the repository in use. Returns ------- dict A mapping with the database's connection parameters. """ info_re = re.search(r"postgresql://(?P[\w_]+):(?P.*?)@" r"(?P[\w_.]+):(?P\d+)/" r"(?P[\w_]+)", repository_url, flags=re.UNICODE) try: db_info = info_re.groupdict() except AttributeError: raise RuntimeError("Could not parse repository url {0!r}".format( repository_url)) else: return db_info def _recreate_postgresql_database(configuration): """Recreate a postgresql database. This function will try to create a new postgresql database for testing purposes. If the database already exists it is deleted and then recreated. Parameters ---------- configuration: _pytest.config.Config The configuration object used by pytest Raises ------ RuntimeError If a connection to the postgresql server cannot be made """ connection = psycopg2.connect( database="postgres", user=configuration.getoption("--database-user-postgresql"), password=configuration.getoption("--database-password-postgresql"), host=configuration.getoption("--database-host-postgresql"), port=configuration.getoption("--database-port-postgresql") ) connection.set_isolation_level( psycopg2.extensions.ISOLATION_LEVEL_AUTOCOMMIT) cursor = connection.cursor() db_name = configuration.getoption("--database-name-postgresql") cursor.execute("DROP DATABASE IF EXISTS {database}".format( database=db_name)) cursor.execute("CREATE DATABASE {database}".format(database=db_name)) cursor.execute( "SELECT COUNT(1) FROM pg_available_extensions WHERE name='postgis'") postgis_available = bool(cursor.fetchone()[0]) cursor.close() connection.close() if postgis_available: _create_postgresql_extension(configuration, extension="postgis") else: _create_postgresql_extension(configuration, extension="plpythonu") def _create_postgresql_extension(configuration, extension): """Create a postgresql extension in a previously created database. Parameters ---------- configuration: _pytest.config.Config The configuration object used by pytest extension: str Name of the extension to be created """ connection = psycopg2.connect( database=configuration.getoption("--database-name-postgresql"), user=configuration.getoption("--database-user-postgresql"), password=configuration.getoption("--database-password-postgresql"), host=configuration.getoption("--database-host-postgresql"), port=configuration.getoption("--database-port-postgresql") ) connection.set_isolation_level( psycopg2.extensions.ISOLATION_LEVEL_AUTOCOMMIT) cursor = connection.cursor() cursor.execute("CREATE EXTENSION {0}".format(extension)) cursor.close() connection.close() def _repository_exists(repository_url, table_name): """Test if the database already exists. Parameters ---------- repository_url: str URL for the repository, as used by SQLAlchemy engines table_name: str Name of the table that is to be used to store pycsw records Returns ------- bool Whether the repository exists or not. """ if repository_url.startswith("sqlite"): repository_path = repository_url.replace("sqlite:///", "") result = os.path.isfile(repository_path) elif repository_url.startswith("postgresql"): db_info = _parse_postgresql_repository_url(repository_url) try: connection = psycopg2.connect(user=db_info["user"], password=db_info["password"], host=db_info["host"], port=db_info["port"], database=db_info["database"]) cursor = connection.cursor() cursor.execute("SELECT COUNT(1) FROM {table_name}".format( table_name=table_name)) except (psycopg2.OperationalError, psycopg2.ProgrammingError): # database or table does not exist yet result = False else: result = True else: raise NotImplementedError return result