pk/atst
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #1050 from dod-ccpo/cloud-provision-interface

Browse Source 
Update CloudProviderInterface for provision job consumption
This commit is contained in:
tomdds 2019-09-03 15:59:55 -04:00 committed by GitHub
commit 85f8c8f9e0
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
7 changed files with 118 additions and 74 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

159
atst/domain/csp/cloud.py
View File

@ -1,84 +1,147 @@
from typing import Dict
from uuid import uuid4
from atst.models.environment_role import CSPRole
from atst.models.user import User
from atst.models.environment import Environment
from atst.models.environment_role import EnvironmentRole
class CloudProviderInterface:
def create_application(self, name): # pragma: no cover
"""Create an application in the cloud with the provided name. Returns
the ID of the created object.
def create_environment(
self, auth_credentials: Dict, user: User, environment: Environment
) -> str:
"""Create a new environment in the CSP.
Arguments:
auth_credentials -- Object containing CSP account credentials
user -- ATAT user authorizing the environment creation
environment -- ATAT Environment model
Returns:
string: ID of created environment
"""
raise NotImplementedError()
def delete_application(self, cloud_id): # pragma: no cover
"""Delete an application in the cloud with the provided cloud_id. Returns
True for success or raises an error.
def create_atat_admin_user(
self, auth_credentials: Dict, csp_environment_id: str
) -> Dict:
"""Creates a new, programmatic user in the CSP. Grants this user full permissions to administer
the CSP.
Arguments:
auth_credentials -- Object containing CSP account credentials
csp_environment_id -- ID of the CSP Environment the admin user should be created in
Returns:
object: Object representing new remote admin user, including credentials
Something like:
{
"user_id": string,
"credentials": dict, # structure TBD based on csp
}
"""
raise NotImplementedError()
def create_user(self, user): # pragma: no cover
"""Create an account in the CSP for specified user. Returns the ID of
the created user.
def create_environment_baseline(
self, auth_credentials: Dict, csp_environment_id: str
) -> Dict:
"""Provision the necessary baseline entities (such as roles) in the given environment
Arguments:
auth_credentials -- Object containing CSP account credentials
csp_environment_id -- ID of the CSP Environment to provision roles against.
Returns:
dict: Returns dict that associates the resource identities with their ATAT representations.
"""
raise NotImplementedError()
def create_role(self, environment_role): # pragma: no cover
"""Takes an `atst.model.EnvironmentRole` object and allows the
specified user access to the specified cloud entity.
def create_or_update_user(
self, auth_credentials: Dict, user_info: EnvironmentRole, csp_role_id: str
) -> str:
"""Creates a user or updates an existing user's role.
This _does not_ return a token, but is intended to perform any necessary
setup to allow a token to be generated in the future (for example,
add the user to the cloud entity in some fashion).
Arguments:
auth_credentials -- Object containing CSP account credentials
user_info -- instance of EnvironmentRole containing user data
if it has a csp_user_id it will try to update that user
csp_role_id -- The id of the role the user should be given in the CSP
Returns:
string: Returns the interal csp_user_id of the created/updated user account
"""
raise NotImplementedError()
def delete_role(self, environment_role): # pragma: no cover
"""Takes an `atst.model.EnvironmentRole` object and performs any action
necessary in the CSP to remove the specified user from the specified
environment. This method does not return anything.
def suspend_user(self, auth_credentials: Dict, csp_user_id: str) -> bool:
"""Revoke all privileges for a user. Used to prevent user access while a full
delete is being processed.
Arguments:
auth_credentials -- Object containing CSP account credentials
csp_user_id -- CSP internal user identifier
Returns:
bool -- True on success
"""
raise NotImplementedError()
def get_access_token(self, environment_role): # pragma: no cover
"""Takes an `atst.model.EnvironmentRole` object and returns a federated
access token that gives the specified user access to the specified
environment with the proper permissions.
def delete_user(self, auth_credentials: Dict, csp_user_id: str) -> bool:
"""Given the csp-internal id for a user, initiate user deletion.
Arguments:
auth_credentials -- Object containing CSP account credentials
csp_user_id -- CSP internal user identifier
Returns:
bool -- True on success
Raises:
TBDException: Some part of user deletion failed
"""
raise NotImplementedError()
def calculator_url(self): # pragma: no cover
"""Returns a URL for the CSP's estimate calculator."""
def get_calculator_url(self) -> str:
"""Returns the calculator url for the CSP.
This will likely be a static property elsewhere once a CSP is chosen.
"""
raise NotImplementedError()
def get_environment_login_url(self, environment) -> str:
"""Returns the login url for a given environment
This may move to be a computed property on the Environment domain object
"""
raise NotImplementedError()
class MockCloudProvider(CloudProviderInterface):
def create_application(self, name):
"""Returns an id that represents what would be an application in the
cloud."""
def create_environment(self, auth_credentials, user, environment):
return uuid4().hex
def delete_application(self, name):
"""Returns an id that represents what would be an application in the
cloud."""
return True
def create_atat_admin_user(self, auth_credentials, csp_environment_id):
return {"id": uuid4().hex, "credentials": {}}
def create_user(self, user):
"""Returns an id that represents what would be an user in the cloud."""
return uuid4().hex
def create_environment_baseline(self, auth_credentials, csp_environment_id):
return {
CSPRole.BASIC_ACCESS: uuid4().hex,
CSPRole.NETWORK_ADMIN: uuid4().hex,
CSPRole.BUSINESS_READ: uuid4().hex,
CSPRole.TECHNICAL_READ: uuid4().hex,
}
def create_role(self, environment_role):
# Currently, there is nothing to mock out, so just do nothing.
def create_or_update_user(self, auth_credentials, user_info, csp_role_id):
return {"id": uuid4().hex}
def suspend_user(self, auth_credentials, csp_user_id):
pass
def delete_role(self, environment_role):
# Currently nothing to do.
def delete_user(self, auth_credentials, csp_user_id):
pass
def get_access_token(self, environment_role):
# for now, just create a mock token using the user and environment
# cloud IDs and the name of the role in the environment
user_id = environment_role.application_role.user.cloud_id or ""
env_id = environment_role.environment.cloud_id or ""
role_details = environment_role.role
return "::".join([user_id, env_id, role_details])
def calculator_url(self):
def get_calculator_url(self):
return "https://www.rackspace.com/en-us/calculator"
def get_environment_login_url(self, environment):
"""Returns the login url for a given environment
"""
return "https://www.mycloud.com/my-env-login"

