diff options
author | Lauro Neto <lauro.neto@openbossa.org> | 2010-01-15 15:39:41 -0300 |
---|---|---|
committer | Marcelo Lira <marcelo.lira@openbossa.org> | 2010-01-15 20:12:25 -0300 |
commit | 72ca828dde351cc3b715eec5447fe4ea5c46ec01 (patch) | |
tree | e8cc0a754e3953fd629c432b9ac44eac22f7a8b5 /tests/samplebinding/event_loop_thread_test.py | |
parent | 007d8cf6955f0f99a41bad8c69c9da7789d6bbe8 (diff) |
Adding new tests related to threads/GIL locking
- Event loop - calling virtual methods from C++
- Event loop with thread - calling virtuals from C++ along with
accessing the binding from another python
thread
- Thread locking - blocker C++ method that is unlocked from another
python thread
For these tests, a new ObjectType subclass was added, Bucket, which
is just a container for the producer/consumer tests and has the
lock/unlock method.
Reviewed by Marcelo Lira <marcelo.lira@openbossa.org>
Diffstat (limited to 'tests/samplebinding/event_loop_thread_test.py')
-rwxr-xr-x | tests/samplebinding/event_loop_thread_test.py | 68 |
1 files changed, 68 insertions, 0 deletions
diff --git a/tests/samplebinding/event_loop_thread_test.py b/tests/samplebinding/event_loop_thread_test.py new file mode 100755 index 000000000..205c87930 --- /dev/null +++ b/tests/samplebinding/event_loop_thread_test.py @@ -0,0 +1,68 @@ +#!/usr/bin/env python + +import time +import threading +import unittest +from random import random + +from sample import ObjectType, Event + + +class Producer(ObjectType): + + def __init__(self): + ObjectType.__init__(self) + self.data = None + self.read = False + + def event(self, event): + self.data = random() + + while not self.read: + time.sleep(0.01) + + return True + + +class Collector(threading.Thread): + + def __init__(self, objects): + threading.Thread.__init__(self) + self.max_runs = len(objects) + self.objects = objects + self.data = [] + + def run(self): + i = 0 + while i < self.max_runs: + if self.objects[i].data is not None: + self.data.append(self.objects[i].data) + self.objects[i].read = True + i += 1 + time.sleep(0.01) + + +class TestEventLoopWithThread(unittest.TestCase): + '''Communication between a python thread and an simple + event loop in C++''' + + def testBasic(self): + '''Allowing threads and calling virtuals from C++''' + number = 10 + objs = [Producer() for x in range(number)] + thread = Collector(objs) + + thread.start() + + evaluated = ObjectType.processEvent(objs, + Event(Event.BASIC_EVENT)) + + thread.join() + + producer_data = [x.data for x in objs] + self.assertEqual(evaluated, number) + self.assertEqual(producer_data, thread.data) + + +if __name__ == '__main__': + unittest.main() |