<reponame>kabochya/memq
'''
Created on Feb 3, 2021

@author: ambudsharma
'''
import unittest
from io import BytesIO
from memq.inputstream import DataInputStream
import base64
from memq.memqlogmessageiterator import Deserializer, MemqLogMessage,\
    MemqLogMessageIterator

class Test(unittest.TestCase):


    def setUp(self):
        pass


    def tearDown(self):
        pass


    def test_uncompressed_batch_iterator(self):
        # raw base64 uncompressed Batch bytes from Java client
        base64str = "<KEY>"
        content = base64.b64decode(base64str)
        test = BytesIO(content)
        test.seek(0)
        stream = DataInputStream(test)
        
        header_deserializer = Deserializer()
        value_deserializer = Deserializer()
        curr_notification_obj = { 
            MemqLogMessage.INTERNAL_FIELD_TOPIC:"test",
            MemqLogMessage.INTERNAL_FIELD_OBJECT_SIZE:content.__sizeof__(),
            MemqLogMessage.INTERNAL_FIELD_NOTIFICATION_PARTITION_ID:0,
            MemqLogMessage.INTERNAL_FIELD_NOTIFICATION_READ_TIMESTAMP:0,
            MemqLogMessage.INTERNAL_FIELD_NOTIFICATION_PARTITION_OFFSET:0  
        }
        itr = MemqLogMessageIterator("test", stream, curr_notification_obj, header_deserializer, value_deserializer)
        
        counter = 0
        while itr.has_next():
            message = itr.next()
            counter += 1
            assert "test1231231" == message.value.decode('ascii')
        
        assert counter == 4, "There should be 4 messages read from this batch"

    def test_gzip_batch_iterator(self):
        base64str = "<KEY>
        content = base64.b64decode(base64str)
        test = BytesIO(content)
        stream = DataInputStream(test)
        
        header_deserializer = Deserializer()
        value_deserializer = Deserializer()
        curr_notification_obj = { 
            MemqLogMessage.INTERNAL_FIELD_TOPIC:"test",
            MemqLogMessage.INTERNAL_FIELD_OBJECT_SIZE:content.__sizeof__(),
            MemqLogMessage.INTERNAL_FIELD_NOTIFICATION_PARTITION_ID:0,
            MemqLogMessage.INTERNAL_FIELD_NOTIFICATION_READ_TIMESTAMP:0,
            MemqLogMessage.INTERNAL_FIELD_NOTIFICATION_PARTITION_OFFSET:0  
        }
        itr = MemqLogMessageIterator("test", stream, curr_notification_obj, header_deserializer, value_deserializer)
        
        counter = 0
        while itr.has_next():
            message = itr.next()
            counter += 1
            assert "test1231231" == message.value.decode('ascii')
        
        assert counter == 6, "There should be 6 messages read from this batch"
        
    def test_zstd_batch_iterator(self):
        base64str = "<KEY>
        content = base64.b64decode(base64str)
        test = BytesIO(content)
        stream = DataInputStream(test)
        
        header_deserializer = Deserializer()
        value_deserializer = Deserializer()
        curr_notification_obj = { 
            MemqLogMessage.INTERNAL_FIELD_TOPIC:"test",
            MemqLogMessage.INTERNAL_FIELD_OBJECT_SIZE:content.__sizeof__(),
            MemqLogMessage.INTERNAL_FIELD_NOTIFICATION_PARTITION_ID:0,
            MemqLogMessage.INTERNAL_FIELD_NOTIFICATION_READ_TIMESTAMP:0,
            MemqLogMessage.INTERNAL_FIELD_NOTIFICATION_PARTITION_OFFSET:0  
        }
        itr = MemqLogMessageIterator("test", stream, curr_notification_obj, header_deserializer, value_deserializer)
        
        counter = 0
        while itr.has_next():
            message = itr.next()
            counter += 1
            assert "test1231231" == message.value.decode('ascii')
        
        assert counter == 10, "There should be 10 messages read from this batch"

if __name__ == "__main__":
    #import sys;sys.argv = ['', 'Test.testName']
    unittest.main()
