pyobjc / pyobjc-core / PyObjCTest / test_locking.py

"""
Test locking objects and interaction with @synchronized() statements

These tests take an annoyingly long time to ensure that we'd hit a race condition when
locking doesn't actually lock. It should be possible to find a faster mechanism for this.
"""
import sys
from PyObjCTools.TestSupport import *
import objc
import threading
import time

from PyObjCTest.locking import OC_LockTest

NSAutoreleasePool = objc.lookUpClass("NSAutoreleasePool")

class OtherThread (threading.Thread):
    def __init__(self, obj):
        self.obj = obj

        threading.Thread.__init__(self)

    def run(self):
        pool = NSAutoreleasePool.alloc().init()

        lck = objc.object_lock(self.obj)

        for i in range(6):
            time.sleep(0.05)
            lck.lock()
            if self.obj.isLocked():
                self.obj.appendToList_("LOCK FOUND")

            self.obj.setLocked_(True)
            self.obj.appendToList_("thread %d a"%(i,))
            time.sleep(0.5)
            self.obj.appendToList_("thread %d b"%(i,))
            self.obj.setLocked_(False)
            lck.unlock()

        del pool

class ObjCThread (threading.Thread):
    def __init__(self, obj):
        self.obj = obj

        threading.Thread.__init__(self)

    def run(self):
        pool = NSAutoreleasePool.alloc().init()

        native = OC_LockTest.alloc().init()
        native.threadFunc_(self.obj)

        del native
        del pool


class BaseClass (objc.lookUpClass('NSObject')):
    def initWithList_(self, list):
        self = super(BaseClass, self).init()
        if self is None:
            return None

        self.list = list
        self._locked = False
        return self

    def isLocked(self):
        return self._locked

    def setLocked_(self, value):
        self._locked = value


    def appendToList_(self, value):
        self.list.append(value)

class TestLockingBasic (TestCase):

    def testBasicLocking(self):
        lst = []

        obj = BaseClass.alloc().initWithList_(lst)
        lck = objc.object_lock(obj)

        thr = OtherThread(obj)
        thr.start()
        for i in range(5):
            time.sleep(0.1)
            lck.lock()
            self.assertFalse(obj.isLocked())
            obj.setLocked_(True)
            obj.appendToList_("mainthread")
            obj.setLocked_(False)
            lck.unlock()

        thr.join()

        self.assertNotIn("LOCK FOUND", lst)
        for idx in range(len(lst)):
            if lst[idx].endswith(' a'):
                self.assertTrue(lst[idx+1].endswith(' b'))

    def testObjectiveCLocking(self):
        lst = []
        lst = []

        obj = BaseClass.alloc().initWithList_(lst)
        lck = objc.object_lock(obj)

        thr = ObjCThread(obj)
        thr.start()
        for i in range(5):
            time.sleep(0.1)
            lck.lock()
            self.assertFalse(obj.isLocked())
            obj.setLocked_(True)
            obj.appendToList_("mainthread")
            time.sleep(0.5)
            obj.setLocked_(False)
            lck.unlock()

        thr.join()

        self.assertNotIn("LOCK FOUND", lst)
        for idx in range(len(lst)):
            if lst[idx].endswith(' a'):
                self.assertTrue(lst[idx+1].endswith(' b'))


class TestLockingWithStatement (TestCase):

    def testBasicLocking(self):
        lst = []
        lst = []

        obj = BaseClass.alloc().initWithList_(lst)

        thr = OtherThread(obj)
        thr.start()
        for i in range(5):
            time.sleep(0.1)
            with objc.object_lock(obj):
                self.assertFalse(obj.isLocked())
                obj.setLocked_(True)
                obj.appendToList_("mainthread")
                time.sleep(0.5)
                obj.setLocked_(False)

        thr.join()

        self.assertNotIn("LOCK FOUND", lst)
        for idx in range(len(lst)):
            if lst[idx].endswith(' a'):
                self.assertTrue(lst[idx+1].endswith(' b'))


if __name__ == "__main__":
    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.