Logo Search packages:      
Sourcecode: ubuntuone-client version File versions

test_hashqueue.py

#
# Author: Facundo Batista <facundo@canonical.com>
#
# Copyright 2009 Canonical Ltd.
#
# This program is free software: you can redistribute it and/or modify it
# under the terms of the GNU General Public License version 3, as published
# by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranties of
# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
# PURPOSE.  See the GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program.  If not, see <http://www.gnu.org/licenses/>.
'''Tests the Hashs Queue.'''

from __future__ import with_statement

import os
import shutil
import unittest
import Queue
import random
import time

from twisted.trial.unittest import TestCase as TwistedTestCase
from twisted.internet import defer

from ubuntuone.syncdaemon import hash_queue
from ubuntuone.storageprotocol.hash import \
            content_hash_factory, crc32

TESTS_DIR = os.path.join(os.getcwd(), "tmp")

00037 class HasherTests(TwistedTestCase):
    '''Test the whole stuff to receive signals.'''

00040     def setUp(self):
        '''Setup the test.'''
        try:
            os.mkdir(TESTS_DIR)
        except OSError:
            # already there, remove it to clean and create again
            shutil.rmtree(TESTS_DIR)
            os.mkdir(TESTS_DIR)

        self.timeout = 2

00051     def tearDown(self):
        '''Clean up the tests.'''
        shutil.rmtree(TESTS_DIR)

00055     def test_live_process(self):
        '''Check that the hasher lives and dies.'''
        # create the hasher
        mark = object()
        queue = Queue.Queue()
        class Helper(object):
            '''helper class'''
            def push(self, *args):
                '''callback'''
        receiver = Helper()
        hasher = hash_queue._Hasher(queue, mark, receiver)
        hasher.start()

        # it's aliveeeeeeee!
        self.assertTrue(hasher.isAlive())

        # stop it, and release the processor to let the other thread run
        queue.put(mark)
        time.sleep(.1)

        # "I see dead threads"
        self.assertFalse(hasher.isAlive())


00079     def test_called_back(self):
        '''Tests that the hasher produces correct info.'''
        # create the hasher
        mark = object()
        queue = Queue.Queue()
        d = defer.Deferred()
        class Helper(object):
            '''helper class'''
            def push(self, *args):
                '''callback'''
                d.callback(True)
        receiver = Helper()
        hasher = hash_queue._Hasher(queue, mark, receiver)
        hasher.start()

        # send what to hash
        testfile = os.path.join(TESTS_DIR, "testfile")
        with open(testfile, "w") as fh:
            fh.write("foobar")
        queue.put(testfile)

        # calculate what we should receive
        realh = content_hash_factory()
        realh.hash_object.update("foobar")
        should_be = realh.content_hash()

        # release the processor and check
        return d

00108     def test_order(self):
        '''The hasher should return in order.'''
        # calculate what we should receive
        should_be = []
        for i in range(10):
            hasher = content_hash_factory()
            text = "supercalifragilistico"+str(i)
            hasher.hash_object.update(text)
            tfile = os.path.join(TESTS_DIR, "tfile"+str(i))
            with open(tfile, "w") as fh:
                fh.write("supercalifragilistico"+str(i))
            should_be.append(("HQ_HASH_NEW", tfile, hasher.content_hash(),
                              crc32(text), len(text), os.stat(tfile)))

        # create the hasher
        mark = object()
        queue = Queue.Queue()
        d = defer.Deferred()

        class Helper(object):
            '''helper class'''
            # class-closure, cannot use self, pylint: disable-msg=E0213
            def __init__(innerself):
                innerself.store = []
            def push(innerself, *args):
                '''callback'''
                innerself.store.append(args)
                if len(innerself.store) == 10:
                    if innerself.store == should_be:
                        d.callback(True)
                    else:
                        d.errback(Exception("are different!"))
        receiver = Helper()

        hasher = hash_queue._Hasher(queue, mark, receiver)
        hasher.start()

        # send what to hash
        for i in range(10):
            tfile = os.path.join(TESTS_DIR, "tfile"+str(i))
            queue.put(tfile)
        return d

