<commit_before>#!/usr/bin/python3
# This file is part of the GOsa framework.
#
#  http://gosa-project.org
#
# Copyright:
#  (C) 2016 GONICUS GmbH, Germany, http://www.gonicus.de
#
# See the LICENSE file in the project's top-level directory for details.
from contextlib import contextmanager
from unittest import mock, TestCase
import pytest
import threading
from io import StringIO

import time
from gosa.common.components.scheduler.scheduler import *
from gosa.common.components.scheduler.events import *
from gosa.common.components.scheduler.jobstores.ram_store import *
from tests.helper import slow


class CallHandler:
    def __init__(self):
        self.lock = threading.Lock()
        self.i = 0

    def count(self):
        self.lock.acquire()
        self.i += 1
        self.lock.release()

    def get_count(self):
        return self.i


def process(handler):
    handler.count()


class SchedulerTestCase(TestCase):
    def test_SchedulerAlreadyRunningError(self):
        err = SchedulerAlreadyRunningError()
        assert str(err) == "Scheduler is already running"

    def test_configure(self):
        s = Scheduler()
        s.shutdown() # Does nothing
        
        gconfig = {"gosa.common.components.scheduler.jobstore.jobstoredefault.class": "gosa.common.components.scheduler.jobstores.ram_store:RAMJobStore"}
        options = {}
        
        s.configure(gconfig=gconfig)
        s.start()
        
        # If the jobstore "jobstoredefault" does not exist, an exception
        # will be raised.
        s.add_date_job(process, "2016-12-13", args=(mock.MagicMock(),), jobstore="jobstoredefault")
        
        with pytest.raises(SchedulerAlreadyRunningError):
            s.start()
        with pytest.raises(SchedulerAlreadyRunningError):
            s.configure()
        s.shutdown()
        
    def test_add_job(self):
        # Adding jobs before and after start()
        def dummy(): pass
        s = Scheduler()
        s.add_jobstore(RAMJobStore(), "default")
        with StringIO("") as buf:
            s.print_jobs(out=buf)
            assert buf.getvalue() == os.linesep.join("""Jobstore default:
    No scheduled jobs""".splitlines())+os.linesep
        trigger = SimpleTrigger("2016-12-12")
        j1 = s.add_job(trigger, dummy, (), {})
        assert isinstance(j1, Job)
        s.start()
        j2 = s.add_job(trigger, dummy, (), {})
        assert isinstance(j2, Job)
        assert len(s.get_jobs()) == 2
        with StringIO("") as buf:
            s.print_jobs(out=buf)
            assert buf.getvalue() == os.linesep.join("""Jobstore default:
    dummy (trigger: date[2016-12-12 00:00:00], next run at: 2016-12-12 00:00:00)
    dummy (trigger: date[2016-12-12 00:00:00], next run at: 2016-12-12 00:00:00)""".splitlines())+os.linesep
        s.shutdown()

    def test_jobstores(self):
        s = Scheduler()
        s.add_jobstore(RAMJobStore(), "ram1")
        with pytest.raises(KeyError):
            s.add_jobstore(RAMJobStore(), "ram1")
        s.remove_jobstore("ram1")
        with pytest.raises(KeyError):
            s.remove_jobstore("ram2")

    def test_listeners(self):
        s = Scheduler()
        with mock.patch.object(s, "_main_loop"):
            s.start()
            m_thread = mock.MagicMock()
            m_thread.isAlive.return_value = True
            s._thread = m_thread

            dummyCallback = mock.MagicMock()
            s.add_listener(dummyCallback)
            job = s.add_date_job(process, "2016-12-12", args=(mock.MagicMock(),))
            assert dummyCallback.called
            e = dummyCallback.call_args[0][0]
            assert isinstance(e, JobStoreEvent)
            assert e.alias == "default"
            assert e.job == job
            s.remove_listener(dummyCallback)
            s.shutdown()

    def test_get_job_by_id(self):
        s = Scheduler()
        s.start()
        job = s.add_date_job(process, "2016-12-12", args=(mock.MagicMock(),))
        assert s.get_job_by_id(job.uuid) == job
        assert s.get_job_by_id("unknown") == None
        s.shutdown()

    def test_unschedule_job(self):
        s = Scheduler()
        s.start()
        job = s.add_date_job(process, "2016-12-12", args=(mock.MagicMock(),))
        assert len(s.get_jobs()) == 1
        s.unschedule_job(job)
        assert len(s.get_jobs()) == 0
        with pytest.raises(KeyError):
            s.unschedule_job(mock.MagicMock())
        s.shutdown()

    def test_reschedule_date_job(self):

        now = datetime.datetime.now()
        with mock.patch("gosa.common.components.scheduler.scheduler.Event", wraps=Event) as m_event:
            m_event.return_value

            s = Scheduler()

            callback = mock.MagicMock()

            time1 = now + datetime.timedelta(seconds=1)
            time2 = now + datetime.timedelta(seconds=3)

            s.start()
            m_event.return_value.wait.assert_called_with()

            #m_datetime.datetime.now.return_value = now
            job = s.add_date_job(callback, time1)
            time.sleep(0.1)
            assert round(m_event.return_value.wait.call_args[0][0]) == 1

            s.reschedule_date_job(job, time2)
            time.sleep(1)
            assert not callback.called
            assert round(m_event.return_value.wait.call_args[0][0]) == 2


        s.shutdown()


    def test_unschedule_func(self):
        s = Scheduler()
        s.start()
        s.add_date_job(process, "2016-12-12", args=(mock.MagicMock(),))
        assert len(s.get_jobs()) == 1
        s.unschedule_func(process)
        assert len(s.get_jobs()) == 0
        with pytest.raises(KeyError):
            s.unschedule_func(mock.MagicMock())
        s.shutdown()

    def assert_jobs(self, s, handler, calls=1):<commit_msg>make test more robust<commit_after>#!/usr/bin/python3
