<commit_before># Copyright (c) 2012 - 2014 <NAME>, Hupfeldt IT
# All rights reserved. This work is under a BSD license, see LICENSE.TXT.

from __future__ import print_function
import os
from os.path import join as jp
from collections import OrderedDict

from jenkinsflow.api_base import BuildResult, Progress, UnknownJobException, ApiInvocationMixin
from jenkinsflow.mocked import hyperspeed

from .base_test_api import TestJob, TestJenkins
from jenkinsflow.test.cfg import ApiType

here = os.path.abspath(os.path.dirname(__file__))


class MockApi(TestJenkins):
    job_xml_template = jp(here, 'job.xml.tenjin')
    api_type = ApiType.MOCK

    def __init__(self, job_name_prefix, baseurl):
        super(MockApi, self).__init__(job_name_prefix)
        self.baseurl = baseurl
        self.username = 'dummy'
        self.password = '<PASSWORD>'

    def job(self, name, exec_time, max_fails, expect_invocations, expect_order, initial_buildno=None, invocation_delay=0.1, params=None,
            script=None, unknown_result=False, final_result=None, serial=False, print_env=False, flow_created=False, create_job=None,
            disappearing=False, non_existing=False, kill=False):
        job_name = self.job_name_prefix + name
        assert not self.test_jobs.get(job_name)
        job = MockJob(name=job_name, exec_time=exec_time, max_fails=max_fails, expect_invocations=expect_invocations, expect_order=expect_order,
                      initial_buildno=initial_buildno, invocation_delay=invocation_delay, unknown_result=unknown_result,
                      final_result=final_result, serial=serial, params=params, flow_created=flow_created, create_job=create_job,
                      disappearing=disappearing, non_existing=non_existing, kill=kill)
        self.test_jobs[job_name] = job

    def flow_job(self, name=None, params=None):
        # Don't create flow jobs when mocked
        return self.flow_job_name(name)

    # --- Mock API ---

    def poll(self):
        pass

    def quick_poll(self):
        pass

    def queue_poll(self):
        pass

    def get_job(self, name):
        try:
            job = self.test_jobs[name]
            if job.non_existing:
                raise UnknownJobException(name)
            return job
        except KeyError:
            raise UnknownJobException(name)

    def set_build_description(self, job_name, build_number, description, replace=False, separator='\n'):
        pass


class MockJob(TestJob):
    def __init__(self, name, exec_time, max_fails, expect_invocations, expect_order, initial_buildno, invocation_delay, unknown_result,
                 final_result, serial, params, flow_created, create_job, disappearing, non_existing, kill):
        super(MockJob, self).__init__(exec_time=exec_time, max_fails=max_fails, expect_invocations=expect_invocations, expect_order=expect_order,
                                      initial_buildno=initial_buildno, invocation_delay=invocation_delay, unknown_result=unknown_result, final_result=final_result,
                                      serial=serial, print_env=False, flow_created=flow_created, create_job=create_job, disappearing=disappearing,
                                      non_existing=non_existing, kill=kill)
        self.name = name
        self.public_uri = self.baseurl = 'http://hupfeldtit.dk/job/' + self.name
        self.build_number = None
        self.initial_build_number = initial_buildno
        self.last_build_number = initial_buildno
        self.params = params
        self.just_invoked = False
        self._invocation_url = 0
        self._invocations = OrderedDict()
        self.queued_why = "Why am I queued?"

    def job_status(self):
        latest_build_number = self._get_last_build_number_or_none()
        if self.start_time <= hyperspeed.time() < self.end_time:
            assert latest_build_number
            return (BuildResult.UNKNOWN, Progress.RUNNING, latest_build_number)

        if self.invocation_time <= hyperspeed.time() < self.start_time:
            return (BuildResult.UNKNOWN, Progress.QUEUD, latest_build_number)

        # Job is finished
        if self.invocation_number <= self.max_fails:
            return (BuildResult.FAILURE, Progress.IDLE, latest_build_number)
        if self.final_result is None:
            return (BuildResult.SUCCESS, Progress.IDLE, latest_build_number)
        return (BuildResult[self.final_result.name], Progress.IDLE, latest_build_number)

    def poll(self):
        # If has been invoked and started running or already (supposed to be) finished
        if self.just_invoked and hyperspeed.time() >= self.start_time:
            self.just_invoked = False
            self.build_number = self.last_build_number + 1 if self.last_build_number else 1<commit_msg>Fix mock api typo<commit_after># Copyright (c) 2012 - 2014 <NAME>, Hupfeldt IT
