Create new AliasModel for CSP datalcasses, ignore credentials when converting to dictionary.This will allow all of our dataclasses to convert automatically between python style snake_case and the camelCase that the Azure APIs use. This also allows us to default to that behavior while specifying aliases for any fields as necessary.Additionally, any dataclass including the creds schema will have those creds removed from their dict representation. This can help keep creds out of logs as well as making the dataclasses more consumable for API usage.

This commit is contained in:
tomdds 2020-01-13 16:40:17 -05:00
parent 2ac333e0b7
commit 7c22922d6d
3 changed files with 52 additions and 36 deletions

View File

@ -7,6 +7,7 @@ from pydantic import BaseModel
from atst.models.user import User
from atst.models.environment import Environment
from atst.models.environment_role import EnvironmentRole
from atst.utils import snake_to_camel
class GeneralCSPException(Exception):
@ -142,10 +143,33 @@ class BaselineProvisionException(GeneralCSPException):
)
class BaseCSPPayload(BaseModel):
class AliasModel(BaseModel):
"""
This provides automatic camel <-> snake conversion for serializing to/from json
You can override the alias generation in subclasses by providing a Config that defines
a fields property with a dict mapping variables to their cast names, for cases like:
* some_url:someURL
* user_object_id:objectId
"""
class Config:
alias_generator = snake_to_camel
allow_population_by_field_name = True
class BaseCSPPayload(AliasModel):
# {"username": "mock-cloud", "pass": "shh"}
creds: Dict
def dict(self, *args, **kwargs):
exclude = {"creds"}
if "exclude" not in kwargs:
kwargs["exclude"] = exclude
else:
kwargs["exclude"].update(exclude)
return super().dict(*args, **kwargs)
class TenantCSPPayload(BaseCSPPayload):
user_id: str
@ -156,52 +180,23 @@ class TenantCSPPayload(BaseCSPPayload):
country_code: str
password_recovery_email_address: str
class Config:
fields = {
"user_id": "userId",
"domain_name": "domainName",
"first_name": "firstName",
"last_name": "lastName",
"country_code": "countryCode",
"password_recovery_email_address": "passwordRecoveryEmailAddress",
}
allow_population_by_field_name = True
class TenantCSPResult(BaseModel):
class TenantCSPResult(AliasModel):
user_id: str
tenant_id: str
user_object_id: str
class Config:
allow_population_by_field_name = True
fields = {
"user_id": "userId",
"tenant_id": "tenantId",
"user_object_id": "objectId",
}
class BillingProfileAddress(BaseModel):
address: Dict
"""
"address": {
"firstName": "string",
"lastName": "string",
"companyName": "string",
"addressLine1": "string",
"addressLine2": "string",
"addressLine3": "string",
"city": "string",
"region": "string",
"country": "string",
"postalCode": "string"
},
"""
class BillingProfileAddress(AliasModel):
class BillingProfileCLINBudget(BaseModel):
clinBudget: Dict
class BillingProfileCLINBudget(AliasModel):
clin_budget: Dict
"""
"clinBudget": {
"amount": 0,

View File

@ -25,6 +25,11 @@ def camel_to_snake(camel_cased):
return re.sub("([a-z0-9])([A-Z])", r"\1_\2", s1).lower()
def snake_to_camel(snake_cased):
parts = snake_cased.split("_")
return f"{parts[0]}{''.join([w.capitalize() for w in parts[1:]])}"
def pick(keys, dct):
_keys = set(keys)
return {k: v for (k, v) in dct.items() if k in _keys}

View File

@ -3,7 +3,11 @@ from unittest.mock import Mock
from uuid import uuid4
from atst.domain.csp.cloud import AzureCloudProvider, TenantCSPResult
from atst.domain.csp.cloud import (
AzureCloudProvider,
TenantCSPResult,
TenantCSPPayload,
)
from tests.mock_azure import mock_azure, AUTH_CREDENTIALS
from tests.factories import EnvironmentFactory, ApplicationFactory
@ -136,7 +140,19 @@ def test_create_tenant(mock_azure: AzureCloudProvider):
}
mock_result.status_code = 200
mock_azure.sdk.requests.post.return_value = mock_result
result = mock_azure.create_tenant(None, suffix=2)
payload = TenantCSPPayload(
**dict(
creds={"username": "mock-cloud", "pass": "shh"},
user_id="123",
password="123",
domain_name="123",
first_name="john",
last_name="doe",
country_code="US",
password_recovery_email_address="password@email.com",
)
)
result = mock_azure.create_tenant(payload)
print(result)
body: TenantCSPResult = result.get("body")
assert body.tenant_id == "60ff9d34-82bf-4f21-b565-308ef0533435"