<commit_before>import os
import time
import pickle
import unittest
import Queue

import sys; sys.path.append('..')
from __writer import Writer
from __message import Message


def get_queue(maxsize=1024):
    return Queue.Queue(maxsize=maxsize)


class TestWriter(unittest.TestCase):
    def test_periodic_scheduling_removing_from_queue(self):
        class MyWriter(Writer):
            def write(self, msg):
                return msg

        q = get_queue()
        q.put(Message(content=1))
        q.put(Message(content=2))

        conf = {'interval' : 1}
        mywriter = MyWriter(q, conf=conf)
        mywriter.start()

        #waits interval to write messages
        time.sleep(2)

        self.assertEqual(0, q.qsize())

    def test_processed_messages_counting(self):
        class MyWriter(Writer):
            def write(self, msg):
                return True

        q = get_queue()
        q.put(Message(content=1))
        q.put(Message(content=2))

        mywriter = MyWriter(q)
        mywriter.start()

        #force processing messages
        mywriter.process()
        mywriter.process()

        self.assertEqual(2, mywriter.processed)

    def test_checkpoint_saving(self):
        checkpoint_path = '/tmp/wcheckpoint'
        class MyWriter(Writer):
            def write(self, msg):
                return True

        q = get_queue()
        q.put(Message(content='foo', checkpoint='foo'))
        q.put(Message(content='bar', checkpoint='bar'))

        conf = {'checkpoint_enabled' : True,
                'checkpoint_path' : checkpoint_path,
                'checkpoint_interval' : 1}
        mywriter = MyWriter(q, conf=conf)
        mywriter.start()

        #should process message 1
        mywriter.process()

        #waits for checkpoint_interval
        time.sleep(2)

        self.assertEqual('foo', mywriter.last_checkpoint)

        #should process message 2
        mywriter.process()

        #waits for checkpoint_interval
        time.sleep(2)

        self.assertEqual(2, mywriter.processed)
        self.assertEqual('bar', mywriter.last_checkpoint)

        f = open(checkpoint_path, 'r+')
        self.assertEqual('bar', pickle.load(f))
        f.close()

        os.remove(checkpoint_path)

    def test_restore_last_checkpoint(self):
        checkpoint_path = '/tmp/wcheckpoint'
        f = open(checkpoint_path, 'w+')
        pickle.dump('foo', f)
        f.close()

        class MyWriter(Writer):
            def write(self, msg):
                return True

        q = get_queue()

        conf = {'checkpoint_enabled' : True,
                'checkpoint_path' : checkpoint_path}
        mywriter = MyWriter(q, conf=conf)
        self.assertEqual('foo', mywriter.last_checkpoint)

        os.remove(checkpoint_path)

    def test_store_discarded_messages_due_to_some_fail(self):
        class MyWriter(Writer):
            def write(self, msg):
                if msg == 'foo2' or msg == 'foo3':
                    return False
                return True

        q = get_queue(3)
        q.put(Message(content='foo1'))
        q.put(Message(content='foo2'))
        q.put(Message(content='foo3'))

        mywriter = MyWriter(q)
        mywriter.start()

        #force processing messages
        mywriter.process()
        mywriter.process()
        mywriter.process()

        self.assertEqual(2, mywriter.discarded)


if __name__ == "__main__":
    unittest.main()
<commit_msg>fixing test<commit_after>import os
import time
import pickle
import unittest
import Queue

import sys; sys.path.append('..')
from __writer import Writer
from __message import Message


def get_queue(maxsize=1024):
    return Queue.Queue(maxsize=maxsize)


class TestWriter(unittest.TestCase):
    def test_periodic_scheduling_removing_from_queue(self):
        class MyWriter(Writer):
            def write(self, msg):
                return msg

        q = get_queue()
        q.put(Message(content=1))
        q.put(Message(content=2))

        conf = {'interval' : 1}
        mywriter = MyWriter(q, conf=conf)
        mywriter.start()

        #waits interval to write messages
        time.sleep(2)

        self.assertEqual(0, q.qsize())

    def test_processed_messages_counting(self):
        class MyWriter(Writer):
            def write(self, msg):
                return True

        q = get_queue()
        q.put(Message(content=1))
        q.put(Message(content=2))

        mywriter = MyWriter(q)
        mywriter.start()

        #force processing messages
        mywriter.process()
        mywriter.process()

        self.assertEqual(2, mywriter.processed)

    def test_checkpoint_saving(self):
        checkpoint_path = '/tmp/wcheckpoint'
        class MyWriter(Writer):
            def write(self, msg):
                return True

        q = get_queue()
        q.put(Message(content='foo', checkpoint='foo'))
        q.put(Message(content='bar', checkpoint='bar'))

        conf = {'checkpoint_enabled' : True,
                'checkpoint_path' : checkpoint_path,
                'checkpoint_interval' : 1}
        mywriter = MyWriter(q, conf=conf)
        mywriter.start()

        #should process message 1
        mywriter.process()

        #waits for checkpoint_interval
        time.sleep(2)

        self.assertEqual('foo', mywriter.last_checkpoint)

        #should process message 2
        mywriter.process()

        #waits for checkpoint_interval
        time.sleep(2)

        self.assertEqual(2, mywriter.processed)
        self.assertEqual('bar', mywriter.last_checkpoint)

        f = open(checkpoint_path, 'r+')
        self.assertEqual('bar', pickle.load(f))
        f.close()

        os.remove(checkpoint_path)

    def test_restore_last_checkpoint(self):
        checkpoint_path = '/tmp/wcheckpoint'
        f = open(checkpoint_path, 'w+')
        pickle.dump('foo', f)
        f.close()

        class MyWriter(Writer):
            def write(self, msg):
                return True

        q = get_queue()

        conf = {'checkpoint_enabled' : True,
                'checkpoint_path' : checkpoint_path}
        mywriter = MyWriter(q, conf=conf)
        self.assertEqual('foo', mywriter.last_checkpoint)

        os.remove(checkpoint_path)

    def test_store_discarded_messages_due_to_some_fail(self):
        class MyWriter(Writer):
            def write(self, msg):
                if msg == 'foo2' or msg == 'foo3':
                    return False
                return True

        q = get_queue(3)
        q.put(Message(content='foo1'))
        q.put(Message(content='foo2'))
        q.put(Message(content='foo3'))

        mywriter = MyWriter(q, conf={'blockable': False})
        mywriter.start()

        #force processing messages
        mywriter.process()
        mywriter.process()
        mywriter.process()

        self.assertEqual(2, mywriter.discarded)


if __name__ == "__main__":
    unittest.main()

{'fraction_non_alphanumeric': 0.10851033087704089, 'fraction_numerical': 0.007224389539083947, 'mean_word_length': 2.1578467153284673, 'pattern_counts': {'":': 2, '<': 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': '4406808', 'n_tokens_mistral': 2100, 'n_tokens_neox': 1974, 'n_words': 411}