# All rights reserved. This work is under a BSD license, see LICENSE.TXT.

from __future__ import print_function
import os
from os.path import join as jp
from collections import OrderedDict

from jenkinsflow.api_base import BuildResult, Progress, UnknownJobException, ApiInvocationMixin
from jenkinsflow.mocked import hyperspeed

from .base_test_api import TestJob, TestJenkins
from jenkinsflow.test.cfg import ApiType

here = os.path.abspath(os.path.dirname(__file__))


class MockApi(TestJenkins):
    job_xml_template = jp(here, 'job.xml.tenjin')
    api_type = ApiType.MOCK

    def __init__(self, job_name_prefix, baseurl):
        super(MockApi, self).__init__(job_name_prefix)
        self.baseurl = baseurl
        self.username = 'dummy'
        self.password = '<PASSWORD>'

    def job(self, name, exec_time, max_fails, expect_invocations, expect_order, initial_buildno=None, invocation_delay=0.1, params=None,
            script=None, unknown_result=False, final_result=None, serial=False, print_env=False, flow_created=False, create_job=None,
            disappearing=False, non_existing=False, kill=False):
        job_name = self.job_name_prefix + name
        assert not self.test_jobs.get(job_name)
        job = MockJob(name=job_name, exec_time=exec_time, max_fails=max_fails, expect_invocations=expect_invocations, expect_order=expect_order,
                      initial_buildno=initial_buildno, invocation_delay=invocation_delay, unknown_result=unknown_result,
                      final_result=final_result, serial=serial, params=params, flow_created=flow_created, create_job=create_job,
                      disappearing=disappearing, non_existing=non_existing, kill=kill)
        self.test_jobs[job_name] = job

    def flow_job(self, name=None, params=None):
        # Don't create flow jobs when mocked
        return self.flow_job_name(name)

    # --- Mock API ---

    def poll(self):
        pass

    def quick_poll(self):
        pass

    def queue_poll(self):
        pass

    def get_job(self, name):
        try:
            job = self.test_jobs[name]
            if job.non_existing:
                raise UnknownJobException(name)
            return job
        except KeyError:
            raise UnknownJobException(name)

    def set_build_description(self, job_name, build_number, description, replace=False, separator='\n'):
        pass


class MockJob(TestJob):
    def __init__(self, name, exec_time, max_fails, expect_invocations, expect_order, initial_buildno, invocation_delay, unknown_result,
                 final_result, serial, params, flow_created, create_job, disappearing, non_existing, kill):
        super(MockJob, self).__init__(exec_time=exec_time, max_fails=max_fails, expect_invocations=expect_invocations, expect_order=expect_order,
                                      initial_buildno=initial_buildno, invocation_delay=invocation_delay, unknown_result=unknown_result, final_result=final_result,
                                      serial=serial, print_env=False, flow_created=flow_created, create_job=create_job, disappearing=disappearing,
                                      non_existing=non_existing, kill=kill)
        self.name = name
        self.public_uri = self.baseurl = 'http://hupfeldtit.dk/job/' + self.name
        self.build_number = None
        self.initial_build_number = initial_buildno
        self.last_build_number = initial_buildno
        self.params = params
        self.just_invoked = False
        self._invocation_url = 0
        self._invocations = OrderedDict()
        self.queued_why = "Why am I queued?"

    def job_status(self):
        latest_build_number = self._get_last_build_number_or_none()
        if self.start_time <= hyperspeed.time() < self.end_time:
            assert latest_build_number
            return (BuildResult.UNKNOWN, Progress.RUNNING, latest_build_number)

        if self.invocation_time <= hyperspeed.time() < self.start_time:
            return (BuildResult.UNKNOWN, Progress.QUEUED, latest_build_number)

        # Job is finished
        if self.invocation_number <= self.max_fails:
            return (BuildResult.FAILURE, Progress.IDLE, latest_build_number)
        if self.final_result is None:
            return (BuildResult.SUCCESS, Progress.IDLE, latest_build_number)
        return (BuildResult[self.final_result.name], Progress.IDLE, latest_build_number)

    def poll(self):
        # If has been invoked and started running or already (supposed to be) finished
        if self.just_invoked and hyperspeed.time() >= self.start_time:
            self.just_invoked = False
            self.build_number = self.last_build_number + 1 if self.last_build_number else 1
{'fraction_non_alphanumeric': 0.0810782572353986, 'fraction_numerical': 0.0027165395465468603, 'mean_word_length': 3.1223083548664943, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 4}, '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': '5256491', 'n_tokens_mistral': 2874, 'n_tokens_neox': 2779, 'n_words': 608}