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

from __future__ import absolute_import

import logging

from tests.support.helpers import slowTest
from tests.support.unit import skipIf
from tests.unit.utils.scheduler.base import SchedulerTestsBase

try:
    import dateutil.parser

    HAS_DATEUTIL_PARSER = True
except ImportError:
    HAS_DATEUTIL_PARSER = False

log = logging.getLogger(__name__)


@skipIf(
    HAS_DATEUTIL_PARSER is False, "The 'dateutil.parser' library is not available",
)
class SchedulerSkipTest(SchedulerTestsBase):
    """
    Validate the pkg module
    """

    def setUp(self):
        super(SchedulerSkipTest, self).setUp()
        self.schedule.opts["loop_interval"] = 1

    def tearDown(self):
        self.schedule.reset()
        super(SchedulerSkipTest, self).tearDown()

    @slowTest
    def test_skip(self):
        """
        verify that scheduled job is skipped at the specified time
        """
        job_name = "test_skip"
        job = {
            "schedule": {
                job_name: {
                    "function": "test.ping",
                    "when": ["11/29/2017 4pm", "11/29/2017 5pm"],
                }
            }
        }

        # Add job to schedule
        self.schedule.opts.update(job)

        run_time = dateutil.parser.parse("11/29/2017 4:00pm")
        self.schedule.skip_job(
            job_name,
            {
                "time": run_time.strftime("%Y-%m-%dT%H:%M:%S"),
                "time_fmt": "%Y-%m-%dT%H:%M:%S",
            },
        )

        # Run 11/29/2017 at 4pm
        self.schedule.eval(now=run_time)
        ret = self.schedule.job_status(job_name)
        self.assertNotIn("_last_run", ret)
        self.assertEqual(ret["_skip_reason"], "skip_explicit")
        self.assertEqual(ret["_skipped_time"], run_time)

        # Run 11/29/2017 at 5pm
        run_time = dateutil.parser.parse("11/29/2017 5:00pm")
        self.schedule.eval(now=run_time)
        ret = self.schedule.job_status(job_name)
        self.assertEqual(ret["_last_run"], run_time)

    @slowTest
    def test_skip_during_range(self):
        """
        verify that scheduled job is skipped during the specified range
        """
        job_name = "test_skip_during_range"
        job = {
            "schedule": {
                job_name: {
                    "function": "test.ping",
                    "hours": "1",
                    "skip_during_range": {
                        "start": "11/29/2017 2pm",
                        "end": "11/29/2017 3pm",
                    },
                }
            }
        }

        # Add job to schedule
        self.schedule.opts.update(job)

        # eval at 1:30pm to prime.
        run_time = dateutil.parser.parse("11/29/2017 1:30pm")
        self.schedule.eval(now=run_time)
        ret = self.schedule.job_status(job_name)

        # eval at 2:30pm, will not run during range.
        run_time = dateutil.parser.parse("11/29/2017 2:30pm")
        self.schedule.eval(now=run_time)
        ret = self.schedule.job_status(job_name)
        self.assertNotIn("_last_run", ret)
        self.assertEqual(ret["_skip_reason"], "in_skip_range")
        self.assertEqual(ret["_skipped_time"], run_time)

        # eval at 3:30pm, will run.
        run_time = dateutil.parser.parse("11/29/2017 3:30pm")
        self.schedule.eval(now=run_time)
        ret = self.schedule.job_status(job_name)
        self.assertEqual(ret["_last_run"], run_time)

    def test_skip_during_range_invalid_datestring(self):
        """
        verify that scheduled job is not not and returns the right error string
        """
        run_time = dateutil.parser.parse("11/29/2017 2:30pm")

        job_name1 = "skip_during_range_invalid_datestring1"
        job1 = {
            "schedule": {
                job_name1: {
                    "function": "test.ping",
                    "hours": "1",
                    "_next_fire_time": run_time,
                    "skip_during_range": {"start": "25pm", "end": "3pm"},
                }
            }
        }

        job_name2 = "skip_during_range_invalid_datestring2"
        job2 = {
            "schedule": {
                job_name2: {
                    "function": "test.ping",
                    "hours": "1",
                    "_next_fire_time": run_time,
                    "skip_during_range": {"start": "2pm", "end": "25pm"},
                }
            }
        }

        # Add job1 to schedule
        self.schedule.opts.update(job1)

        # Eval
        self.schedule.eval(now=run_time)

        # Check the first job
        ret = self.schedule.job_status(job_name1)
        _expected = (
            "Invalid date string for start in "
            "skip_during_range. Ignoring "
            "job {0}."
        ).format(job_name1)
        self.assertEqual(ret["_error"], _expected)

        # Clear out schedule
        self.schedule.opts["schedule"] = {}

        # Add job2 to schedule
        self.schedule.opts.update(job2)

        # Eval
        self.schedule.eval(now=run_time)

        # Check the second job
        ret = self.schedule.job_status(job_name2)
        _expected = (
            "Invalid date string for end in " "skip_during_range. Ignoring " "job {0}."
        ).format(job_name2)
        self.assertEqual(ret["_error"], _expected)

    @slowTest
    def test_skip_during_range_global(self):<commit_msg>Drop Py2 and six on tests/unit/utils/scheduler/test_skip.py<commit_after>import logging

