Gecko [ www.ctm-euromeuble.fr ]

Name	Size	Permission	Date
__pycache__	[ DIR ]	drwxr-sr-x	2025-06-04 09:23
builtin	[ DIR ]	drwxr-sr-x	2025-06-04 09:23
data	[ DIR ]	drwxr-sr-x	2025-06-04 09:23
data01	[ DIR ]	drwxr-sr-x	2025-06-04 09:23
data02	[ DIR ]	drwxr-sr-x	2025-06-04 09:23
data03	[ DIR ]	drwxr-sr-x	2025-06-04 09:23
extension	[ DIR ]	drwxr-sr-x	2025-06-04 09:23
frozen	[ DIR ]	drwxr-sr-x	2025-06-04 09:23
import_	[ DIR ]	drwxr-sr-x	2025-06-04 09:23
namespace_pkgs	[ DIR ]	drwxr-sr-x	2025-06-04 09:23
namespacedata01	[ DIR ]	drwxr-sr-x	2025-06-04 09:23
partial	[ DIR ]	drwxr-sr-x	2025-06-04 09:23
source	[ DIR ]	drwxr-sr-x	2025-06-04 09:23
zipdata01	[ DIR ]	drwxr-sr-x	2025-06-04 09:23
zipdata02	[ DIR ]	drwxr-sr-x	2025-06-04 09:23
__init__.py	142 B	-rw-r--r--	2025-06-04 09:23
__main__.py	58 B	-rw-r--r--	2025-06-04 09:23
abc.py	2.22 KB	-rw-r--r--	2025-06-04 09:23
fixtures.py	7.28 KB	-rw-r--r--	2025-06-04 09:23
stubs.py	233 B	-rw-r--r--	2025-06-04 09:23
test_abc.py	34.19 KB	-rw-r--r--	2025-06-04 09:23
test_api.py	18.59 KB	-rw-r--r--	2025-06-04 09:23
test_files.py	1011 B	-rw-r--r--	2025-06-04 09:23
test_lazy.py	4.81 KB	-rw-r--r--	2025-06-04 09:23
test_locks.py	4.41 KB	-rw-r--r--	2025-06-04 09:23
test_main.py	8.75 KB	-rw-r--r--	2025-06-04 09:23
test_metadata_api.py	11.41 KB	-rw-r--r--	2025-06-04 09:23
test_namespace_pkgs.py	12.11 KB	-rw-r--r--	2025-06-04 09:23
test_open.py	2.31 KB	-rw-r--r--	2025-06-04 09:23
test_path.py	1.88 KB	-rw-r--r--	2025-06-04 09:23
test_pkg_import.py	2.69 KB	-rw-r--r--	2025-06-04 09:23
test_read.py	1.94 KB	-rw-r--r--	2025-06-04 09:23
test_reader.py	4.18 KB	-rw-r--r--	2025-06-04 09:23
test_resource.py	8.18 KB	-rw-r--r--	2025-06-04 09:23
test_spec.py	30.88 KB	-rw-r--r--	2025-06-04 09:23
test_threaded_import.py	9.49 KB	-rw-r--r--	2025-06-04 09:23
test_util.py	34.67 KB	-rw-r--r--	2025-06-04 09:23
test_windows.py	7.26 KB	-rw-r--r--	2025-06-04 09:23
test_zip.py	2.61 KB	-rw-r--r--	2025-06-04 09:23
threaded_import_hangers.py	1.45 KB	-rw-r--r--	2025-06-04 09:23
update-zips.py	1.38 KB	-rwxr-xr-x	2025-06-04 09:23
util.py	18.22 KB	-rw-r--r--	2025-06-04 09:23

Rename

from . 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

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

# _is_owned() unsupported
    test__is_owned = None
    # acquire(blocking=False) unsupported
    test_try_acquire = None
    test_try_acquire_contended = None
    # `with` unsupported
    test_with = None
    # acquire(timeout=...) unsupported
    test_timeout = None
    # _release_save() unsupported
    test_release_save_unacquired = None
    # lock status in repr unsupported
    test_repr = None
    test_locked_repr = None

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 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 = _acquire(a)
            barrier.wait()
            rb = _acquire(b)
            results.append((ra, rb))
            if rb:
                b.release()
            if ra:
                a.release()
        lock_tests.Bunch(f, NTHREADS).wait_for_finished()
        self.assertEqual(len(results), NTHREADS)
        return results

def test_deadlock(self):
        results = self.run_deadlock_avoidance_test(True)
        # 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.
        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__':
    unittets.main()