@ -8,11 +8,10 @@ import attr
from zope . interface import implementer
from twisted . internet import address , threads , udp
from twisted . internet . _resolver import HostResolution
from twisted . internet . address import IPv4Address
from twisted . internet . defer import Deferred
from twisted . internet . _resolver import SimpleResolverComplexifier
from twisted . internet . defer import Deferred , fail , succeed
from twisted . internet . error import DNSLookupError
from twisted . internet . interfaces import IReactorPluggableNameResolver
from twisted . internet . interfaces import IReactorPluggableNameResolver , IResolverSimple
from twisted . python . failure import Failure
from twisted . test . proto_helpers import MemoryReactorClock
from twisted . web . http import unquote
@ -227,30 +226,16 @@ class ThreadedMemoryReactorClock(MemoryReactorClock):
def __init__ ( self ) :
self . _udp = [ ]
self . lookups = { }
class Resolver ( object ) :
def resolveHostName (
_self ,
resolutionReceiver ,
hostName ,
portNumber = 0 ,
addressTypes = None ,
transportSemantics = ' TCP ' ,
) :
resolution = HostResolution ( hostName )
resolutionReceiver . resolutionBegan ( resolution )
if hostName not in self . lookups :
raise DNSLookupError ( " OH NO " )
lookups = self . lookups = { }
resolutionReceiver . addressResolved (
IPv4Address ( ' TCP ' , self . lookups [ hostName ] , portNumber )
)
resolutionReceiver . resolutionComplete ( )
return resolution
@implementer ( IResolverSimple )
class FakeResolver ( object ) :
def getHostByName ( self , name , timeout = None ) :
if name not in lookups :
return fail ( DNSLookupError ( " OH NO: unknown %s " % ( name , ) ) )
return succeed ( lookups [ name ] )
self . nameResolver = Resolver ( )
self . nameResolver = SimpleResolverComplexifier ( FakeResolver ( ) )
super ( ThreadedMemoryReactorClock , self ) . __init__ ( )
def listenUDP ( self , port , protocol , interface = ' ' , maxPacketSize = 8196 ) :