<commit_before>from __future__ import absolute_import, division

import json
import logging
import requests
import time

from datetime import datetime
from hashlib import sha1
from flask import current_app

from changes.backends.base import BaseBackend, UnrecoverableException
from changes.config import db, queue
from changes.constants import Result, Status
from changes.db.utils import create_or_update, get_or_create
from changes.events import publish_logchunk_update
from changes.models import (
    AggregateTestSuite, TestResult, TestResultManager, TestSuite,
    LogSource, LogChunk, Node, JobPhase, JobStep
)

LOG_CHUNK_SIZE = 4096


def chunked(iterator, chunk_size):
    """
    Given an iterator, chunk it up into ~chunk_size, but be aware of newline
    termination as an intended goal.
    """
    result = ''
    for chunk in iterator:
        result += chunk
        while len(result) >= chunk_size:
            newline_pos = result.rfind('\n', 0, chunk_size)
            if newline_pos == -1:
                newline_pos = chunk_size
            else:
                newline_pos += 1
            yield result[:newline_pos]
            result = result[newline_pos:]
    if result:
        yield result


class NotFound(Exception):
    pass


class JenkinsBuilder(BaseBackend):
    provider = 'jenkins'

    def __init__(self, base_url=None, job_name=None, token=None, *args, **kwargs):
        super(JenkinsBuilder, self).__init__(*args, **kwargs)
        self.base_url = base_url or self.app.config['JENKINS_URL']
        self.token = token or self.app.config['JENKINS_TOKEN']
        self.sync_artifacts = self.app.config['JENKINS_SYNC_ARTIFACTS']
        self.logger = logging.getLogger('jenkins')
        self.job_name = job_name

    def _get_raw_response(self, path, method='GET', params=None, **kwargs):
        url = '{}/{}'.format(self.base_url, path.lstrip('/'))

        if params is None:
            params = {}

        params.setdefault('token', self.token)

        self.logger.info('Fetching %r', url)
        resp = getattr(requests, method.lower())(url, params=params, **kwargs)

        if resp.status_code == 404:
            raise NotFound
        elif not (200 <= resp.status_code < 300):
            raise Exception('Invalid response. Status code was %s' % resp.status_code)

        return resp.text

    def _get_json_response(self, path, *args, **kwargs):
        path = '{}/api/json/'.format(path.strip('/'))

        data = self._get_raw_response(path, *args, **kwargs)
        if not data:
            return

        try:
            return json.loads(data)
        except ValueError:
            raise Exception('Invalid JSON data')

    _get_response = _get_json_response

    def _parse_parameters(self, json):
        params = {}
        for action in json['actions']:
            params.update(
                (p['name'], p.get('value'))
                for p in action.get('parameters', [])
            )
        return params

    def _sync_job_from_queue(self, job):
        # TODO(dcramer): when we hit a NotFound in the queue, maybe we should
        # attempt to scrape the list of jobs for a matching CHANGES_BID, as this
        # doesnt explicitly mean that the job doesnt exist
        try:
            item = self._get_response('/queue/item/{}'.format(
                job.data['item_id']))
        except NotFound:
            job.status = Status.finished
            job.result = Result.unknown
            db.session.add(job)
            return

        if item.get('executable'):
            build_no = item['executable']['number']
            job.data['queued'] = False
            job.data['build_no'] = build_no

        if item['blocked']:
            job.status = Status.queued
            db.session.add(job)
        elif item.get('cancelled') and not job.data.get('build_no'):
            job.status = Status.finished
            job.result = Result.aborted
            db.session.add(job)
        elif item.get('executable'):
            for x in xrange(6):
                # There's a possible race condition where the item has been
                # assigned an ID, yet the API responds as if the job does
                # not exist
                try:
                    self._sync_job_from_active(job, fail_on_404=True)
                except NotFound:
                    time.sleep(0.3)
                else:
                    break

    def _sync_job_from_active(self, job, fail_on_404=False):
        try:
            item = self._get_response('/job/{}/{}'.format(
                job.data['job_name'], job.data['build_no']))
        except NotFound:
            if fail_on_404:
                raise
            job.date_finished = datetime.utcnow()
            job.status = Status.finished
            job.result = Result.unknown
            db.session.add(job)
            return

        should_finish = False

        # XXX(dcramer): timestamp implies creation date, so lets just assume
        # we were able to track it immediately
        if not job.date_started:
            job.date_started = datetime.utcnow()

        if item['building']:
            job.status = Status.in_progress
        else:
            should_finish = True
            job.date_finished = datetime.utcnow()

            if item['result'] == 'SUCCESS':
                job.result = Result.passed
            elif item['result'] == 'ABORTED':
                job.result = Result.aborted
            elif item['result'] in ('FAILURE', 'REGRESSION', 'UNSTABLE'):
                job.result = Result.failed
            else:
                raise ValueError('Invalid job result: %s' % (item['result'],))

        if item['duration']:
            job.duration = item['duration']

        job.data.update({
            'backend': {
                'uri': item['url'],
                'label': item['fullDisplayName'],
            }
        })

        db.session.add(job)
        db.session.commit()

        node, _ = get_or_create(Node, where={
            'label': item['builtOn'],
        })

        jobphase, created = get_or_create(JobPhase, where={
            'job': job,
            'label': job.data['job_name'],
        }, defaults={
            'project_id': job.project_id,
            'repository_id': job.repository_id,
            'date_started': job.date_started,
            'status': job.status,
            'result': job.result,
        })

        jobstep, created = get_or_create(JobStep, where={
            'phase': jobphase,
            'label': item['fullDisplayName'],
        }, defaults={
            'job': job,
            'project_id': job.project_id,
            'node_id': node.id,
            'repository_id': job.repository_id,
            'date_started': job.date_started,
            'status': job.status,
            'result': job.result,
        })

        # TODO(dcramer): ideally we could fire off jobs to sync test results
        # and console logs
        try:
            self._sync_test_results(job)
        except Exception:
            db.session.rollback()
            current_app.logger.exception('Unable to sync test results for job %r', job.id.hex)

        if should_finish:
            # FIXME(dcramer): we're waiting until the job is complete to sync
            # logs due to our inability to correctly identify start offsets
            # if we're supposed to be finishing, lets ensure we actually
            # get the entirety of the log
            try:
                start = time.time()
                while self._sync_console_log(job):
                    if time.time() - start > 15:
                        raise Exception('Took too long to sync log')
                    continue
            except Exception:<commit_msg>Only sync test results if we are finished<commit_after>from __future__ import absolute_import, division

