Read config to determine mock CSP type
This commit is contained in:
parent
fd65a3a972
commit
6c0420d6c5
@ -4,22 +4,24 @@ from .reports import MockReportingProvider
|
|||||||
|
|
||||||
|
|
||||||
class MockCSP:
|
class MockCSP:
|
||||||
def __init__(self, app):
|
def __init__(self, app, test_mode=False):
|
||||||
self.cloud = MockCloudProvider()
|
self.cloud = MockCloudProvider(
|
||||||
|
app.config, with_delay=(not test_mode), with_failure=(not test_mode)
|
||||||
|
)
|
||||||
self.files = MockUploader(app)
|
self.files = MockUploader(app)
|
||||||
self.reports = MockReportingProvider()
|
self.reports = MockReportingProvider()
|
||||||
|
|
||||||
|
|
||||||
class AzureCSP:
|
class AzureCSP:
|
||||||
def __init__(self, app):
|
def __init__(self, app):
|
||||||
self.cloud = MockCloudProvider()
|
self.cloud = MockCloudProvider(app.config)
|
||||||
self.files = AzureUploader(app.config)
|
self.files = AzureUploader(app.config)
|
||||||
self.reports = MockReportingProvider()
|
self.reports = MockReportingProvider()
|
||||||
|
|
||||||
|
|
||||||
class AwsCSP:
|
class AwsCSP:
|
||||||
def __init__(self, app):
|
def __init__(self, app):
|
||||||
self.cloud = MockCloudProvider()
|
self.cloud = MockCloudProvider(app.config)
|
||||||
self.files = AwsUploader(app.config)
|
self.files = AwsUploader(app.config)
|
||||||
self.reports = MockReportingProvider()
|
self.reports = MockReportingProvider()
|
||||||
|
|
||||||
@ -29,5 +31,7 @@ def make_csp_provider(app, csp=None):
|
|||||||
app.csp = AwsCSP(app)
|
app.csp = AwsCSP(app)
|
||||||
elif csp == "azure":
|
elif csp == "azure":
|
||||||
app.csp = AzureCSP(app)
|
app.csp = AzureCSP(app)
|
||||||
|
elif csp == "mock-test":
|
||||||
|
app.csp = MockCSP(app, test_mode=True)
|
||||||
else:
|
else:
|
||||||
app.csp = MockCSP(app)
|
app.csp = MockCSP(app)
|
||||||
|
@ -12,6 +12,9 @@ class GeneralCSPException(Exception):
|
|||||||
|
|
||||||
|
|
||||||
class CloudProviderInterface:
|
class CloudProviderInterface:
|
||||||
|
def root_creds() -> Dict:
|
||||||
|
raise NotImplementedError()
|
||||||
|
|
||||||
def create_environment(
|
def create_environment(
|
||||||
self, auth_credentials: Dict, user: User, environment: Environment
|
self, auth_credentials: Dict, user: User, environment: Environment
|
||||||
) -> str:
|
) -> str:
|
||||||
@ -128,7 +131,7 @@ class MockCloudProvider(CloudProviderInterface):
|
|||||||
ENV_CREATE_FAILURE_PCT = 12
|
ENV_CREATE_FAILURE_PCT = 12
|
||||||
ATAT_ADMIN_CREATE_FAILURE_PCT = 12
|
ATAT_ADMIN_CREATE_FAILURE_PCT = 12
|
||||||
|
|
||||||
def __init__(self, with_delay=True, with_failure=True):
|
def __init__(self, config, with_delay=True, with_failure=True):
|
||||||
from time import sleep
|
from time import sleep
|
||||||
import random
|
import random
|
||||||
|
|
||||||
@ -137,6 +140,9 @@ class MockCloudProvider(CloudProviderInterface):
|
|||||||
self._sleep = sleep
|
self._sleep = sleep
|
||||||
self._random = random
|
self._random = random
|
||||||
|
|
||||||
|
def root_creds(self):
|
||||||
|
return self._auth_credentials
|
||||||
|
|
||||||
def create_environment(self, auth_credentials, user, environment):
|
def create_environment(self, auth_credentials, user, environment):
|
||||||
self._authorize(auth_credentials)
|
self._authorize(auth_credentials)
|
||||||
|
|
||||||
@ -159,7 +165,7 @@ class MockCloudProvider(CloudProviderInterface):
|
|||||||
GeneralCSPException("Could not create admin user."),
|
GeneralCSPException("Could not create admin user."),
|
||||||
)
|
)
|
||||||
|
|
||||||
return {"id": self._id(), "credentials": {}}
|
return {"id": self._id(), "credentials": self._auth_credentials}
|
||||||
|
|
||||||
def create_environment_baseline(self, auth_credentials, csp_environment_id):
|
def create_environment_baseline(self, auth_credentials, csp_environment_id):
|
||||||
self._authorize(auth_credentials)
|
self._authorize(auth_credentials)
|
||||||
@ -172,10 +178,10 @@ class MockCloudProvider(CloudProviderInterface):
|
|||||||
)
|
)
|
||||||
|
|
||||||
return {
|
return {
|
||||||
CSPRole.BASIC_ACCESS: self._id(),
|
CSPRole.BASIC_ACCESS.value: self._id(),
|
||||||
CSPRole.NETWORK_ADMIN: self._id(),
|
CSPRole.NETWORK_ADMIN.value: self._id(),
|
||||||
CSPRole.BUSINESS_READ: self._id(),
|
CSPRole.BUSINESS_READ.value: self._id(),
|
||||||
CSPRole.TECHNICAL_READ: self._id(),
|
CSPRole.TECHNICAL_READ.value: self._id(),
|
||||||
}
|
}
|
||||||
|
|
||||||
def create_or_update_user(self, auth_credentials, user_info, csp_role_id):
|
def create_or_update_user(self, auth_credentials, user_info, csp_role_id):
|
||||||
|
@ -6,3 +6,4 @@ CRL_STORAGE_CONTAINER = tests/fixtures/crl
|
|||||||
WTF_CSRF_ENABLED = false
|
WTF_CSRF_ENABLED = false
|
||||||
STORAGE_PROVIDER=LOCAL
|
STORAGE_PROVIDER=LOCAL
|
||||||
PRESERVE_CONTEXT_ON_EXCEPTION = false
|
PRESERVE_CONTEXT_ON_EXCEPTION = false
|
||||||
|
CSP=mock-test
|
||||||
|
@ -2,12 +2,12 @@ import pytest
|
|||||||
|
|
||||||
from atst.domain.csp import MockCloudProvider
|
from atst.domain.csp import MockCloudProvider
|
||||||
|
|
||||||
CREDENTIALS = MockCloudProvider()._auth_credentials
|
CREDENTIALS = MockCloudProvider(config={})._auth_credentials
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture
|
@pytest.fixture
|
||||||
def mock_csp():
|
def mock_csp():
|
||||||
return MockCloudProvider(with_delay=False, with_failure=False)
|
return MockCloudProvider(config={}, with_delay=False, with_failure=False)
|
||||||
|
|
||||||
|
|
||||||
def test_create_environment(mock_csp: MockCloudProvider):
|
def test_create_environment(mock_csp: MockCloudProvider):
|
||||||
|
Loading…
x
Reference in New Issue
Block a user