11from twilio .base .exceptions import TwilioRestException
2- from twilio .http import HttpClient
2+ from twilio .http import HttpClient , AsyncHttpClient
33from twilio .http .request import Request
44import platform
55from twilio import __version__
@@ -15,6 +15,7 @@ class Holodeck(HttpClient):
1515 def __init__ (self ):
1616 self ._holograms = []
1717 self ._requests = []
18+ self .is_async = False
1819
1920 def mock (self , response , request = None ):
2021 request = request or Request ()
@@ -87,3 +88,86 @@ def request(
8788 message += "No Holograms loaded"
8889
8990 raise TwilioRestException (404 , url , message , method = method )
91+
92+
93+ class AsyncHolodeck (AsyncHttpClient ):
94+ """Async version of Holodeck for async testing"""
95+
96+ def __init__ (self ):
97+ # Don't call super().__init__() to avoid logger requirement
98+ self ._holograms = []
99+ self ._requests = []
100+ self .is_async = True
101+
102+ def mock (self , response , request = None ):
103+ request = request or Request ()
104+ self ._holograms .append (Hologram (request , response ))
105+
106+ @property
107+ def requests (self ):
108+ return self ._requests
109+
110+ def add_standard_headers (self , request ):
111+ standard_headers = {
112+ "User-Agent" : "twilio-python/{} ({} {}) Python/{}" .format (
113+ __version__ ,
114+ platform .system (),
115+ platform .machine (),
116+ platform .python_version (),
117+ ),
118+ "X-Twilio-Client" : "python-{}" .format (__version__ ),
119+ "Accept" : "application/json" ,
120+ "Accept-Charset" : "utf-8" ,
121+ }
122+
123+ if request .method == "POST" and "Content-Type" not in standard_headers :
124+ standard_headers ["Content-Type" ] = "application/x-www-form-urlencoded"
125+
126+ standard_headers .update (request .headers )
127+ request .headers = standard_headers
128+ return request
129+
130+ def assert_has_request (self , request ):
131+ for req in self .requests :
132+ if req == request or req == self .add_standard_headers (request ):
133+ return
134+
135+ message = "\n Holodeck never received a request matching: \n + {}\n " .format (
136+ request
137+ )
138+ if self ._requests :
139+ message += "Requests received:\n "
140+ message += "\n " .join (" * {}" .format (r ) for r in self .requests )
141+ else :
142+ message += "No Requests received"
143+
144+ raise AssertionError (message )
145+
146+ async def request (
147+ self ,
148+ method ,
149+ url ,
150+ params = None ,
151+ data = None ,
152+ headers = None ,
153+ auth = None ,
154+ timeout = None ,
155+ allow_redirects = False ,
156+ ):
157+ """Async request method"""
158+ request = Request (method , url , auth , params , data , headers )
159+
160+ self ._requests .append (request )
161+
162+ for hologram in self ._holograms :
163+ if hologram .request == request :
164+ return hologram .response
165+
166+ message = "\n Holodeck has no hologram for: {}\n " .format (request )
167+ if self ._holograms :
168+ message += "Holograms loaded:\n "
169+ message += "\n " .join (" - {}" .format (h .request ) for h in self ._holograms )
170+ else :
171+ message += "No Holograms loaded"
172+
173+ raise TwilioRestException (404 , url , message , method = method )
0 commit comments