<commit_before># -*- coding: utf-8 -*-
"""
    pytest_pipeline.core
    ~~~~~~~~~~~~~~~~~~~~

    Core pipeline test classes.

    :copyright: (c) 2014 <NAME> <<EMAIL>>
    :license: BSD

"""

import os
import shlex
import subprocess
import sys
import time
from uuid import uuid4

import pytest

from . import mark


class PipelineRun(object):

    def __init__(self, cmd, stdout=sys.stdout, stderr=sys.stderr, poll_time=1):
        self.cmd = cmd
        self.stdout = stdout
        self.stderr = stderr
        self.poll_time = poll_time
        self._process = None
        self._toks = shlex.split(cmd)

    def __repr__(self):
        return "{0}(...)".format(self.__class__.__name__)

    @property
    def run_id(self):
        if not hasattr(self, "_run_id"):
            self._run_id = str(uuid4())
        return self._run_id

    @property
    def exit_code(self):
        if self._process is not None:
            return self._process.returncode

    def launch_process(self, restart=False):

        if isinstance(self.stdout, basestring):
            self.stdout = open(self.stdout, "w")
        elif self.stdout is None:
            self.stdout = open(os.devnull, "w")

        if isinstance(self.stderr, basestring):
            self.stderr = open(self.stderr, "w")
        elif self.stderr is None:
            self.stderr = open(os.devnull, "w")

        if restart and self._process is not None:
            self._process.kill()
            self._process = None

        self._process = subprocess.Popen(self._toks, stdout=self.stdout,
                                         stderr=self.stderr)

    def launch_process_and_wait(self, restart=False):
        self.launch_process(restart)
        while self._process.poll() is None:
            time.sleep(self.poll_time)


class MetaPipelineTest(type):

    def __init__(cls, name, bases, dct):
        cls.test_id = str(uuid4())
        cls.test_dirname = cls.__name__ + "_" + cls.test_id
        cls = pytest.mark.usefixtures("_autogenerated_test_dir")(cls)
        cls._before_runs = []
        cls._before_runs_done = []

        if not cls.__name__ == "PipelineTest":
            for attr, val in dct.iteritems():
                if not callable(val) or not hasattr(val, "_pipeline"):
                    continue
                dct[attr] = pytest.mark.incremental(val)
                if not val._pipeline.get("phase") == mark.BEFORE_RUN:
                    continue
                cls._before_runs.append(val)
                cls._before_runs_done.append(False)

            if not isinstance(getattr(cls, "pipeline_run", None), PipelineRun):
                raise ValueError("Test class '{0}' does not have a proper 'pipeline_run' "
                                 "attribute".format(cls))

        super(MetaPipelineTest, cls).__init__(name, bases, dct)


class PipelineTest(object):

    __metaclass__ = MetaPipelineTest

    def __repr__(self):
        return "{0}(...)".format(self.test_id)

    def _run_before_run_methods(self):
        if not all(self._before_runs_done):
            for idx, bfunc in enumerate(self._before_runs):
                bfunc(self)
                self._before_runs_done[idx] = True<commit_msg>Update MetaPipelineTest to allow proper subclassing of classes<commit_after># -*- coding: utf-8 -*-
"""
    pytest_pipeline.core
    ~~~~~~~~~~~~~~~~~~~~

    Core pipeline test classes.

    :copyright: (c) 2014 <NAME> <<EMAIL>>
    :license: BSD

"""

import os
import shlex
import subprocess
import sys
import time
from uuid import uuid4

import pytest

from . import mark


class PipelineRun(object):

    def __init__(self, cmd, stdout=sys.stdout, stderr=sys.stderr, poll_time=1):
        self.cmd = cmd
        self.stdout = stdout
        self.stderr = stderr
        self.poll_time = poll_time
        self._process = None
        self._toks = shlex.split(cmd)

    def __repr__(self):
        return "{0}(...)".format(self.__class__.__name__)

    @property
    def run_id(self):
        if not hasattr(self, "_run_id"):
            self._run_id = str(uuid4())
        return self._run_id

    @property
    def exit_code(self):
        if self._process is not None:
            return self._process.returncode

    def launch_process(self, restart=False):

        if isinstance(self.stdout, basestring):
            self.stdout = open(self.stdout, "w")
        elif self.stdout is None:
            self.stdout = open(os.devnull, "w")

        if isinstance(self.stderr, basestring):
            self.stderr = open(self.stderr, "w")
        elif self.stderr is None:
            self.stderr = open(os.devnull, "w")

        if restart and self._process is not None:
            self._process.kill()
            self._process = None

        self._process = subprocess.Popen(self._toks, stdout=self.stdout,
                                         stderr=self.stderr)

    def launch_process_and_wait(self, restart=False):
        self.launch_process(restart)
        while self._process.poll() is None:
            time.sleep(self.poll_time)


class MetaPipelineTest(type):

    def __init__(cls, name, bases, dct):
        cls.test_id = str(uuid4())
        cls.test_dirname = cls.__name__ + "_" + cls.test_id
        cls = pytest.mark.usefixtures("_autogenerated_test_dir")(cls)
        cls._before_runs = []
        cls._before_runs_done = []

        for attr, val in dct.iteritems():

            if attr == "pipeline_run" and not \
                    isinstance(getattr(cls, "pipeline_run", None), PipelineRun):
                raise ValueError("Test class '{0}' does not have a proper "
                                 "'pipeline_run' attribute".format(cls))

            if not callable(val) or not hasattr(val, "_pipeline"):
                continue

            dct[attr] = pytest.mark.incremental(val)

            if not val._pipeline.get("phase") == mark.BEFORE_RUN:
                continue

            cls._before_runs.append(val)
            cls._before_runs_done.append(False)

        super(MetaPipelineTest, cls).__init__(name, bases, dct)


class PipelineTest(object):

    __metaclass__ = MetaPipelineTest

    def __repr__(self):
        return "{0}(...)".format(self.test_id)

    def _run_before_run_methods(self):
        if not all(self._before_runs_done):
            for idx, bfunc in enumerate(self._before_runs):
                bfunc(self)
                self._before_runs_done[idx] = True
{'fraction_non_alphanumeric': 0.11400651465798045, 'fraction_numerical': 0.0037226617031177293, 'mean_word_length': 2.1392405063291138, 'pattern_counts': {'":': 1, '<': 9, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4002958', 'n_tokens_mistral': 1961, 'n_tokens_neox': 1916, 'n_words': 435}