diff --git a/atst/domain/csp/cloud.py b/atst/domain/csp/cloud.py index 7b67e7d3..ea512b08 100644 --- a/atst/domain/csp/cloud.py +++ b/atst/domain/csp/cloud.py @@ -124,8 +124,9 @@ class MockCloudProvider(CloudProviderInterface): AUTH_EXCEPTION = GeneralCSPException("Authentication failure.") NETWORK_EXCEPTION = GeneralCSPException("Network failure.") - NETWORK_FAILURE_PCT = 12 + NETWORK_FAILURE_PCT = 7 ENV_CREATE_FAILURE_PCT = 12 + ATAT_ADMIN_CREATE_FAILURE_PCT = 12 def __init__(self, with_delay=True, with_failure=True): from time import sleep @@ -137,19 +138,39 @@ class MockCloudProvider(CloudProviderInterface): self._random = random def create_environment(self, auth_credentials, user, environment): - self._delay(1, 5) self._authorize(auth_credentials) self._delay(1, 5) self._maybe_throw(self.NETWORK_FAILURE_PCT, self.NETWORK_EXCEPTION) - self._maybe_throw(self.ENV_CREATE_FAILURE_PCT, GeneralCSPException("Could not create environment.")) + self._maybe_throw( + self.ENV_CREATE_FAILURE_PCT, + GeneralCSPException("Could not create environment."), + ) return self._id() def create_atat_admin_user(self, auth_credentials, csp_environment_id): + self._authorize(auth_credentials) + + self._delay(1, 5) + self._maybe_throw(self.NETWORK_FAILURE_PCT, self.NETWORK_EXCEPTION) + self._maybe_throw( + self.ATAT_ADMIN_CREATE_FAILURE_PCT, + GeneralCSPException("Could not create admin user."), + ) + return {"id": self._id(), "credentials": {}} def create_environment_baseline(self, auth_credentials, csp_environment_id): + self._authorize(auth_credentials) + + self._delay(1, 5) + self._maybe_throw(self.NETWORK_FAILURE_PCT, self.NETWORK_EXCEPTION) + self._maybe_throw( + self.ATAT_ADMIN_CREATE_FAILURE_PCT, + GeneralCSPException("Could not create environment baseline."), + ) + return { CSPRole.BASIC_ACCESS: self._id(), CSPRole.NETWORK_ADMIN: self._id(), @@ -158,13 +179,22 @@ class MockCloudProvider(CloudProviderInterface): } def create_or_update_user(self, auth_credentials, user_info, csp_role_id): + self._authorize(auth_credentials) + + self._delay(1, 5) + self._maybe_throw(self.NETWORK_FAILURE_PCT, self.NETWORK_EXCEPTION) + self._maybe_throw( + self.ATAT_ADMIN_CREATE_FAILURE_PCT, + GeneralCSPException("Could not create user."), + ) + return {"id": self._id()} def suspend_user(self, auth_credentials, csp_user_id): - pass + return self._maybe(12) def delete_user(self, auth_credentials, csp_user_id): - pass + return self._maybe(12) def get_calculator_url(self): return "https://www.rackspace.com/en-us/calculator" @@ -182,13 +212,18 @@ class MockCloudProvider(CloudProviderInterface): duration = self._random.randrange(min_secs, max_secs) self._sleep(duration) + def _maybe(self, pct): + return not self._with_failure or self._random.randrange(0, 100) < pct + def _maybe_throw(self, pct, exc): - if self._with_failure and self._random.randrange(0, 100) < pct: + if self._with_failure and self._maybe(pct): raise exc + @property def _auth_credentials(self): return {"username": "mock-cloud", "pass": "shh"} def _authorize(self, credentials): - if credentials != self._auth_credentials(): + self._delay(1, 5) + if credentials != self._auth_credentials: raise self.AUTH_EXCEPTION diff --git a/tests/domain/test_mock_csp.py b/tests/domain/test_mock_csp.py index a16b347f..5dd063c8 100644 --- a/tests/domain/test_mock_csp.py +++ b/tests/domain/test_mock_csp.py @@ -2,6 +2,7 @@ import pytest from atst.domain.csp import MockCloudProvider +CREDENTIALS = MockCloudProvider()._auth_credentials @pytest.fixture def mock_csp(): @@ -9,6 +10,29 @@ def mock_csp(): def test_create_environment(mock_csp: MockCloudProvider): - credentials = mock_csp._auth_credentials() - environment_id = mock_csp.create_environment(credentials, {}, {}) + environment_id = mock_csp.create_environment(CREDENTIALS, {}, {}) assert isinstance(environment_id, str) + + +def test_create_admin_user(mock_csp: MockCloudProvider): + admin_user = mock_csp.create_atat_admin_user(CREDENTIALS, "env_id") + assert isinstance(admin_user["id"], str) + assert isinstance(admin_user["credentials"], dict) + + +def test_create_environment_baseline(mock_csp: MockCloudProvider): + baseline = mock_csp.create_atat_admin_user(CREDENTIALS, "env_id") + assert isinstance(baseline, dict) + + +def test_create_or_update_user(mock_csp: MockCloudProvider): + user_dict = mock_csp.create_or_update_user(CREDENTIALS, {}, "csp_role_id") + assert isinstance(user_dict["id"], str) + + +def test_suspend_user(mock_csp: MockCloudProvider): + assert mock_csp.suspend_user(CREDENTIALS, "csp_user_id") + + +def test_delete_user(mock_csp: MockCloudProvider): + assert mock_csp.delete_user(CREDENTIALS, "csp_user_id")