|
| 1 | +# coding: utf-8 |
| 2 | + |
1 | 3 | """ |
2 | | -Integration test for Bandwidth's Phone Number Lookup API |
3 | | -""" |
| 4 | + Bandwidth |
4 | 5 |
|
5 | | -import json |
6 | | -import time |
7 | | -import unittest |
| 6 | + Bandwidth's Communication APIs |
8 | 7 |
|
9 | | -import bandwidth |
10 | | -from bandwidth.api import phone_number_lookup_api |
11 | | -from bandwidth.models.lookup_request import LookupRequest |
12 | | -from bandwidth.models.create_lookup_response import CreateLookupResponse |
13 | | -from bandwidth.models.lookup_status import LookupStatus |
14 | | -from bandwidth.models.lookup_result import LookupResult |
15 | | -from bandwidth.models.lookup_status_enum import LookupStatusEnum |
16 | | -from bandwidth.models.tn_lookup_request_error import TnLookupRequestError |
17 | | -from bandwidth.exceptions import ApiException, UnauthorizedException, ForbiddenException |
18 | | -from test.utils.env_variables import * |
| 8 | + The version of the OpenAPI document: 1.0.0 |
| 9 | + |
| 10 | + Generated by OpenAPI Generator (https://openapi-generator.tech) |
| 11 | +
|
| 12 | + Do not edit the manually. |
| 13 | +""" # noqa: E501 |
19 | 14 |
|
20 | | -from hamcrest.core import * |
21 | | -from hamcrest.library import * |
22 | 15 |
|
23 | | -from .bwmatchers.one_of_string import is_one_of_string |
| 16 | +import unittest |
| 17 | +from datetime import date |
| 18 | +from uuid import UUID |
| 19 | +from time import sleep |
| 20 | + |
| 21 | +from hamcrest import * |
| 22 | +from bandwidth import ApiClient, Configuration |
| 23 | +from bandwidth.api.phone_number_lookup_api import PhoneNumberLookupApi |
| 24 | +from bandwidth.models.async_lookup_request import AsyncLookupRequest |
| 25 | +from bandwidth.models.create_async_bulk_lookup_response import CreateAsyncBulkLookupResponse |
| 26 | +from bandwidth.models.link_schema import LinkSchema |
| 27 | +from bandwidth.models.create_async_bulk_lookup_response_data import CreateAsyncBulkLookupResponseData |
| 28 | +from bandwidth.models.in_progress_lookup_status_enum import InProgressLookupStatusEnum |
| 29 | +from bandwidth.models.sync_lookup_request import SyncLookupRequest |
| 30 | +from bandwidth.models.create_sync_lookup_response import CreateSyncLookupResponse |
| 31 | +from bandwidth.models.completed_lookup_status_enum import CompletedLookupStatusEnum |
| 32 | +from bandwidth.models.lookup_result import LookupResult |
| 33 | +from bandwidth.models.line_type_enum import LineTypeEnum |
| 34 | +from bandwidth.models.latest_message_delivery_status_enum import LatestMessageDeliveryStatusEnum |
| 35 | +from bandwidth.models.get_async_bulk_lookup_response import GetAsyncBulkLookupResponse |
24 | 36 |
|
| 37 | +from test.utils.env_variables import * |
25 | 38 |
|
26 | | -class TestPhoneNumberLookupIntegration(unittest.TestCase): |
27 | | - """Phone Number Lookup API integration test |
28 | | - """ |
| 39 | +class TestPhoneNumberLookupApi(unittest.TestCase): |
| 40 | + """PhoneNumberLookupApi unit test stubs""" |
29 | 41 |
|
30 | 42 | def setUp(self) -> None: |
31 | | - configuration = bandwidth.Configuration( |
| 43 | + configuration = Configuration( |
32 | 44 | username=BW_USERNAME, |
33 | 45 | password=BW_PASSWORD |
34 | 46 | ) |
35 | | - api_client = bandwidth.ApiClient(configuration) |
36 | | - self.api_instance = phone_number_lookup_api.PhoneNumberLookupApi(api_client) |
37 | | - self.account_id = BW_ACCOUNT_ID |
| 47 | + api_client = ApiClient(configuration) |
| 48 | + self.api = PhoneNumberLookupApi(api_client) |
38 | 49 |
|
39 | | - def validateResult(self, result: LookupResult, e_164_format: str) -> None: |
40 | | - """Verify a successful phone number lookup LookupResult object |
| 50 | + self.phone_numbers = [BW_NUMBER, USER_NUMBER] |
41 | 51 |
|
42 | | - Args: |
43 | | - result (LookupResult): Result derived from LookupStatus result list |
44 | | - e_164_format (str): Phone number in e164 format ex: +19195551234 |
45 | | - line_provider (str): Line service provider ex: Verizon |
46 | | - """ |
47 | | - |
48 | | - # if result has 1 of these attributes it should have the other |
49 | | - if result.mobile_country_code or result.mobile_network_code: |
50 | | - self.assertIs(type(result.mobile_country_code), str) |
51 | | - self.assertIs(type(result.mobile_network_code), str) |
52 | | - |
53 | | - assert_that(result, has_properties( |
54 | | - 'response_code', 0, |
55 | | - 'e_164_format', e_164_format, |
56 | | - 'country', is_one_of_string(["US", "Canada"]), |
57 | | - 'line_type', is_one_of_string(["Mobile", "Fixed"]) |
58 | | - ) |
59 | | - ) |
60 | | - self.assertIs(type(result.line_provider), str) |
| 52 | + def tearDown(self) -> None: |
| 53 | + pass |
61 | 54 |
|
62 | | - def pollLookupStatus(self, request_id: str) -> LookupStatus: |
63 | | - """Poll LookupRequest for 'COMPLETE' status |
| 55 | + def test_create_get_async_bulk_lookup(self) -> None: |
| 56 | + """Test case for create_async_bulk_lookup |
64 | 57 |
|
65 | | - Args: |
66 | | - request_id (str): LookupResult.request_id value to query |
67 | | -
|
68 | | - Raises: |
69 | | - Exception: Tries 5 times and raises a general exception if the query takes more than 5 attempts to minimize run time. |
70 | | -
|
71 | | - Returns: |
72 | | - LookupStatus: LookupStatus in 'COMPLETE' state |
73 | | - """ |
74 | | - get_lookup_status_response: LookupStatus = self.api_instance.get_lookup_status( |
75 | | - self.account_id, request_id) |
76 | | - get_lookup_status_response_attempts = 1 |
77 | | - while get_lookup_status_response.status != LookupStatusEnum('COMPLETE'): |
78 | | - # Raise an error if it takes more than 5 requests to get COMPLETE status |
79 | | - if get_lookup_status_response_attempts == 5: |
80 | | - raise Exception( |
81 | | - f'Took too long to get phone number lookup \'COMPLETE\' status. Aborting test after {get_lookup_status_response_attempts} attempts.') |
82 | | - time.sleep(2) |
83 | | - |
84 | | - get_lookup_status_response: LookupStatus = self.api_instance.get_lookup_status( |
85 | | - self.account_id, request_id) |
86 | | - get_lookup_status_response_attempts += 1 |
87 | | - |
88 | | - return get_lookup_status_response |
89 | | - |
90 | | - def assertAuthException(self, context: ApiException, expectedException: ApiException, expected_status_code: int) -> None: |
91 | | - """Validates that an auth exception (401 or 403) is properly formatted |
92 | | -
|
93 | | - Args: |
94 | | - context (ApiException): Exception to validate |
95 | | - expectedException (ApiException): Expected exception type |
96 | | - expected_status_code (int): Expected status code |
| 58 | + Create Asynchronous Bulk Number Lookup |
97 | 59 | """ |
98 | | - self.assertIs(type(context.exception), expectedException) |
99 | | - self.assertIs(type(context.exception.status), int) |
100 | | - self.assertEqual(context.exception.status, expected_status_code) |
101 | | - self.assertIs(type(context.exception.body), str) |
102 | | - |
103 | | - def testSuccessfulPhoneNumberLookup(self) -> None: |
104 | | - """Test Phone Number Lookup API |
| 60 | + request = AsyncLookupRequest(phone_numbers=self.phone_numbers) |
| 61 | + |
| 62 | + response = self.api.create_async_bulk_lookup_with_http_info(BW_ACCOUNT_ID, request) |
| 63 | + assert_that(response, is_not(none())) |
| 64 | + assert_that(response.status_code, equal_to(202)) |
| 65 | + assert_that(response.data, is_not(none())) |
| 66 | + assert_that(response.data, instance_of(CreateAsyncBulkLookupResponse)) |
| 67 | + assert_that(response.data.links, is_not(none())) |
| 68 | + assert_that(response.data.links[0], instance_of(LinkSchema)) |
| 69 | + assert_that(response.data.links[0].rel, instance_of(str)) |
| 70 | + assert_that(response.data.links[0].href, instance_of(str)) |
| 71 | + assert_that(response.data.links[0].method, instance_of(str)) |
| 72 | + assert_that(response.data.data, is_not(none())) |
| 73 | + assert_that(response.data.data, instance_of(CreateAsyncBulkLookupResponseData)) |
| 74 | + assert_that(response.data.data.request_id, instance_of(UUID)) |
| 75 | + assert_that(response.data.data.status, instance_of(InProgressLookupStatusEnum)) |
| 76 | + assert_that(response.data.errors, is_not(none())) |
| 77 | + assert_that(response.data.errors, instance_of(list)) |
| 78 | + |
| 79 | + request_id = response.data.data.request_id |
| 80 | + sleep(10) |
| 81 | + |
| 82 | + response = self.api.get_async_bulk_lookup_with_http_info(BW_ACCOUNT_ID, request_id) |
| 83 | + assert_that(response, is_not(none())) |
| 84 | + assert_that(response.status_code, equal_to(200)) |
| 85 | + assert_that(response.data, is_not(none())) |
| 86 | + assert_that(response.data, instance_of(GetAsyncBulkLookupResponse)) |
| 87 | + print(response.data) |
| 88 | + assert_that(response.data.links, is_not(none())) |
| 89 | + assert_that(response.data.links[0], instance_of(LinkSchema)) |
| 90 | + assert_that(response.data.links[0].rel, instance_of(str)) |
| 91 | + assert_that(response.data.links[0].href, instance_of(str)) |
| 92 | + assert_that(response.data.links[0].method, instance_of(str)) |
| 93 | + assert_that(response.data.data, is_not(none())) |
| 94 | + assert_that(response.data.data.request_id, instance_of(UUID)) |
| 95 | + assert_that(response.data.data.status, instance_of(InProgressLookupStatusEnum)) |
| 96 | + assert_that(response.data.data.status, equal_to(InProgressLookupStatusEnum('COMPLETE'))) |
| 97 | + assert_that(response.data.data.results, instance_of(list)) |
| 98 | + assert_that(response.data.data.results[0], instance_of(LookupResult)) |
| 99 | + assert_that(response.data.data.results[0].phone_number, instance_of(str)) |
| 100 | + assert_that(response.data.data.results[0].line_type, instance_of(LineTypeEnum)) |
| 101 | + assert_that(response.data.data.results[0].messaging_provider, instance_of(str)) |
| 102 | + assert_that(response.data.data.results[0].voice_provider, instance_of(str)) |
| 103 | + assert_that(response.data.data.results[0].country_code_a3, instance_of(str)) |
| 104 | + |
| 105 | + def test_create_sync_lookup(self) -> None: |
| 106 | + """Test case for create_sync_lookup |
| 107 | +
|
| 108 | + Create Synchronous Number Lookup |
105 | 109 | """ |
106 | | - lookup_request = LookupRequest( |
107 | | - tns=[ |
108 | | - BW_NUMBER, |
109 | | - VZW_NUMBER, |
110 | | - ATT_NUMBER, |
111 | | - T_MOBILE_NUMBER, |
112 | | - # BW_INVALID_TN_LOOKUP_NUMBER |
113 | | - ], |
114 | | - ) |
115 | | - |
116 | | - # Create the lookup request and validate the response |
117 | | - create_lookup_response: CreateLookupResponse = self.api_instance.create_lookup( |
118 | | - self.account_id, lookup_request |
119 | | - ) |
120 | | - |
121 | | - self.assertIs(type(create_lookup_response.status), LookupStatusEnum) |
122 | | - self.assertEqual(create_lookup_response.status, LookupStatusEnum("IN_PROGRESS")) |
123 | | - self.assertIs(type(create_lookup_response.request_id), str) |
124 | | - self.assertRegex(create_lookup_response.request_id, |
125 | | - r'^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$') |
126 | | - |
127 | | - # Check the status code for the GET LookupStatus response |
128 | | - get_lookup_status_response: LookupStatus = self.api_instance.get_lookup_status( |
129 | | - self.account_id, |
130 | | - create_lookup_response.request_id |
131 | | - ) |
132 | | - |
133 | | - get_lookup_status_response: LookupStatus = self.pollLookupStatus( |
134 | | - create_lookup_response.request_id) |
135 | | - |
136 | | - self.assertEqual(get_lookup_status_response.request_id, |
137 | | - create_lookup_response.request_id) |
138 | | - self.assertIs(type(get_lookup_status_response), LookupStatus) |
139 | | - |
140 | | - # Assert that each result is of type LookupResult |
141 | | - for i in range(len(get_lookup_status_response.result)): |
142 | | - self.assertIs(type(get_lookup_status_response.result[i]), LookupResult) |
143 | | - |
144 | | - # Check the information for a Bandwidth TN |
145 | | - bw_lookup_result = get_lookup_status_response.result[0] |
146 | | - self.validateResult(bw_lookup_result, BW_NUMBER) |
147 | | - |
148 | | - # Check the information for a Verizon TN |
149 | | - vzw_lookup_result = get_lookup_status_response.result[1] |
150 | | - self.validateResult(vzw_lookup_result, VZW_NUMBER) |
151 | | - |
152 | | - # Check the information for an AT&T TN |
153 | | - att_lookup_result = get_lookup_status_response.result[2] |
154 | | - self.validateResult(att_lookup_result, ATT_NUMBER) |
155 | | - |
156 | | - # Check the information for a T-Mobile TN |
157 | | - t_mobile_lookup_result = get_lookup_status_response.result[3] |
158 | | - self.validateResult(t_mobile_lookup_result, T_MOBILE_NUMBER) |
159 | | - |
160 | | - # The only way to get a failed number is if the api call to the downstream service fails - so there is no way to force this in our testing currently |
161 | | - # check the failed_telephone_number list |
162 | | - # self.assertIs(type(get_lookup_status_response.failed_telephone_numbers), list) |
163 | | - # self.assertIn(BW_INVALID_TN_LOOKUP_NUMBER, get_lookup_status_response.failed_telephone_numbers) |
164 | | - |
165 | | - def testFailedPhoneNumberLookup(self) -> None: |
166 | | - """Test Phone Number Lookup API with bad data to force an error |
167 | | - """ |
168 | | - with self.assertRaises(ApiException) as context: |
169 | | - lookup_request = LookupRequest( |
170 | | - tns=[ |
171 | | - 'not a number', |
172 | | - ], |
173 | | - ) |
174 | | - self.api_instance.create_lookup(self.account_id, lookup_request) |
175 | | - |
176 | | - self.assertIs(type(context.exception.status), int) |
177 | | - self.assertIs(type(context.exception.body), str) |
178 | | - |
179 | | - # initialize TnLookupRequestError model |
180 | | - error = TnLookupRequestError(message=(json.loads(context.exception.body))['message']) |
181 | | - self.assertIs(type(error), TnLookupRequestError) |
182 | | - |
183 | | - def testDuplicatePhoneNumberLookup(self) -> None: |
184 | | - """Test a request with a duplicate number. Should throw a 400 Bad Request error. |
185 | | - """ |
186 | | - with self.assertRaises(ApiException) as context: |
187 | | - lookup_request = LookupRequest( |
188 | | - tns=[ |
189 | | - BW_NUMBER, |
190 | | - BW_NUMBER |
191 | | - ], |
192 | | - ) |
193 | | - self.api_instance.create_lookup(self.account_id, lookup_request) |
194 | | - |
195 | | - self.assertIs(type(context.exception.status), int) |
196 | | - self.assertEqual(context.exception.status, 400) |
197 | | - self.assertIs(type(context.exception.body), str) |
198 | | - |
199 | | - def testUnauthorizedRequest(self) -> None: |
200 | | - """Validate an unauthorized (401) request |
201 | | - """ |
202 | | - configuration = bandwidth.Configuration( |
203 | | - username='bad_username', |
204 | | - password='bad_password' |
205 | | - ) |
206 | | - unauthorized_api_client = bandwidth.ApiClient(configuration) |
207 | | - unauthorized_api_instance = phone_number_lookup_api.PhoneNumberLookupApi( |
208 | | - unauthorized_api_client) |
209 | | - lookup_request = LookupRequest( |
210 | | - tns=[ |
211 | | - BW_NUMBER |
212 | | - ], |
213 | | - ) |
214 | | - |
215 | | - with self.assertRaises(UnauthorizedException) as context: |
216 | | - unauthorized_api_instance.create_lookup(self.account_id, lookup_request) |
217 | | - |
218 | | - self.assertAuthException(context, UnauthorizedException, 401) |
219 | | - |
220 | | - def testForbiddenRequest(self) -> None: |
221 | | - """Validate a forbidden (403) request |
222 | | - """ |
223 | | - configuration = bandwidth.Configuration( |
224 | | - username=FORBIDDEN_USERNAME, |
225 | | - password=FORBIDDEN_PASSWORD |
226 | | - ) |
227 | | - forbidden_api_client = bandwidth.ApiClient(configuration) |
228 | | - forbidden_api_instance = phone_number_lookup_api.PhoneNumberLookupApi(forbidden_api_client) |
229 | | - lookup_request = LookupRequest( |
230 | | - tns=[ |
231 | | - BW_NUMBER |
232 | | - ], |
233 | | - ) |
234 | | - |
235 | | - # This API throws a 401 when a user provides valid credentials with the `TN Lookup` role disabled |
236 | | - # with self.assertRaises(ForbiddenException) as context: |
237 | | - with self.assertRaises(UnauthorizedException) as context: |
238 | | - forbidden_api_instance.create_lookup(self.account_id, lookup_request) |
239 | | - |
240 | | - # self.validateAuthException(context, ForbiddenException, 403) |
241 | | - self.assertAuthException(context, UnauthorizedException, 401) |
242 | | - |
| 110 | + request = SyncLookupRequest(phone_numbers=self.phone_numbers) |
| 111 | + |
| 112 | + response = self.api.create_sync_lookup_with_http_info(BW_ACCOUNT_ID, request) |
| 113 | + assert_that(response, is_not(none())) |
| 114 | + assert_that(response.status_code, equal_to(200)) |
| 115 | + assert_that(response.data, is_not(none())) |
| 116 | + assert_that(response.data, instance_of(CreateSyncLookupResponse)) |
| 117 | + assert_that(response.data.links, is_not(none())) |
| 118 | + assert_that(response.data.links[0], instance_of(LinkSchema)) |
| 119 | + assert_that(response.data.links[0].rel, instance_of(str)) |
| 120 | + assert_that(response.data.links[0].href, instance_of(str)) |
| 121 | + assert_that(response.data.links[0].method, instance_of(str)) |
| 122 | + assert_that(response.data.data, is_not(none())) |
| 123 | + assert_that(response.data.data.request_id, instance_of(UUID)) |
| 124 | + assert_that(response.data.data.status, instance_of(CompletedLookupStatusEnum)) |
| 125 | + assert_that(response.data.data.status, equal_to(CompletedLookupStatusEnum('COMPLETE'))) |
| 126 | + assert_that(response.data.data.results, instance_of(list)) |
| 127 | + assert_that(response.data.data.results[0], instance_of(LookupResult)) |
| 128 | + assert_that(response.data.data.results[0].phone_number, instance_of(str)) |
| 129 | + assert_that(response.data.data.results[0].line_type, instance_of(LineTypeEnum)) |
| 130 | + assert_that(response.data.data.results[0].messaging_provider, instance_of(str)) |
| 131 | + assert_that(response.data.data.results[0].voice_provider, instance_of(str)) |
| 132 | + assert_that(response.data.data.results[0].country_code_a3, instance_of(str)) |
| 133 | + |
243 | 134 |
|
244 | 135 | if __name__ == '__main__': |
245 | 136 | unittest.main() |
0 commit comments