twisted/test/test_internet.py

   1 # Copyright (c) Twisted Matrix Laboratories.
   2 # See LICENSE for details.
   3
   4 """
   5 Tests for lots of functionality provided by L{twisted.internet}.
   6 """
   7
   8 import os
   9 import sys
  10 import time
  11
  12 from twisted.trial import unittest
  13 from twisted.internet import reactor, protocol, error, abstract, defer
  14 from twisted.internet import interfaces, base
  15
  16 try:
  17     from twisted.internet import ssl
  18 except ImportError:
  19     ssl = None
  20 if ssl and not ssl.supported:
  21     ssl = None
  22
  23 from twisted.internet.defer import Deferred, maybeDeferred
  24 from twisted.python import util, runtime
  25
  26
  27
  28 class ThreePhaseEventTests(unittest.TestCase):
  29     """
  30     Tests for the private implementation helpers for system event triggers.
  31     """
  32     def setUp(self):
  33         """
  34         Create a trigger, an argument, and an event to be used by tests.
  35         """
  36         self.trigger = lambda x: None
  37         self.arg = object()
  38         self.event = base._ThreePhaseEvent()
  39
  40
  41     def test_addInvalidPhase(self):
  42         """
  43         L{_ThreePhaseEvent.addTrigger} should raise L{KeyError} when called
  44         with an invalid phase.
  45         """
  46         self.assertRaises(
  47             KeyError,
  48             self.event.addTrigger, 'xxx', self.trigger, self.arg)
  49
  50
  51     def test_addBeforeTrigger(self):
  52         """
  53         L{_ThreePhaseEvent.addTrigger} should accept C{'before'} as a phase, a
  54         callable, and some arguments and add the callable with the arguments to
  55         the before list.
  56         """
  57         self.event.addTrigger('before', self.trigger, self.arg)
  58         self.assertEqual(
  59             self.event.before,
  60             [(self.trigger, (self.arg,), {})])
  61
  62
  63     def test_addDuringTrigger(self):
  64         """
  65         L{_ThreePhaseEvent.addTrigger} should accept C{'during'} as a phase, a
  66         callable, and some arguments and add the callable with the arguments to
  67         the during list.
  68         """
  69         self.event.addTrigger('during', self.trigger, self.arg)
  70         self.assertEqual(
  71             self.event.during,
  72             [(self.trigger, (self.arg,), {})])
  73
  74
  75     def test_addAfterTrigger(self):
  76         """
  77         L{_ThreePhaseEvent.addTrigger} should accept C{'after'} as a phase, a
  78         callable, and some arguments and add the callable with the arguments to
  79         the after list.
  80         """
  81         self.event.addTrigger('after', self.trigger, self.arg)
  82         self.assertEqual(
  83             self.event.after,
  84             [(self.trigger, (self.arg,), {})])
  85
  86
  87     def test_removeTrigger(self):
  88         """
  89         L{_ThreePhaseEvent.removeTrigger} should accept an opaque object
  90         previously returned by L{_ThreePhaseEvent.addTrigger} and remove the
  91         associated trigger.
  92         """
  93         handle = self.event.addTrigger('before', self.trigger, self.arg)
  94         self.event.removeTrigger(handle)
  95         self.assertEqual(self.event.before, [])
  96
  97
  98     def test_removeNonexistentTrigger(self):
  99         """
 100         L{_ThreePhaseEvent.removeTrigger} should raise L{ValueError} when given
 101         an object not previously returned by L{_ThreePhaseEvent.addTrigger}.
 102         """
 103         self.assertRaises(ValueError, self.event.removeTrigger, object())
 104
 105
 106     def test_removeRemovedTrigger(self):
 107         """
 108         L{_ThreePhaseEvent.removeTrigger} should raise L{ValueError} the second
 109         time it is called with an object returned by
 110         L{_ThreePhaseEvent.addTrigger}.
 111         """
 112         handle = self.event.addTrigger('before', self.trigger, self.arg)
 113         self.event.removeTrigger(handle)
 114         self.assertRaises(ValueError, self.event.removeTrigger, handle)
 115
 116
 117     def test_removeAlmostValidTrigger(self):
 118         """
 119         L{_ThreePhaseEvent.removeTrigger} should raise L{ValueError} if it is
 120         given a trigger handle which resembles a valid trigger handle aside
 121         from its phase being incorrect.
 122         """
 123         self.assertRaises(
 124             KeyError,
 125             self.event.removeTrigger, ('xxx', self.trigger, (self.arg,), {}))
 126
 127
 128     def test_fireEvent(self):
 129         """
 130         L{_ThreePhaseEvent.fireEvent} should call I{before}, I{during}, and
 131         I{after} phase triggers in that order.
 132         """
 133         events = []
 134         self.event.addTrigger('after', events.append, ('first', 'after'))
 135         self.event.addTrigger('during', events.append, ('first', 'during'))
 136         self.event.addTrigger('before', events.append, ('first', 'before'))
 137         self.event.addTrigger('before', events.append, ('second', 'before'))
 138         self.event.addTrigger('during', events.append, ('second', 'during'))
 139         self.event.addTrigger('after', events.append, ('second', 'after'))
 140
 141         self.assertEqual(events, [])
 142         self.event.fireEvent()
 143         self.assertEqual(events,
 144                          [('first', 'before'), ('second', 'before'),
 145                           ('first', 'during'), ('second', 'during'),
 146                           ('first', 'after'), ('second', 'after')])
 147
 148
 149     def test_asynchronousBefore(self):
 150         """
 151         L{_ThreePhaseEvent.fireEvent} should wait for any L{Deferred} returned
 152         by a I{before} phase trigger before proceeding to I{during} events.
 153         """
 154         events = []
 155         beforeResult = Deferred()
 156         self.event.addTrigger('before', lambda: beforeResult)
 157         self.event.addTrigger('during', events.append, 'during')
 158         self.event.addTrigger('after', events.append, 'after')
 159
 160         self.assertEqual(events, [])
 161         self.event.fireEvent()
 162         self.assertEqual(events, [])
 163         beforeResult.callback(None)
 164         self.assertEqual(events, ['during', 'after'])
 165
 166
 167     def test_beforeTriggerException(self):
 168         """
 169         If a before-phase trigger raises a synchronous exception, it should be
 170         logged and the remaining triggers should be run.
 171         """
 172         events = []
 173
 174         class DummyException(Exception):
 175             pass
 176
 177         def raisingTrigger():
 178             raise DummyException()
 179
 180         self.event.addTrigger('before', raisingTrigger)
 181         self.event.addTrigger('before', events.append, 'before')
 182         self.event.addTrigger('during', events.append, 'during')
 183         self.event.fireEvent()
 184         self.assertEqual(events, ['before', 'during'])
 185         errors = self.flushLoggedErrors(DummyException)
 186         self.assertEqual(len(errors), 1)
 187
 188
 189     def test_duringTriggerException(self):
 190         """
 191         If a during-phase trigger raises a synchronous exception, it should be
 192         logged and the remaining triggers should be run.
 193         """
 194         events = []
 195
 196         class DummyException(Exception):
 197             pass
 198
 199         def raisingTrigger():
 200             raise DummyException()
 201
 202         self.event.addTrigger('during', raisingTrigger)
 203         self.event.addTrigger('during', events.append, 'during')
 204         self.event.addTrigger('after', events.append, 'after')
 205         self.event.fireEvent()
 206         self.assertEqual(events, ['during', 'after'])
 207         errors = self.flushLoggedErrors(DummyException)
 208         self.assertEqual(len(errors), 1)
 209
 210
 211     def test_synchronousRemoveAlreadyExecutedBefore(self):
 212         """
 213         If a before-phase trigger tries to remove another before-phase trigger
 214         which has already run, a warning should be emitted.
 215         """
 216         events = []
 217
 218         def removeTrigger():
 219             self.event.removeTrigger(beforeHandle)
 220
 221         beforeHandle = self.event.addTrigger('before', events.append, ('first', 'before'))
 222         self.event.addTrigger('before', removeTrigger)
 223         self.event.addTrigger('before', events.append, ('second', 'before'))
 224         self.assertWarns(
 225             DeprecationWarning,
 226             "Removing already-fired system event triggers will raise an "
 227             "exception in a future version of Twisted.",
 228             __file__,
 229             self.event.fireEvent)
 230         self.assertEqual(events, [('first', 'before'), ('second', 'before')])
 231
 232
 233     def test_synchronousRemovePendingBefore(self):
 234         """
 235         If a before-phase trigger removes another before-phase trigger which
 236         has not yet run, the removed trigger should not be run.
 237         """
 238         events = []
 239         self.event.addTrigger(
 240             'before', lambda: self.event.removeTrigger(beforeHandle))
 241         beforeHandle = self.event.addTrigger(
 242             'before', events.append, ('first', 'before'))
 243         self.event.addTrigger('before', events.append, ('second', 'before'))
 244         self.event.fireEvent()
 245         self.assertEqual(events, [('second', 'before')])
 246
 247
 248     def test_synchronousBeforeRemovesDuring(self):
 249         """
 250         If a before-phase trigger removes a during-phase trigger, the
 251         during-phase trigger should not be run.
 252         """
 253         events = []
 254         self.event.addTrigger(
 255             'before', lambda: self.event.removeTrigger(duringHandle))
 256         duringHandle = self.event.addTrigger('during', events.append, 'during')
 257         self.event.addTrigger('after', events.append, 'after')
 258         self.event.fireEvent()
 259         self.assertEqual(events, ['after'])
 260
 261
 262     def test_asynchronousBeforeRemovesDuring(self):
 263         """
 264         If a before-phase trigger returns a L{Deferred} and later removes a
 265         during-phase trigger before the L{Deferred} fires, the during-phase
 266         trigger should not be run.
 267         """
 268         events = []
 269         beforeResult = Deferred()
 270         self.event.addTrigger('before', lambda: beforeResult)
 271         duringHandle = self.event.addTrigger('during', events.append, 'during')
 272         self.event.addTrigger('after', events.append, 'after')
 273         self.event.fireEvent()
 274         self.event.removeTrigger(duringHandle)
 275         beforeResult.callback(None)
 276         self.assertEqual(events, ['after'])
 277
 278
 279     def test_synchronousBeforeRemovesConspicuouslySimilarDuring(self):
 280         """
 281         If a before-phase trigger removes a during-phase trigger which is
 282         identical to an already-executed before-phase trigger aside from their
 283         phases, no warning should be emitted and the during-phase trigger
 284         should not be run.
 285         """
 286         events = []
 287         def trigger():
 288             events.append('trigger')
 289         self.event.addTrigger('before', trigger)
 290         self.event.addTrigger(
 291             'before', lambda: self.event.removeTrigger(duringTrigger))
 292         duringTrigger = self.event.addTrigger('during', trigger)
 293         self.event.fireEvent()
 294         self.assertEqual(events, ['trigger'])
 295
 296
 297     def test_synchronousRemovePendingDuring(self):
 298         """
 299         If a during-phase trigger removes another during-phase trigger which
 300         has not yet run, the removed trigger should not be run.
 301         """
 302         events = []
 303         self.event.addTrigger(
 304             'during', lambda: self.event.removeTrigger(duringHandle))
 305         duringHandle = self.event.addTrigger(
 306             'during', events.append, ('first', 'during'))
 307         self.event.addTrigger(
 308             'during', events.append, ('second', 'during'))
 309         self.event.fireEvent()
 310         self.assertEqual(events, [('second', 'during')])
 311
 312
 313     def test_triggersRunOnce(self):
 314         """
 315         A trigger should only be called on the first call to
 316         L{_ThreePhaseEvent.fireEvent}.
 317         """
 318         events = []
 319         self.event.addTrigger('before', events.append, 'before')
 320         self.event.addTrigger('during', events.append, 'during')
 321         self.event.addTrigger('after', events.append, 'after')
 322         self.event.fireEvent()
 323         self.event.fireEvent()
 324         self.assertEqual(events, ['before', 'during', 'after'])
 325
 326
 327     def test_finishedBeforeTriggersCleared(self):
 328         """
 329         The temporary list L{_ThreePhaseEvent.finishedBefore} should be emptied
 330         and the state reset to C{'BASE'} before the first during-phase trigger
 331         executes.
 332         """
 333         events = []
 334         def duringTrigger():
 335             events.append('during')
 336             self.assertEqual(self.event.finishedBefore, [])
 337             self.assertEqual(self.event.state, 'BASE')
 338         self.event.addTrigger('before', events.append, 'before')
 339         self.event.addTrigger('during', duringTrigger)
 340         self.event.fireEvent()
 341         self.assertEqual(events, ['before', 'during'])
 342
 343
 344
 345 class SystemEventTestCase(unittest.TestCase):
 346     """
 347     Tests for the reactor's implementation of the C{fireSystemEvent},
 348     C{addSystemEventTrigger}, and C{removeSystemEventTrigger} methods of the
 349     L{IReactorCore} interface.
 350
 351     @ivar triggers: A list of the handles to triggers which have been added to
 352         the reactor.
 353     """
 354     def setUp(self):
 355         """
 356         Create an empty list in which to store trigger handles.
 357         """
 358         self.triggers = []
 359
 360
 361     def tearDown(self):
 362         """
 363         Remove all remaining triggers from the reactor.
 364         """
 365         while self.triggers:
 366             trigger = self.triggers.pop()
 367             try:
 368                 reactor.removeSystemEventTrigger(trigger)
 369             except (ValueError, KeyError):
 370                 pass
 371
 372
 373     def addTrigger(self, event, phase, func):
 374         """
 375         Add a trigger to the reactor and remember it in C{self.triggers}.
 376         """
 377         t = reactor.addSystemEventTrigger(event, phase, func)
 378         self.triggers.append(t)
 379         return t
 380
 381
 382     def removeTrigger(self, trigger):
 383         """
 384         Remove a trigger by its handle from the reactor and from
 385         C{self.triggers}.
 386         """
 387         reactor.removeSystemEventTrigger(trigger)
 388         self.triggers.remove(trigger)
 389
 390
 391     def _addSystemEventTriggerTest(self, phase):
 392         eventType = 'test'
 393         events = []
 394         def trigger():
 395             events.append(None)
 396         self.addTrigger(phase, eventType, trigger)
 397         self.assertEqual(events, [])
 398         reactor.fireSystemEvent(eventType)
 399         self.assertEqual(events, [None])
 400
 401
 402     def test_beforePhase(self):
 403         """
 404         L{IReactorCore.addSystemEventTrigger} should accept the C{'before'}
 405         phase and not call the given object until the right event is fired.
 406         """
 407         self._addSystemEventTriggerTest('before')
 408
 409
 410     def test_duringPhase(self):
 411         """
 412         L{IReactorCore.addSystemEventTrigger} should accept the C{'during'}
 413         phase and not call the given object until the right event is fired.
 414         """
 415         self._addSystemEventTriggerTest('during')
 416
 417
 418     def test_afterPhase(self):
 419         """
 420         L{IReactorCore.addSystemEventTrigger} should accept the C{'after'}
 421         phase and not call the given object until the right event is fired.
 422         """
 423         self._addSystemEventTriggerTest('after')
 424
 425
 426     def test_unknownPhase(self):
 427         """
 428         L{IReactorCore.addSystemEventTrigger} should reject phases other than
 429         C{'before'}, C{'during'}, or C{'after'}.
 430         """
 431         eventType = 'test'
 432         self.assertRaises(
 433             KeyError, self.addTrigger, 'xxx', eventType, lambda: None)
 434
 435
 436     def test_beforePreceedsDuring(self):
 437         """
 438         L{IReactorCore.addSystemEventTrigger} should call triggers added to the
 439         C{'before'} phase before it calls triggers added to the C{'during'}
 440         phase.
 441         """
 442         eventType = 'test'
 443         events = []
 444         def beforeTrigger():
 445             events.append('before')
 446         def duringTrigger():
 447             events.append('during')
 448         self.addTrigger('before', eventType, beforeTrigger)
 449         self.addTrigger('during', eventType, duringTrigger)
 450         self.assertEqual(events, [])
 451         reactor.fireSystemEvent(eventType)
 452         self.assertEqual(events, ['before', 'during'])
 453
 454
 455     def test_duringPreceedsAfter(self):
 456         """
 457         L{IReactorCore.addSystemEventTrigger} should call triggers added to the
 458         C{'during'} phase before it calls triggers added to the C{'after'}
 459         phase.
 460         """
 461         eventType = 'test'
 462         events = []
 463         def duringTrigger():
 464             events.append('during')
 465         def afterTrigger():
 466             events.append('after')
 467         self.addTrigger('during', eventType, duringTrigger)
 468         self.addTrigger('after', eventType, afterTrigger)
 469         self.assertEqual(events, [])
 470         reactor.fireSystemEvent(eventType)
 471         self.assertEqual(events, ['during', 'after'])
 472
 473
 474     def test_beforeReturnsDeferred(self):
 475         """
 476         If a trigger added to the C{'before'} phase of an event returns a
 477         L{Deferred}, the C{'during'} phase should be delayed until it is called
 478         back.
 479         """
 480         triggerDeferred = Deferred()
 481         eventType = 'test'
 482         events = []
 483         def beforeTrigger():
 484             return triggerDeferred
 485         def duringTrigger():
 486             events.append('during')
 487         self.addTrigger('before', eventType, beforeTrigger)
 488         self.addTrigger('during', eventType, duringTrigger)
 489         self.assertEqual(events, [])
 490         reactor.fireSystemEvent(eventType)
 491         self.assertEqual(events, [])
 492         triggerDeferred.callback(None)
 493         self.assertEqual(events, ['during'])
 494
 495
 496     def test_multipleBeforeReturnDeferred(self):
 497         """
 498         If more than one trigger added to the C{'before'} phase of an event
 499         return L{Deferred}s, the C{'during'} phase should be delayed until they
 500         are all called back.
 501         """
 502         firstDeferred = Deferred()
 503         secondDeferred = Deferred()
 504         eventType = 'test'
 505         events = []
 506         def firstBeforeTrigger():
 507             return firstDeferred
 508         def secondBeforeTrigger():
 509             return secondDeferred
 510         def duringTrigger():
 511             events.append('during')
 512         self.addTrigger('before', eventType, firstBeforeTrigger)
 513         self.addTrigger('before', eventType, secondBeforeTrigger)
 514         self.addTrigger('during', eventType, duringTrigger)
 515         self.assertEqual(events, [])
 516         reactor.fireSystemEvent(eventType)
 517         self.assertEqual(events, [])
 518         firstDeferred.callback(None)
 519         self.assertEqual(events, [])
 520         secondDeferred.callback(None)
 521         self.assertEqual(events, ['during'])
 522
 523
 524     def test_subsequentBeforeTriggerFiresPriorBeforeDeferred(self):
 525         """
 526         If a trigger added to the C{'before'} phase of an event calls back a
 527         L{Deferred} returned by an earlier trigger in the C{'before'} phase of
 528         the same event, the remaining C{'before'} triggers for that event
 529         should be run and any further L{Deferred}s waited on before proceeding
 530         to the C{'during'} events.
 531         """
 532         eventType = 'test'
 533         events = []
 534         firstDeferred = Deferred()
 535         secondDeferred = Deferred()
 536         def firstBeforeTrigger():
 537             return firstDeferred
 538         def secondBeforeTrigger():
 539             firstDeferred.callback(None)
 540         def thirdBeforeTrigger():
 541             events.append('before')
 542             return secondDeferred
 543         def duringTrigger():
 544             events.append('during')
 545         self.addTrigger('before', eventType, firstBeforeTrigger)
 546         self.addTrigger('before', eventType, secondBeforeTrigger)
 547         self.addTrigger('before', eventType, thirdBeforeTrigger)
 548         self.addTrigger('during', eventType, duringTrigger)
 549         self.assertEqual(events, [])
 550         reactor.fireSystemEvent(eventType)
 551         self.assertEqual(events, ['before'])
 552         secondDeferred.callback(None)
 553         self.assertEqual(events, ['before', 'during'])
 554
 555
 556     def test_removeSystemEventTrigger(self):
 557         """
 558         A trigger removed with L{IReactorCore.removeSystemEventTrigger} should
 559         not be called when the event fires.
 560         """
 561         eventType = 'test'
 562         events = []
 563         def firstBeforeTrigger():
 564             events.append('first')
 565         def secondBeforeTrigger():
 566             events.append('second')
 567         self.addTrigger('before', eventType, firstBeforeTrigger)
 568         self.removeTrigger(
 569             self.addTrigger('before', eventType, secondBeforeTrigger))
 570         self.assertEqual(events, [])
 571         reactor.fireSystemEvent(eventType)
 572         self.assertEqual(events, ['first'])
 573
 574
 575     def test_removeNonExistentSystemEventTrigger(self):
 576         """
 577         Passing an object to L{IReactorCore.removeSystemEventTrigger} which was
 578         not returned by a previous call to
 579         L{IReactorCore.addSystemEventTrigger} or which has already been passed
 580         to C{removeSystemEventTrigger} should result in L{TypeError},
 581         L{KeyError}, or L{ValueError} being raised.
 582         """
 583         b = self.addTrigger('during', 'test', lambda: None)
 584         self.removeTrigger(b)
 585         self.assertRaises(
 586             TypeError, reactor.removeSystemEventTrigger, None)
 587         self.assertRaises(
 588             ValueError, reactor.removeSystemEventTrigger, b)
 589         self.assertRaises(
 590             KeyError,
 591             reactor.removeSystemEventTrigger,
 592             (b[0], ('xxx',) + b[1][1:]))
 593
 594
 595     def test_interactionBetweenDifferentEvents(self):
 596         """
 597         L{IReactorCore.fireSystemEvent} should behave the same way for a
 598         particular system event regardless of whether Deferreds are being
 599         waited on for a different system event.
 600         """
 601         events = []
 602
 603         firstEvent = 'first-event'
 604         firstDeferred = Deferred()
 605         def beforeFirstEvent():
 606             events.append(('before', 'first'))
 607             return firstDeferred
 608         def afterFirstEvent():
 609             events.append(('after', 'first'))
 610
 611         secondEvent = 'second-event'
 612         secondDeferred = Deferred()
 613         def beforeSecondEvent():
 614             events.append(('before', 'second'))
 615             return secondDeferred
 616         def afterSecondEvent():
 617             events.append(('after', 'second'))
 618
 619         self.addTrigger('before', firstEvent, beforeFirstEvent)
 620         self.addTrigger('after', firstEvent, afterFirstEvent)
 621         self.addTrigger('before', secondEvent, beforeSecondEvent)
 622         self.addTrigger('after', secondEvent, afterSecondEvent)
 623
 624         self.assertEqual(events, [])
 625
 626         # After this, firstEvent should be stuck before 'during' waiting for
 627         # firstDeferred.
 628         reactor.fireSystemEvent(firstEvent)
 629         self.assertEqual(events, [('before', 'first')])
 630
 631         # After this, secondEvent should be stuck before 'during' waiting for
 632         # secondDeferred.
 633         reactor.fireSystemEvent(secondEvent)
 634         self.assertEqual(events, [('before', 'first'), ('before', 'second')])
 635
 636         # After this, firstEvent should have finished completely, but
 637         # secondEvent should be at the same place.
 638         firstDeferred.callback(None)
 639         self.assertEqual(events, [('before', 'first'), ('before', 'second'),
 640                                   ('after', 'first')])
 641
 642         # After this, secondEvent should have finished completely.
 643         secondDeferred.callback(None)
 644         self.assertEqual(events, [('before', 'first'), ('before', 'second'),
 645                                   ('after', 'first'), ('after', 'second')])
 646
 647
 648
 649 class TimeTestCase(unittest.TestCase):
 650     """
 651     Tests for the IReactorTime part of the reactor.
 652     """
 653
 654
 655     def test_seconds(self):
 656         """
 657         L{twisted.internet.reactor.seconds} should return something
 658         like a number.
 659
 660         1. This test specifically does not assert any relation to the
 661            "system time" as returned by L{time.time} or
 662            L{twisted.python.runtime.seconds}, because at some point we
 663            may find a better option for scheduling calls than
 664            wallclock-time.
 665         2. This test *also* does not assert anything about the type of
 666            the result, because operations may not return ints or
 667            floats: For example, datetime-datetime == timedelta(0).
 668         """
 669         now = reactor.seconds()
 670         self.assertEqual(now-now+now, now)
 671
 672
 673     def test_callLaterUsesReactorSecondsInDelayedCall(self):
 674         """
 675         L{reactor.callLater} should use the reactor's seconds factory
 676         to produce the time at which the DelayedCall will be called.
 677         """
 678         oseconds = reactor.seconds
 679         reactor.seconds = lambda: 100
 680         try:
 681             call = reactor.callLater(5, lambda: None)
 682             self.assertEqual(call.getTime(), 105)
 683         finally:
 684             reactor.seconds = oseconds
 685
 686
 687     def test_callLaterUsesReactorSecondsAsDelayedCallSecondsFactory(self):
 688         """
 689         L{reactor.callLater} should propagate its own seconds factory
 690         to the DelayedCall to use as its own seconds factory.
 691         """
 692         oseconds = reactor.seconds
 693         reactor.seconds = lambda: 100
 694         try:
 695             call = reactor.callLater(5, lambda: None)
 696             self.assertEqual(call.seconds(), 100)
 697         finally:
 698             reactor.seconds = oseconds
 699
 700
 701     def test_callLater(self):
 702         """
 703         Test that a DelayedCall really calls the function it is
 704         supposed to call.
 705         """
 706         d = Deferred()
 707         reactor.callLater(0, d.callback, None)
 708         d.addCallback(self.assertEqual, None)
 709         return d
 710
 711
 712     def test_cancelDelayedCall(self):
 713         """
 714         Test that when a DelayedCall is cancelled it does not run.
 715         """
 716         called = []
 717         def function():
 718             called.append(None)
 719         call = reactor.callLater(0, function)
 720         call.cancel()
 721
 722         # Schedule a call in two "iterations" to check to make sure that the
 723         # above call never ran.
 724         d = Deferred()
 725         def check():
 726             try:
 727                 self.assertEqual(called, [])
 728             except:
 729                 d.errback()
 730             else:
 731                 d.callback(None)
 732         reactor.callLater(0, reactor.callLater, 0, check)
 733         return d
 734
 735
 736     def test_cancelCancelledDelayedCall(self):
 737         """
 738         Test that cancelling a DelayedCall which has already been cancelled
 739         raises the appropriate exception.
 740         """
 741         call = reactor.callLater(0, lambda: None)
 742         call.cancel()
 743         self.assertRaises(error.AlreadyCancelled, call.cancel)
 744
 745
 746     def test_cancelCalledDelayedCallSynchronous(self):
 747         """
 748         Test that cancelling a DelayedCall in the DelayedCall's function as
 749         that function is being invoked by the DelayedCall raises the
 750         appropriate exception.
 751         """
 752         d = Deferred()
 753         def later():
 754             try:
 755                 self.assertRaises(error.AlreadyCalled, call.cancel)
 756             except:
 757                 d.errback()
 758             else:
 759                 d.callback(None)
 760         call = reactor.callLater(0, later)
 761         return d
 762
 763
 764     def test_cancelCalledDelayedCallAsynchronous(self):
 765         """
 766         Test that cancelling a DelayedCall after it has run its function
 767         raises the appropriate exception.
 768         """
 769         d = Deferred()
 770         def check():
 771             try:
 772                 self.assertRaises(error.AlreadyCalled, call.cancel)
 773             except:
 774                 d.errback()
 775             else:
 776                 d.callback(None)
 777         def later():
 778             reactor.callLater(0, check)
 779         call = reactor.callLater(0, later)
 780         return d
 781
 782
 783     def testCallLaterTime(self):
 784         d = reactor.callLater(10, lambda: None)
 785         try:
 786             self.failUnless(d.getTime() - (time.time() + 10) < 1)
 787         finally:
 788             d.cancel()
 789
 790     def testCallLaterOrder(self):
 791         l = []
 792         l2 = []
 793         def f(x):
 794             l.append(x)
 795         def f2(x):
 796             l2.append(x)
 797         def done():
 798             self.assertEqual(l, range(20))
 799         def done2():
 800             self.assertEqual(l2, range(10))
 801
 802         for n in range(10):
 803             reactor.callLater(0, f, n)
 804         for n in range(10):
 805             reactor.callLater(0, f, n+10)
 806             reactor.callLater(0.1, f2, n)
 807
 808         reactor.callLater(0, done)
 809         reactor.callLater(0.1, done2)
 810         d = Deferred()
 811         reactor.callLater(0.2, d.callback, None)
 812         return d
 813
 814     testCallLaterOrder.todo = "See bug 1396"
 815     testCallLaterOrder.skip = "Trial bug, todo doesn't work! See bug 1397"
 816     def testCallLaterOrder2(self):
 817         # This time destroy the clock resolution so that it fails reliably
 818         # even on systems that don't have a crappy clock resolution.
 819
 820         def seconds():
 821             return int(time.time())
 822
 823         base_original = base.seconds
 824         runtime_original = runtime.seconds
 825         base.seconds = seconds
 826         runtime.seconds = seconds
 827
 828         def cleanup(x):
 829             runtime.seconds = runtime_original
 830             base.seconds = base_original
 831             return x
 832         return maybeDeferred(self.testCallLaterOrder).addBoth(cleanup)
 833
 834     testCallLaterOrder2.todo = "See bug 1396"
 835     testCallLaterOrder2.skip = "Trial bug, todo doesn't work! See bug 1397"
 836
 837     def testDelayedCallStringification(self):
 838         # Mostly just make sure str() isn't going to raise anything for
 839         # DelayedCalls within reason.
 840         dc = reactor.callLater(0, lambda x, y: None, 'x', y=10)
 841         str(dc)
 842         dc.reset(5)
 843         str(dc)
 844         dc.cancel()
 845         str(dc)
 846
 847         dc = reactor.callLater(0, lambda: None, x=[({'hello': u'world'}, 10j), reactor], *range(10))
 848         str(dc)
 849         dc.cancel()
 850         str(dc)
 851
 852         def calledBack(ignored):
 853             str(dc)
 854         d = Deferred().addCallback(calledBack)
 855         dc = reactor.callLater(0, d.callback, None)
 856         str(dc)
 857         return d
 858
 859
 860     def testDelayedCallSecondsOverride(self):
 861         """
 862         Test that the C{seconds} argument to DelayedCall gets used instead of
 863         the default timing function, if it is not None.
 864         """
 865         def seconds():
 866             return 10
 867         dc = base.DelayedCall(5, lambda: None, (), {}, lambda dc: None,
 868                               lambda dc: None, seconds)
 869         self.assertEqual(dc.getTime(), 5)
 870         dc.reset(3)
 871         self.assertEqual(dc.getTime(), 13)
 872
 873
 874 class CallFromThreadTests(unittest.TestCase):
 875     def testWakeUp(self):
 876         # Make sure other threads can wake up the reactor
 877         d = Deferred()
 878         def wake():
 879             time.sleep(0.1)
 880             # callFromThread will call wakeUp for us
 881             reactor.callFromThread(d.callback, None)
 882         reactor.callInThread(wake)
 883         return d
 884
 885     if interfaces.IReactorThreads(reactor, None) is None:
 886         testWakeUp.skip = "Nothing to wake up for without thread support"
 887
 888     def _stopCallFromThreadCallback(self):
 889         self.stopped = True
 890
 891     def _callFromThreadCallback(self, d):
 892         reactor.callFromThread(self._callFromThreadCallback2, d)
 893         reactor.callLater(0, self._stopCallFromThreadCallback)
 894
 895     def _callFromThreadCallback2(self, d):
 896         try:
 897             self.assert_(self.stopped)
 898         except:
 899             # Send the error to the deferred
 900             d.errback()
 901         else:
 902             d.callback(None)
 903
 904     def testCallFromThreadStops(self):
 905         """
 906         Ensure that callFromThread from inside a callFromThread
 907         callback doesn't sit in an infinite loop and lets other
 908         things happen too.
 909         """
 910         self.stopped = False
 911         d = defer.Deferred()
 912         reactor.callFromThread(self._callFromThreadCallback, d)
 913         return d
 914
 915
 916 class DelayedTestCase(unittest.TestCase):
 917     def setUp(self):
 918         self.finished = 0
 919         self.counter = 0
 920         self.timers = {}
 921         self.deferred = defer.Deferred()
 922
 923     def tearDown(self):
 924         for t in self.timers.values():
 925             t.cancel()
 926
 927     def checkTimers(self):
 928         l1 = self.timers.values()
 929         l2 = list(reactor.getDelayedCalls())
 930
 931         # There should be at least the calls we put in.  There may be other
 932         # calls that are none of our business and that we should ignore,
 933         # though.
 934
 935         missing = []
 936         for dc in l1:
 937             if dc not in l2:
 938                 missing.append(dc)
 939         if missing:
 940             self.finished = 1
 941         self.failIf(missing, "Should have been missing no calls, instead was missing " + repr(missing))
 942
 943     def callback(self, tag):
 944         del self.timers[tag]
 945         self.checkTimers()
 946
 947     def addCallback(self, tag):
 948         self.callback(tag)
 949         self.addTimer(15, self.callback)
 950
 951     def done(self, tag):
 952         self.finished = 1
 953         self.callback(tag)
 954         self.deferred.callback(None)
 955
 956     def addTimer(self, when, callback):
 957         self.timers[self.counter] = reactor.callLater(when * 0.01, callback,
 958                                                       self.counter)
 959         self.counter += 1
 960         self.checkTimers()
 961
 962     def testGetDelayedCalls(self):
 963         if not hasattr(reactor, "getDelayedCalls"):
 964             return
 965         # This is not a race because we don't do anything which might call
 966         # the reactor until we have all the timers set up. If we did, this
 967         # test might fail on slow systems.
 968         self.checkTimers()
 969         self.addTimer(35, self.done)
 970         self.addTimer(20, self.callback)
 971         self.addTimer(30, self.callback)
 972         which = self.counter
 973         self.addTimer(29, self.callback)
 974         self.addTimer(25, self.addCallback)
 975         self.addTimer(26, self.callback)
 976
 977         self.timers[which].cancel()
 978         del self.timers[which]
 979         self.checkTimers()
 980
 981         self.deferred.addCallback(lambda x : self.checkTimers())
 982         return self.deferred
 983
 984
 985     def test_active(self):
 986         """
 987         L{IDelayedCall.active} returns False once the call has run.
 988         """
 989         dcall = reactor.callLater(0.01, self.deferred.callback, True)
 990         self.assertEqual(dcall.active(), True)
 991
 992         def checkDeferredCall(success):
 993             self.assertEqual(dcall.active(), False)
 994             return success
 995
 996         self.deferred.addCallback(checkDeferredCall)
 997
 998         return self.deferred
 999
