Skip to content

Commit

Permalink
Python Orgs Api Changes
Browse files Browse the repository at this point in the history
  • Loading branch information
@AsabuHere
AsabuHere committed Sep 17, 2024
1 parent b4c5734 commit 3e246e4
Show file tree
Hide file tree
Showing 17 changed files with 1,488 additions and 197 deletions.
9 changes: 9 additions & 0 deletions tests/cluster/test_cluster.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,15 @@ def setUp(self):
)
self.voice_twiml = VoiceResponse()


def test_token_fetch(self):
token = self.client.preview_iam.token.create(
grant_type = GRANT_TYPE,
client_id = CLIENT_ID,
client_secret = CLIENT_SECRET)
print(f'{token}')


def test_send_text_message(self):
msg = self.client.messages.create(
to=self.to_number, from_=self.from_number, body="hello world"
Expand Down
156 changes: 78 additions & 78 deletions tests/cluster/test_webhook.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -29,81 +29,81 @@ def process_request(self):
)


# class WebhookTest(unittest.TestCase):
# def setUp(self):
# api_key = os.environ["TWILIO_API_KEY"]
# api_secret = os.environ["TWILIO_API_SECRET"]
# account_sid = os.environ["TWILIO_ACCOUNT_SID"]
# self.client = Client(api_key, api_secret, account_sid)
#
# portNumber = 7777
# self.validation_server = HTTPServer(("", portNumber), RequestHandler)
# self.tunnel = ngrok.connect(portNumber)
# self.flow_sid = ""
# _thread.start_new_thread(self.start_http_server, ())
#
# def start_http_server(self):
# self.validation_server.serve_forever()
#
# def tearDown(self):
# self.client.studio.v2.flows(self.flow_sid).delete()
# ngrok.kill()
# self.validation_server.shutdown()
# self.validation_server.server_close()
#
# def create_studio_flow(self, url, method):
# flow = self.client.studio.v2.flows.create(
# friendly_name="Python Cluster Test Flow",
# status="published",
# definition={
# "description": "Studio Flow",
# "states": [
# {
# "name": "Trigger",
# "type": "trigger",
# "transitions": [
# {
# "next": "httpRequest",
# "event": "incomingRequest",
# },
# ],
# "properties": {},
# },
# {
# "name": "httpRequest",
# "type": "make-http-request",
# "transitions": [],
# "properties": {
# "method": method,
# "content_type": "application/x-www-form-urlencoded;charset=utf-8",
# "url": url,
# },
# },
# ],
# "initial_state": "Trigger",
# "flags": {
# "allow_concurrent_calls": True,
# },
# },
# )
# return flow
#
# def validate(self, method):
# flow = self.create_studio_flow(url=self.tunnel.public_url, method=method)
# self.flow_sid = flow.sid
# time.sleep(5)
# self.client.studio.v2.flows(self.flow_sid).executions.create(
# to="to", from_="from"
# )
#
# def test_get(self):
# time.sleep(5)
# self.validate("GET")
# time.sleep(5)
# self.assertEqual(RequestHandler.is_request_valid, True)
#
# def test_post(self):
# time.sleep(5)
# self.validate("POST")
# time.sleep(5)
# self.assertEqual(RequestHandler.is_request_valid, True)
class WebhookTest(unittest.TestCase):
def setUp(self):
api_key = os.environ["TWILIO_API_KEY"]
api_secret = os.environ["TWILIO_API_SECRET"]
account_sid = os.environ["TWILIO_ACCOUNT_SID"]
self.client = Client(api_key, api_secret, account_sid)

portNumber = 7777
self.validation_server = HTTPServer(("", portNumber), RequestHandler)
self.tunnel = ngrok.connect(portNumber)
self.flow_sid = ""
_thread.start_new_thread(self.start_http_server, ())

def start_http_server(self):
self.validation_server.serve_forever()

def tearDown(self):
self.client.studio.v2.flows(self.flow_sid).delete()
ngrok.kill()
self.validation_server.shutdown()
self.validation_server.server_close()

def create_studio_flow(self, url, method):
flow = self.client.studio.v2.flows.create(
friendly_name="Python Cluster Test Flow",
status="published",
definition={
"description": "Studio Flow",
"states": [
{
"name": "Trigger",
"type": "trigger",
"transitions": [
{
"next": "httpRequest",
"event": "incomingRequest",
},
],
"properties": {},
},
{
"name": "httpRequest",
"type": "make-http-request",
"transitions": [],
"properties": {
"method": method,
"content_type": "application/x-www-form-urlencoded;charset=utf-8",
"url": url,
},
},
],
"initial_state": "Trigger",
"flags": {
"allow_concurrent_calls": True,
},
},
)
return flow

def validate(self, method):
flow = self.create_studio_flow(url=self.tunnel.public_url, method=method)
self.flow_sid = flow.sid
time.sleep(5)
self.client.studio.v2.flows(self.flow_sid).executions.create(
to="to", from_="from"
)

def test_get(self):
time.sleep(5)
self.validate("GET")
time.sleep(5)
self.assertEqual(RequestHandler.is_request_valid, True)

