isowned unsupported acquireblockingfalse unsupported with unsupported acquiretimeout unsupported releasesave unsupported recursioncount unsupported lock status in repr unsupported try to acquire the lock return true on success false on deadlock try lock acquire except self deadlockerror return false else return true def f a b pairs pop ra acquirea barrier wait rb acquireb results appendra rb if rb b release if ra a release with locktests bunchf nthreads pass self assertequallenresults nthreads return results def testdeadlockself results self rundeadlockavoidancetesttrue at least one of the threads detected a potential deadlock on its second acquire call it may be several of them because the deadlock avoidance mechanism is conservative nbdeadlocks results counttrue false self assertgreaterequalnbdeadlocks 1 self assertequalresults counttrue true lenresults nbdeadlocks def testnodeadlockself results self rundeadlockavoidancetestfalse self assertequalresults counttrue false 0 self assertequalresults counttrue true lenresults deadlockerrors kind splitinit bootstrap deadlockerror for kind splitinit in init items frozendeadlockavoidancetests sourcedeadlockavoidancetests testutil testbothdeadlockavoidancetests locktypelocktypes deadlockerrordeadlockerrors class lifetimetests property def bootstrapself return self init bootstrap def testlocklifetimeself name xyzzy self assertnotinname self bootstrap modulelocks lock self bootstrap getmodulelockname self assertinname self bootstrap modulelocks wr weakref reflock del lock support gccollect self assertnotinname self bootstrap modulelocks self assertisnonewr def testalllocksself support gccollect self assertequal0 lenself bootstrap modulelocks self bootstrap modulelocks frozenlifetimetests sourcelifetimetests testutil testbothlifetimetests initinit def setupmodule threadinfo threadinghelper threadingsetup unittest addmodulecleanupthreadinghelper threadingcleanup threadinfo if name main unittest main _is_owned unsupported acquire blocking false unsupported with unsupported acquire timeout unsupported _release_save unsupported _recursion_count unsupported lock status in repr unsupported try to acquire the lock return true on success false on deadlock at least one of the threads detected a potential deadlock on its second acquire call it may be several of them because the deadlock avoidance mechanism is conservative
from test.test_importlib import util as test_util

init = test_util.import_importlib('importlib')

import sys
import threading
import unittest
import weakref

from test import support
from test.support import threading_helper
from test import lock_tests


threading_helper.requires_working_threading(module=True)


class ModuleLockAsRLockTests:
    locktype = classmethod(lambda cls: cls.LockType("some_lock"))

    
    test__is_owned = None
    
    test_try_acquire = None
    test_try_acquire_contended = None
    
    test_with = None
    
    test_timeout = None
    
    test_release_save_unacquired = None
    
    test_recursion_count = None
    
    test_repr = None
    test_locked_repr = None

    def tearDown(self):
        for splitinit in init.values():
            splitinit._bootstrap._blocking_on.clear()


LOCK_TYPES = {kind: splitinit._bootstrap._ModuleLock
              for kind, splitinit in init.items()}

(Frozen_ModuleLockAsRLockTests,
 Source_ModuleLockAsRLockTests
 ) = test_util.test_both(ModuleLockAsRLockTests, lock_tests.RLockTests,
                         LockType=LOCK_TYPES)


class DeadlockAvoidanceTests:

    def setUp(self):
        try:
            self.old_switchinterval = sys.getswitchinterval()
            support.setswitchinterval(0.000001)
        except AttributeError:
            self.old_switchinterval = None

    def tearDown(self):
        if self.old_switchinterval is not None:
            sys.setswitchinterval(self.old_switchinterval)

    def run_deadlock_avoidance_test(self, create_deadlock):
        NLOCKS = 10
        locks = [self.LockType(str(i)) for i in range(NLOCKS)]
        pairs = [(locks[i], locks[(i+1)%NLOCKS]) for i in range(NLOCKS)]
        if create_deadlock:
            NTHREADS = NLOCKS
        else:
            NTHREADS = NLOCKS - 1
        barrier = threading.Barrier(NTHREADS)
        results = []

        def _acquire(lock):
            
            try:
                lock.acquire()
            except self.DeadlockError:
                return False
            else:
                return True

        def f():
            a, b = pairs.pop()
            ra = _acquire(a)
            barrier.wait()
            rb = _acquire(b)
            results.append((ra, rb))
            if rb:
                b.release()
            if ra:
                a.release()
        with lock_tests.Bunch(f, NTHREADS):
            pass
        self.assertEqual(len(results), NTHREADS)
        return results

    def test_deadlock(self):
        results = self.run_deadlock_avoidance_test(True)
        
        
        
        nb_deadlocks = results.count((True, False))
        self.assertGreaterEqual(nb_deadlocks, 1)
        self.assertEqual(results.count((True, True)), len(results) - nb_deadlocks)

    def test_no_deadlock(self):
        results = self.run_deadlock_avoidance_test(False)
        self.assertEqual(results.count((True, False)), 0)
        self.assertEqual(results.count((True, True)), len(results))


DEADLOCK_ERRORS = {kind: splitinit._bootstrap._DeadlockError
                   for kind, splitinit in init.items()}

(Frozen_DeadlockAvoidanceTests,
 Source_DeadlockAvoidanceTests
 ) = test_util.test_both(DeadlockAvoidanceTests,
                         LockType=LOCK_TYPES,
                         DeadlockError=DEADLOCK_ERRORS)


class LifetimeTests:

    @property
    def bootstrap(self):
        return self.init._bootstrap

    def test_lock_lifetime(self):
        name = "xyzzy"
        self.assertNotIn(name, self.bootstrap._module_locks)
        lock = self.bootstrap._get_module_lock(name)
        self.assertIn(name, self.bootstrap._module_locks)
        wr = weakref.ref(lock)
        del lock
        support.gc_collect()
        self.assertNotIn(name, self.bootstrap._module_locks)
        self.assertIsNone(wr())

    def test_all_locks(self):
        support.gc_collect()
        self.assertEqual(0, len(self.bootstrap._module_locks),
                         self.bootstrap._module_locks)


(Frozen_LifetimeTests,
 Source_LifetimeTests
 ) = test_util.test_both(LifetimeTests, init=init)


def setUpModule():
    thread_info = threading_helper.threading_setup()
    unittest.addModuleCleanup(threading_helper.threading_cleanup, *thread_info)


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