<commit_before># -*- coding: utf-8 -*-

import logging
import sys
import os
import traceback
import textwrap
from typing import Dict, List

import pytest

from openfisca_core.tools import assert_near
from openfisca_core.simulation_builder import SimulationBuilder
from openfisca_core.errors import SituationParsingError, VariableNotFound

log = logging.getLogger(__name__)


def import_yaml():
    import yaml
    try:
        from yaml import CLoader as Loader
    except ImportError:
        log.warning(
            ' '
            'libyaml is not installed in your environment, this can make your '
            'test suite slower to run. Once you have installed libyaml, run `pip '
            'uninstall pyyaml && pip install pyyaml --no-cache-dir` so that it is used in your '
            'Python environment.')
        from yaml import SafeLoader as Loader
    return yaml, Loader


TEST_KEYWORDS = {'absolute_error_margin', 'description', 'extensions', 'ignore_variables', 'input', 'keywords', 'name', 'only_variables', 'output', 'period', 'reforms', 'relative_error_margin'}

yaml, Loader = import_yaml()

_tax_benefit_system_cache: Dict = {}


def run_tests(tax_benefit_system, paths, options = None):
    """
    Runs all the YAML tests contained in a file or a directory.

    If `path` is a directory, subdirectories will be recursively explored.

    :param TaxBenefitSystem tax_benefit_system: the tax-benefit system to use to run the tests
    :param (str/list) paths: A path, or a list of paths, towards the files or directories containing the tests to run. If a path is a directory, subdirectories will be recursively explored.
    :param dict options: See more details below.

    :raises AssertionError: if a test does not pass

    :return: the number of sucessful tests excecuted

    **Testing options**:

    +-------------------------------+-----------+-------------------------------------------+
    | Key                           | Type      | Role                                      |
    +===============================+===========+===========================================+
    | verbose                       | ``bool``  |                                           |
    +-------------------------------+-----------+ See :any:`openfisca_test` options doc +
    | name_filter                   | ``str``   |                                           |
    +-------------------------------+-----------+-------------------------------------------+

    """

    argv = ["--capture", "no"]

    if options.get('pdb'):
        argv.append('--pdb')

    if isinstance(paths, str):
        paths = [paths]

    if options is None:
        options = {}

    return pytest.main([*argv, *paths] if True else paths, plugins = [OpenFiscaPlugin(tax_benefit_system, options)])


class YamlFile(pytest.File):

    def __init__(self, path, fspath, parent, tax_benefit_system, options):
        super(YamlFile, self).__init__(path, parent)
        self.tax_benefit_system = tax_benefit_system
        self.options = options

    def collect(self):
        try:
            tests = yaml.load(self.fspath.open(), Loader = Loader)
        except (yaml.scanner.ScannerError, yaml.parser.ParserError, TypeError):
            message = os.linesep.join([
                traceback.format_exc(),
                f"'{self.fspath}' is not a valid YAML file. Check the stack trace above for more details.",
                ])
            raise ValueError(message)

        if not isinstance(tests, list):
            tests: List[Dict] = [tests]

        for test in tests:
            if not self.should_ignore(test):
                yield YamlItem.from_parent(self,
                    name = '',
                    baseline_tax_benefit_system = self.tax_benefit_system,
                    test = test, options = self.options)

    def should_ignore(self, test):
        name_filter = self.options.get('name_filter')
        return (
            name_filter is not None
            and name_filter not in os.path.splitext(self.fspath.basename)[0]
            and name_filter not in test.get('name', '')
            and name_filter not in test.get('keywords', [])
            )


class YamlItem(pytest.Item):
    """
    Terminal nodes of the test collection tree.
    """

    def __init__(self, name, parent, baseline_tax_benefit_system, test, options):
        super(YamlItem, self).__init__(name, parent)
        self.baseline_tax_benefit_system = baseline_tax_benefit_system
        self.options = options
        self.test = test
        self.simulation = None
        self.tax_benefit_system = None

    def runtest(self):
        self.name = self.test.get('name', '')
        if not self.test.get('output'):
            raise ValueError("Missing key 'output' in test '{}' in file '{}'".format(self.name, self.fspath))

        if not TEST_KEYWORDS.issuperset(self.test.keys()):
            unexpected_keys = set(self.test.keys()).difference(TEST_KEYWORDS)
            raise ValueError("Unexpected keys {} in test '{}' in file '{}'".format(unexpected_keys, self.name, self.fspath))

        self.tax_benefit_system = _get_tax_benefit_system(self.baseline_tax_benefit_system, self.test.get('reforms', []), self.test.get('extensions', []))

        builder = SimulationBuilder()
        input = self.test.get('input', {})
        period = self.test.get('period')
        verbose = self.options.get('verbose')
        performance_graph = self.options.get('performance_graph')
        performance_tables = self.options.get('performance_tables')

        try:
            builder.set_default_period(period)
            self.simulation = builder.build_from_dict(self.tax_benefit_system, input)
        except (VariableNotFound, SituationParsingError):
            raise
        except Exception as e:
            error_message = os.linesep.join([str(e), '', f"Unexpected error raised while parsing '{self.fspath}'"])
            raise ValueError(error_message).with_traceback(sys.exc_info()[2]) from e  # Keep the stack trace from the root error<commit_msg>Control max_spiral_loops via YAML test files<commit_after># -*- coding: utf-8 -*-

import logging
import sys
import os
import traceback
import textwrap
from typing import Dict, List

import pytest