def test_post(self):
time.sleep(5)
self.validate("POST")
time.sleep(5)
self.assertEqual(RequestHandler.is_request_valid, True)
30 changes: 21 additions & 9 deletions twilio/base/client_base.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -70,7 +70,7 @@ def request(
timeout: Optional[float] = None,
allow_redirects: bool = False,
is_oauth: bool = False,
domain: Optional[str] = None
domain: Optional[str] = None,
) -> Response:
"""
Makes a request to the Twilio API using the configured http client
Expand All @@ -87,15 +87,21 @@ def request(
:returns: Response from the Twilio API
"""

print('*****')
if not is_oauth:
auth = self.get_auth(auth)
headers = self.get_headers(method, headers)
uri = self.get_hostname(uri)
if is_oauth:
OauthTokenBase = dynamic_import("twilio.base.oauth_token_base", "OauthTokenBase")
token = OauthTokenBase().get_oauth_token(domain, "v1", self.username, self.password)
headers['Authorization'] = f'Bearer {token}'
headers.get('Authorization')
OauthTokenBase = dynamic_import(
"twilio.base.oauth_token_base", "OauthTokenBase"
)
token = OauthTokenBase().get_oauth_token(
domain, "v1", self.username, self.password
)
headers["Authorization"] = f"Bearer {token}"
headers.get("Authorization")

return self.http_client.request(
method,
Expand Down Expand Up @@ -145,10 +151,14 @@ async def request_async(
headers = self.get_headers(method, headers)
uri = self.get_hostname(uri)
if is_oauth:
OauthTokenBase = dynamic_import("twilio.base.oauth_token_base", "OauthTokenBase")
token = OauthTokenBase().get_oauth_token(domain, "v1", self.username, self.password)
headers['Authorization'] = f'Bearer {token}'
headers.get('Authorization')
OauthTokenBase = dynamic_import(
"twilio.base.oauth_token_base", "OauthTokenBase"
)
token = OauthTokenBase().get_oauth_token(
domain, "v1", self.username, self.password
)
headers["Authorization"] = f"Bearer {token}"
headers.get("Authorization")

return await self.http_client.request(
method,
Expand Down Expand Up @@ -247,7 +257,9 @@ def __repr__(self) -> str:
"""
return "<Twilio {}>".format(self.account_sid)


def dynamic_import(module_name, class_name):
from importlib import import_module

module = import_module(module_name)
return getattr(module, class_name)
2 changes: 1 addition & 1 deletion twilio/base/domain.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -94,5 +94,5 @@ async def request_async(
auth=auth,
timeout=timeout,
allow_redirects=allow_redirects,
is_oauth=is_oauth
is_oauth=is_oauth,
)
3 changes: 2 additions & 1 deletion twilio/base/exceptions.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -62,7 +62,8 @@ def get_uri(code: int) -> str:
"\n\n{twilio_returned}\n\n{message}\n".format(
red_error=red("HTTP Error"),
request_was=white("Your request was:"),
http_line=teal("%s %s" % (self.method, self.uri)),
http_line=teal("%s %s" % (self.method, self.uri, self.data, self.uri)),
http_line=teal("%s %s" % (self.data, self.headers)),
twilio_returned=white("Twilio returned the following information:"),
message=blue(str(self.msg)),
)
Expand Down
27 changes: 20 additions & 7 deletions twilio/base/oauth_token_base.py
View file
Original file line number Diff line number Diff line change
@@ -1,24 +1,37 @@
from twilio.http.token_manager_initializer import TokenManagerInitializer


# Dynamic import utility function
def dynamic_import(module_name, class_name):
from importlib import import_module

module = import_module(module_name)
return getattr(module, class_name)


class OauthTokenBase:
def get_oauth_token(self, domain: str, version: str, username: str, password: str):
Domain = dynamic_import("twilio.base.domain", "Domain")
Version = dynamic_import("twilio.base.version", "Version")
BearerTokenHTTPClient = dynamic_import("twilio.http.bearer_token_http_client", "BearerTokenHTTPClient")
OrgTokenManager = dynamic_import("twilio.http.orgs_token_manager", "OrgTokenManager")
BearerTokenHTTPClient = dynamic_import(
"twilio.http.bearer_token_http_client", "BearerTokenHTTPClient"
)
OrgTokenManager = dynamic_import(
"twilio.http.orgs_token_manager", "OrgTokenManager"
)
Client = dynamic_import("twilio.rest", "Client")
try:
orgs_token_manager = TokenManagerInitializer.get_token_manager()
return BearerTokenHTTPClient(orgs_token_manager).get_access_token(Version(Domain(Client(username, password), domain), version))
return BearerTokenHTTPClient(orgs_token_manager).get_access_token(
Version(Domain(Client(username, password), domain), version)
)
except Exception:
orgs_token_manager = OrgTokenManager(grant_type='client_credentials',
client_id=username,
client_secret=password)
orgs_token_manager = OrgTokenManager(
grant_type="client_credentials",
client_id=username,
client_secret=password,
)
TokenManagerInitializer().set_token_manager(orgs_token_manager)
return BearerTokenHTTPClient(orgs_token_manager).get_access_token(Version(Domain(Client(username, password), domain), version))
return BearerTokenHTTPClient(orgs_token_manager).get_access_token(
Version(Domain(Client(username, password), domain), version)
)
Loading

0 comments on commit 3e246e4

Please sign in to comment.