<commit_before>import forge
import functools
from unittest import TestCase
from flux import timeline as timeline_module
from flux.sequence import Sequence

class TimelineTestBase(TestCase):
    def setUp(self):
        super(TimelineTestBase, self).setUp()
        self.timeline = timeline_module.Timeline()
        self.timeline.set_time_factor(0)

class TimelineAPITest(TimelineTestBase):
    def test__time_does_not_progress(self):
        self.assertEquals(self.timeline.time(), self.timeline.time())
    def test__sleep(self):
        start_time = self.timeline.time()
        sleep_time = 6.5
        self.timeline.sleep(sleep_time)
        self.assertEquals(self.timeline.time(), start_time + sleep_time)
    def test__sleep_wait_all_scheduled(self):
        start_time = self.timeline.time()
        self.timeline.schedule_callback(100, setattr, self, "called1", True)
        self.timeline.schedule_callback(200, setattr, self, "called2", True)
        self.timeline.sleep_wait_all_scheduled()
        self.assertEquals(self.timeline.time(), start_time + 200)
        self.assertTrue(self.called1)
        self.assertTrue(self.called2)
    def test__cannot_sleep_negative_seconds(self):
        with self.assertRaises(ValueError):
            self.timeline.sleep(-0.1)
    def test__cannot_sleep_negative_time_factor(self):
        with self.assertRaises(ValueError):
            self.timeline.set_time_factor(-0.1)
    def test__cannot_schedule_negative_delay(self):
        with self.assertRaises(ValueError):
            self.timeline.schedule_callback(-0.1, lambda : None)
    def test__cannot_set_past_time(self):
        self.timeline.set_time(self.timeline.time() + 10)
        with self.assertRaises(ValueError):
            self.timeline.set_time(self.timeline.time() - 1)

class TimeFactorTest(TestCase):
    def setUp(self):
        super(TimeFactorTest, self).setUp()
        self._time = 1337.0
        self.forge = forge.Forge()
        self.forge.replace_with(timeline_module, "_real_time", self.time)
        self.forge.replace_with(timeline_module, "_real_sleep", self.sleep)
        self.timeline = timeline_module.Timeline()
    def tearDown(self):
        self.forge.restore_all_replacements()
        self.forge.verify()
        self.forge.reset()
        super(TimeFactorTest, self).tearDown()
    def time(self):
        return self._time
    def sleep(self, seconds):
        assert seconds >= 0
        self._time += seconds
    def test__default_factor(self):
        start_time = self.timeline.time()
        self.sleep(10)
        self.assertEquals(self.timeline.time(), start_time + 10)
    def test__factor_change(self):
        expected_time = self.timeline.time()
        for index, (factor, sleep) in enumerate([
            (1, 20),
            (2.5, 3),
            (0.5, 70),
            ]):
            vtime_before_factor_change = self.timeline.time()
            self.timeline.set_time_factor(factor)
            vtime_after_factor_change = self.timeline.time()
            self.assertEquals(vtime_before_factor_change, vtime_after_factor_change, "set_time_factor() unexpectedly changed virtual time!")<commit_msg>Fix sleep behavior<commit_after>import forge
import functools
from unittest import TestCase
from flux import timeline as timeline_module
from flux.sequence import Sequence

class TimelineTestBase(TestCase):
    def setUp(self):
        super(TimelineTestBase, self).setUp()
        self.timeline = timeline_module.Timeline()
        self.timeline.set_time_factor(0)

class TimelineAPITest(TimelineTestBase):
    def test__time_does_not_progress(self):
        self.assertEquals(self.timeline.time(), self.timeline.time())
    def test__sleep(self):
        start_time = self.timeline.time()
        sleep_time = 6.5
        self.timeline.sleep(sleep_time)
        self.assertEquals(self.timeline.time(), start_time + sleep_time)
    def test__sleep_wait_all_scheduled(self):
        start_time = self.timeline.time()
        self.timeline.schedule_callback(100, setattr, self, "called1", True)
        self.timeline.schedule_callback(200, setattr, self, "called2", True)
        self.timeline.sleep_wait_all_scheduled()
        self.assertEquals(self.timeline.time(), start_time + 200)
        self.assertTrue(self.called1)
        self.assertTrue(self.called2)
    def test__cannot_sleep_negative_seconds(self):
        with self.assertRaises(ValueError):
            self.timeline.sleep(-0.1)
    def test__cannot_sleep_negative_time_factor(self):
        with self.assertRaises(ValueError):
            self.timeline.set_time_factor(-0.1)
    def test__cannot_schedule_negative_delay(self):
        with self.assertRaises(ValueError):
            self.timeline.schedule_callback(-0.1, lambda : None)
    def test__cannot_set_past_time(self):
        self.timeline.set_time(self.timeline.time() + 10)
        with self.assertRaises(ValueError):
            self.timeline.set_time(self.timeline.time() - 1)

class TimeFactorTest(TestCase):
    def setUp(self):
        super(TimeFactorTest, self).setUp()
        self._time = 1337.0
        self._callback_calls = []
        self.forge = forge.Forge()
        self.forge.replace_with(timeline_module, "_real_time", self.time)
        self.forge.replace_with(timeline_module, "_real_sleep", self.sleep)
        self.timeline = timeline_module.Timeline()
    def tearDown(self):
        self.forge.restore_all_replacements()
        self.forge.verify()
        self.forge.reset()
        super(TimeFactorTest, self).tearDown()
    def callback(self):
        self._callback_calls.append(self.timeline.time())
    def time(self):
        return self._time
    def sleep(self, seconds):
        assert seconds >= 0
        self._time += seconds
    def test__default_factor(self):
        start_time = self.timeline.time()
        self.sleep(10)
        self.assertEquals(self.timeline.time(), start_time + 10)
    def test__scheduled_callbacks(self):
        start_time = self.timeline.time()
        schedule_delay = 5
        schedule_time = start_time + schedule_delay
        self.timeline.schedule_callback(schedule_delay, self.callback)
        self.sleep(10)
        # need to trigger the callbacks
        self.timeline.sleep(0)
        self.assertEquals(self.timeline.time(), start_time + 10)
        self.assertEquals(self._callback_calls, [schedule_time])
    def test__factor_change(self):
        expected_time = self.timeline.time()
        for index, (factor, sleep) in enumerate([
            (1, 20),
            (2.5, 3),
            (0.5, 70),
            ]):
            vtime_before_factor_change = self.timeline.time()
            self.timeline.set_time_factor(factor)
            vtime_after_factor_change = self.timeline.time()
            self.assertEquals(vtime_before_factor_change, vtime_after_factor_change, "set_time_factor() unexpectedly changed virtual time!")
{'fraction_non_alphanumeric': 0.1046073601854535, 'fraction_numerical': 0.013909011880614315, 'mean_word_length': 3.008710801393728, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, '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': '5968122', 'n_tokens_mistral': 2161, 'n_tokens_neox': 2078, 'n_words': 350}