forked from Azure/azure-sdk-for-python
-
Notifications
You must be signed in to change notification settings - Fork 2
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
communication identity related codes replaced to tests folder from sh…
…ared (Azure#21716) Co-authored-by: Aigerim Beishenbekova <[email protected]>
- Loading branch information
Showing
6 changed files
with
89 additions
and
82 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
27 changes: 27 additions & 0 deletions
27
sdk/communication/azure-communication-identity/tests/asynctestcase.py
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,27 @@ | ||
|
|
||
| # coding: utf-8 | ||
| # ------------------------------------------------------------------------- | ||
| # Copyright (c) Microsoft Corporation. All rights reserved. | ||
| # Licensed under the MIT License. See License.txt in the project root for | ||
| # license information. | ||
| # -------------------------------------------------------------------------- | ||
| import functools | ||
| import asyncio | ||
| from azure_devtools.scenario_tests.utilities import trim_kwargs_from_test_function | ||
| from testcase import CommunicationIdentityTestCase | ||
|
|
||
| class AsyncCommunicationIdentityTestCase(CommunicationIdentityTestCase): | ||
|
|
||
| @staticmethod | ||
| def await_prepared_test(test_fn): | ||
| """Synchronous wrapper for async test methods. Used to avoid making changes | ||
| upstream to AbstractPreparer (which doesn't await the functions it wraps) | ||
| """ | ||
|
|
||
| @functools.wraps(test_fn) | ||
| def run(test_class_instance, *args, **kwargs): | ||
| trim_kwargs_from_test_function(test_fn, kwargs) | ||
| loop = asyncio.get_event_loop() | ||
| return loop.run_until_complete(test_fn(test_class_instance, **kwargs)) | ||
|
|
||
| return run |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
56 changes: 56 additions & 0 deletions
56
sdk/communication/azure-communication-identity/tests/testcase.py
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,56 @@ | ||
|
|
||
| # ------------------------------------------------------------------------- | ||
| # Copyright (c) Microsoft Corporation. All rights reserved. | ||
| # Licensed under the MIT License. See License.txt in the project root for | ||
| # license information. | ||
| # -------------------------------------------------------------------------- | ||
| import os | ||
|
|
||
| from azure.communication.identity._shared.utils import parse_connection_str | ||
| from _shared.testcase import CommunicationTestCase | ||
| from msal import PublicClientApplication | ||
|
|
||
| class CommunicationIdentityTestCase(CommunicationTestCase): | ||
|
|
||
| def __init__(self, method_name, *args, **kwargs): | ||
| super(CommunicationIdentityTestCase, self).__init__(method_name, *args, **kwargs) | ||
|
|
||
| def setUp(self): | ||
| super(CommunicationIdentityTestCase, self).setUp() | ||
| if self.is_playback(): | ||
| self.connection_str = "endpoint=https://sanitized/;accesskey=fake===" | ||
| self.m365_app_id = "sanitized" | ||
| self.m365_aad_authority = "sanitized" | ||
| self.m365_aad_tenant = "sanitized" | ||
| self.m365_scope = "sanitized" | ||
| self.msal_username = "sanitized" | ||
| self.msal_password = "sanitized" | ||
| self.expired_teams_token = "sanitized" | ||
| self.skip_get_token_for_teams_user_tests = "false" | ||
| else: | ||
| self.connection_str = os.getenv('COMMUNICATION_LIVETEST_DYNAMIC_CONNECTION_STRING') | ||
| self.m365_app_id = os.getenv('COMMUNICATION_M365_APP_ID') | ||
| self.m365_aad_authority = os.getenv('COMMUNICATION_M365_AAD_AUTHORITY') | ||
| self.m365_aad_tenant = os.getenv('COMMUNICATION_M365_AAD_TENANT') | ||
| self.m365_scope = os.getenv('COMMUNICATION_M365_SCOPE') | ||
| self.msal_username = os.getenv('COMMUNICATION_MSAL_USERNAME') | ||
| self.msal_password = os.getenv('COMMUNICATION_MSAL_PASSWORD') | ||
| self.expired_teams_token = os.getenv('COMMUNICATION_EXPIRED_TEAMS_TOKEN') | ||
| endpoint, _ = parse_connection_str(self.connection_str) | ||
| self._resource_name = endpoint.split(".")[0] | ||
| self.scrubber.register_name_pair(self._resource_name, "sanitized") | ||
| self.skip_get_token_for_teams_user_tests = os.getenv('SKIP_INT_IDENTITY_EXCHANGE_TOKEN_TEST') | ||
|
|
||
| def generate_teams_user_aad_token(self): | ||
| if self.is_playback(): | ||
| teams_user_aad_token = "sanitized" | ||
| else: | ||
| msal_app = PublicClientApplication( | ||
| client_id=self.m365_app_id, | ||
| authority="{}/{}".format(self.m365_aad_authority, self.m365_aad_tenant)) | ||
| result = msal_app.acquire_token_by_username_password(username=self.msal_username, password=self.msal_password, scopes=[self.m365_scope]) | ||
| teams_user_aad_token = result["access_token"] | ||
| return teams_user_aad_token | ||
|
|
||
| def skip_get_token_for_teams_user_test(self): | ||
| return str(self.skip_get_token_for_teams_user_tests).lower() == 'true' |