import json
import logging
import requests
import time

from datetime import datetime
from hashlib import sha1
from flask import current_app

from changes.backends.base import BaseBackend, UnrecoverableException
from changes.config import db, queue
from changes.constants import Result, Status
from changes.db.utils import create_or_update, get_or_create
from changes.events import publish_logchunk_update
from changes.models import (
    AggregateTestSuite, TestResult, TestResultManager, TestSuite,
    LogSource, LogChunk, Node, JobPhase, JobStep
)

LOG_CHUNK_SIZE = 4096


def chunked(iterator, chunk_size):
    """
    Given an iterator, chunk it up into ~chunk_size, but be aware of newline
    termination as an intended goal.
    """
    result = ''
    for chunk in iterator:
        result += chunk
        while len(result) >= chunk_size:
            newline_pos = result.rfind('\n', 0, chunk_size)
            if newline_pos == -1:
                newline_pos = chunk_size
            else:
                newline_pos += 1
            yield result[:newline_pos]
            result = result[newline_pos:]
    if result:
        yield result


class NotFound(Exception):
    pass


class JenkinsBuilder(BaseBackend):
    provider = 'jenkins'

    def __init__(self, base_url=None, job_name=None, token=None, *args, **kwargs):
        super(JenkinsBuilder, self).__init__(*args, **kwargs)
        self.base_url = base_url or self.app.config['JENKINS_URL']
        self.token = token or self.app.config['JENKINS_TOKEN']
        self.sync_artifacts = self.app.config['JENKINS_SYNC_ARTIFACTS']
        self.logger = logging.getLogger('jenkins')
        self.job_name = job_name

    def _get_raw_response(self, path, method='GET', params=None, **kwargs):
        url = '{}/{}'.format(self.base_url, path.lstrip('/'))

        if params is None:
            params = {}

        params.setdefault('token', self.token)

        self.logger.info('Fetching %r', url)
        resp = getattr(requests, method.lower())(url, params=params, **kwargs)

        if resp.status_code == 404:
            raise NotFound
        elif not (200 <= resp.status_code < 300):
            raise Exception('Invalid response. Status code was %s' % resp.status_code)

        return resp.text

    def _get_json_response(self, path, *args, **kwargs):
        path = '{}/api/json/'.format(path.strip('/'))

        data = self._get_raw_response(path, *args, **kwargs)
        if not data:
            return

        try:
            return json.loads(data)
        except ValueError:
            raise Exception('Invalid JSON data')

    _get_response = _get_json_response

    def _parse_parameters(self, json):
        params = {}
        for action in json['actions']:
            params.update(
                (p['name'], p.get('value'))
                for p in action.get('parameters', [])
            )
        return params

    def _sync_job_from_queue(self, job):
        # TODO(dcramer): when we hit a NotFound in the queue, maybe we should
        # attempt to scrape the list of jobs for a matching CHANGES_BID, as this
        # doesnt explicitly mean that the job doesnt exist
        try:
            item = self._get_response('/queue/item/{}'.format(
                job.data['item_id']))
        except NotFound:
            job.status = Status.finished
            job.result = Result.unknown
            db.session.add(job)
            return

        if item.get('executable'):
            build_no = item['executable']['number']
            job.data['queued'] = False
            job.data['build_no'] = build_no

        if item['blocked']:
            job.status = Status.queued
            db.session.add(job)
        elif item.get('cancelled') and not job.data.get('build_no'):
            job.status = Status.finished
            job.result = Result.aborted
            db.session.add(job)
        elif item.get('executable'):
            for x in xrange(6):
                # There's a possible race condition where the item has been
                # assigned an ID, yet the API responds as if the job does
                # not exist
                try:
                    self._sync_job_from_active(job, fail_on_404=True)
                except NotFound:
                    time.sleep(0.3)
                else:
                    break

    def _sync_job_from_active(self, job, fail_on_404=False):
        try:
            item = self._get_response('/job/{}/{}'.format(
                job.data['job_name'], job.data['build_no']))
        except NotFound:
            if fail_on_404:
                raise
            job.date_finished = datetime.utcnow()
            job.status = Status.finished
            job.result = Result.unknown
            db.session.add(job)
            return

        should_finish = False

        # XXX(dcramer): timestamp implies creation date, so lets just assume
        # we were able to track it immediately
        if not job.date_started:
            job.date_started = datetime.utcnow()

        if item['building']:
            job.status = Status.in_progress
        else:
            should_finish = True
            job.date_finished = datetime.utcnow()

            if item['result'] == 'SUCCESS':
                job.result = Result.passed
            elif item['result'] == 'ABORTED':
                job.result = Result.aborted
            elif item['result'] in ('FAILURE', 'REGRESSION', 'UNSTABLE'):
                job.result = Result.failed
            else:
                raise ValueError('Invalid job result: %s' % (item['result'],))

        if item['duration']:
            job.duration = item['duration']

        job.data.update({
            'backend': {
                'uri': item['url'],
                'label': item['fullDisplayName'],
            }
        })

        db.session.add(job)
        db.session.commit()

        node, _ = get_or_create(Node, where={
            'label': item['builtOn'],
        })

        jobphase, created = get_or_create(JobPhase, where={
            'job': job,
            'label': job.data['job_name'],
        }, defaults={
            'project_id': job.project_id,
            'repository_id': job.repository_id,
            'date_started': job.date_started,
            'status': job.status,
            'result': job.result,
        })

        jobstep, created = get_or_create(JobStep, where={
            'phase': jobphase,
            'label': item['fullDisplayName'],
        }, defaults={
            'job': job,
            'project_id': job.project_id,
            'node_id': node.id,
            'repository_id': job.repository_id,
            'date_started': job.date_started,
            'status': job.status,
            'result': job.result,
        })

        if should_finish:
            # TODO(dcramer): ideally we could fire off jobs to sync test results
            # and console logs
            try:
                self._sync_test_results(job)
            except Exception:
                db.session.rollback()
                current_app.logger.exception('Unable to sync test results for job %r', job.id.hex)

            # FIXME(dcramer): we're waiting until the job is complete to sync
            # logs due to our inability to correctly identify start offsets
            # if we're supposed to be finishing, lets ensure we actually
            # get the entirety of the log
            try:
                start = time.time()
                while self._sync_console_log(job):
                    if time.time() - start > 15:
                        raise Exception('Took too long to sync log')
                    continue
            except Exception:
{'fraction_non_alphanumeric': 0.10286374727109285, 'fraction_numerical': 0.003980993964299473, 'mean_word_length': 1.8704386288241799, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 7, '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': '6627705', 'n_tokens_mistral': 4365, 'n_tokens_neox': 4185, 'n_words': 1134}