8
atst/domain/environment_roles.py
View File

@ -1,5 +1,3 @@
from flask import current_app as app
from atst.database import db
from atst.models import EnvironmentRole, ApplicationRole
@ -10,10 +8,6 @@ class EnvironmentRoles(object):
env_role = EnvironmentRole(
application_role=application_role, environment=environment, role=role
)
# TODO: move cloud_id behavior to invitation acceptance
# if not user.cloud_id:
# user.cloud_id = app.csp.cloud.create_user(user)
app.csp.cloud.create_role(env_role)
return env_role
@classmethod
@ -45,7 +39,7 @@ class EnvironmentRoles(object):
def delete(cls, application_role_id, environment_id):
existing_env_role = EnvironmentRoles.get(application_role_id, environment_id)
if existing_env_role:
app.csp.cloud.delete_role(existing_env_role)
# TODO: Set status to pending_delete
db.session.delete(existing_env_role)
db.session.commit()
return True

4
atst/domain/environments.py
View File

@ -1,4 +1,3 @@
from flask import current_app as app
from sqlalchemy.orm.exc import NoResultFound
from atst.database import db
@ -13,7 +12,6 @@ class Environments(object):
@classmethod
def create(cls, application, name):
environment = Environment(application=application, name=name)
environment.cloud_id = app.csp.cloud.create_application(environment.name)
db.session.add(environment)
db.session.commit()
return environment
@ -101,6 +99,6 @@ class Environments(object):
if commit:
db.session.commit()
app.csp.cloud.delete_application(environment.cloud_id)
# TODO: How do we work around environment deletion being a largely manual process in the CSPs
return environment

2
atst/routes/__init__.py
View File

@ -131,4 +131,4 @@ def csp_environment_access():
@bp.route("/jedi-csp-calculator")
def jedi_csp_calculator():
return redirect(app.csp.cloud.calculator_url())
return redirect(app.csp.cloud.get_calculator_url())

4
atst/routes/applications/__init__.py
View File

@ -30,6 +30,6 @@ def access_environment(environment_id):
env_role = EnvironmentRoles.get_by_user_and_environment(
g.current_user.id, environment_id
)
token = app.csp.cloud.get_access_token(env_role)
login_url = app.csp.cloud.get_environment_login_url(env_role.environment)
return redirect(url_for("atst.csp_environment_access", token=token))
return redirect(url_for("atst.csp_environment_access", login_url=login_url))

14
tests/domain/test_environment_roles.py
View File

@ -1,5 +1,4 @@
import pytest
from unittest.mock import MagicMock
from atst.domain.environment_roles import EnvironmentRoles
@ -19,10 +18,6 @@ def environment(application_role):
def test_create(application_role, environment, monkeypatch):
mock_create_role = MagicMock()
monkeypatch.setattr(
"atst.domain.environment_roles.app.csp.cloud.create_role", mock_create_role
)
environment_role = EnvironmentRoles.create(
application_role, environment, "network admin"
@ -30,7 +25,6 @@ def test_create(application_role, environment, monkeypatch):
assert environment_role.application_role == application_role
assert environment_role.environment == environment
assert environment_role.role == "network admin"
mock_create_role.assert_called_with(environment_role)
def test_get(application_role, environment):
@ -55,16 +49,10 @@ def test_get_by_user_and_environment(application_role, environment):
def test_delete(application_role, environment, monkeypatch):
mock_delete_role = MagicMock()
monkeypatch.setattr(
"atst.domain.environment_roles.app.csp.cloud.delete_role", mock_delete_role
)
environment_role = EnvironmentRoleFactory.create(
EnvironmentRoleFactory.create(
application_role=application_role, environment=environment
)
assert EnvironmentRoles.delete(application_role.id, environment.id)
mock_delete_role.assert_called_with(environment_role)
assert not EnvironmentRoles.delete(application_role.id, environment.id)

1
tests/domain/test_environments.py
View File

@ -15,6 +15,7 @@ from tests.factories import (
)
@pytest.mark.skip(reason="Reinstate and update once jobs api is up")
def test_create_environments():
application = ApplicationFactory.create()
environments = Environments.create_many(application, ["Staging", "Production"])