from tests.support.helpers import slowTest
from tests.support.unit import skipIf
from tests.unit.utils.scheduler.base import SchedulerTestsBase

try:
    import dateutil.parser

    HAS_DATEUTIL_PARSER = True
except ImportError:
    HAS_DATEUTIL_PARSER = False

log = logging.getLogger(__name__)


@skipIf(
    HAS_DATEUTIL_PARSER is False, "The 'dateutil.parser' library is not available",
)
class SchedulerSkipTest(SchedulerTestsBase):
    """
    Validate the pkg module
    """

    def setUp(self):
        super().setUp()
        self.schedule.opts["loop_interval"] = 1

    def tearDown(self):
        self.schedule.reset()
        super().tearDown()

    @slowTest
    def test_skip(self):
        """
        verify that scheduled job is skipped at the specified time
        """
        job_name = "test_skip"
        job = {
            "schedule": {
                job_name: {
                    "function": "test.ping",
                    "when": ["11/29/2017 4pm", "11/29/2017 5pm"],
                }
            }
        }

        # Add job to schedule
        self.schedule.opts.update(job)

        run_time = dateutil.parser.parse("11/29/2017 4:00pm")
        self.schedule.skip_job(
            job_name,
            {
                "time": run_time.strftime("%Y-%m-%dT%H:%M:%S"),
                "time_fmt": "%Y-%m-%dT%H:%M:%S",
            },
        )

        # Run 11/29/2017 at 4pm
        self.schedule.eval(now=run_time)
        ret = self.schedule.job_status(job_name)
        self.assertNotIn("_last_run", ret)
        self.assertEqual(ret["_skip_reason"], "skip_explicit")
        self.assertEqual(ret["_skipped_time"], run_time)

        # Run 11/29/2017 at 5pm
        run_time = dateutil.parser.parse("11/29/2017 5:00pm")
        self.schedule.eval(now=run_time)
        ret = self.schedule.job_status(job_name)
        self.assertEqual(ret["_last_run"], run_time)

    @slowTest
    def test_skip_during_range(self):
        """
        verify that scheduled job is skipped during the specified range
        """
        job_name = "test_skip_during_range"
        job = {
            "schedule": {
                job_name: {
                    "function": "test.ping",
                    "hours": "1",
                    "skip_during_range": {
                        "start": "11/29/2017 2pm",
                        "end": "11/29/2017 3pm",
                    },
                }
            }
        }

        # Add job to schedule
        self.schedule.opts.update(job)

        # eval at 1:30pm to prime.
        run_time = dateutil.parser.parse("11/29/2017 1:30pm")
        self.schedule.eval(now=run_time)
        ret = self.schedule.job_status(job_name)

        # eval at 2:30pm, will not run during range.
        run_time = dateutil.parser.parse("11/29/2017 2:30pm")
        self.schedule.eval(now=run_time)
        ret = self.schedule.job_status(job_name)
        self.assertNotIn("_last_run", ret)
        self.assertEqual(ret["_skip_reason"], "in_skip_range")
        self.assertEqual(ret["_skipped_time"], run_time)

        # eval at 3:30pm, will run.
        run_time = dateutil.parser.parse("11/29/2017 3:30pm")
        self.schedule.eval(now=run_time)
        ret = self.schedule.job_status(job_name)
        self.assertEqual(ret["_last_run"], run_time)

    def test_skip_during_range_invalid_datestring(self):
        """
        verify that scheduled job is not not and returns the right error string
        """
        run_time = dateutil.parser.parse("11/29/2017 2:30pm")

        job_name1 = "skip_during_range_invalid_datestring1"
        job1 = {
            "schedule": {
                job_name1: {
                    "function": "test.ping",
                    "hours": "1",
                    "_next_fire_time": run_time,
                    "skip_during_range": {"start": "25pm", "end": "3pm"},
                }
            }
        }

        job_name2 = "skip_during_range_invalid_datestring2"
        job2 = {
            "schedule": {
                job_name2: {
                    "function": "test.ping",
                    "hours": "1",
                    "_next_fire_time": run_time,
                    "skip_during_range": {"start": "2pm", "end": "25pm"},
                }
            }
        }

        # Add job1 to schedule
        self.schedule.opts.update(job1)

        # Eval
        self.schedule.eval(now=run_time)

        # Check the first job
        ret = self.schedule.job_status(job_name1)
        _expected = (
            "Invalid date string for start in " "skip_during_range. Ignoring " "job {}."
        ).format(job_name1)
        self.assertEqual(ret["_error"], _expected)

        # Clear out schedule
        self.schedule.opts["schedule"] = {}

        # Add job2 to schedule
        self.schedule.opts.update(job2)

        # Eval
        self.schedule.eval(now=run_time)

        # Check the second job
        ret = self.schedule.job_status(job_name2)
        _expected = (
            "Invalid date string for end in " "skip_during_range. Ignoring " "job {}."
        ).format(job_name2)
        self.assertEqual(ret["_error"], _expected)

    @slowTest
    def test_skip_during_range_global(self):
{'fraction_non_alphanumeric': 0.1169568861160257, 'fraction_numerical': 0.029239221529006425, 'mean_word_length': 1.800521512385919, 'pattern_counts': {'":': 50, '<': 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': '7646027', 'n_tokens_mistral': 3532, 'n_tokens_neox': 3213, 'n_words': 617}