<commit_before># ***** BEGIN LICENSE BLOCK *****
#
# For copyright and licensing please refer to COPYING.
#
# ***** END LICENSE BLOCK *****

"""
Pika simplebuffer tests
"""

import os
import sys
import nose
sys.path.append('..')
sys.path.append(os.path.join('..', '..'))

import nose
import pika.simplebuffer


def test_simplebuffer():
    """
    >>> b = pika.simplebuffer.SimpleBuffer("abcdef")
    >>> b.read_and_consume(3)
    'abc'
    >>> b.write(None, '')
    >>> b.read(0)
    ''
    >>> repr(b)
    "<SimpleBuffer of 3 bytes, 6 total size, 'def'>"
    >>> str(b)
    "<SimpleBuffer of 3 bytes, 6 total size, 'def'>"
    >>> b.flush()
    >>> b.read(1)
    ''
    """

class TestBuffer(object):
    def test_init_with_data(self):
        buffer = pika.simplebuffer.SimpleBuffer('abc')
        nose.tools.eq_(buffer.buf.tell(), 3)

    def test_init_nodata(self):
        buffer = pika.simplebuffer.SimpleBuffer()
        nose.tools.eq_(buffer.buf.tell(), 0)

    def test_write_onestring(self):
        buffer = pika.simplebuffer.SimpleBuffer()
        buffer.write('abc')
        nose.tools.eq_(len(buffer), 3)

    def test_write_multistring(self):
        buffer = pika.simplebuffer.SimpleBuffer()
        buffer.write("abc", "def", "ghi", "jkl", "mno")
        nose.tools.eq_(len(buffer), 15)


if __name__ == "__main__":
    nose.runmodule(argv=['-v', '--with-coverage', '-s', '--with-doctest', '--doctest-tests'])
<commit_msg>simplebuffer_test no longer depends on nose. Also, minor refactoring and added a large number of unit tests. Removed a doctest which was obviated by added unit tests.<commit_after># ***** BEGIN LICENSE BLOCK *****
#
# For copyright and licensing please refer to COPYING.
#
# ***** END LICENSE BLOCK *****

"""
Pika simplebuffer tests
"""

import os
import sys
import unittest
from mock import Mock
sys.path.append('..')
sys.path.append(os.path.join('..', '..'))

import pika.simplebuffer

class TestBuffer(unittest.TestCase):
    def setUp(self):
        self.buffer = pika.simplebuffer.SimpleBuffer()

    def test_init_with_data(self):
        buffer = pika.simplebuffer.SimpleBuffer('abc')
        self.assertEqual(buffer.buf.tell(), 3)

    def test_init_nodata(self):
        self.assertEqual(self.buffer.buf.tell(), 0)

    def test_write_onestring(self):
        self.buffer.write('abc')
        self.assertEqual(len(self.buffer), 3)

    def test_write_multistring(self):
        self.buffer.write("abc", "def", "ghi", "jkl", "mno")
        self.assertEqual(len(self.buffer), 15)

    def test_write_multi_empty(self):
        self.buffer.write('abc', '', 'def', '')
        self.assertEqual(len(self.buffer), 6)

    def test_read(self):
        self.buffer.write('abc')
        res = self.buffer.read(1)

        self.assertEqual(res, 'a')
        self.assertEqual(len(self.buffer), 3)

    def test_read_zerosize(self):
        self.buffer.write('abc')
        res = self.buffer.read(0)

        self.assertEqual(res, '')
        self.assertEqual(len(self.buffer), 3)

    def test_read_nonesize(self):
        self.buffer.write('abcdef')
        res = self.buffer.read()

        self.assertEqual(res, 'abcdef')
        self.assertEqual(len(self.buffer), 6)

    def test_read_and_consume(self):
        self.buffer.write('abcdef')
        result = self.buffer.read_and_consume(3)

        self.assertEqual(result, 'abc')
        self.assertEqual(len(self.buffer), 3)

    def test_read_and_consume_nosize(self):
        """
        read_and_consume called w/ no size arg should raise a TypeError

        """
        self.assertRaises(TypeError, self.buffer.read_and_consume)

    def test_read_and_consume_toobig(self):
        """
        read_and_consume asserts that incoming 'size' arg is smaller than
        the buffer size.

        """
        self.buffer.write('abc')
        self.assertRaises(AssertionError, self.buffer.read_and_consume, 10)

    def test_send_to_socket(self):
        self.buffer.write('abc')
        sd = Mock(spec=['send'])
        sd.send = Mock(return_value=3)
        bytes = self.buffer.send_to_socket(sd)

        self.assertEqual(bytes, 3)
        self.assertEqual(self.buffer.buf.tell(), 3)
        self.assertEqual(self.buffer.offset, 3)
        self.assertEqual(len(self.buffer), 0)

    def test_flush(self):
        self.buffer.write('abc')
        self.assertEqual(len(self.buffer), 3)
        self.buffer.flush()

        self.assertEqual(len(self.buffer), 0)

    def test_nonzero(self):
        self.buffer.write('abc')
        self.assertTrue(self.buffer)
        self.buffer.flush()
        self.assertFalse(self.buffer)

    def test_str(self):
        self.buffer.write('abcdef')
        self.buffer.read_and_consume(3)
        self.assertEqual(str(self.buffer), "<SimpleBuffer of 3 bytes, 6 total size, 'def'>")





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

{'fraction_non_alphanumeric': 0.15095502156500307, 'fraction_numerical': 0.007599096323680428, 'mean_word_length': 2.896, 'pattern_counts': {'":': 2, '<': 6, '<?xml version=': 0, '>': 30, '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': '1341408', 'n_tokens_mistral': 1647, 'n_tokens_neox': 1568, 'n_words': 307}