lelo / tests / test.py

# coding: utf-8
# Author: João S. O. Bueno

import sys, os
sys.path.insert(0, "..")

from lelo import parallel
import unittest

class Tests(unittest.TestCase):
    def test_can_import(self):

    def test_can_execute_func_out_of_process(self):
        from multiprocessing import Queue, Process
        from lelo._lelo import _xecuter
        queue = Queue(1)

        def sum_(a, b):
            return a + b

        proc  = Process(target=_xecuter,
            args=(queue, sum_, (2, 3), {}))

        self.assertEqual(queue.get(), 5)

    def test_proc_wrapper_works(self):

        from lelo._lelo import Wrapper, LazyCallWrapper
        def sum_(a,b):
            return a + b

        wrapped = Wrapper(sum_)
        result = wrapped(2,3)
        self.assertEqual(type(result), LazyCallWrapper)
        self.assertEqual(result._value, 5)

    def test_lazy_class_factory(self):
        from lelo._lelo import MetaParallel
        X = MetaParallel("X", (object,), {})
        x = X()
        # check for operator access and work
        object.__setattr__(x, "_value", 10)
        self.assertEqual(x + 0, 10)
        # check for attribute access
        object.__setattr__(x, "_value", {"y": 10})
        self.assertEqual(list(x.keys()), ["y"])

    def test_parallel_function_creation(self):
        from types import MethodType
        def soma(a, b):
            return a + b
        x = getattr(soma, "__call__")
        self.assertTrue(x.__class__, MethodType)

    def test_parallel_execution(self):
        def soma(a, b):
            return a + b
        result = soma(2,3)
        #print result.__repr__
        self.assertEqual(result, 5)

    def test_async_exec(self):
        import time
            from urllib import urlopen
        except ImportError:
            from urllib.request import urlopen

        url = "http://example.com"

        def retr_html(url):
            return urlopen(url).read()

        fast = parallel(retr_html)
        t0 = time.time()
        res_0 = fast(url)
        t0 = time.time() - t0

        t1 = time.time()
        res_1 = retr_html(url)
        t1 = time.time() - t1
        #print "sync: %f, async: %f" % (t1, t0)

        self.assertTrue(t0 < t1)
        self.assertEqual(len(res_0), len(res_1))

    def test_async_iterator(self):
        from types import GeneratorType
        def fibonacci(limit):
            n = 1
            n1 = 0
            counter = 0
            while counter < limit:
                n, n1 = n1, n + n1
                counter += 1
                yield n1

        self.assertEqual(list(fibonacci(10)), [1, 1, 2, 3, 5, 8, 13, 21, 34, 55])

if __name__ == "__main__":

Tip: Filter by directory path e.g. /media app.js to search for public/media/app.js.
Tip: Use camelCasing e.g. ProjME to search for ProjectModifiedEvent.java.
Tip: Filter by extension type e.g. /repo .js to search for all .js files in the /repo directory.
Tip: Separate your search with spaces e.g. /ssh pom.xml to search for src/ssh/pom.xml.
Tip: Use ↑ and ↓ arrow keys to navigate and return to view the file.
Tip: You can also navigate files with Ctrl+j (next) and Ctrl+k (previous) and view the file with Ctrl+o.
Tip: You can also navigate files with Alt+j (next) and Alt+k (previous) and view the file with Alt+o.