'''
Created on Feb 3, 2021

@author: ambudsharma
'''
import unittest
from io import BytesIO
from memq.inputstream import DataInputStream
import base64
from memq.memqlogmessageiterator import Deserializer, MemqLogMessage,\
    MemqLogMessageIterator

class Test(unittest.TestCase):


    def setUp(self):
        pass


    def tearDown(self):
        pass


    def test_uncompressed_batch_iterator(self):
        # raw base64 uncompressed Batch bytes from Java client
        base64str = "<KEY>"
        content = base64.b64decode(base64str)
        test = BytesIO(content)
        test.seek(0)
        stream = DataInputStream(test)
        
        header_deserializer = Deserializer()
        value_deserializer = Deserializer()
        curr_notification_obj = { 
            MemqLogMessage.INTERNAL_FIELD_TOPIC:"test",
            MemqLogMessage.INTERNAL_FIELD_OBJECT_SIZE:content.__sizeof__(),
            MemqLogMessage.INTERNAL_FIELD_NOTIFICATION_PARTITION_ID:0,
            MemqLogMessage.INTERNAL_FIELD_NOTIFICATION_READ_TIMESTAMP:0,
            MemqLogMessage.INTERNAL_FIELD_NOTIFICATION_PARTITION_OFFSET:0  
        }
        itr = MemqLogMessageIterator("test", stream, curr_notification_obj, header_deserializer, value_deserializer)
        
        counter = 0
        while itr.has_next():
            message = itr.next()
            counter += 1
            assert "test1231231" == message.value.decode('ascii')
        
        assert counter == 4, "There should be 4 messages read from this batch"

    def test_gzip_batch_iterator(self):
        base64str = "<KEY>
        content = base64.b64decode(base64str)
        test = BytesIO(content)
        stream = DataInputStream(test)
        
        header_deserializer = Deserializer()
        value_deserializer = Deserializer()
        curr_notification_obj = { 
            MemqLogMessage.INTERNAL_FIELD_TOPIC:"test",
            MemqLogMessage.INTERNAL_FIELD_OBJECT_SIZE:content.__sizeof__(),
            MemqLogMessage.INTERNAL_FIELD_NOTIFICATION_PARTITION_ID:0,
            MemqLogMessage.INTERNAL_FIELD_NOTIFICATION_READ_TIMESTAMP:0,
            MemqLogMessage.INTERNAL_FIELD_NOTIFICATION_PARTITION_OFFSET:0  
        }
        itr = MemqLogMessageIterator("test", stream, curr_notification_obj, header_deserializer, value_deserializer)
        
        counter = 0
        while itr.has_next():
            message = itr.next()
            counter += 1
            assert "test1231231" == message.value.decode('ascii')
        
        assert counter == 6, "There should be 6 messages read from this batch"
        
    def test_zstd_batch_iterator(self):
        base64str = "<KEY>
        content = base64.b64decode(base64str)
        test = BytesIO(content)
        stream = DataInputStream(test)
        
        header_deserializer = Deserializer()
        value_deserializer = Deserializer()
        curr_notification_obj = { 
            MemqLogMessage.INTERNAL_FIELD_TOPIC:"test",
            MemqLogMessage.INTERNAL_FIELD_OBJECT_SIZE:content.__sizeof__(),
            MemqLogMessage.INTERNAL_FIELD_NOTIFICATION_PARTITION_ID:0,
            MemqLogMessage.INTERNAL_FIELD_NOTIFICATION_READ_TIMESTAMP:0,
            MemqLogMessage.INTERNAL_FIELD_NOTIFICATION_PARTITION_OFFSET:0  
        }
        itr = MemqLogMessageIterator("test", stream, curr_notification_obj, header_deserializer, value_deserializer)
        
        counter = 0
        while itr.has_next():
            message = itr.next()
            counter += 1
            assert "test1231231" == message.value.decode('ascii')
        
        assert counter == 10, "There should be 10 messages read from this batch"

if __name__ == "__main__":
    #import sys;sys.argv = ['', 'Test.testName']
    unittest.main()
Created on Feb 3, 2021  @author: ambudsharma # raw base64 uncompressed Batch bytes from Java client #import sys;sys.argv = ['', 'Test.testName']