~ubuntu-branches/ubuntu/quantal/python-gevent/quantal

« back to all changes in this revision

Viewing changes to greentest/test_threading_2.py

• browse files at revision 9
• compare with another revision
• download diff
• download tarball
• view history from revision 9

Show diffs side-by-side

added

removed

greentest/test_threading_2.py

1

1

# testing gevent's Event, Lock, RLock, Semaphore, BoundedSemaphore with standard test_threading

2

2

from __future__ import with_statement

3

 

from gevent import monkey; monkey.patch_all()

 

3

 

 

4

setup = '''from gevent import monkey; monkey.patch_all()

4

5

from gevent.event import Event

5

6

from gevent.coros import RLock, Semaphore, BoundedSemaphore

6

7

from gevent.thread import allocate_lock as Lock

7

 

 

8

 

import test.test_support

9

 

from test.test_support import verbose

10

 

import random

11

 

import re

12

 

import sys

13

8

import threading

14

 

import thread

15

 

import time

16

 

import unittest

17

 

import weakref

18

 

 

19

9

threading.Event = Event

20

10

threading.Lock = Lock

21

11

threading.RLock = RLock

27

17

    threading.Thread.name = property(lambda self: self.getName())

28

18

if not hasattr(threading.Thread, 'is_alive'):

29

19

    threading.Thread.is_alive = threading.Thread.isAlive

 

20

if not hasattr(threading.Thread, 'daemon'):

 

21

    threading.Thread.daemon = property(threading.Thread.isDaemon, threading.Thread.setDaemon)

30

22

if not hasattr(threading._Condition, 'notify_all'):

31

23

    threading._Condition.notify_all = threading._Condition.notifyAll

 

24

'''

 

25

 

 

26

exec setup

 

27

 

 

28

setup_3 = '\n'.join('            %s' % line for line in setup.split('\n'))

 

29

setup_4 = '\n'.join('                %s' % line for line in setup.split('\n'))

 

30

setup_5 = '\n'.join('                    %s' % line for line in setup.split('\n'))

 

31

 

 

32

 

 

33

import test.test_support

 

34

from test.test_support import verbose

 

35

import random

 

36

import re

 

37

import sys

 

38

import threading

 

39

import thread

 

40

import time

 

41

import unittest

 

42

import weakref

32

43

 

33

44

import lock_tests

34

45

 

153

164

        self.test_various_ops()

154

165

        threading.stack_size(0)

155

166

 

156

 

    def BOGUS_test_foreign_thread(self):

 

167

    def test_foreign_thread(self):

157

168

        # Check that a "foreign" thread can use the threading module.

158

169

        def f(mutex):

159

170

            # Calling current_thread() forces an entry for the foreign

271

282

 

272

283

            import subprocess

273

284

            rc = subprocess.call([sys.executable, "-c", """if 1:

 

285

%s

274

286

                import ctypes, sys, time, thread

275

287

 

276

288

                # This lock is used as a simple event variable.

294

306

                thread.start_new_thread(waitingThread, ())

295

307

                ready.acquire()  # Be sure the other thread is waiting.

296

308

                sys.exit(42)

297

 

                """])

 

309

                """ % setup_4])

298

310

            self.assertEqual(rc, 42)

299

311

 

300

312

    def test_finalize_with_trace(self):

302

314

        # Avoid a deadlock when sys.settrace steps into threading._shutdown

303

315

        import subprocess

304

316

        rc = subprocess.call([sys.executable, "-c", """if 1:

 

317

%s

305

318

            import sys, threading

306

319

 

307

320

            # A deadlock-killer, to prevent the

321

334

                return func

322

335

 

323

336

            sys.settrace(func)

324

 

            """])

 

337

            """ % setup_3])

325

338

        self.failIf(rc == 2, "interpreted was blocked")

326

339

        self.failUnless(rc == 0, "Unexpected error")

327

340

 

331

344

            # Raising SystemExit skipped threading._shutdown

332

345

            import subprocess

333

346

            p = subprocess.Popen([sys.executable, "-c", """if 1:

 

347

%s

334

348

                    import threading

335

349

                    from time import sleep

336

350

 

342

356

 

343

357

                    threading.Thread(target=child).start()

344

358

                    raise SystemExit

345

 

                """],

 

359

                """ % setup_5],

346

360

                stdout=subprocess.PIPE,

347

361

                stderr=subprocess.PIPE)

348

362

            stdout, stderr = p.communicate()

349

 

            self.assertEqual(stdout.strip(),

350

 

                "Woke up, sleep function is: <built-in function sleep>")

 

363

            stdout = stdout.strip()

 

364

            assert re.match('^Woke up, sleep function is: <.*?sleep.*?>$', stdout), repr(stdout)

351

365

            stderr = re.sub(r"^\[\d+ refs\]", "", stderr, re.MULTILINE).strip()

352

366

            self.assertEqual(stderr, "")

353

367

 

407

421

 

408

422

    def _run_and_join(self, script):

409

423

        script = """if 1:

 

424

%s

410

425

            import sys, os, time, threading

411

426

 

412

427

            # a thread, which waits for the main program to terminate

413

428

            def joiningfunc(mainthread):

414

429

                mainthread.join()

415

430

                print 'end of thread'

416

 

        \n""" + script

 

431

        \n""" % setup_3 + script

417

432

 

418

433

        import subprocess

419

434

        p = subprocess.Popen([sys.executable, "-c", script], stdout=subprocess.PIPE)

462

477

            return

463

478

        # Skip platforms with known problems forking from a worker thread.

464

479

        # See http://bugs.python.org/issue3863.

465

 

        if sys.platform in ('freebsd4', 'freebsd5', 'freebsd6', 'os2emx'):

466

 

            print >>sys.stderr, ('Skipping test_3_join_in_forked_from_thread'

467

 

                                 ' due to known OS bugs on'), sys.platform

468

 

            return

 

480

        # skip disable because I think the bug shouldn't apply to gevent -- denis

 

481

        #if sys.platform in ('freebsd4', 'freebsd5', 'freebsd6', 'os2emx'):

 

482

        #    print >>sys.stderr, ('Skipping test_3_join_in_forked_from_thread'

 

483

        #                         ' due to known OS bugs on'), sys.platform

 

484

        #    return

469

485

        script = """if 1:

470

486

            main_thread = threading.current_thread()

471

487

            def worker():

• Older »

Loggerhead is a web-based interface for Breezy
Version: 2.0.1