<commit_before># Copyright [1999-2015] Wellcome Trust Sanger Institute and the EMBL-European Bioinformatics Institute
# Copyright [2016-2020] EMBL-European Bioinformatics Institute
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Local hook and fixture implementations for the Continuous Integration Test (CITest) suite."""

import json
import os
import re
from typing import Dict

import pytest
from _pytest.config.argparsing import Parser
from _pytest.fixtures import FixtureRequest
from _pytest.main import Session
from _pytest.python import Metafunc
from _pytest.runner import CallInfo, TestReport

from ensembl.compara.citest import TestDB


def pytest_addoption(parser: Parser) -> None:
    """Register argparse-style options for the test run.

    Args:
        parser: Command line option parser.
    """
    parser.addoption("--pipeline", action="store", metavar="NAME", dest="pipeline_name",
                     help="name of the pipeline to be tested")
    parser.addoption("--json-file", action="store", metavar="JSON_FILE", dest="json_file",
                     help="CITest configuration JSON file")
    parser.addoption("--ref-url", action="store", metavar="URL", dest="ref_url",
                     help="URL to the reference database")
    parser.addoption("--target-url", action="store", required=True, metavar="URL", dest="target_url",
                     help="URL to the target database")


def pytest_generate_tests(metafunc: Metafunc) -> None:
    """Generate (multiple) parametrized calls to a test function.

    Args:
        metafunc: Container of the test configuration values.
    """
    # --pipeline or --json-file argument required
    pipeline_name = metafunc.config.getoption("pipeline_name")
    if pipeline_name:
        json_file = os.path.join(
            os.environ['ENSEMBL_CVS_ROOT_DIR'],
            *["ensembl-compara", "src", "python", "conf", "citest", "pipelines", pipeline_name + ".json"]
        )
    else:
        json_file = metafunc.config.getoption("json_file")
    ref_url = metafunc.config.getoption("ref_url")
    target_url = metafunc.config.getoption("target_url")
    test_data = get_test_data(json_file, ref_url, target_url)
    # Load the parameters for each test
    metafunc.parametrize("db_test_data", test_data["database"], indirect=True)
    # TODO: parametrize all files tests (once class is available)


@pytest.fixture()
def db_test_data(request: FixtureRequest) -> Dict:
    """Fixture to retrieve the database test data.

    Args:
        request: Special fixture providing information of the requesting test function.

    Returns:
        Dictionary with the database test data. Keys "db_test_handler", "test" and "table" will always be
        present.
    """
    return request.param


def pytest_sessionstart(session: Session) -> None:
    """Add required variables to the session before entering the run test loop.

    Args:
        session: Pytest Session object.
    """
    session.report = {}


def pytest_sessionfinish(session: Session, exitstatus: int) -> None:
    """Generate a custom report before returning the exit status to the system.

    Args:
        session: Pytest Session object.
        existatus: Status which pytest will return to the system.
    """
    report_list = [report for report in session.report.values()]
    # Save full report in a JSON file
    pipeline_name = session.config.getoption("pipeline_name")
    if pipeline_name:
        report_file = pipeline_name + ".report.json"
    else:
        json_file = os.path.basename(session.config.getoption("json_file"))
        report_file = json_file.rsplit(".", 1)[0] + ".report.json"
    # Make sure not to overwrite previous reports
    if os.path.isfile(report_file):
        i = 1
        while os.path.isfile("{}.{}".format(report_file, i)):
            i += 1
        report_file = "{}.{}".format(report_file, i)
    with open(report_file, "w") as f:
        json_report = {} # type: Dict
        for report in report_list:
            test_run = re.sub(r'\[.+\]', '', report.location[-1])
            test_report = json_report.setdefault(test_run, {})
            table_report = test_report.setdefault(report.test_params["table"], {})
            subtest = report.test_params["test"]
            info = {"args":   {k: v for k, v in report.test_params.items() if k not in ["table", "test"]},
                    "status": report.outcome.capitalize()}
            if report.failed:
                info["error"] = {"message": report.longrepr.reprcrash.message,
                                 "details": report.error_info}
            table_report.setdefault(subtest, []).append(info)
        json.dump(json_report, f, indent=4, sort_keys=True)
    # Print summary
    total = len(report_list)
    failed = sum(1 for report in report_list if report.failed)
    print("\n{} out of {} tests ok".format(total-failed, total))


@pytest.hookimpl(tryfirst=True, hookwrapper=True)
def pytest_runtest_makereport(item: pytest.Item, call: CallInfo) -> TestReport:
    """Extend the test report with custom information.

    Args:
        item: Pytest Item.
        call: Result/Exception info a function invocation.

    Returns:
        Updated test report.
    """
    outcome = yield
    report = outcome.get_result()
    if report.when == "call":
        # Update the test report adding our custom report_info
        report.error_info = getattr(item, "error_info", 0)
        if "db_test_data" in item.fixturenames:
            report.test_params = item.funcargs["db_test_data"]
            del report.test_params["db_test_handler"]
        # TODO: Once class is available, uncomment following lines
        # elif "files_test_data" in item.fixturenames:
        #     report.test_params = item.funcargs["files_test_data"]
        #     del report.test_params["files_test_handler"]
        item.session.report[item] = report


def get_test_data(json_file: str, ref_url: str, target_url: str) -> Dict:
    """Load the test data from the JSON file.

    The reference and target database URLs will be added to the test data if any database test is performed.

    Args:
        json_file: CITest configuration JSON file for a specific pipeline/database.
        ref_url: URL to the reference database.
        target_url: URL to the target database.

    Returns:
        Dictionary with the test data for both database and files tests.
    """
    # Load the JSON file with the tests to run and their parameters
    with open(json_file) as f:
        data = json.load(f)
    params = {"database": [], "files": []} # type: Dict
    # Load the configuration for all database tests
    if data["database"]:
        # If no reference database is passed, use the default one defined in the JSON file
        if not ref_url:
            ref_url = data["reference_db"]
        db_test_handler = TestDB.TestDB(ref_url=ref_url, target_url=target_url)
        for table in data["database"]:
            for test in data["database"][table]:
                # Add extra parameters required for the test
                test["db_test_handler"] = db_test_handler
                test["table"] = table
                params["database"].append(test)
    # TODO: load all files tests (once class is available)
    return params
<commit_msg>improved help text<commit_after># Copyright [1999-2015] Wellcome Trust Sanger Institute and the EMBL-European Bioinformatics Institute
# Copyright [2016-2020] EMBL-European Bioinformatics Institute
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Local hook and fixture implementations for the Continuous Integration Test (CITest) suite."""

