azure csp tests catch exceptions resulting from raised requests exceptions
This commit is contained in:
@@ -4,8 +4,6 @@ from uuid import uuid4
|
||||
import pendulum
|
||||
import pydantic
|
||||
import pytest
|
||||
#import requests
|
||||
from secrets import token_urlsafe
|
||||
|
||||
from tests.factories import ApplicationFactory, EnvironmentFactory
|
||||
from tests.mock_azure import AUTH_CREDENTIALS, mock_azure
|
||||
@@ -67,7 +65,6 @@ from atst.domain.csp.cloud.models import (
|
||||
BILLING_ACCOUNT_NAME = "52865e4c-52e8-5a6c-da6b-c58f0814f06f:7ea5de9d-b8ce-4901-b1c5-d864320c7b03_2019-05-31"
|
||||
|
||||
|
||||
|
||||
def mock_management_group_create(mock_azure, spec_dict):
|
||||
mock_azure.sdk.managementgroups.ManagementGroupsAPI.return_value.management_groups.create_or_update.return_value.result.return_value = (
|
||||
spec_dict
|
||||
@@ -103,6 +100,7 @@ def mock_get_secret(azure, val=None):
|
||||
|
||||
return azure
|
||||
|
||||
|
||||
def test_create_application_succeeds(mock_azure: AzureCloudProvider):
|
||||
application = ApplicationFactory.create()
|
||||
mock_management_group_create(mock_azure, {"id": "Test Id"})
|
||||
@@ -141,6 +139,7 @@ def test_create_policy_definition_succeeds(mock_azure: AzureCloudProvider):
|
||||
parameters=mock_policy_definition,
|
||||
)
|
||||
|
||||
|
||||
def test_create_tenant(mock_azure: AzureCloudProvider):
|
||||
|
||||
mock_result = Mock()
|
||||
@@ -151,10 +150,11 @@ def test_create_tenant(mock_azure: AzureCloudProvider):
|
||||
}
|
||||
mock_result.status_code = 200
|
||||
|
||||
mock_azure.sdk.requests.post.return_value = mock_result
|
||||
mock_azure.sdk.requests.post.side_effect = [
|
||||
mock_azure.sdk.requests.exceptions.ConnectionError,
|
||||
mock_result
|
||||
mock_azure.sdk.requests.exceptions.Timeout,
|
||||
mock_azure.sdk.requests.exceptions.HTTPError,
|
||||
mock_result,
|
||||
]
|
||||
|
||||
payload = TenantCSPPayload(
|
||||
@@ -172,79 +172,13 @@ def test_create_tenant(mock_azure: AzureCloudProvider):
|
||||
|
||||
with pytest.raises(ConnectionException):
|
||||
mock_azure.create_tenant(payload)
|
||||
|
||||
result: TenantCSPResult = mock_azure.create_tenant(payload)
|
||||
assert result.tenant_id == "60ff9d34-82bf-4f21-b565-308ef0533435"
|
||||
|
||||
|
||||
def test_create_tenant_req_mock(mock_azure: AzureCloudProvider, requests_mock):
|
||||
url = f"{mock_azure.sdk.cloud.endpoints.resource_manager}providers/Microsoft.SignUp/createTenant"
|
||||
requests_mock.register_uri(
|
||||
"POST", "?".join([url, "api-version=2020-01-01-preview"]),
|
||||
json={
|
||||
"objectId": "0a5f4926-e3ee-4f47-a6e3-8b0a30a40e3d",
|
||||
"tenantId": "60ff9d34-82bf-4f21-b565-308ef0533435",
|
||||
"userId": "1153801116406515559",
|
||||
},
|
||||
)
|
||||
mock_azure = mock_get_secret(mock_azure)
|
||||
payload = TenantCSPPayload(
|
||||
**dict(
|
||||
user_id="admin",
|
||||
password="JediJan13$coot", # pragma: allowlist secret
|
||||
domain_name="jediccpospawnedtenant2",
|
||||
first_name="Tedry",
|
||||
last_name="Tenet",
|
||||
country_code="US",
|
||||
password_recovery_email_address="thomas@promptworks.com",
|
||||
)
|
||||
)
|
||||
result: TenantCSPResult = mock_azure.create_tenant(payload)
|
||||
assert result.tenant_id == "60ff9d34-82bf-4f21-b565-308ef0533435"
|
||||
|
||||
|
||||
def test_create_tenant_req_mock_fail(mock_azure: AzureCloudProvider, requests_mock):
|
||||
url = f"{mock_azure.sdk.cloud.endpoints.resource_manager}providers/Microsoft.SignUp/createTenant"
|
||||
requests_mock.register_uri(
|
||||
"POST", "?".join([url, "api-version=2020-01-01-preview"]),
|
||||
exc=requests.exceptions.ConnectionError,
|
||||
)
|
||||
mock_azure = mock_get_secret(mock_azure)
|
||||
payload = TenantCSPPayload(
|
||||
**dict(
|
||||
user_id="admin",
|
||||
password="JediJan13$coot", # pragma: allowlist secret
|
||||
domain_name="jediccpospawnedtenant2",
|
||||
first_name="Tedry",
|
||||
last_name="Tenet",
|
||||
country_code="US",
|
||||
password_recovery_email_address="thomas@promptworks.com",
|
||||
)
|
||||
)
|
||||
|
||||
with pytest.raises(ConnectionException):
|
||||
assert mock_azure.create_tenant(payload)
|
||||
mock_azure.create_tenant(payload)
|
||||
with pytest.raises(UnknownServerException):
|
||||
mock_azure.create_tenant(payload)
|
||||
|
||||
|
||||
# def test_create_tenant_fails(mock_azure: AzureCloudProvider):
|
||||
# mock_result = Mock()
|
||||
# mock_result.status_code = 200
|
||||
# mock_azure.sdk.requests.post.return_value = mock_result
|
||||
# payload = TenantCSPPayload(
|
||||
# **dict(
|
||||
# user_id="admin",
|
||||
# password="JediJan13$coot", # pragma: allowlist secret
|
||||
# domain_name="jediccpospawnedtenant2",
|
||||
# first_name="Tedry",
|
||||
# last_name="Tenet",
|
||||
# country_code="US",
|
||||
# password_recovery_email_address="thomas@promptworks.com",
|
||||
# )
|
||||
# )
|
||||
# mock_azure = mock_get_secret(mock_azure)
|
||||
|
||||
# with pytest.raises(pydantic.ValidationError):
|
||||
# assert mock_azure.create_tenant(payload)
|
||||
result: TenantCSPResult = mock_azure.create_tenant(payload)
|
||||
assert result.tenant_id == "60ff9d34-82bf-4f21-b565-308ef0533435"
|
||||
|
||||
|
||||
def test_create_billing_profile_creation(mock_azure: AzureCloudProvider):
|
||||
@@ -258,7 +192,14 @@ def test_create_billing_profile_creation(mock_azure: AzureCloudProvider):
|
||||
"Retry-After": "10",
|
||||
}
|
||||
mock_result.status_code = 202
|
||||
mock_azure.sdk.requests.post.return_value = mock_result
|
||||
|
||||
mock_azure.sdk.requests.post.side_effect = [
|
||||
mock_azure.sdk.requests.exceptions.ConnectionError,
|
||||
mock_azure.sdk.requests.exceptions.Timeout,
|
||||
mock_azure.sdk.requests.exceptions.HTTPError,
|
||||
mock_result,
|
||||
]
|
||||
|
||||
payload = BillingProfileCreationCSPPayload(
|
||||
**dict(
|
||||
address=dict(
|
||||
@@ -274,6 +215,13 @@ def test_create_billing_profile_creation(mock_azure: AzureCloudProvider):
|
||||
billing_account_name=BILLING_ACCOUNT_NAME,
|
||||
)
|
||||
)
|
||||
with pytest.raises(ConnectionException):
|
||||
mock_azure.create_billing_profile_creation(payload)
|
||||
with pytest.raises(ConnectionException):
|
||||
mock_azure.create_billing_profile_creation(payload)
|
||||
with pytest.raises(UnknownServerException):
|
||||
mock_azure.create_billing_profile_creation(payload)
|
||||
|
||||
body: BillingProfileCreationCSPResult = mock_azure.create_billing_profile_creation(
|
||||
payload
|
||||
)
|
||||
@@ -316,7 +264,12 @@ def test_validate_billing_profile_creation(mock_azure: AzureCloudProvider):
|
||||
},
|
||||
"type": "Microsoft.Billing/billingAccounts/billingProfiles",
|
||||
}
|
||||
mock_azure.sdk.requests.get.return_value = mock_result
|
||||
mock_azure.sdk.requests.get.side_effect = [
|
||||
mock_azure.sdk.requests.exceptions.ConnectionError,
|
||||
mock_azure.sdk.requests.exceptions.Timeout,
|
||||
mock_azure.sdk.requests.exceptions.HTTPError,
|
||||
mock_result,
|
||||
]
|
||||
|
||||
payload = BillingProfileVerificationCSPPayload(
|
||||
**dict(
|
||||
@@ -324,6 +277,12 @@ def test_validate_billing_profile_creation(mock_azure: AzureCloudProvider):
|
||||
billing_profile_verify_url="https://management.azure.com/providers/Microsoft.Billing/billingAccounts/7c89b735-b22b-55c0-ab5a-c624843e8bf6:de4416ce-acc6-44b1-8122-c87c4e903c91_2019-05-31/operationResults/createBillingProfile_478d5706-71f9-4a8b-8d4e-2cbaca27a668?api-version=2019-10-01-preview",
|
||||
)
|
||||
)
|
||||
with pytest.raises(ConnectionException):
|
||||
mock_azure.create_billing_profile_verification(payload)
|
||||
with pytest.raises(ConnectionException):
|
||||
mock_azure.create_billing_profile_verification(payload)
|
||||
with pytest.raises(UnknownServerException):
|
||||
mock_azure.create_billing_profile_verification(payload)
|
||||
|
||||
body: BillingProfileVerificationCSPResult = mock_azure.create_billing_profile_verification(
|
||||
payload
|
||||
@@ -356,7 +315,12 @@ def test_create_billing_profile_tenant_access(mock_azure: AzureCloudProvider):
|
||||
"type": "Microsoft.Billing/billingRoleAssignments",
|
||||
}
|
||||
|
||||
mock_azure.sdk.requests.post.return_value = mock_result
|
||||
mock_azure.sdk.requests.post.side_effect = [
|
||||
mock_azure.sdk.requests.exceptions.ConnectionError,
|
||||
mock_azure.sdk.requests.exceptions.Timeout,
|
||||
mock_azure.sdk.requests.exceptions.HTTPError,
|
||||
mock_result,
|
||||
]
|
||||
|
||||
payload = BillingProfileTenantAccessCSPPayload(
|
||||
**dict(
|
||||
@@ -366,6 +330,12 @@ def test_create_billing_profile_tenant_access(mock_azure: AzureCloudProvider):
|
||||
billing_profile_name="KQWI-W2SU-BG7-TGB",
|
||||
)
|
||||
)
|
||||
with pytest.raises(ConnectionException):
|
||||
mock_azure.create_billing_profile_tenant_access(payload)
|
||||
with pytest.raises(ConnectionException):
|
||||
mock_azure.create_billing_profile_tenant_access(payload)
|
||||
with pytest.raises(UnknownServerException):
|
||||
mock_azure.create_billing_profile_tenant_access(payload)
|
||||
|
||||
body: BillingProfileTenantAccessCSPResult = mock_azure.create_billing_profile_tenant_access(
|
||||
payload
|
||||
@@ -388,7 +358,12 @@ def test_create_task_order_billing_creation(mock_azure: AzureCloudProvider):
|
||||
"Retry-After": "10",
|
||||
}
|
||||
|
||||
mock_azure.sdk.requests.patch.return_value = mock_result
|
||||
mock_azure.sdk.requests.patch.side_effect = [
|
||||
mock_azure.sdk.requests.exceptions.ConnectionError,
|
||||
mock_azure.sdk.requests.exceptions.Timeout,
|
||||
mock_azure.sdk.requests.exceptions.HTTPError,
|
||||
mock_result,
|
||||
]
|
||||
|
||||
payload = TaskOrderBillingCreationCSPPayload(
|
||||
**dict(
|
||||
@@ -397,6 +372,12 @@ def test_create_task_order_billing_creation(mock_azure: AzureCloudProvider):
|
||||
billing_profile_name="KQWI-W2SU-BG7-TGB",
|
||||
)
|
||||
)
|
||||
with pytest.raises(ConnectionException):
|
||||
mock_azure.create_task_order_billing_creation(payload)
|
||||
with pytest.raises(ConnectionException):
|
||||
mock_azure.create_task_order_billing_creation(payload)
|
||||
with pytest.raises(UnknownServerException):
|
||||
mock_azure.create_task_order_billing_creation(payload)
|
||||
|
||||
body: TaskOrderBillingCreationCSPResult = mock_azure.create_task_order_billing_creation(
|
||||
payload
|
||||
@@ -450,7 +431,12 @@ def test_create_task_order_billing_verification(mock_azure):
|
||||
},
|
||||
"type": "Microsoft.Billing/billingAccounts/billingProfiles",
|
||||
}
|
||||
mock_azure.sdk.requests.get.return_value = mock_result
|
||||
mock_azure.sdk.requests.get.side_effect = [
|
||||
mock_azure.sdk.requests.exceptions.ConnectionError,
|
||||
mock_azure.sdk.requests.exceptions.Timeout,
|
||||
mock_azure.sdk.requests.exceptions.HTTPError,
|
||||
mock_result,
|
||||
]
|
||||
|
||||
payload = TaskOrderBillingVerificationCSPPayload(
|
||||
**dict(
|
||||
@@ -458,6 +444,12 @@ def test_create_task_order_billing_verification(mock_azure):
|
||||
task_order_billing_verify_url="https://management.azure.com/providers/Microsoft.Billing/billingAccounts/7c89b735-b22b-55c0-ab5a-c624843e8bf6:de4416ce-acc6-44b1-8122-c87c4e903c91_2019-05-31/operationResults/createBillingProfile_478d5706-71f9-4a8b-8d4e-2cbaca27a668?api-version=2019-10-01-preview",
|
||||
)
|
||||
)
|
||||
with pytest.raises(ConnectionException):
|
||||
mock_azure.create_task_order_billing_verification(payload)
|
||||
with pytest.raises(ConnectionException):
|
||||
mock_azure.create_task_order_billing_verification(payload)
|
||||
with pytest.raises(UnknownServerException):
|
||||
mock_azure.create_task_order_billing_verification(payload)
|
||||
|
||||
body: TaskOrderBillingVerificationCSPResult = mock_azure.create_task_order_billing_verification(
|
||||
payload
|
||||
@@ -486,7 +478,12 @@ def test_create_billing_instruction(mock_azure: AzureCloudProvider):
|
||||
"type": "Microsoft.Billing/billingAccounts/billingProfiles/billingInstructions",
|
||||
}
|
||||
|
||||
mock_azure.sdk.requests.put.return_value = mock_result
|
||||
mock_azure.sdk.requests.put.side_effect = [
|
||||
mock_azure.sdk.requests.exceptions.ConnectionError,
|
||||
mock_azure.sdk.requests.exceptions.Timeout,
|
||||
mock_azure.sdk.requests.exceptions.HTTPError,
|
||||
mock_result,
|
||||
]
|
||||
|
||||
payload = BillingInstructionCSPPayload(
|
||||
**dict(
|
||||
@@ -500,6 +497,12 @@ def test_create_billing_instruction(mock_azure: AzureCloudProvider):
|
||||
billing_profile_name="KQWI-W2SU-BG7-TGB",
|
||||
)
|
||||
)
|
||||
with pytest.raises(ConnectionException):
|
||||
mock_azure.create_billing_instruction(payload)
|
||||
with pytest.raises(ConnectionException):
|
||||
mock_azure.create_billing_instruction(payload)
|
||||
with pytest.raises(UnknownServerException):
|
||||
mock_azure.create_billing_instruction(payload)
|
||||
body: BillingInstructionCSPResult = mock_azure.create_billing_instruction(payload)
|
||||
assert body.reported_clin_name == "TO1:CLIN001"
|
||||
|
||||
@@ -516,7 +519,12 @@ def test_create_product_purchase(mock_azure: AzureCloudProvider):
|
||||
"Retry-After": "10",
|
||||
}
|
||||
|
||||
mock_azure.sdk.requests.post.return_value = mock_result
|
||||
mock_azure.sdk.requests.post.side_effect = [
|
||||
mock_azure.sdk.requests.exceptions.ConnectionError,
|
||||
mock_azure.sdk.requests.exceptions.Timeout,
|
||||
mock_azure.sdk.requests.exceptions.HTTPError,
|
||||
mock_result,
|
||||
]
|
||||
|
||||
payload = ProductPurchaseCSPPayload(
|
||||
**dict(
|
||||
@@ -525,6 +533,12 @@ def test_create_product_purchase(mock_azure: AzureCloudProvider):
|
||||
billing_profile_name="KQWI-W2SU-BG7-TGB",
|
||||
)
|
||||
)
|
||||
with pytest.raises(ConnectionException):
|
||||
mock_azure.create_product_purchase(payload)
|
||||
with pytest.raises(ConnectionException):
|
||||
mock_azure.create_product_purchase(payload)
|
||||
with pytest.raises(UnknownServerException):
|
||||
mock_azure.create_product_purchase(payload)
|
||||
|
||||
body: ProductPurchaseCSPResult = mock_azure.create_product_purchase(payload)
|
||||
assert (
|
||||
@@ -561,7 +575,12 @@ def test_create_product_purchase_verification(mock_azure):
|
||||
"type": "Microsoft.Billing/billingAccounts/billingProfiles/invoiceSections/products",
|
||||
}
|
||||
|
||||
mock_azure.sdk.requests.get.return_value = mock_result
|
||||
mock_azure.sdk.requests.get.side_effect = [
|
||||
mock_azure.sdk.requests.exceptions.ConnectionError,
|
||||
mock_azure.sdk.requests.exceptions.Timeout,
|
||||
mock_azure.sdk.requests.exceptions.HTTPError,
|
||||
mock_result,
|
||||
]
|
||||
|
||||
payload = ProductPurchaseVerificationCSPPayload(
|
||||
**dict(
|
||||
@@ -569,6 +588,12 @@ def test_create_product_purchase_verification(mock_azure):
|
||||
product_purchase_verify_url="https://management.azure.com/providers/Microsoft.Billing/billingAccounts/7c89b735-b22b-55c0-ab5a-c624843e8bf6:de4416ce-acc6-44b1-8122-c87c4e903c91_2019-05-31/operationResults/createBillingProfile_478d5706-71f9-4a8b-8d4e-2cbaca27a668?api-version=2019-10-01-preview",
|
||||
)
|
||||
)
|
||||
with pytest.raises(ConnectionException):
|
||||
mock_azure.create_product_purchase_verification(payload)
|
||||
with pytest.raises(ConnectionException):
|
||||
mock_azure.create_product_purchase_verification(payload)
|
||||
with pytest.raises(UnknownServerException):
|
||||
mock_azure.create_product_purchase_verification(payload)
|
||||
|
||||
body: ProductPurchaseVerificationCSPResult = mock_azure.create_product_purchase_verification(
|
||||
payload
|
||||
@@ -588,12 +613,25 @@ def test_create_tenant_principal_app(mock_azure: AzureCloudProvider):
|
||||
mock_result.ok = True
|
||||
mock_result.json.return_value = {"appId": "appId", "id": "id"}
|
||||
|
||||
mock_azure.sdk.requests.post.return_value = mock_result
|
||||
mock_azure.sdk.requests.post.side_effect = [
|
||||
mock_azure.sdk.requests.exceptions.ConnectionError,
|
||||
mock_azure.sdk.requests.exceptions.Timeout,
|
||||
mock_azure.sdk.requests.exceptions.HTTPError,
|
||||
mock_result,
|
||||
]
|
||||
|
||||
mock_azure = mock_get_secret(mock_azure)
|
||||
|
||||
payload = TenantPrincipalAppCSPPayload(
|
||||
**{"tenant_id": "6d2d2d6c-a6d6-41e1-8bb1-73d11475f8f4"}
|
||||
)
|
||||
with pytest.raises(ConnectionException):
|
||||
mock_azure.create_tenant_principal_app(payload)
|
||||
with pytest.raises(ConnectionException):
|
||||
mock_azure.create_tenant_principal_app(payload)
|
||||
with pytest.raises(UnknownServerException):
|
||||
mock_azure.create_tenant_principal_app(payload)
|
||||
|
||||
result: TenantPrincipalAppCSPResult = mock_azure.create_tenant_principal_app(
|
||||
payload
|
||||
)
|
||||
@@ -613,7 +651,13 @@ def test_create_tenant_principal(mock_azure: AzureCloudProvider):
|
||||
mock_result.ok = True
|
||||
mock_result.json.return_value = {"id": "principal_id"}
|
||||
|
||||
mock_azure.sdk.requests.post.return_value = mock_result
|
||||
mock_azure.sdk.requests.post.side_effect = [
|
||||
mock_azure.sdk.requests.exceptions.ConnectionError,
|
||||
mock_azure.sdk.requests.exceptions.Timeout,
|
||||
mock_azure.sdk.requests.exceptions.HTTPError,
|
||||
mock_result,
|
||||
]
|
||||
|
||||
mock_azure = mock_get_secret(mock_azure)
|
||||
|
||||
payload = TenantPrincipalCSPPayload(
|
||||
@@ -622,6 +666,12 @@ def test_create_tenant_principal(mock_azure: AzureCloudProvider):
|
||||
"principal_app_id": "appId",
|
||||
}
|
||||
)
|
||||
with pytest.raises(ConnectionException):
|
||||
mock_azure.create_tenant_principal(payload)
|
||||
with pytest.raises(ConnectionException):
|
||||
mock_azure.create_tenant_principal(payload)
|
||||
with pytest.raises(UnknownServerException):
|
||||
mock_azure.create_tenant_principal(payload)
|
||||
|
||||
result: TenantPrincipalCSPResult = mock_azure.create_tenant_principal(payload)
|
||||
|
||||
@@ -640,7 +690,12 @@ def test_create_tenant_principal_credential(mock_azure: AzureCloudProvider):
|
||||
mock_result.ok = True
|
||||
mock_result.json.return_value = {"secretText": "new secret key"}
|
||||
|
||||
mock_azure.sdk.requests.post.return_value = mock_result
|
||||
mock_azure.sdk.requests.post.side_effect = [
|
||||
mock_azure.sdk.requests.exceptions.ConnectionError,
|
||||
mock_azure.sdk.requests.exceptions.Timeout,
|
||||
mock_azure.sdk.requests.exceptions.HTTPError,
|
||||
mock_result,
|
||||
]
|
||||
|
||||
mock_azure = mock_get_secret(mock_azure)
|
||||
|
||||
@@ -651,6 +706,12 @@ def test_create_tenant_principal_credential(mock_azure: AzureCloudProvider):
|
||||
"principal_app_object_id": "appObjId",
|
||||
}
|
||||
)
|
||||
with pytest.raises(ConnectionException):
|
||||
mock_azure.create_tenant_principal_credential(payload)
|
||||
with pytest.raises(ConnectionException):
|
||||
mock_azure.create_tenant_principal_credential(payload)
|
||||
with pytest.raises(UnknownServerException):
|
||||
mock_azure.create_tenant_principal_credential(payload)
|
||||
|
||||
result: TenantPrincipalCredentialCSPResult = mock_azure.create_tenant_principal_credential(
|
||||
payload
|
||||
@@ -676,12 +737,23 @@ def test_create_admin_role_definition(mock_azure: AzureCloudProvider):
|
||||
]
|
||||
}
|
||||
|
||||
mock_azure.sdk.requests.get.return_value = mock_result
|
||||
mock_azure.sdk.requests.get.side_effect = [
|
||||
mock_azure.sdk.requests.exceptions.ConnectionError,
|
||||
mock_azure.sdk.requests.exceptions.Timeout,
|
||||
mock_azure.sdk.requests.exceptions.HTTPError,
|
||||
mock_result,
|
||||
]
|
||||
mock_azure = mock_get_secret(mock_azure)
|
||||
|
||||
payload = AdminRoleDefinitionCSPPayload(
|
||||
**{"tenant_id": "6d2d2d6c-a6d6-41e1-8bb1-73d11475f8f4"}
|
||||
)
|
||||
with pytest.raises(ConnectionException):
|
||||
mock_azure.create_admin_role_definition(payload)
|
||||
with pytest.raises(ConnectionException):
|
||||
mock_azure.create_admin_role_definition(payload)
|
||||
with pytest.raises(UnknownServerException):
|
||||
mock_azure.create_admin_role_definition(payload)
|
||||
|
||||
result: AdminRoleDefinitionCSPResult = mock_azure.create_admin_role_definition(
|
||||
payload
|
||||
@@ -702,7 +774,12 @@ def test_create_tenant_admin_ownership(mock_azure: AzureCloudProvider):
|
||||
mock_result.ok = True
|
||||
mock_result.json.return_value = {"id": "id"}
|
||||
|
||||
mock_azure.sdk.requests.put.return_value = mock_result
|
||||
mock_azure.sdk.requests.put.side_effect = [
|
||||
mock_azure.sdk.requests.exceptions.ConnectionError,
|
||||
mock_azure.sdk.requests.exceptions.Timeout,
|
||||
mock_azure.sdk.requests.exceptions.HTTPError,
|
||||
mock_result,
|
||||
]
|
||||
|
||||
payload = TenantAdminOwnershipCSPPayload(
|
||||
**{
|
||||
@@ -710,6 +787,12 @@ def test_create_tenant_admin_ownership(mock_azure: AzureCloudProvider):
|
||||
"user_object_id": "971efe4d-1e80-4e39-b3b9-4e5c63ad446d",
|
||||
}
|
||||
)
|
||||
with pytest.raises(ConnectionException):
|
||||
mock_azure.create_tenant_admin_ownership(payload)
|
||||
with pytest.raises(ConnectionException):
|
||||
mock_azure.create_tenant_admin_ownership(payload)
|
||||
with pytest.raises(UnknownServerException):
|
||||
mock_azure.create_tenant_admin_ownership(payload)
|
||||
|
||||
result: TenantAdminOwnershipCSPResult = mock_azure.create_tenant_admin_ownership(
|
||||
payload
|
||||
@@ -730,7 +813,12 @@ def test_create_tenant_principal_ownership(mock_azure: AzureCloudProvider):
|
||||
mock_result.ok = True
|
||||
mock_result.json.return_value = {"id": "id"}
|
||||
|
||||
mock_azure.sdk.requests.put.return_value = mock_result
|
||||
mock_azure.sdk.requests.put.side_effect = [
|
||||
mock_azure.sdk.requests.exceptions.ConnectionError,
|
||||
mock_azure.sdk.requests.exceptions.Timeout,
|
||||
mock_azure.sdk.requests.exceptions.HTTPError,
|
||||
mock_result,
|
||||
]
|
||||
|
||||
payload = TenantPrincipalOwnershipCSPPayload(
|
||||
**{
|
||||
@@ -738,6 +826,12 @@ def test_create_tenant_principal_ownership(mock_azure: AzureCloudProvider):
|
||||
"principal_id": "971efe4d-1e80-4e39-b3b9-4e5c63ad446d",
|
||||
}
|
||||
)
|
||||
with pytest.raises(ConnectionException):
|
||||
mock_azure.create_tenant_principal_ownership(payload)
|
||||
with pytest.raises(ConnectionException):
|
||||
mock_azure.create_tenant_principal_ownership(payload)
|
||||
with pytest.raises(UnknownServerException):
|
||||
mock_azure.create_tenant_principal_ownership(payload)
|
||||
|
||||
result: TenantPrincipalOwnershipCSPResult = mock_azure.create_tenant_principal_ownership(
|
||||
payload
|
||||
@@ -758,7 +852,12 @@ def test_create_principal_admin_role(mock_azure: AzureCloudProvider):
|
||||
mock_result.ok = True
|
||||
mock_result.json.return_value = {"id": "id"}
|
||||
|
||||
mock_azure.sdk.requests.post.return_value = mock_result
|
||||
mock_azure.sdk.requests.post.side_effect = [
|
||||
mock_azure.sdk.requests.exceptions.ConnectionError,
|
||||
mock_azure.sdk.requests.exceptions.Timeout,
|
||||
mock_azure.sdk.requests.exceptions.HTTPError,
|
||||
mock_result,
|
||||
]
|
||||
|
||||
payload = PrincipalAdminRoleCSPPayload(
|
||||
**{
|
||||
@@ -767,6 +866,12 @@ def test_create_principal_admin_role(mock_azure: AzureCloudProvider):
|
||||
"admin_role_def_id": uuid4().hex,
|
||||
}
|
||||
)
|
||||
with pytest.raises(ConnectionException):
|
||||
mock_azure.create_principal_admin_role(payload)
|
||||
with pytest.raises(ConnectionException):
|
||||
mock_azure.create_principal_admin_role(payload)
|
||||
with pytest.raises(UnknownServerException):
|
||||
mock_azure.create_principal_admin_role(payload)
|
||||
|
||||
result: PrincipalAdminRoleCSPResult = mock_azure.create_principal_admin_role(
|
||||
payload
|
||||
@@ -790,7 +895,14 @@ def test_create_subscription_creation(mock_azure: AzureCloudProvider):
|
||||
"Retry-After": 10,
|
||||
}
|
||||
mock_result.json.return_value = {}
|
||||
mock_azure.sdk.requests.put.return_value = mock_result
|
||||
|
||||
mock_azure.sdk.requests.put.side_effect = [
|
||||
mock_azure.sdk.requests.exceptions.ConnectionError,
|
||||
mock_azure.sdk.requests.exceptions.Timeout,
|
||||
mock_azure.sdk.requests.exceptions.HTTPError,
|
||||
mock_result,
|
||||
]
|
||||
|
||||
management_group_id = str(uuid4())
|
||||
payload = SubscriptionCreationCSPPayload(
|
||||
**dict(
|
||||
@@ -802,6 +914,12 @@ def test_create_subscription_creation(mock_azure: AzureCloudProvider):
|
||||
invoice_section_name="6HMZ-2HLO-PJA-TGB",
|
||||
)
|
||||
)
|
||||
with pytest.raises(ConnectionException):
|
||||
mock_azure.create_subscription_creation(payload)
|
||||
with pytest.raises(ConnectionException):
|
||||
mock_azure.create_subscription_creation(payload)
|
||||
with pytest.raises(UnknownServerException):
|
||||
mock_azure.create_subscription_creation(payload)
|
||||
|
||||
result: SubscriptionCreationCSPResult = mock_azure.create_subscription_creation(
|
||||
payload
|
||||
@@ -823,7 +941,13 @@ def test_create_subscription_verification(mock_azure: AzureCloudProvider):
|
||||
mock_result.json.return_value = {
|
||||
"subscriptionLink": "/subscriptions/60fbbb72-0516-4253-ab18-c92432ba3230"
|
||||
}
|
||||
mock_azure.sdk.requests.get.return_value = mock_result
|
||||
|
||||
mock_azure.sdk.requests.get.side_effect = [
|
||||
mock_azure.sdk.requests.exceptions.ConnectionError,
|
||||
mock_azure.sdk.requests.exceptions.Timeout,
|
||||
mock_azure.sdk.requests.exceptions.HTTPError,
|
||||
mock_result,
|
||||
]
|
||||
|
||||
payload = SubscriptionVerificationCSPPayload(
|
||||
**dict(
|
||||
@@ -831,6 +955,12 @@ def test_create_subscription_verification(mock_azure: AzureCloudProvider):
|
||||
subscription_verify_url="https://verify.me",
|
||||
)
|
||||
)
|
||||
with pytest.raises(ConnectionException):
|
||||
mock_azure.create_subscription_verification(payload)
|
||||
with pytest.raises(ConnectionException):
|
||||
mock_azure.create_subscription_verification(payload)
|
||||
with pytest.raises(UnknownServerException):
|
||||
mock_azure.create_subscription_verification(payload)
|
||||
|
||||
result: SuscriptionVerificationCSPResult = mock_azure.create_subscription_verification(
|
||||
payload
|
||||
@@ -859,7 +989,14 @@ def test_get_reporting_data(mock_azure: AzureCloudProvider):
|
||||
"type": "Microsoft.CostManagement/query",
|
||||
}
|
||||
mock_result.ok = True
|
||||
mock_azure.sdk.requests.post.return_value = mock_result
|
||||
|
||||
mock_azure.sdk.requests.post.side_effect = [
|
||||
mock_azure.sdk.requests.exceptions.ConnectionError,
|
||||
mock_azure.sdk.requests.exceptions.Timeout,
|
||||
mock_azure.sdk.requests.exceptions.HTTPError,
|
||||
mock_result,
|
||||
]
|
||||
|
||||
mock_azure = mock_get_secret(mock_azure)
|
||||
|
||||
# Subset of a profile's CSP data that we care about for reporting
|
||||
@@ -873,14 +1010,19 @@ def test_get_reporting_data(mock_azure: AzureCloudProvider):
|
||||
],
|
||||
},
|
||||
}
|
||||
|
||||
data: CostManagementQueryCSPResult = mock_azure.get_reporting_data(
|
||||
ReportingCSPPayload(
|
||||
from_date=pendulum.now().subtract(years=1).add(days=1).format("YYYY-MM-DD"),
|
||||
to_date=pendulum.now().format("YYYY-MM-DD"),
|
||||
**csp_data,
|
||||
)
|
||||
payload = ReportingCSPPayload(
|
||||
from_date=pendulum.now().subtract(years=1).add(days=1).format("YYYY-MM-DD"),
|
||||
to_date=pendulum.now().format("YYYY-MM-DD"),
|
||||
**csp_data,
|
||||
)
|
||||
with pytest.raises(ConnectionException):
|
||||
mock_azure.get_reporting_data(payload)
|
||||
with pytest.raises(ConnectionException):
|
||||
mock_azure.get_reporting_data(payload)
|
||||
with pytest.raises(UnknownServerException):
|
||||
mock_azure.get_reporting_data(payload)
|
||||
|
||||
data: CostManagementQueryCSPResult = mock_azure.get_reporting_data(payload)
|
||||
|
||||
assert isinstance(data, CostManagementQueryCSPResult)
|
||||
assert data.name == "e82d0cda-2ffb-4476-a98a-425c83c216f9"
|
||||
@@ -890,7 +1032,14 @@ def test_get_reporting_data(mock_azure: AzureCloudProvider):
|
||||
def test_get_reporting_data_malformed_payload(mock_azure: AzureCloudProvider):
|
||||
mock_result = Mock()
|
||||
mock_result.ok = True
|
||||
mock_azure.sdk.requests.post.return_value = mock_result
|
||||
|
||||
mock_azure.sdk.requests.post.side_effect = [
|
||||
mock_azure.sdk.requests.exceptions.ConnectionError,
|
||||
mock_azure.sdk.requests.exceptions.Timeout,
|
||||
mock_azure.sdk.requests.exceptions.HTTPError,
|
||||
mock_result,
|
||||
]
|
||||
|
||||
mock_azure = mock_get_secret(mock_azure)
|
||||
|
||||
# Malformed csp_data payloads that should throw pydantic validation errors
|
||||
@@ -946,7 +1095,12 @@ def test_create_active_directory_user(mock_azure: AzureCloudProvider):
|
||||
mock_result = Mock()
|
||||
mock_result.ok = True
|
||||
mock_result.json.return_value = {"id": "id"}
|
||||
mock_azure.sdk.requests.post.return_value = mock_result
|
||||
mock_azure.sdk.requests.post.side_effect = [
|
||||
mock_azure.sdk.requests.exceptions.ConnectionError,
|
||||
mock_azure.sdk.requests.exceptions.Timeout,
|
||||
mock_azure.sdk.requests.exceptions.HTTPError,
|
||||
mock_result,
|
||||
]
|
||||
|
||||
payload = UserCSPPayload(
|
||||
tenant_id=uuid4().hex,
|
||||
@@ -955,6 +1109,12 @@ def test_create_active_directory_user(mock_azure: AzureCloudProvider):
|
||||
email="test@testerson.test",
|
||||
password="asdfghjkl", # pragma: allowlist secret
|
||||
)
|
||||
with pytest.raises(ConnectionException):
|
||||
mock_azure._create_active_directory_user("token", payload)
|
||||
with pytest.raises(ConnectionException):
|
||||
mock_azure._create_active_directory_user("token", payload)
|
||||
with pytest.raises(UnknownServerException):
|
||||
mock_azure._create_active_directory_user("token", payload)
|
||||
|
||||
result = mock_azure._create_active_directory_user("token", payload)
|
||||
|
||||
@@ -964,8 +1124,12 @@ def test_create_active_directory_user(mock_azure: AzureCloudProvider):
|
||||
def test_update_active_directory_user_email(mock_azure: AzureCloudProvider):
|
||||
mock_result = Mock()
|
||||
mock_result.ok = True
|
||||
mock_azure.sdk.requests.patch.return_value = mock_result
|
||||
|
||||
mock_azure.sdk.requests.patch.side_effect = [
|
||||
mock_azure.sdk.requests.exceptions.ConnectionError,
|
||||
mock_azure.sdk.requests.exceptions.Timeout,
|
||||
mock_azure.sdk.requests.exceptions.HTTPError,
|
||||
mock_result,
|
||||
]
|
||||
payload = UserCSPPayload(
|
||||
tenant_id=uuid4().hex,
|
||||
display_name="Test Testerson",
|
||||
@@ -973,6 +1137,12 @@ def test_update_active_directory_user_email(mock_azure: AzureCloudProvider):
|
||||
email="test@testerson.test",
|
||||
password="asdfghjkl", # pragma: allowlist secret
|
||||
)
|
||||
with pytest.raises(ConnectionException):
|
||||
mock_azure._update_active_directory_user_email("token", uuid4().hex, payload)
|
||||
with pytest.raises(ConnectionException):
|
||||
mock_azure._update_active_directory_user_email("token", uuid4().hex, payload)
|
||||
with pytest.raises(UnknownServerException):
|
||||
mock_azure._update_active_directory_user_email("token", uuid4().hex, payload)
|
||||
|
||||
result = mock_azure._update_active_directory_user_email(
|
||||
"token", uuid4().hex, payload
|
||||
@@ -1007,5 +1177,4 @@ def test_create_user(mock_azure: AzureCloudProvider):
|
||||
)
|
||||
|
||||
result = mock_azure.create_user(payload)
|
||||
|
||||
assert result.id == "id"
|
||||
|
||||
Reference in New Issue
Block a user