1000
1001
1002 resolve_helper = """
1003 import %(reactor)s
1004 %(reactor)s.install()
1005 from twisted.internet import reactor
1006
1007 class Foo:
1008     def __init__(self):
1009         reactor.callWhenRunning(self.start)
1010         self.timer = reactor.callLater(3, self.failed)
1011     def start(self):
1012         reactor.resolve('localhost').addBoth(self.done)
1013     def done(self, res):
1014         print 'done', res
1015         reactor.stop()
1016     def failed(self):
1017         print 'failed'
1018         self.timer = None
1019         reactor.stop()
1020 f = Foo()
1021 reactor.run()
1022 """
1023
1024 class ChildResolveProtocol(protocol.ProcessProtocol):
1025     def __init__(self, onCompletion):
1026         self.onCompletion = onCompletion
1027
1028     def connectionMade(self):
1029         self.output = []
1030         self.error = []
1031
1032     def outReceived(self, out):
1033         self.output.append(out)
1034
1035     def errReceived(self, err):
1036         self.error.append(err)
1037
1038     def processEnded(self, reason):
1039         self.onCompletion.callback((reason, self.output, self.error))
1040         self.onCompletion = None
1041
1042
1043 class Resolve(unittest.TestCase):
1044     def testChildResolve(self):
1045         # I've seen problems with reactor.run under gtk2reactor. Spawn a
1046         # child which just does reactor.resolve after the reactor has
1047         # started, fail if it does not complete in a timely fashion.
1048         helperPath = os.path.abspath(self.mktemp())
1049         helperFile = open(helperPath, 'w')
1050
1051         # Eeueuuggg
1052         reactorName = reactor.__module__
1053
1054         helperFile.write(resolve_helper % {'reactor': reactorName})
1055         helperFile.close()
1056
1057         env = os.environ.copy()
1058         env['PYTHONPATH'] = os.pathsep.join(sys.path)
1059
1060         helperDeferred = Deferred()
1061         helperProto = ChildResolveProtocol(helperDeferred)
1062
1063         reactor.spawnProcess(helperProto, sys.executable, ("python", "-u", helperPath), env)
1064
1065         def cbFinished((reason, output, error)):
1066             # If the output is "done 127.0.0.1\n" we don't really care what
1067             # else happened.
1068             output = ''.join(output)
1069             if output != 'done 127.0.0.1\n':
1070                 self.fail((
1071                     "The child process failed to produce the desired results:\n"
1072                     "   Reason for termination was: %r\n"
1073                     "   Output stream was: %r\n"
1074                     "   Error stream was: %r\n") % (reason.getErrorMessage(), output, ''.join(error)))
1075
1076         helperDeferred.addCallback(cbFinished)
1077         return helperDeferred
1078
1079 if not interfaces.IReactorProcess(reactor, None):
1080     Resolve.skip = "cannot run test: reactor doesn't support IReactorProcess"
1081
1082
1083
1084 class CallFromThreadTestCase(unittest.TestCase):
1085     """
1086     Task scheduling from threads tests.
1087     """
1088     if interfaces.IReactorThreads(reactor, None) is None:
1089         skip = "Nothing to test without thread support"
1090
1091     def setUp(self):
1092         self.counter = 0
1093         self.deferred = Deferred()
1094
1095
1096     def schedule(self, *args, **kwargs):
1097         """
1098         Override in subclasses.
1099         """
1100         reactor.callFromThread(*args, **kwargs)
1101
1102
1103     def test_lotsOfThreadsAreScheduledCorrectly(self):
1104         """
1105         L{IReactorThreads.callFromThread} can be used to schedule a large
1106         number of calls in the reactor thread.
1107         """
1108         def addAndMaybeFinish():
1109             self.counter += 1
1110             if self.counter == 100:
1111                 self.deferred.callback(True)
1112
1113         for i in xrange(100):
1114             self.schedule(addAndMaybeFinish)
1115
1116         return self.deferred
1117
1118
1119     def test_threadsAreRunInScheduledOrder(self):
1120         """
1121         Callbacks should be invoked in the order they were scheduled.
1122         """
1123         order = []
1124
1125         def check(_):
1126             self.assertEqual(order, [1, 2, 3])
1127
1128         self.deferred.addCallback(check)
1129         self.schedule(order.append, 1)
1130         self.schedule(order.append, 2)
1131         self.schedule(order.append, 3)
1132         self.schedule(reactor.callFromThread, self.deferred.callback, None)
1133
1134         return self.deferred
1135
1136
1137     def test_scheduledThreadsNotRunUntilReactorRuns(self):
1138         """
1139         Scheduled tasks should not be run until the reactor starts running.
1140         """
1141         def incAndFinish():
1142             self.counter = 1
1143             self.deferred.callback(True)
1144         self.schedule(incAndFinish)
1145
1146         # Callback shouldn't have fired yet.
1147         self.assertEqual(self.counter, 0)
1148
1149         return self.deferred
1150
1151
1152
1153 class MyProtocol(protocol.Protocol):
1154     """
1155     Sample protocol.
1156     """
1157
1158 class MyFactory(protocol.Factory):
1159     """
1160     Sample factory.
1161     """
1162
1163     protocol = MyProtocol
1164
1165
1166 class ProtocolTestCase(unittest.TestCase):
1167
1168     def testFactory(self):
1169         factory = MyFactory()
1170         protocol = factory.buildProtocol(None)
1171         self.assertEqual(protocol.factory, factory)
1172         self.assert_( isinstance(protocol, factory.protocol) )
1173
1174
1175 class DummyProducer(object):
1176     """
1177     Very uninteresting producer implementation used by tests to ensure the
1178     right methods are called by the consumer with which it is registered.
1179
1180     @type events: C{list} of C{str}
1181     @ivar events: The producer/consumer related events which have happened to
1182     this producer.  Strings in this list may be C{'resume'}, C{'stop'}, or
1183     C{'pause'}.  Elements are added as they occur.
1184     """
1185
1186     def __init__(self):
1187         self.events = []
1188
1189
1190     def resumeProducing(self):
1191         self.events.append('resume')
1192
1193
1194     def stopProducing(self):
1195         self.events.append('stop')
1196
1197
1198     def pauseProducing(self):
1199         self.events.append('pause')
1200
1201
1202
1203 class SillyDescriptor(abstract.FileDescriptor):
1204     """
1205     A descriptor whose data buffer gets filled very fast.
1206
1207     Useful for testing FileDescriptor's IConsumer interface, since
1208     the data buffer fills as soon as at least four characters are
1209     written to it, and gets emptied in a single doWrite() cycle.
1210     """
1211     bufferSize = 3
1212     connected = True
1213
1214     def writeSomeData(self, data):
1215         """
1216         Always write all data.
1217         """
1218         return len(data)
1219
1220
1221     def startWriting(self):
1222         """
1223         Do nothing: bypass the reactor.
1224         """
1225     stopWriting = startWriting
1226
1227
1228
1229 class ReentrantProducer(DummyProducer):
1230     """
1231     Similar to L{DummyProducer}, but with a resumeProducing method which calls
1232     back into an L{IConsumer} method of the consumer against which it is
1233     registered.
1234
1235     @ivar consumer: The consumer with which this producer has been or will
1236     be registered.
1237
1238     @ivar methodName: The name of the method to call on the consumer inside
1239     C{resumeProducing}.
1240
1241     @ivar methodArgs: The arguments to pass to the consumer method invoked in
1242     C{resumeProducing}.
1243     """
1244     def __init__(self, consumer, methodName, *methodArgs):
1245         super(ReentrantProducer, self).__init__()
1246         self.consumer = consumer
1247         self.methodName = methodName
1248         self.methodArgs = methodArgs
1249
1250
1251     def resumeProducing(self):
1252         super(ReentrantProducer, self).resumeProducing()
1253         getattr(self.consumer, self.methodName)(*self.methodArgs)
1254
1255
1256
1257 class TestProducer(unittest.TestCase):
1258     """
1259     Test abstract.FileDescriptor's consumer interface.
1260     """
1261     def test_doubleProducer(self):
1262         """
1263         Verify that registering a non-streaming producer invokes its
1264         resumeProducing() method and that you can only register one producer
1265         at a time.
1266         """
1267         fd = abstract.FileDescriptor()
1268         fd.connected = 1
1269         dp = DummyProducer()
1270         fd.registerProducer(dp, 0)
1271         self.assertEqual(dp.events, ['resume'])
1272         self.assertRaises(RuntimeError, fd.registerProducer, DummyProducer(), 0)
1273
1274
1275     def test_unconnectedFileDescriptor(self):
1276         """
1277         Verify that registering a producer when the connection has already
1278         been closed invokes its stopProducing() method.
1279         """
1280         fd = abstract.FileDescriptor()
1281         fd.disconnected = 1
1282         dp = DummyProducer()
1283         fd.registerProducer(dp, 0)
1284         self.assertEqual(dp.events, ['stop'])
1285
1286
1287     def _dontPausePullConsumerTest(self, methodName):
1288         descriptor = SillyDescriptor()
1289         producer = DummyProducer()
1290         descriptor.registerProducer(producer, streaming=False)
1291         self.assertEqual(producer.events, ['resume'])
1292         del producer.events[:]
1293
1294         # Fill up the descriptor's write buffer so we can observe whether or
1295         # not it pauses its producer in that case.
1296         getattr(descriptor, methodName)('1234')
1297
1298         self.assertEqual(producer.events, [])
1299
1300
1301     def test_dontPausePullConsumerOnWrite(self):
1302         """
1303         Verify that FileDescriptor does not call producer.pauseProducing() on a
1304         non-streaming pull producer in response to a L{IConsumer.write} call
1305         which results in a full write buffer. Issue #2286.
1306         """
1307         return self._dontPausePullConsumerTest('write')
1308
1309
1310     def test_dontPausePullConsumerOnWriteSequence(self):
1311         """
1312         Like L{test_dontPausePullConsumerOnWrite}, but for a call to
1313         C{writeSequence} rather than L{IConsumer.write}.
1314
1315         C{writeSequence} is not part of L{IConsumer}, but
1316         L{abstract.FileDescriptor} has supported consumery behavior in response
1317         to calls to L{writeSequence} forever.
1318         """
1319         return self._dontPausePullConsumerTest('writeSequence')
1320
1321
1322     def _reentrantStreamingProducerTest(self, methodName):
1323         descriptor = SillyDescriptor()
1324         producer = ReentrantProducer(descriptor, methodName, 'spam')
1325         descriptor.registerProducer(producer, streaming=True)
1326
1327         # Start things off by filling up the descriptor's buffer so it will
1328         # pause its producer.
1329         getattr(descriptor, methodName)('spam')
1330
1331         # Sanity check - make sure that worked.
1332         self.assertEqual(producer.events, ['pause'])
1333         del producer.events[:]
1334
1335         # After one call to doWrite, the buffer has been emptied so the
1336         # FileDescriptor should resume its producer.  That will result in an
1337         # immediate call to FileDescriptor.write which will again fill the
1338         # buffer and result in the producer being paused.
1339         descriptor.doWrite()
1340         self.assertEqual(producer.events, ['resume', 'pause'])
1341         del producer.events[:]
1342
1343         # After a second call to doWrite, the exact same thing should have
1344         # happened.  Prior to the bugfix for which this test was written,
1345         # FileDescriptor would have incorrectly believed its producer was
1346         # already resumed (it was paused) and so not resume it again.
1347         descriptor.doWrite()
1348         self.assertEqual(producer.events, ['resume', 'pause'])
1349
1350
1351     def test_reentrantStreamingProducerUsingWrite(self):
1352         """
1353         Verify that FileDescriptor tracks producer's paused state correctly.
1354         Issue #811, fixed in revision r12857.
1355         """
1356         return self._reentrantStreamingProducerTest('write')
1357
1358
1359     def test_reentrantStreamingProducerUsingWriteSequence(self):
1360         """
1361         Like L{test_reentrantStreamingProducerUsingWrite}, but for calls to
1362         C{writeSequence}.
1363
1364         C{writeSequence} is B{not} part of L{IConsumer}, however
1365         C{abstract.FileDescriptor} has supported consumery behavior in response
1366         to calls to C{writeSequence} forever.
1367         """
1368         return self._reentrantStreamingProducerTest('writeSequence')
1369
1370
1371
1372 class PortStringification(unittest.TestCase):
1373     if interfaces.IReactorTCP(reactor, None) is not None:
1374         def testTCP(self):
1375             p = reactor.listenTCP(0, protocol.ServerFactory())
1376             portNo = p.getHost().port
1377             self.assertNotEqual(str(p).find(str(portNo)), -1,
1378                                 "%d not found in %s" % (portNo, p))
1379             return p.stopListening()
1380
1381     if interfaces.IReactorUDP(reactor, None) is not None:
1382         def testUDP(self):
1383             p = reactor.listenUDP(0, protocol.DatagramProtocol())
1384             portNo = p.getHost().port
1385             self.assertNotEqual(str(p).find(str(portNo)), -1,
1386                                 "%d not found in %s" % (portNo, p))
1387             return p.stopListening()
1388
1389     if interfaces.IReactorSSL(reactor, None) is not None and ssl:
1390         def testSSL(self, ssl=ssl):
1391             pem = util.sibpath(__file__, 'server.pem')
1392             p = reactor.listenSSL(0, protocol.ServerFactory(), ssl.DefaultOpenSSLContextFactory(pem, pem))
1393             portNo = p.getHost().port
1394             self.assertNotEqual(str(p).find(str(portNo)), -1,
1395                                 "%d not found in %s" % (portNo, p))
1396             return p.stopListening()