text: # =================================================================
  #
  # Authors: Ricardo Garcia Silva <ricardo.garcia.silva@gmail.com>
  #
  # 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
  from six.moves 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",
  ])
  
  
  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:
  
        * <test_name>,<request_query_string>
  
      * 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.SafeConfigParser()
      with codecs.open(config_path, encoding="utf-8") as fh:
          config.readfp(fh)
      suite_name = config_path.split(os.path.sep)[-2]
      suite_dirs = _get_suite_dirs(suite_name)
      data_dir = suite_dirs.data_tests_dir
      if data_dir is not None:  # suite has its own database
          repository_url = _get_repository_url(request.config, suite_name,
                                               tests_directory)
      else:  # suite uses the CITE database
          data_dir = _get_cite_suite_data_dir()
          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)
      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_data_dir():
      """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")
      data_dir = data_tests_dir if os.path.isdir(data_tests_dir) else None
      return data_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")
      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)
      return SuiteDirs(get_tests_dir=gets_dir,
                       post_tests_dir=posts_dir,
                       data_tests_dir=data_dir,
                       expected_results_dir=expected_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: SafeConfigParser
          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):
      """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.
  
      """
  
      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...")
          admin.load_records(context=StaticContext(), database=repository_url,
                             table=table_name, xml_dirpath=data_dir)
  
  
  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<user>[\w_]+):(?P<password>.*?)@"
                          r"(?P<host>[\w_.]+):(?P<port>\d+)/"
                          r"(?P<database>[\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
  