00151     def test_large_content(self):
        '''The hasher works ok for a lot of info.'''
        # calculate what we should receive
        should_be = []
        testinfo = "".join(chr(random.randint(0, 255)) for i in range(100000))
        hasher = content_hash_factory()
        hasher.hash_object.update(testinfo)
        testfile = os.path.join(TESTS_DIR, "testfile")
        testhash = hasher.content_hash()

        # create the hasher
        mark = object()
        queue = Queue.Queue()
        d = defer.Deferred()

        class Helper(object):
            '''helper class'''
            def push(self, event, path, newhash, crc, size, stat):
                '''callback'''
                if event != "HQ_HASH_NEW":
                    d.errback(Exception("envent is not HQ_HASH_NEW"))
                elif path != testfile:
                    d.errback(Exception("path is not the original one"))
                elif newhash != testhash:
                    d.errback(Exception("the hashes are different!"))
                else:
                    d.callback(True)
        receiver = Helper()

        hasher = hash_queue._Hasher(queue, mark, receiver)
        hasher.start()

        # send what to hash
        with open(testfile, "w") as fh:
            fh.write(testinfo)
        queue.put(testfile)
        return d


00190 class HashQueueTests(TwistedTestCase):
    '''Test the whole stuff to receive signals.'''

00193     def setUp(self):
        '''Setup the test.'''
        try:
            os.mkdir(TESTS_DIR)
        except OSError:
            # already there, remove it to clean and create again
            shutil.rmtree(TESTS_DIR)
            os.mkdir(TESTS_DIR)

        self.timeout = 2

00204     def tearDown(self):
        '''Clean up the tests.'''
        shutil.rmtree(TESTS_DIR)

00208     def test_called_back(self):
        '''Tests that the hasher produces correct info.'''
        # create the hasher
        d = defer.Deferred()
        class Helper(object):
            '''helper class'''
            def push(self, *args):
                '''callback'''
                d.callback(True)
        receiver = Helper()
        hq = hash_queue.HashQueue(receiver)

        # send what to hash
        testfile = os.path.join(TESTS_DIR, "testfile")
        with open(testfile, "w") as fh:
            fh.write("foobar")
        hq.insert(testfile)

        # calculate what we should receive
        realh = content_hash_factory()
        realh.hash_object.update("foobar")
        should_be = realh.content_hash()
        return d

00232     def test_order(self):
        '''The hasher should return in order.'''
        # calculate what we should receive
        should_be = []
        for i in range(10):
            hasher = content_hash_factory()
            text = "supercalifragilistico"+str(i)
            hasher.hash_object.update(text)
            tfile = os.path.join(TESTS_DIR, "tfile"+str(i))
            with open(tfile, "w") as fh:
                fh.write("supercalifragilistico"+str(i))
            should_be.append(("HQ_HASH_NEW", tfile, hasher.content_hash(),
                              crc32(text), len(text), os.stat(tfile)))

        d = defer.Deferred()
        class Helper(object):
            '''helper class'''
            # class-closure, cannot use self, pylint: disable-msg=E0213
            def __init__(innerself):
                innerself.store = []
            def push(innerself, *args):
                '''callback'''
                innerself.store.append(args)
                if len(innerself.store) == 10:
                    if innerself.store[:-1] == should_be[:-1]:
                        d.callback(True)
                    else:
                        d.errback(Exception("are different! "))
        receiver = Helper()

        hq = hash_queue.HashQueue(receiver)

        # send what to hash
        for i in range(10):
            tfile = os.path.join(TESTS_DIR, "tfile"+str(i))
            hq.insert(tfile)
        return d


def test_suite():
    # pylint: disable-msg=C0111
    return unittest.TestLoader().loadTestsFromName(__name__)

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

Generated by  Doxygen 1.6.0   Back to index