# This file is part of the GOsa framework.
#
#  http://gosa-project.org
#
# Copyright:
#  (C) 2016 GONICUS GmbH, Germany, http://www.gonicus.de
#
# See the LICENSE file in the project's top-level directory for details.
from contextlib import contextmanager
from unittest import mock, TestCase
import pytest
import threading
from io import StringIO

import time
from gosa.common.components.scheduler.scheduler import *
from gosa.common.components.scheduler.events import *
from gosa.common.components.scheduler.jobstores.ram_store import *
from tests.helper import slow


class CallHandler:
    def __init__(self):
        self.lock = threading.Lock()
        self.i = 0

    def count(self):
        self.lock.acquire()
        self.i += 1
        self.lock.release()

    def get_count(self):
        return self.i


def process(handler):
    handler.count()


class SchedulerTestCase(TestCase):
    def test_SchedulerAlreadyRunningError(self):
        err = SchedulerAlreadyRunningError()
        assert str(err) == "Scheduler is already running"

    def test_configure(self):
        s = Scheduler()
        s.shutdown() # Does nothing
        
        gconfig = {"gosa.common.components.scheduler.jobstore.jobstoredefault.class": "gosa.common.components.scheduler.jobstores.ram_store:RAMJobStore"}
        options = {}
        
        s.configure(gconfig=gconfig)
        s.start()
        
        # If the jobstore "jobstoredefault" does not exist, an exception
        # will be raised.
        s.add_date_job(process, "2016-12-13", args=(mock.MagicMock(),), jobstore="jobstoredefault")
        
        with pytest.raises(SchedulerAlreadyRunningError):
            s.start()
        with pytest.raises(SchedulerAlreadyRunningError):
            s.configure()
        s.shutdown()
        
    def test_add_job(self):
        # Adding jobs before and after start()
        def dummy(): pass
        s = Scheduler()
        s.add_jobstore(RAMJobStore(), "default")
        with StringIO("") as buf:
            s.print_jobs(out=buf)
            assert buf.getvalue() == os.linesep.join("""Jobstore default:
    No scheduled jobs""".splitlines())+os.linesep
        trigger = SimpleTrigger("2016-12-12")
        j1 = s.add_job(trigger, dummy, (), {})
        assert isinstance(j1, Job)
        s.start()
        j2 = s.add_job(trigger, dummy, (), {})
        assert isinstance(j2, Job)
        assert len(s.get_jobs()) == 2
        with StringIO("") as buf:
            s.print_jobs(out=buf)
            assert buf.getvalue() == os.linesep.join("""Jobstore default:
    dummy (trigger: date[2016-12-12 00:00:00], next run at: 2016-12-12 00:00:00)
    dummy (trigger: date[2016-12-12 00:00:00], next run at: 2016-12-12 00:00:00)""".splitlines())+os.linesep
        s.shutdown()

    def test_jobstores(self):
        s = Scheduler()
        s.add_jobstore(RAMJobStore(), "ram1")
        with pytest.raises(KeyError):
            s.add_jobstore(RAMJobStore(), "ram1")
        s.remove_jobstore("ram1")
        with pytest.raises(KeyError):
            s.remove_jobstore("ram2")

    def test_listeners(self):
        s = Scheduler()
        with mock.patch.object(s, "_main_loop"):
            s.start()
            m_thread = mock.MagicMock()
            m_thread.isAlive.return_value = True
            s._thread = m_thread

            dummyCallback = mock.MagicMock()
            s.add_listener(dummyCallback)
            job = s.add_date_job(process, "2016-12-12", args=(mock.MagicMock(),))
            assert dummyCallback.called
            e = dummyCallback.call_args[0][0]
            assert isinstance(e, JobStoreEvent)
            assert e.alias == "default"
            assert e.job == job
            s.remove_listener(dummyCallback)
            s.shutdown()

    def test_get_job_by_id(self):
        s = Scheduler()
        s.start()
        job = s.add_date_job(process, "2016-12-12", args=(mock.MagicMock(),))
        assert s.get_job_by_id(job.uuid) == job
        assert s.get_job_by_id("unknown") == None
        s.shutdown()

    def test_unschedule_job(self):
        s = Scheduler()
        s.start()
        job = s.add_date_job(process, "2016-12-12", args=(mock.MagicMock(),))
        assert len(s.get_jobs()) == 1
        s.unschedule_job(job)
        assert len(s.get_jobs()) == 0
        with pytest.raises(KeyError):
            s.unschedule_job(mock.MagicMock())
        s.shutdown()

    def test_reschedule_date_job(self):

        now = datetime.datetime.now()
        with mock.patch("gosa.common.components.scheduler.scheduler.Event", wraps=Event) as m_event:
            m_event.return_value

            s = Scheduler()

            callback = mock.MagicMock()

            time1 = now + datetime.timedelta(seconds=1)

            s.start()
            time.sleep(0.1)
            m_event.return_value.wait.assert_called_with()

            job = s.add_date_job(callback, time1)
            time.sleep(0.1)
            assert round(m_event.return_value.wait.call_args[0][0]) == 1

            time2 = datetime.datetime.now() + datetime.timedelta(seconds=3)
            s.reschedule_date_job(job, time2)
            time.sleep(0.1)
            assert round(m_event.return_value.wait.call_args[0][0]) == 3

            # make sure that the callback really does not ge called on initial time
            time.sleep(1)
            assert not callback.called



        s.shutdown()


    def test_unschedule_func(self):
        s = Scheduler()
        s.start()
        s.add_date_job(process, "2016-12-12", args=(mock.MagicMock(),))
        assert len(s.get_jobs()) == 1
        s.unschedule_func(process)
        assert len(s.get_jobs()) == 0
        with pytest.raises(KeyError):
            s.unschedule_func(mock.MagicMock())
        s.shutdown()

    def assert_jobs(self, s, handler, calls=1):
{'fraction_non_alphanumeric': 0.12305055698371893, 'fraction_numerical': 0.024678663239074552, 'mean_word_length': 2.3137421919363996, 'pattern_counts': {'":': 2, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '936197', 'n_tokens_mistral': 3879, 'n_tokens_neox': 3618, 'n_words': 721}