<commit_before>"""
Stages in a Job
"""

import logging
import json
import shlex
import subprocess

import requests.exceptions

from dockci.exceptions import (AlreadyRunError,
                               DockerUnreachableError,
                               StageFailedError,
                               )
from dockci.server import pika_conn, redis_pool
from dockci.stage_io import StageIO
from dockci.util import bytes_str


class JobStageBase(object):
    """
    A logged stage to a job
    """

    def __init__(self, job):
        self.job = job
        self.returncode = None

    def runnable(self, handle):
        """ Executeable portion of the stage """
        raise NotImplementedError("You must override the 'runnable' method")

    def run(self, expected_rc=0):
        """
        Start the child process, streaming it's output to the associated file,
        and block until it returns. Returns True if the return code matches
        ``expected_rc``. If ``expected_rc`` is None, always return true
        """
        # pylint:disable=no-member
        logging.getLogger('dockci.job.stages').debug(
            "Starting '%s' job stage for job '%s'",
            self.slug, self.job.slug,
        )

        if self.returncode is not None:
            raise AlreadyRunError(self)

        from dockci.models.job import JobStageTmp
        stage = JobStageTmp(job=self.job, slug=self.slug)

        self.job.job_output_path().ensure_dir()
        with pika_conn() as pika_conn_:
            with redis_pool() as redis_pool_:
                with StageIO.open(
                    self,
                    redis_pool=redis_pool_,
                    pika_conn=pika_conn_,
                ) as handle:
                    self.job.db_session.add(stage)
                    self.job.db_session.commit()

                    try:
                        self.returncode = self.runnable(handle)

                    except StageFailedError as ex:
                        if not ex.handled:
                            handle.write(("FAILED: %s\n" % ex).encode())
                            handle.flush()

                        return False

        if expected_rc is None:
            return True

        success = self.returncode == expected_rc
        if not success:
            logging.getLogger('dockci.job.stages').debug(
                "Stage '%s' expected return of '%s'. Actually got '%s'",
                self.slug, expected_rc, self.returncode,
            )

        return success


class JobStage(JobStageBase):
    """ Ad-hoc job stage """<commit_msg>Add todo for stage status<commit_after>"""
Stages in a Job
"""

import logging
import json
import shlex
import subprocess

import requests.exceptions

from dockci.exceptions import (AlreadyRunError,
                               DockerUnreachableError,
                               StageFailedError,
                               )
from dockci.server import pika_conn, redis_pool
from dockci.stage_io import StageIO
from dockci.util import bytes_str


class JobStageBase(object):
    """
    A logged stage to a job
    """

    def __init__(self, job):
        self.job = job
        self.returncode = None

    def runnable(self, handle):
        """ Executeable portion of the stage """
        raise NotImplementedError("You must override the 'runnable' method")

    def run(self, expected_rc=0):
        """
        Start the child process, streaming it's output to the associated file,
        and block until it returns. Returns True if the return code matches
        ``expected_rc``. If ``expected_rc`` is None, always return true
        """
        # pylint:disable=no-member
        logging.getLogger('dockci.job.stages').debug(
            "Starting '%s' job stage for job '%s'",
            self.slug, self.job.slug,
        )

        if self.returncode is not None:
            raise AlreadyRunError(self)

        from dockci.models.job import JobStageTmp
        stage = JobStageTmp(job=self.job, slug=self.slug)

        self.job.job_output_path().ensure_dir()
        with pika_conn() as pika_conn_:
            with redis_pool() as redis_pool_:
                with StageIO.open(
                    self,
                    redis_pool=redis_pool_,
                    pika_conn=pika_conn_,
                ) as handle:
                    self.job.db_session.add(stage)
                    self.job.db_session.commit()

                    try:
                        self.returncode = self.runnable(handle)

                    except StageFailedError as ex:
                        if not ex.handled:
                            handle.write(("FAILED: %s\n" % ex).encode())
                            handle.flush()

                        return False

        if expected_rc is None:
            return True

        # TODO start/end status to rabbit
        success = self.returncode == expected_rc
        if not success:
            logging.getLogger('dockci.job.stages').debug(
                "Stage '%s' expected return of '%s'. Actually got '%s'",
                self.slug, expected_rc, self.returncode,
            )

        return success


class JobStage(JobStageBase):
    """ Ad-hoc job stage """
{'fraction_non_alphanumeric': 0.08517952635599695, 'fraction_numerical': 0.0003819709702062643, 'mean_word_length': 1.6422805247225025, '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': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3929729', 'n_tokens_mistral': 1432, 'n_tokens_neox': 1362, 'n_words': 387}