From 596227a7738a5342274486e30489239d539b11d1 Mon Sep 17 00:00:00 2001 From: Aigerim Date: Thu, 11 Nov 2021 12:13:35 +0100 Subject: [PATCH] communication identity related codes replaced to tests folder from shared (#21716) Co-authored-by: Aigerim Beishenbekova --- .../tests/_shared/testcase.py | 41 +------------- .../tests/_shared/utils.py | 35 ------------ .../tests/asynctestcase.py | 27 +++++++++ .../test_communication_identity_client.py | 8 +-- ...est_communication_identity_client_async.py | 4 +- .../tests/testcase.py | 56 +++++++++++++++++++ 6 files changed, 89 insertions(+), 82 deletions(-) create mode 100644 sdk/communication/azure-communication-identity/tests/asynctestcase.py create mode 100644 sdk/communication/azure-communication-identity/tests/testcase.py diff --git a/sdk/communication/azure-communication-identity/tests/_shared/testcase.py b/sdk/communication/azure-communication-identity/tests/_shared/testcase.py index c63fa91aa4e5..1e200ec23015 100644 --- a/sdk/communication/azure-communication-identity/tests/_shared/testcase.py +++ b/sdk/communication/azure-communication-identity/tests/_shared/testcase.py @@ -5,12 +5,9 @@ # license information. # -------------------------------------------------------------------------- import re -import os from devtools_testutils import AzureTestCase from azure_devtools.scenario_tests import RecordingProcessor, ReplayableTest from azure_devtools.scenario_tests.utilities import is_text_payload -from azure.communication.identity._shared.utils import parse_connection_str -from _shared.utils import generate_teams_user_aad_token class ResponseReplacerProcessor(RecordingProcessor): def __init__(self, keys=None, replacement="sanitized"): self._keys = keys if keys else [] @@ -74,40 +71,4 @@ class CommunicationTestCase(AzureTestCase): FILTER_HEADERS = ReplayableTest.FILTER_HEADERS + ['x-azure-ref', 'x-ms-content-sha256', 'location'] def __init__(self, method_name, *args, **kwargs): - super(CommunicationTestCase, self).__init__(method_name, *args, **kwargs) - - def setUp(self): - super(CommunicationTestCase, 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: - teams_user_aad_token = generate_teams_user_aad_token(m365_app_id=self.m365_app_id, m365_aad_authority=self.m365_aad_authority, m365_aad_tenant=self.m365_aad_tenant, msal_username=self.msal_username, msal_password=self.msal_password, m365_scope=self.m365_scope) - 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' \ No newline at end of file + super(CommunicationTestCase, self).__init__(method_name, *args, **kwargs) \ No newline at end of file diff --git a/sdk/communication/azure-communication-identity/tests/_shared/utils.py b/sdk/communication/azure-communication-identity/tests/_shared/utils.py index d64239af32a5..6e92a813ab29 100644 --- a/sdk/communication/azure-communication-identity/tests/_shared/utils.py +++ b/sdk/communication/azure-communication-identity/tests/_shared/utils.py @@ -5,7 +5,6 @@ # ------------------------------------------------------------------------- from azure.core.pipeline.policies import HttpLoggingPolicy -from msal import PublicClientApplication def get_http_logging_policy(**kwargs): http_logging_policy = HttpLoggingPolicy(**kwargs) @@ -15,37 +14,3 @@ def get_http_logging_policy(**kwargs): } ) return http_logging_policy - -def generate_teams_user_aad_token( - m365_app_id, # type: str - m365_aad_authority, # type: str - m365_aad_tenant, # type: str - msal_username, # type: str - msal_password, # type: str - m365_scope # type: str -): - # type: (...) -> str - """Returns issued AAD access token of a Teams User by MSAL library - :param m365_app_id: the application id of M365 - :type m365_app_id: str - :param m365_aad_authority: the AAD authority of M365 - :type m365_aad_authority: str - :param m365_aad_tenant: the tenant ID of M365 application - :type m365_aad_tenant: str - :param msal_username: the username for authenticating via MSAL library - :type msal_username: str - :param msal_password: the password for authenticating via MSAL library - :type msal_password: str - :param m365_scope: the scope of M365 application - :type m365_scope: str - :return: an AAD access token of a Teams User - :rtype: str - """ - msal_app = PublicClientApplication( - client_id=m365_app_id, - authority="{}/{}".format(m365_aad_authority, m365_aad_tenant)) - result = msal_app.acquire_token_by_username_password( - username=msal_username, - password=msal_password, - scopes=[m365_scope]) - return result["access_token"] diff --git a/sdk/communication/azure-communication-identity/tests/asynctestcase.py b/sdk/communication/azure-communication-identity/tests/asynctestcase.py new file mode 100644 index 000000000000..31d2c2ab6c75 --- /dev/null +++ b/sdk/communication/azure-communication-identity/tests/asynctestcase.py @@ -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 diff --git a/sdk/communication/azure-communication-identity/tests/test_communication_identity_client.py b/sdk/communication/azure-communication-identity/tests/test_communication_identity_client.py index 059c4cb2af8e..a88aee6e5b35 100644 --- a/sdk/communication/azure-communication-identity/tests/test_communication_identity_client.py +++ b/sdk/communication/azure-communication-identity/tests/test_communication_identity_client.py @@ -10,10 +10,8 @@ from azure.communication.identity import CommunicationTokenScope from azure.core.credentials import AccessToken from _shared.helper import URIIdentityReplacer -from _shared.testcase import ( - CommunicationTestCase, - BodyReplacerProcessor -) +from _shared.testcase import BodyReplacerProcessor +from testcase import CommunicationIdentityTestCase from _shared.communication_service_preparer import CommunicationPreparer from _shared.utils import get_http_logging_policy from azure.identity import DefaultAzureCredential @@ -26,7 +24,7 @@ def __init__(self): def get_token(self, *args): return self.token -class CommunicationIdentityClientTest(CommunicationTestCase): +class CommunicationIdentityClientTest(CommunicationIdentityTestCase): def setUp(self): super(CommunicationIdentityClientTest, self).setUp() self.recording_processors.extend([ diff --git a/sdk/communication/azure-communication-identity/tests/test_communication_identity_client_async.py b/sdk/communication/azure-communication-identity/tests/test_communication_identity_client_async.py index 51774d55a731..b4cde9c67227 100644 --- a/sdk/communication/azure-communication-identity/tests/test_communication_identity_client_async.py +++ b/sdk/communication/azure-communication-identity/tests/test_communication_identity_client_async.py @@ -13,7 +13,7 @@ from azure_devtools.scenario_tests import RecordingProcessor from devtools_testutils import ResourceGroupPreparer from _shared.helper import URIIdentityReplacer -from _shared.asynctestcase import AsyncCommunicationTestCase +from asynctestcase import AsyncCommunicationIdentityTestCase from _shared.testcase import BodyReplacerProcessor from _shared.communication_service_preparer import CommunicationPreparer from _shared.utils import get_http_logging_policy @@ -25,7 +25,7 @@ def __init__(self): async def get_token(self, *args): return self.token -class CommunicationIdentityClientTestAsync(AsyncCommunicationTestCase): +class CommunicationIdentityClientTestAsync(AsyncCommunicationIdentityTestCase): def setUp(self): super(CommunicationIdentityClientTestAsync, self).setUp() self.recording_processors.extend([ diff --git a/sdk/communication/azure-communication-identity/tests/testcase.py b/sdk/communication/azure-communication-identity/tests/testcase.py new file mode 100644 index 000000000000..f61086e21bf8 --- /dev/null +++ b/sdk/communication/azure-communication-identity/tests/testcase.py @@ -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' \ No newline at end of file