from openfisca_core.tools import assert_near
from openfisca_core.simulation_builder import SimulationBuilder
from openfisca_core.errors import SituationParsingError, VariableNotFound

log = logging.getLogger(__name__)


def import_yaml():
    import yaml
    try:
        from yaml import CLoader as Loader
    except ImportError:
        log.warning(
            ' '
            'libyaml is not installed in your environment, this can make your '
            'test suite slower to run. Once you have installed libyaml, run `pip '
            'uninstall pyyaml && pip install pyyaml --no-cache-dir` so that it is used in your '
            'Python environment.')
        from yaml import SafeLoader as Loader
    return yaml, Loader


TEST_KEYWORDS = {'absolute_error_margin', 'description', 'extensions', 'ignore_variables', 'input', 'keywords', 'name', 'only_variables', 'output', 'period', 'reforms', 'relative_error_margin'}

yaml, Loader = import_yaml()

_tax_benefit_system_cache: Dict = {}


def run_tests(tax_benefit_system, paths, options = None):
    """
    Runs all the YAML tests contained in a file or a directory.

    If `path` is a directory, subdirectories will be recursively explored.

    :param TaxBenefitSystem tax_benefit_system: the tax-benefit system to use to run the tests
    :param (str/list) paths: A path, or a list of paths, towards the files or directories containing the tests to run. If a path is a directory, subdirectories will be recursively explored.
    :param dict options: See more details below.

    :raises AssertionError: if a test does not pass

    :return: the number of sucessful tests excecuted

    **Testing options**:

    +-------------------------------+-----------+-------------------------------------------+
    | Key                           | Type      | Role                                      |
    +===============================+===========+===========================================+
    | verbose                       | ``bool``  |                                           |
    +-------------------------------+-----------+ See :any:`openfisca_test` options doc +
    | name_filter                   | ``str``   |                                           |
    +-------------------------------+-----------+-------------------------------------------+

    """

    argv = ["--capture", "no"]

    if options.get('pdb'):
        argv.append('--pdb')

    if isinstance(paths, str):
        paths = [paths]

    if options is None:
        options = {}

    return pytest.main([*argv, *paths] if True else paths, plugins = [OpenFiscaPlugin(tax_benefit_system, options)])


class YamlFile(pytest.File):

    def __init__(self, path, fspath, parent, tax_benefit_system, options):
        super(YamlFile, self).__init__(path, parent)
        self.tax_benefit_system = tax_benefit_system
        self.options = options

    def collect(self):
        try:
            tests = yaml.load(self.fspath.open(), Loader = Loader)
        except (yaml.scanner.ScannerError, yaml.parser.ParserError, TypeError):
            message = os.linesep.join([
                traceback.format_exc(),
                f"'{self.fspath}' is not a valid YAML file. Check the stack trace above for more details.",
                ])
            raise ValueError(message)

        if not isinstance(tests, list):
            tests: List[Dict] = [tests]

        for test in tests:
            if not self.should_ignore(test):
                yield YamlItem.from_parent(self,
                    name = '',
                    baseline_tax_benefit_system = self.tax_benefit_system,
                    test = test, options = self.options)

    def should_ignore(self, test):
        name_filter = self.options.get('name_filter')
        return (
            name_filter is not None
            and name_filter not in os.path.splitext(self.fspath.basename)[0]
            and name_filter not in test.get('name', '')
            and name_filter not in test.get('keywords', [])
            )


class YamlItem(pytest.Item):
    """
    Terminal nodes of the test collection tree.
    """

    def __init__(self, name, parent, baseline_tax_benefit_system, test, options):
        super(YamlItem, self).__init__(name, parent)
        self.baseline_tax_benefit_system = baseline_tax_benefit_system
        self.options = options
        self.test = test
        self.simulation = None
        self.tax_benefit_system = None

    def runtest(self):
        self.name = self.test.get('name', '')
        if not self.test.get('output'):
            raise ValueError("Missing key 'output' in test '{}' in file '{}'".format(self.name, self.fspath))

        if not TEST_KEYWORDS.issuperset(self.test.keys()):
            unexpected_keys = set(self.test.keys()).difference(TEST_KEYWORDS)
            raise ValueError("Unexpected keys {} in test '{}' in file '{}'".format(unexpected_keys, self.name, self.fspath))

        self.tax_benefit_system = _get_tax_benefit_system(self.baseline_tax_benefit_system, self.test.get('reforms', []), self.test.get('extensions', []))

        builder = SimulationBuilder()
        input = self.test.get('input', {})
        period = self.test.get('period')
        max_spiral_loops = self.test.get('max_spiral_loops')
        verbose = self.options.get('verbose')
        performance_graph = self.options.get('performance_graph')
        performance_tables = self.options.get('performance_tables')

        try:
            builder.set_default_period(period)
            self.simulation = builder.build_from_dict(self.tax_benefit_system, input)
        except (VariableNotFound, SituationParsingError):
            raise
        except Exception as e:
            error_message = os.linesep.join([str(e), '', f"Unexpected error raised while parsing '{self.fspath}'"])
            raise ValueError(error_message).with_traceback(sys.exc_info()[2]) from e  # Keep the stack trace from the root error

        if max_spiral_loops:
            self.simulation.max_spiral_loops = max_spiral_loops
{'fraction_non_alphanumeric': 0.145464871004381, 'fraction_numerical': 0.00048677592081778357, 'mean_word_length': 2.2126661454261143, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 14, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '110542', 'n_tokens_mistral': 3454, 'n_tokens_neox': 3297, 'n_words': 910}