import json
import os
import re
from typing import Dict

import pytest
from _pytest.config.argparsing import Parser
from _pytest.fixtures import FixtureRequest
from _pytest.main import Session
from _pytest.python import Metafunc
from _pytest.runner import CallInfo, TestReport

from ensembl.compara.citest import TestDB


def pytest_addoption(parser: Parser) -> None:
    """Register argparse-style options for the test run.

    Args:
        parser: Command line option parser.
    """
    parser.addoption("--pipeline", action="store", metavar="NAME", dest="pipeline_name",
                     help="name of the pipeline to be tested (without '_conf' suffix)")
    parser.addoption("--json-file", action="store", metavar="JSON_FILE", dest="json_file",
                     help="CITest configuration JSON file")
    parser.addoption("--ref-url", action="store", metavar="URL", dest="ref_url",
                     help="URL to the reference database")
    parser.addoption("--target-url", action="store", required=True, metavar="URL", dest="target_url",
                     help="URL to the target database")


def pytest_generate_tests(metafunc: Metafunc) -> None:
    """Generate (multiple) parametrized calls to a test function.

    Args:
        metafunc: Container of the test configuration values.
    """
    # --pipeline or --json-file argument required
    pipeline_name = metafunc.config.getoption("pipeline_name")
    if pipeline_name:
        json_file = os.path.join(
            os.environ['ENSEMBL_CVS_ROOT_DIR'],
            *["ensembl-compara", "src", "python", "conf", "citest", "pipelines", pipeline_name + ".json"]
        )
    else:
        json_file = metafunc.config.getoption("json_file")
    ref_url = metafunc.config.getoption("ref_url")
    target_url = metafunc.config.getoption("target_url")
    test_data = get_test_data(json_file, ref_url, target_url)
    # Load the parameters for each test
    metafunc.parametrize("db_test_data", test_data["database"], indirect=True)
    # TODO: parametrize all files tests (once class is available)


@pytest.fixture()
def db_test_data(request: FixtureRequest) -> Dict:
    """Fixture to retrieve the database test data.

    Args:
        request: Special fixture providing information of the requesting test function.

    Returns:
        Dictionary with the database test data. Keys "db_test_handler", "test" and "table" will always be
        present.
    """
    return request.param


def pytest_sessionstart(session: Session) -> None:
    """Add required variables to the session before entering the run test loop.

    Args:
        session: Pytest Session object.
    """
    session.report = {}


def pytest_sessionfinish(session: Session, exitstatus: int) -> None:
    """Generate a custom report before returning the exit status to the system.

    Args:
        session: Pytest Session object.
        existatus: Status which pytest will return to the system.
    """
    report_list = [report for report in session.report.values()]
    # Save full report in a JSON file
    pipeline_name = session.config.getoption("pipeline_name")
    if pipeline_name:
        report_file = pipeline_name + ".report.json"
    else:
        json_file = os.path.basename(session.config.getoption("json_file"))
        report_file = json_file.rsplit(".", 1)[0] + ".report.json"
    # Make sure not to overwrite previous reports
    if os.path.isfile(report_file):
        i = 1
        while os.path.isfile("{}.{}".format(report_file, i)):
            i += 1
        report_file = "{}.{}".format(report_file, i)
    with open(report_file, "w") as f:
        json_report = {} # type: Dict
        for report in report_list:
            test_run = re.sub(r'\[.+\]', '', report.location[-1])
            test_report = json_report.setdefault(test_run, {})
            table_report = test_report.setdefault(report.test_params["table"], {})
            subtest = report.test_params["test"]
            info = {"args":   {k: v for k, v in report.test_params.items() if k not in ["table", "test"]},
                    "status": report.outcome.capitalize()}
            if report.failed:
                info["error"] = {"message": report.longrepr.reprcrash.message,
                                 "details": report.error_info}
            table_report.setdefault(subtest, []).append(info)
        json.dump(json_report, f, indent=4, sort_keys=True)
    # Print summary
    total = len(report_list)
    failed = sum(1 for report in report_list if report.failed)
    print("\n{} out of {} tests ok".format(total-failed, total))


@pytest.hookimpl(tryfirst=True, hookwrapper=True)
def pytest_runtest_makereport(item: pytest.Item, call: CallInfo) -> TestReport:
    """Extend the test report with custom information.

    Args:
        item: Pytest Item.
        call: Result/Exception info a function invocation.

    Returns:
        Updated test report.
    """
    outcome = yield
    report = outcome.get_result()
    if report.when == "call":
        # Update the test report adding our custom report_info
        report.error_info = getattr(item, "error_info", 0)
        if "db_test_data" in item.fixturenames:
            report.test_params = item.funcargs["db_test_data"]
            del report.test_params["db_test_handler"]
        # TODO: Once class is available, uncomment following lines
        # elif "files_test_data" in item.fixturenames:
        #     report.test_params = item.funcargs["files_test_data"]
        #     del report.test_params["files_test_handler"]
        item.session.report[item] = report


def get_test_data(json_file: str, ref_url: str, target_url: str) -> Dict:
    """Load the test data from the JSON file.

    The reference and target database URLs will be added to the test data if any database test is performed.

    Args:
        json_file: CITest configuration JSON file for a specific pipeline/database.
        ref_url: URL to the reference database.
        target_url: URL to the target database.

    Returns:
        Dictionary with the test data for both database and files tests.
    """
    # Load the JSON file with the tests to run and their parameters
    with open(json_file) as f:
        data = json.load(f)
    params = {"database": [], "files": []} # type: Dict
    # Load the configuration for all database tests
    if data["database"]:
        # If no reference database is passed, use the default one defined in the JSON file
        if not ref_url:
            ref_url = data["reference_db"]
        db_test_handler = TestDB.TestDB(ref_url=ref_url, target_url=target_url)
        for table in data["database"]:
            for test in data["database"][table]:
                # Add extra parameters required for the test
                test["db_test_handler"] = db_test_handler
                test["table"] = table
                params["database"].append(test)
    # TODO: load all files tests (once class is available)
    return params

{'fraction_non_alphanumeric': 0.10237027546444587, 'fraction_numerical': 0.003587443946188341, 'mean_word_length': 2.97428716904277, 'pattern_counts': {'":': 14, '<': 3, '<?xml version=': 0, '>': 17, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4665376', 'n_tokens_mistral': 4438, 'n_tokens_neox': 4368, 'n_words': 1419}