From 9015e7f59bfbcc4e0ac72299aefb294ac6d3bc26 Mon Sep 17 00:00:00 2001 From: Adrienne Stilp Date: Wed, 30 Nov 2022 10:23:49 -0800 Subject: [PATCH 1/7] Add a new API method to clone a workspace --- anvil_consortium_manager/anvil_api.py | 45 +++++++++++++++++++++++++++ 1 file changed, 45 insertions(+) diff --git a/anvil_consortium_manager/anvil_api.py b/anvil_consortium_manager/anvil_api.py index 68c68b04..051181b1 100644 --- a/anvil_consortium_manager/anvil_api.py +++ b/anvil_consortium_manager/anvil_api.py @@ -251,6 +251,51 @@ def create_workspace( return self.auth_session.post(method, 201, json=body) + def clone_workspace( + self, + existing_workspace_namespace, + existing_workspace_name, + cloned_workspace_namespace, + cloned_workspace_name, + authorization_domains=[], + ): + """Clone an existing workspace on AnVIL. + + Calls the /api/create_workspace POST method. + + Args: + existing_workspace_namespace (str): The namespace (or billing project) of the + existing workspace to clone. + existing_workspace_name (str): The name of the existing workspace to clone. + cloned_workspace_namespace (str): The namespace (or billing project) in which + to create the cloned workspace. + cloned_workspace_name (str): The name of the cloned workspace to create. + authorization_domains (str): If desired, a list of group names that should be + used as the authorization domain for this workspace. This must include the + authorization domains of the existing workspace. + + Returns: + requests.Response + """ + method = "api/workspaces/{namespace}/{name}/clone".format( + namespace=existing_workspace_namespace, + name=existing_workspace_name, + ) + body = { + "namespace": cloned_workspace_namespace, + "name": cloned_workspace_name, + "attributes": {}, + } + + # Add authorization domains. + if authorization_domains: + if not isinstance(authorization_domains, list): + authorization_domains = [authorization_domains] + auth_domain = [{"membersGroupName": g} for g in authorization_domains] + body["authorizationDomain"] = auth_domain + + return self.auth_session.post(method, 201, json=body) + def delete_workspace(self, workspace_namespace, workspace_name): """Delete a workspace on AnVIL. You must be an owner of the workspace to use this method. From 5031aea6ffbea59aeaefbfd75248d8bbe65d9c76 Mon Sep 17 00:00:00 2001 From: Adrienne Stilp Date: Wed, 30 Nov 2022 14:56:19 -0800 Subject: [PATCH 2/7] Add an anvil_clone method for a Workspace This method clones an existing workspace on AnVIL, but does not create or save a record of the new workspace in the app. This is because we need to also specify the workspace type, and that can't easily be done via this method. Therefore, any methods calling anvil_clone should also call somethign like anvil_import or otherwise create the new workspace plus workspace data in the app. --- anvil_consortium_manager/models.py | 35 ++ .../tests/test_models_anvil_api_unit.py | 388 ++++++++++++++++++ 2 files changed, 423 insertions(+) diff --git a/anvil_consortium_manager/models.py b/anvil_consortium_manager/models.py index f1ccc6b4..5a69c001 100644 --- a/anvil_consortium_manager/models.py +++ b/anvil_consortium_manager/models.py @@ -708,6 +708,41 @@ def anvil_delete(self): """Delete the workspace on AnVIL.""" AnVILAPIClient().delete_workspace(self.billing_project.name, self.name) + def anvil_clone( + self, billing_project, workspace_name, additional_authorization_domains=[] + ): + """Clone this workspace to create a new workspace on AnVIL.""" + # Check that the app can create workspaes in this billing project. + if not billing_project.has_app_as_user: + raise ValueError("BillingProject must have has_app_as_user=True.") + # Check that the new workspace does not already exist in the database. + if Workspace.objects.filter( + billing_project=billing_project, name=workspace_name + ).exists(): + raise ValueError( + "Workspace with this BillingProject and Name already exists." + ) + # All checks have passed, so start the cloning process. + # Set up new auth domains using: + # - existing auth domains for the workspace being cloned + # - new auth domains that are specified when cloning. + current_auth_domains = self.authorization_domains.all() + auth_domains = [g.name for g in current_auth_domains] + [ + g.name + for g in additional_authorization_domains + if g not in current_auth_domains + ] + # Clone the workspace on AnVIL. + AnVILAPIClient().clone_workspace( + self.billing_project.name, + self.name, + billing_project.name, + workspace_name, + authorization_domains=auth_domains, + ) + # Do not create the cloned workspace - it can be imported or created elsewhere. + # That way, the workspace_type can be set. + @classmethod def anvil_import( cls, billing_project_name, workspace_name, workspace_type, note="" diff --git a/anvil_consortium_manager/tests/test_models_anvil_api_unit.py b/anvil_consortium_manager/tests/test_models_anvil_api_unit.py index 2706829f..e13646f6 100644 --- a/anvil_consortium_manager/tests/test_models_anvil_api_unit.py +++ b/anvil_consortium_manager/tests/test_models_anvil_api_unit.py @@ -2310,6 +2310,394 @@ def test_anvil_delete_other(self): responses.assert_call_count(self.url_workspace, 1) +class WorkspaceAnVILCloneTest(AnVILAPIMockTestMixin, TestCase): + """Tests of the Workspace.anvil_clone method.""" + + def setUp(self): + super().setUp() + self.workspace = factories.WorkspaceFactory.create() + + def get_api_url(self, billing_project_name, workspace_name): + return ( + self.entry_point + + "/api/workspaces/" + + billing_project_name + + "/" + + workspace_name + + "/clone" + ) + + def get_api_json_response(self, billing_project_name, workspace_name): + """Return a pared down version of the json response from the AnVIL API with only fields we need.""" + json_data = { + "name": workspace_name, + "namespace": billing_project_name, + } + return json_data + + def test_can_clone_workspace_no_auth_domain(self): + """Can clone a workspace with no auth domains.""" + billing_project = factories.BillingProjectFactory.create() + # Add response. + responses.add( + responses.POST, + self.get_api_url(self.workspace.billing_project.name, self.workspace.name), + status=201, # successful response code. + json=self.get_api_json_response(billing_project.name, "test-workspace"), + match=[ + responses.matchers.json_params_matcher( + { + "namespace": billing_project.name, + "name": "test-workspace", + "attributes": {}, + } + ) + ], + ) + self.workspace.anvil_clone(billing_project, "test-workspace") + # No new workspaces were created in the app. + self.assertEqual(models.Workspace.objects.count(), 1) + self.assertIn(self.workspace, models.Workspace.objects.all()) + + def test_can_clone_workspace_one_auth_domain(self): + """Can clone a workspace with one auth domain.""" + auth_domain = factories.ManagedGroupFactory.create() + self.workspace.authorization_domains.add(auth_domain) + billing_project = factories.BillingProjectFactory.create() + # Add response. + responses.add( + responses.POST, + self.get_api_url(self.workspace.billing_project.name, self.workspace.name), + status=201, # successful response code. + json=self.get_api_json_response(billing_project.name, "test-workspace"), + match=[ + responses.matchers.json_params_matcher( + { + "namespace": billing_project.name, + "name": "test-workspace", + "attributes": {}, + "authorizationDomain": [{"membersGroupName": auth_domain.name}], + } + ) + ], + ) + self.workspace.anvil_clone(billing_project, "test-workspace") + + def test_can_clone_workspace_two_auth_domains(self): + """Can clone a workspace with two auth domains.""" + auth_domain_1 = factories.ManagedGroupFactory.create() + auth_domain_2 = factories.ManagedGroupFactory.create() + self.workspace.authorization_domains.add(auth_domain_1, auth_domain_2) + billing_project = factories.BillingProjectFactory.create() + # Add response. + responses.add( + responses.POST, + self.get_api_url(self.workspace.billing_project.name, self.workspace.name), + status=201, # successful response code. + json=self.get_api_json_response(billing_project.name, "test-workspace"), + match=[ + responses.matchers.json_params_matcher( + { + "namespace": billing_project.name, + "name": "test-workspace", + "attributes": {}, + "authorizationDomain": [ + {"membersGroupName": auth_domain_1.name}, + {"membersGroupName": auth_domain_2.name}, + ], + } + ) + ], + ) + self.workspace.anvil_clone(billing_project, "test-workspace") + + def test_can_clone_workspace_add_one_auth_domain(self): + """Can clone a workspace and add one auth domain.""" + auth_domain = factories.ManagedGroupFactory.create() + billing_project = factories.BillingProjectFactory.create() + # Add response. + responses.add( + responses.POST, + self.get_api_url(self.workspace.billing_project.name, self.workspace.name), + status=201, # successful response code. + json=self.get_api_json_response(billing_project.name, "test-workspace"), + match=[ + responses.matchers.json_params_matcher( + { + "namespace": billing_project.name, + "name": "test-workspace", + "attributes": {}, + "authorizationDomain": [{"membersGroupName": auth_domain.name}], + } + ) + ], + ) + # import ipdb; ipdb.set_trace() + self.workspace.anvil_clone( + billing_project, + "test-workspace", + additional_authorization_domains=[auth_domain], + ) + + def test_can_clone_workspace_add_two_auth_domains(self): + """Can clone a workspace and add one auth domain.""" + auth_domain_1 = factories.ManagedGroupFactory.create() + auth_domain_2 = factories.ManagedGroupFactory.create() + billing_project = factories.BillingProjectFactory.create() + # Add response. + responses.add( + responses.POST, + self.get_api_url(self.workspace.billing_project.name, self.workspace.name), + status=201, # successful response code. + json=self.get_api_json_response(billing_project.name, "test-workspace"), + match=[ + responses.matchers.json_params_matcher( + { + "namespace": billing_project.name, + "name": "test-workspace", + "attributes": {}, + "authorizationDomain": [ + {"membersGroupName": auth_domain_1.name}, + {"membersGroupName": auth_domain_2.name}, + ], + } + ) + ], + ) + self.workspace.anvil_clone( + billing_project, + "test-workspace", + additional_authorization_domains=[auth_domain_1, auth_domain_2], + ) + + def test_can_clone_workspace_one_auth_domain_add_one_auth_domain(self): + """Can clone a workspace with one auth domain and add another auth domain.""" + auth_domain = factories.ManagedGroupFactory.create() + self.workspace.authorization_domains.add(auth_domain) + new_auth_domain = factories.ManagedGroupFactory.create() + billing_project = factories.BillingProjectFactory.create() + # Add response. + responses.add( + responses.POST, + self.get_api_url(self.workspace.billing_project.name, self.workspace.name), + status=201, # successful response code. + json=self.get_api_json_response(billing_project.name, "test-workspace"), + match=[ + responses.matchers.json_params_matcher( + { + "namespace": billing_project.name, + "name": "test-workspace", + "attributes": {}, + "authorizationDomain": [ + {"membersGroupName": auth_domain.name}, + {"membersGroupName": new_auth_domain.name}, + ], + } + ) + ], + ) + self.workspace.anvil_clone( + billing_project, + "test-workspace", + additional_authorization_domains=[new_auth_domain], + ) + + def test_can_clone_workspace_one_auth_domain_add_two_auth_domains(self): + """Can clone a workspace with one auth domain and add another auth domain.""" + auth_domain = factories.ManagedGroupFactory.create() + self.workspace.authorization_domains.add(auth_domain) + new_auth_domain_1 = factories.ManagedGroupFactory.create() + new_auth_domain_2 = factories.ManagedGroupFactory.create() + billing_project = factories.BillingProjectFactory.create() + # Add response. + responses.add( + responses.POST, + self.get_api_url(self.workspace.billing_project.name, self.workspace.name), + status=201, # successful response code. + json=self.get_api_json_response(billing_project.name, "test-workspace"), + match=[ + responses.matchers.json_params_matcher( + { + "namespace": billing_project.name, + "name": "test-workspace", + "attributes": {}, + "authorizationDomain": [ + {"membersGroupName": auth_domain.name}, + {"membersGroupName": new_auth_domain_1.name}, + {"membersGroupName": new_auth_domain_2.name}, + ], + } + ) + ], + ) + self.workspace.anvil_clone( + billing_project, + "test-workspace", + additional_authorization_domains=[new_auth_domain_1, new_auth_domain_2], + ) + + def test_can_clone_workspace_one_auth_domain_add_same_auth_domain(self): + """Can clone a workspace with one auth domain and add the same auth domain.""" + auth_domain = factories.ManagedGroupFactory.create() + self.workspace.authorization_domains.add(auth_domain) + billing_project = factories.BillingProjectFactory.create() + # Add response. + responses.add( + responses.POST, + self.get_api_url(self.workspace.billing_project.name, self.workspace.name), + status=201, # successful response code. + json=self.get_api_json_response(billing_project.name, "test-workspace"), + match=[ + responses.matchers.json_params_matcher( + { + "namespace": billing_project.name, + "name": "test-workspace", + "attributes": {}, + "authorizationDomain": [{"membersGroupName": auth_domain.name}], + } + ) + ], + ) + self.workspace.anvil_clone( + billing_project, + "test-workspace", + additional_authorization_domains=[auth_domain], + ) + + def test_error_cloning_workspace_into_billing_project_where_app_is_not_user(self): + """Error when cloning a workspace into a billing project where the app is not a user.""" + billing_project = factories.BillingProjectFactory.create(has_app_as_user=False) + # No API call exected. + with self.assertRaises(ValueError) as e: + self.workspace.anvil_clone(billing_project, "test-workspace") + self.assertIn("has_app_as_user", str(e.exception)) + # No new workspace was created. + self.assertEqual(models.Workspace.objects.count(), 1) + self.assertIn(self.workspace, models.Workspace.objects.all()) + + def test_error_workspace_already_exists_in_app(self): + """Error when the workspace to be cloned already exists in the app.""" + existing_workspace = factories.WorkspaceFactory.create() + # No API call exected. + with self.assertRaises(ValueError) as e: + self.workspace.anvil_clone( + existing_workspace.billing_project, existing_workspace.name + ) + self.assertIn("already exists", str(e.exception)) + # No new workspace was created. + self.assertEqual(models.Workspace.objects.count(), 2) + self.assertIn(self.workspace, models.Workspace.objects.all()) + self.assertIn(existing_workspace, models.Workspace.objects.all()) + + def test_error_workspace_already_exists_in_anvil_but_not_in_app(self): + """Error when the workspace to be cloned already exists in AnVIL but is not in the app.""" + billing_project = factories.BillingProjectFactory.create() + # Add response. + responses.add( + responses.POST, + self.get_api_url(self.workspace.billing_project.name, self.workspace.name), + status=409, # already exists + json={"message": "other"}, + match=[ + responses.matchers.json_params_matcher( + { + "namespace": billing_project.name, + "name": "test-workspace", + "attributes": {}, + } + ) + ], + ) + with self.assertRaises(anvil_api.AnVILAPIError409): + self.workspace.anvil_clone(billing_project, "test-workspace") + # No new workspace was created. + self.assertEqual(models.Workspace.objects.count(), 1) + self.assertIn(self.workspace, models.Workspace.objects.all()) + + def test_error_workspace_does_not_exist_in_anvil(self): + """Error the workspace to clone does not exist on AnVIL.""" + billing_project = factories.BillingProjectFactory.create() + # Add response. + responses.add( + responses.POST, + self.get_api_url(self.workspace.billing_project.name, self.workspace.name), + status=404, # already exists + json={"message": "other"}, + match=[ + responses.matchers.json_params_matcher( + { + "namespace": billing_project.name, + "name": "test-workspace", + "attributes": {}, + } + ) + ], + ) + with self.assertRaises(anvil_api.AnVILAPIError404): + self.workspace.anvil_clone(billing_project, "test-workspace") + # No new workspace was created. + self.assertEqual(models.Workspace.objects.count(), 1) + self.assertIn(self.workspace, models.Workspace.objects.all()) + + def test_error_authorization_domain_exists_in_app_but_not_on_anvil(self): + """Error when the authorization domain exists in the app but not on AnVIL.""" + billing_project = factories.BillingProjectFactory.create() + new_auth_domain = factories.ManagedGroupFactory.create() + # Add response. + responses.add( + responses.POST, + self.get_api_url(self.workspace.billing_project.name, self.workspace.name), + status=404, # resource not found + json={"message": "other"}, + match=[ + responses.matchers.json_params_matcher( + { + "namespace": billing_project.name, + "name": "test-workspace", + "attributes": {}, + "authorizationDomain": [ + {"membersGroupName": new_auth_domain.name} + ], + } + ) + ], + ) + with self.assertRaises(anvil_api.AnVILAPIError404): + self.workspace.anvil_clone( + billing_project, + "test-workspace", + additional_authorization_domains=[new_auth_domain], + ) + # No new workspace was created. + self.assertEqual(models.Workspace.objects.count(), 1) + self.assertIn(self.workspace, models.Workspace.objects.all()) + + def test_other_api_error(self): + """Error when the response is an error for an unknown reason.""" + billing_project = factories.BillingProjectFactory.create() + # Add response. + responses.add( + responses.POST, + self.get_api_url(self.workspace.billing_project.name, self.workspace.name), + status=500, # other + json={"message": "other"}, + match=[ + responses.matchers.json_params_matcher( + { + "namespace": billing_project.name, + "name": "test-workspace", + "attributes": {}, + } + ) + ], + ) + with self.assertRaises(anvil_api.AnVILAPIError500): + self.workspace.anvil_clone(billing_project, "test-workspace") + # No new workspace was created. + self.assertEqual(models.Workspace.objects.count(), 1) + self.assertIn(self.workspace, models.Workspace.objects.all()) + + class WorkspaceAnVILImportAnVILAPIMockTest(AnVILAPIMockTestMixin, TestCase): """Tests for the Workspace.anvil_import method.""" From d0c8b7a982cfb8c48e9a212db60f6e6c10978499 Mon Sep 17 00:00:00 2001 From: Adrienne Stilp Date: Tue, 6 Dec 2022 12:06:26 -0800 Subject: [PATCH 3/7] Add ability to clone a workspace Add a form for cloning a workspace and tets for that form. Add a view, url, and template for cloning a given workspace into a specific workspae type. Add a button to the workspace detail page for cloning that workspace. Modify the anvil_clone method for Workspace to not check if a workspace already exists; all this method does now is make the API call to AnVIL with the proper auth domains. This is because the view to clone the workspace saves the object and then calls anvil_clone. --- anvil_consortium_manager/forms.py | 95 ++ anvil_consortium_manager/models.py | 28 +- .../anvil_consortium_manager/base.html | 2 + .../workspace_clone.html | 23 + .../workspace_detail.html | 16 + anvil_consortium_manager/tests/test_forms.py | 216 ++++ .../tests/test_models_anvil_api_unit.py | 26 +- anvil_consortium_manager/tests/test_views.py | 1095 +++++++++++++++++ anvil_consortium_manager/urls.py | 10 + anvil_consortium_manager/views.py | 152 +++ 10 files changed, 1629 insertions(+), 34 deletions(-) create mode 100644 anvil_consortium_manager/templates/anvil_consortium_manager/workspace_clone.html diff --git a/anvil_consortium_manager/forms.py b/anvil_consortium_manager/forms.py index f6af0cc8..239d6393 100644 --- a/anvil_consortium_manager/forms.py +++ b/anvil_consortium_manager/forms.py @@ -182,6 +182,101 @@ def __init__(self, workspace_choices=[], *args, **kwargs): ) +class WorkspaceCloneForm(forms.ModelForm): + """Form to create a new workspace on AnVIL by cloning an existing workspace.""" + + # Only allow billing groups where we can create a workspace. + billing_project = forms.ModelChoiceField( + queryset=models.BillingProject.objects.filter(has_app_as_user=True), + widget=autocomplete.ModelSelect2( + url="anvil_consortium_manager:billing_projects:autocomplete", + attrs={"data-theme": "bootstrap-5"}, + ), + help_text="""Select the billing project in which the workspace should be created. + Only billing projects where this app is a user are shown.""", + ) + + class Meta: + model = models.Workspace + fields = ( + "billing_project", + "name", + "authorization_domains", + "note", + ) + widgets = { + "billing_project": autocomplete.ModelSelect2( + url="anvil_consortium_manager:billing_projects:autocomplete", + attrs={"data-theme": "bootstrap-5"}, + ), + "authorization_domains": autocomplete.ModelSelect2Multiple( + url="anvil_consortium_manager:managed_groups:autocomplete", + attrs={"data-theme": "bootstrap-5"}, + ), + } + help_texts = { + "billing_project": """Enter the billing project in which the workspace should be created. + Only billing projects that have this app as a user are shown.""", + "name": "Enter the name of the workspace to create.", + "authorization_domains": """Select the authorization domain(s) to use for this workspace. + This cannot be changed after creation.""", + } + + def __init__(self, workspace_to_clone, *args, **kwargs): + super().__init__(*args, **kwargs) + # Save this so we can modify the authorization domains to include this workspace's auth domain(s). + self.workspace_to_clone = workspace_to_clone + # Add the list of required auth domains to the help text. + if self.workspace_to_clone.authorization_domains.exists(): + auth_domain_names = ( + self.workspace_to_clone.authorization_domains.values_list( + "name", flat=True + ) + ) + print(auth_domain_names) + extra_text = " You must also include the authorization domain(s) from the original workspace ({}).".format( + ", ".join(auth_domain_names) + ) + self.fields["authorization_domains"].help_text = ( + self.fields["authorization_domains"].help_text + extra_text + ) + + def clean_authorization_domains(self): + """Verify that all authorization domains from the original workspace are selected.""" + authorization_domains = self.cleaned_data["authorization_domains"] + required_authorization_domains = ( + self.workspace_to_clone.authorization_domains.all() + ) + missing = [ + g for g in required_authorization_domains if g not in authorization_domains + ] + if missing: + msg = "Must contain all original workspace authorization domains: {}".format( + # ", ".join([g.name for g in self.workspace_to_clone.authorization_domains.all()]) + ", ".join(required_authorization_domains.values_list("name", flat=True)) + ) + raise ValidationError(msg) + return authorization_domains + + def clean(self): + # Check for the same case insensitive name in the same billing project. + billing_project = self.cleaned_data.get("billing_project", None) + name = self.cleaned_data.get("name", None) + if ( + billing_project + and name + and models.Workspace.objects.filter( + billing_project=billing_project, + name__iexact=name, + ).exists() + ): + # The workspace already exists - raise a Validation error. + raise ValidationError( + "Workspace with this Billing Project and Name already exists." + ) + return self.cleaned_data + + class DefaultWorkspaceDataForm(forms.ModelForm): """Default (empty) form for the workspace data object.""" diff --git a/anvil_consortium_manager/models.py b/anvil_consortium_manager/models.py index 5a69c001..45afe753 100644 --- a/anvil_consortium_manager/models.py +++ b/anvil_consortium_manager/models.py @@ -708,29 +708,29 @@ def anvil_delete(self): """Delete the workspace on AnVIL.""" AnVILAPIClient().delete_workspace(self.billing_project.name, self.name) - def anvil_clone( - self, billing_project, workspace_name, additional_authorization_domains=[] - ): - """Clone this workspace to create a new workspace on AnVIL.""" + def anvil_clone(self, billing_project, workspace_name, authorization_domains=[]): + """Clone this workspace to create a new workspace on AnVIL. + + If the workspace to clone already has authorization domains, they will be added to + the authorization domains specified in `authorization_domains`.""" # Check that the app can create workspaes in this billing project. if not billing_project.has_app_as_user: raise ValueError("BillingProject must have has_app_as_user=True.") - # Check that the new workspace does not already exist in the database. - if Workspace.objects.filter( - billing_project=billing_project, name=workspace_name - ).exists(): - raise ValueError( - "Workspace with this BillingProject and Name already exists." - ) + # Do not check if the new workspace already exists in the app. + # It may have already been created for some reason. + # if Workspace.objects.filter( + # billing_project=billing_project, name=workspace_name + # ).exists(): + # raise ValueError( + # "Workspace with this BillingProject and Name already exists." + # ) # All checks have passed, so start the cloning process. # Set up new auth domains using: # - existing auth domains for the workspace being cloned # - new auth domains that are specified when cloning. current_auth_domains = self.authorization_domains.all() auth_domains = [g.name for g in current_auth_domains] + [ - g.name - for g in additional_authorization_domains - if g not in current_auth_domains + g.name for g in authorization_domains if g not in current_auth_domains ] # Clone the workspace on AnVIL. AnVILAPIClient().clone_workspace( diff --git a/anvil_consortium_manager/templates/anvil_consortium_manager/base.html b/anvil_consortium_manager/templates/anvil_consortium_manager/base.html index 1294dffb..899993a3 100644 --- a/anvil_consortium_manager/templates/anvil_consortium_manager/base.html +++ b/anvil_consortium_manager/templates/anvil_consortium_manager/base.html @@ -11,6 +11,8 @@ + + {% block css %} diff --git a/anvil_consortium_manager/templates/anvil_consortium_manager/workspace_clone.html b/anvil_consortium_manager/templates/anvil_consortium_manager/workspace_clone.html new file mode 100644 index 00000000..7ee3af36 --- /dev/null +++ b/anvil_consortium_manager/templates/anvil_consortium_manager/workspace_clone.html @@ -0,0 +1,23 @@ +{% extends "anvil_consortium_manager/base.html" %} +{% load static %} +{% load crispy_forms_tags %} + +{% block title %}Clone {{ object }}{% endblock title %} + +{% block content %} + +

Clone {{ object }} on AnVIL

+ +
+ + {% csrf_token %} + {{ workspace_data_formset|crispy }} + {{ form|crispy }} + + +
+{% endblock %} + +{% block inline_javascript %} + {{ form.media }} +{% endblock inline_javascript %} diff --git a/anvil_consortium_manager/templates/anvil_consortium_manager/workspace_detail.html b/anvil_consortium_manager/templates/anvil_consortium_manager/workspace_detail.html index ca8538de..e997fd9a 100644 --- a/anvil_consortium_manager/templates/anvil_consortium_manager/workspace_detail.html +++ b/anvil_consortium_manager/templates/anvil_consortium_manager/workspace_detail.html @@ -29,6 +29,7 @@ {% endblock panel %} {% block after_panel %} +
@@ -68,6 +69,21 @@

Share with a group

+

+

+ + +
    + {% for workspace_type, workspace_name in registered_workspaces.items %} +
  • {{ workspace_name }}
    • + {% endfor %} + +
+
+

+

Update Delete on AnVIL diff --git a/anvil_consortium_manager/tests/test_forms.py b/anvil_consortium_manager/tests/test_forms.py index 1a7718ce..7ba7968e 100644 --- a/anvil_consortium_manager/tests/test_forms.py +++ b/anvil_consortium_manager/tests/test_forms.py @@ -461,6 +461,222 @@ def test_invalid_missing_workspace(self): self.assertEqual(len(form.errors), 1) +class WorkspaceCloneFormTest(TestCase): + """Tests for the WorkspaceCloneForm.""" + + form_class = forms.WorkspaceCloneForm + + def setUp(self): + """Create a workspace to clone for use in tests.""" + self.workspace_to_clone = factories.WorkspaceFactory.create() + + def test_valid_no_required_auth_domains(self): + """Form is valid with a workspace to clone with no auth domains, and no auth domains selected.""" + billing_project = factories.BillingProjectFactory.create() + form_data = { + "billing_project": billing_project, + "name": "test-workspace", + "authorization_domains": [], + } + form = self.form_class(self.workspace_to_clone, data=form_data) + self.assertTrue(form.is_valid()) + + def test_valid_no_required_auth_domains_with_one_selected_auth_domain(self): + """Form is valid with a workspace to clone with no auth domains, and one auth domain selected.""" + billing_project = factories.BillingProjectFactory.create() + new_auth_domain = factories.ManagedGroupFactory.create() + form_data = { + "billing_project": billing_project, + "name": "test-workspace", + "authorization_domains": [new_auth_domain], + } + form = self.form_class(self.workspace_to_clone, data=form_data) + self.assertTrue(form.is_valid()) + + def test_valid_no_required_auth_domains_with_two_selected_auth_domains(self): + """Form is valid with a workspace to clone with no auth domains, and two auth domains selected.""" + billing_project = factories.BillingProjectFactory.create() + new_auth_domain_1 = factories.ManagedGroupFactory.create() + new_auth_domain_2 = factories.ManagedGroupFactory.create() + form_data = { + "billing_project": billing_project, + "name": "test-workspace", + "authorization_domains": [new_auth_domain_1, new_auth_domain_2], + } + form = self.form_class(self.workspace_to_clone, data=form_data) + self.assertTrue(form.is_valid()) + + def test_valid_one_required_auth_domains(self): + """Form is valid with a workspace to clone with one auth domain, and that auth domain selected.""" + auth_domain = factories.ManagedGroupFactory.create() + self.workspace_to_clone.authorization_domains.add(auth_domain) + billing_project = factories.BillingProjectFactory.create() + form_data = { + "billing_project": billing_project, + "name": "test-workspace", + "authorization_domains": [auth_domain], + } + form = self.form_class(self.workspace_to_clone, data=form_data) + self.assertTrue(form.is_valid()) + + def test_invalid_one_required_auth_domains_no_auth_domains_selected(self): + """Form is not valid when no auth domains are selected but workspace to clone has one auth domain.""" + auth_domain = factories.ManagedGroupFactory.create() + self.workspace_to_clone.authorization_domains.add(auth_domain) + billing_project = factories.BillingProjectFactory.create() + form_data = { + "billing_project": billing_project, + "name": "test-workspace", + "authorization_domains": [], + } + form = self.form_class(self.workspace_to_clone, data=form_data) + self.assertFalse(form.is_valid()) + self.assertEqual(len(form.errors), 1) + self.assertIn("authorization_domains", form.errors) + self.assertEqual(len(form.errors["authorization_domains"]), 1) + self.assertIn( + "contain all original workspace authorization domains", + form.errors["authorization_domains"][0], + ) + self.assertIn(auth_domain.name, form.errors["authorization_domains"][0]) + + def test_invalid_one_required_auth_domains_different_auth_domains_selected(self): + """Form is not valid when no auth domains are selected but workspace to clone has one auth domain.""" + auth_domain = factories.ManagedGroupFactory.create() + self.workspace_to_clone.authorization_domains.add(auth_domain) + billing_project = factories.BillingProjectFactory.create() + other_auth_domain = factories.ManagedGroupFactory.create() + form_data = { + "billing_project": billing_project, + "name": "test-workspace", + "authorization_domains": [other_auth_domain], + } + form = self.form_class(self.workspace_to_clone, data=form_data) + self.assertFalse(form.is_valid()) + self.assertEqual(len(form.errors), 1) + self.assertIn("authorization_domains", form.errors) + self.assertEqual(len(form.errors["authorization_domains"]), 1) + self.assertIn( + "contain all original workspace authorization domains", + form.errors["authorization_domains"][0], + ) + self.assertIn(auth_domain.name, form.errors["authorization_domains"][0]) + + def test_valid_one_required_auth_domains_with_extra_selected_auth_domain(self): + """Form is valid with a workspace to clone with one auth domains, and an extra auth domain selected.""" + auth_domain = factories.ManagedGroupFactory.create() + self.workspace_to_clone.authorization_domains.add(auth_domain) + billing_project = factories.BillingProjectFactory.create() + new_auth_domain = factories.ManagedGroupFactory.create() + form_data = { + "billing_project": billing_project, + "name": "test-workspace", + "authorization_domains": [auth_domain, new_auth_domain], + } + form = self.form_class(self.workspace_to_clone, data=form_data) + self.assertTrue(form.is_valid()) + + def test_valid_two_required_auth_domains(self): + """Form is valid with a workspace to clone with two auth domains, and both auth domains selected.""" + auth_domain_1 = factories.ManagedGroupFactory.create() + auth_domain_2 = factories.ManagedGroupFactory.create() + self.workspace_to_clone.authorization_domains.add(auth_domain_1, auth_domain_2) + billing_project = factories.BillingProjectFactory.create() + form_data = { + "billing_project": billing_project, + "name": "test-workspace", + "authorization_domains": [auth_domain_1, auth_domain_2], + } + form = self.form_class(self.workspace_to_clone, data=form_data) + self.assertTrue(form.is_valid()) + + def test_invalid_two_required_auth_domains_no_auth_domains_selected(self): + """Form is not valid when no auth domains are selected but workspace to clone has two auth domain.""" + auth_domain_1 = factories.ManagedGroupFactory.create() + auth_domain_2 = factories.ManagedGroupFactory.create() + self.workspace_to_clone.authorization_domains.add(auth_domain_1, auth_domain_2) + billing_project = factories.BillingProjectFactory.create() + form_data = { + "billing_project": billing_project, + "name": "test-workspace", + "authorization_domains": [], + } + form = self.form_class(self.workspace_to_clone, data=form_data) + self.assertFalse(form.is_valid()) + self.assertEqual(len(form.errors), 1) + self.assertIn("authorization_domains", form.errors) + self.assertEqual(len(form.errors["authorization_domains"]), 1) + self.assertIn( + "contain all original workspace authorization domains", + form.errors["authorization_domains"][0], + ) + self.assertIn(auth_domain_1.name, form.errors["authorization_domains"][0]) + self.assertIn(auth_domain_2.name, form.errors["authorization_domains"][0]) + + def test_invalid_two_required_auth_domains_one_auth_domain_selected(self): + """Form is not valid when no auth domains are selected but workspace to clone has two auth domain.""" + auth_domain_1 = factories.ManagedGroupFactory.create() + auth_domain_2 = factories.ManagedGroupFactory.create() + self.workspace_to_clone.authorization_domains.add(auth_domain_1, auth_domain_2) + billing_project = factories.BillingProjectFactory.create() + form_data = { + "billing_project": billing_project, + "name": "test-workspace", + "authorization_domains": [auth_domain_1], + } + form = self.form_class(self.workspace_to_clone, data=form_data) + self.assertFalse(form.is_valid()) + self.assertEqual(len(form.errors), 1) + self.assertIn("authorization_domains", form.errors) + self.assertEqual(len(form.errors["authorization_domains"]), 1) + self.assertIn( + "contain all original workspace authorization domains", + form.errors["authorization_domains"][0], + ) + self.assertIn(auth_domain_1.name, form.errors["authorization_domains"][0]) + self.assertIn(auth_domain_2.name, form.errors["authorization_domains"][0]) + + def test_invalid_two_required_auth_domains_different_auth_domains_selected(self): + """Form is not valid when different auth domains are selected but workspace to clone has two auth domains.""" + auth_domain_1 = factories.ManagedGroupFactory.create() + auth_domain_2 = factories.ManagedGroupFactory.create() + self.workspace_to_clone.authorization_domains.add(auth_domain_1, auth_domain_2) + billing_project = factories.BillingProjectFactory.create() + other_auth_domain_1 = factories.ManagedGroupFactory.create() + other_auth_domain_2 = factories.ManagedGroupFactory.create() + form_data = { + "billing_project": billing_project, + "name": "test-workspace", + "authorization_domains": [other_auth_domain_1, other_auth_domain_2], + } + form = self.form_class(self.workspace_to_clone, data=form_data) + self.assertFalse(form.is_valid()) + self.assertEqual(len(form.errors), 1) + self.assertIn("authorization_domains", form.errors) + self.assertEqual(len(form.errors["authorization_domains"]), 1) + self.assertIn( + "contain all original workspace authorization domains", + form.errors["authorization_domains"][0], + ) + self.assertIn(auth_domain_1.name, form.errors["authorization_domains"][0]) + self.assertIn(auth_domain_2.name, form.errors["authorization_domains"][0]) + + def test_valid_two_required_auth_domains_with_extra_selected_auth_domain(self): + """Form is valid with a workspace to clone with one auth domains, and an extra auth domain selected.""" + auth_domain_1 = factories.ManagedGroupFactory.create() + auth_domain_2 = factories.ManagedGroupFactory.create() + self.workspace_to_clone.authorization_domains.add(auth_domain_1, auth_domain_2) + billing_project = factories.BillingProjectFactory.create() + new_auth_domain = factories.ManagedGroupFactory.create() + form_data = { + "billing_project": billing_project, + "name": "test-workspace", + "authorization_domains": [auth_domain_1, auth_domain_2, new_auth_domain], + } + form = self.form_class(self.workspace_to_clone, data=form_data) + self.assertTrue(form.is_valid()) + + class GroupGroupMembershipFormTest(TestCase): form_class = forms.GroupGroupMembershipForm diff --git a/anvil_consortium_manager/tests/test_models_anvil_api_unit.py b/anvil_consortium_manager/tests/test_models_anvil_api_unit.py index e13646f6..8d1c1c11 100644 --- a/anvil_consortium_manager/tests/test_models_anvil_api_unit.py +++ b/anvil_consortium_manager/tests/test_models_anvil_api_unit.py @@ -2436,7 +2436,7 @@ def test_can_clone_workspace_add_one_auth_domain(self): self.workspace.anvil_clone( billing_project, "test-workspace", - additional_authorization_domains=[auth_domain], + authorization_domains=[auth_domain], ) def test_can_clone_workspace_add_two_auth_domains(self): @@ -2467,7 +2467,7 @@ def test_can_clone_workspace_add_two_auth_domains(self): self.workspace.anvil_clone( billing_project, "test-workspace", - additional_authorization_domains=[auth_domain_1, auth_domain_2], + authorization_domains=[auth_domain_1, auth_domain_2], ) def test_can_clone_workspace_one_auth_domain_add_one_auth_domain(self): @@ -2499,7 +2499,7 @@ def test_can_clone_workspace_one_auth_domain_add_one_auth_domain(self): self.workspace.anvil_clone( billing_project, "test-workspace", - additional_authorization_domains=[new_auth_domain], + authorization_domains=[new_auth_domain], ) def test_can_clone_workspace_one_auth_domain_add_two_auth_domains(self): @@ -2533,7 +2533,7 @@ def test_can_clone_workspace_one_auth_domain_add_two_auth_domains(self): self.workspace.anvil_clone( billing_project, "test-workspace", - additional_authorization_domains=[new_auth_domain_1, new_auth_domain_2], + authorization_domains=[new_auth_domain_1, new_auth_domain_2], ) def test_can_clone_workspace_one_auth_domain_add_same_auth_domain(self): @@ -2561,7 +2561,7 @@ def test_can_clone_workspace_one_auth_domain_add_same_auth_domain(self): self.workspace.anvil_clone( billing_project, "test-workspace", - additional_authorization_domains=[auth_domain], + authorization_domains=[auth_domain], ) def test_error_cloning_workspace_into_billing_project_where_app_is_not_user(self): @@ -2575,20 +2575,6 @@ def test_error_cloning_workspace_into_billing_project_where_app_is_not_user(self self.assertEqual(models.Workspace.objects.count(), 1) self.assertIn(self.workspace, models.Workspace.objects.all()) - def test_error_workspace_already_exists_in_app(self): - """Error when the workspace to be cloned already exists in the app.""" - existing_workspace = factories.WorkspaceFactory.create() - # No API call exected. - with self.assertRaises(ValueError) as e: - self.workspace.anvil_clone( - existing_workspace.billing_project, existing_workspace.name - ) - self.assertIn("already exists", str(e.exception)) - # No new workspace was created. - self.assertEqual(models.Workspace.objects.count(), 2) - self.assertIn(self.workspace, models.Workspace.objects.all()) - self.assertIn(existing_workspace, models.Workspace.objects.all()) - def test_error_workspace_already_exists_in_anvil_but_not_in_app(self): """Error when the workspace to be cloned already exists in AnVIL but is not in the app.""" billing_project = factories.BillingProjectFactory.create() @@ -2666,7 +2652,7 @@ def test_error_authorization_domain_exists_in_app_but_not_on_anvil(self): self.workspace.anvil_clone( billing_project, "test-workspace", - additional_authorization_domains=[new_auth_domain], + authorization_domains=[new_auth_domain], ) # No new workspace was created. self.assertEqual(models.Workspace.objects.count(), 1) diff --git a/anvil_consortium_manager/tests/test_views.py b/anvil_consortium_manager/tests/test_views.py index 2c572cab..edc7c996 100644 --- a/anvil_consortium_manager/tests/test_views.py +++ b/anvil_consortium_manager/tests/test_views.py @@ -8073,6 +8073,1101 @@ def test_adapter_does_not_create_objects_if_workspace_data_form_invalid(self): self.assertEqual(len(responses.calls), 2) +class WorkspaceCloneTest(AnVILAPIMockTestMixin, TestCase): + """Tests for the WorkspaceClone view.""" + + api_success_code = 201 + + def setUp(self): + """Set up test class.""" + # The superclass uses the responses package to mock API responses. + super().setUp() + self.factory = RequestFactory() + # Create a user with both view and edit permissions. + self.user = User.objects.create_user(username="test", password="test") + self.user.user_permissions.add( + Permission.objects.get( + codename=models.AnVILProjectManagerAccess.VIEW_PERMISSION_CODENAME + ) + ) + self.user.user_permissions.add( + Permission.objects.get( + codename=models.AnVILProjectManagerAccess.EDIT_PERMISSION_CODENAME + ) + ) + self.workspace_to_clone = factories.WorkspaceFactory.create() + self.api_url = self.entry_point + "/api/workspaces/{}/{}/clone".format( + self.workspace_to_clone.billing_project.name, + self.workspace_to_clone.name, + ) + self.workspace_type = DefaultWorkspaceAdapter.type + + def tearDown(self): + """Clean up after tests.""" + # Unregister all adapters. + workspace_adapter_registry._registry = {} + # Register the default adapter. + workspace_adapter_registry.register(DefaultWorkspaceAdapter) + super().tearDown() + + def get_url(self, *args): + """Get the url for the view being tested.""" + return reverse("anvil_consortium_manager:workspaces:clone", args=args) + + def get_view(self): + """Return the view being tested.""" + return views.WorkspaceCreate.as_view() + + def test_view_redirect_not_logged_in(self): + "View redirects to login view when user is not logged in." + response = self.client.get( + self.get_url( + self.workspace_to_clone.billing_project.name, + self.workspace_to_clone.name, + self.workspace_type, + ) + ) + self.assertRedirects( + response, + resolve_url(settings.LOGIN_URL) + + "?next=" + + self.get_url( + self.workspace_to_clone.billing_project.name, + self.workspace_to_clone.name, + self.workspace_type, + ), + ) + + def test_status_code_with_user_permission(self): + """Returns successful response code.""" + self.client.force_login(self.user) + response = self.client.get( + self.get_url( + self.workspace_to_clone.billing_project.name, + self.workspace_to_clone.name, + self.workspace_type, + ) + ) + self.assertEqual(response.status_code, 200) + + def test_access_with_view_permission(self): + """Raises permission denied if user has only view permission.""" + user_with_view_perm = User.objects.create_user( + username="test-other", password="test-other" + ) + user_with_view_perm.user_permissions.add( + Permission.objects.get( + codename=models.AnVILProjectManagerAccess.VIEW_PERMISSION_CODENAME + ) + ) + request = self.factory.get( + self.get_url( + self.workspace_to_clone.billing_project.name, + self.workspace_to_clone.name, + self.workspace_type, + ) + ) + request.user = user_with_view_perm + with self.assertRaises(PermissionDenied): + self.get_view()( + request, + billing_project_slug=self.workspace_to_clone.billing_project.name, + workspace_slug=self.workspace_to_clone.name, + workspace_type=self.workspace_type, + ) + + def test_access_without_user_permission(self): + """Raises permission denied if user has no permissions.""" + user_no_perms = User.objects.create_user( + username="test-none", password="test-none" + ) + request = self.factory.get( + self.get_url( + self.workspace_to_clone.billing_project.name, + self.workspace_to_clone.name, + self.workspace_type, + ) + ) + request.user = user_no_perms + with self.assertRaises(PermissionDenied): + self.get_view()( + request, + billing_project_slug=self.workspace_to_clone.billing_project.name, + workspace_slug=self.workspace_to_clone.name, + workspace_type=self.workspace_type, + ) + + def test_get_workspace_type_not_registered(self): + """Raises 404 with get request if workspace type is not registered with adapter.""" + request = self.factory.get( + self.get_url( + self.workspace_to_clone.billing_project.name, + self.workspace_to_clone.name, + "foo", + ) + ) + request.user = self.user + with self.assertRaises(Http404): + self.get_view()( + request, + billing_project_slug=self.workspace_to_clone.billing_project.name, + workspace_slug=self.workspace_to_clone.name, + workspace_type="foo", + ) + + def test_post_workspace_type_not_registered(self): + """Raises 404 with post request if workspace type is not registered with adapter.""" + request = self.factory.post( + self.get_url( + self.workspace_to_clone.billing_project.name, + self.workspace_to_clone.name, + "foo", + ), + {}, + ) + request.user = self.user + with self.assertRaises(Http404): + self.get_view()( + request, + billing_project_slug=self.workspace_to_clone.billing_project.name, + workspace_slug=self.workspace_to_clone.name, + workspace_type="foo", + ) + + def test_has_form_in_context(self): + """Response includes a form.""" + self.client.force_login(self.user) + response = self.client.get( + self.get_url( + self.workspace_to_clone.billing_project.name, + self.workspace_to_clone.name, + self.workspace_type, + ) + ) + self.assertTrue("form" in response.context_data) + self.assertIsInstance(response.context_data["form"], forms.WorkspaceCloneForm) + + def test_has_formset_in_context(self): + """Response includes a formset for the workspace_data model.""" + self.client.force_login(self.user) + response = self.client.get( + self.get_url( + self.workspace_to_clone.billing_project.name, + self.workspace_to_clone.name, + self.workspace_type, + ) + ) + self.assertTrue("workspace_data_formset" in response.context_data) + formset = response.context_data["workspace_data_formset"] + self.assertIsInstance(formset, BaseInlineFormSet) + self.assertEqual(len(formset.forms), 1) + self.assertIsInstance(formset.forms[0], forms.DefaultWorkspaceDataForm) + + def test_can_create_an_object(self): + """Posting valid data to the form creates an object.""" + billing_project = factories.BillingProjectFactory.create( + name="test-billing-project" + ) + json_data = { + "namespace": "test-billing-project", + "name": "test-workspace", + "attributes": {}, + } + responses.add( + responses.POST, + self.api_url, + status=self.api_success_code, + match=[responses.matchers.json_params_matcher(json_data)], + ) + self.client.force_login(self.user) + response = self.client.post( + self.get_url( + self.workspace_to_clone.billing_project.name, + self.workspace_to_clone.name, + self.workspace_type, + ), + { + "billing_project": billing_project.pk, + "name": "test-workspace", + # Default workspace data for formset. + "workspacedata-TOTAL_FORMS": 1, + "workspacedata-INITIAL_FORMS": 0, + "workspacedata-MIN_NUM_FORMS": 1, + "workspacedata-MAX_NUM_FORMS": 1, + }, + ) + self.assertEqual(response.status_code, 302) + new_object = models.Workspace.objects.latest("pk") + self.assertIsInstance(new_object, models.Workspace) + self.assertEqual( + new_object.workspace_type, + DefaultWorkspaceAdapter().get_type(), + ) + responses.assert_call_count(self.api_url, 1) + # History is added. + self.assertEqual(new_object.history.count(), 1) + self.assertEqual(new_object.history.latest().history_type, "+") + + def test_can_create_object_with_auth_domains(self): + """Posting valid data to the form creates an object.""" + auth_domain = factories.ManagedGroupFactory.create() + self.workspace_to_clone.authorization_domains.add(auth_domain) + billing_project = factories.BillingProjectFactory.create( + name="test-billing-project" + ) + json_data = { + "namespace": "test-billing-project", + "name": "test-workspace", + "attributes": {}, + "authorizationDomain": [{"membersGroupName": auth_domain.name}], + } + responses.add( + responses.POST, + self.api_url, + status=self.api_success_code, + match=[responses.matchers.json_params_matcher(json_data)], + ) + self.client.force_login(self.user) + response = self.client.post( + self.get_url( + self.workspace_to_clone.billing_project.name, + self.workspace_to_clone.name, + self.workspace_type, + ), + { + "billing_project": billing_project.pk, + "name": "test-workspace", + "authorization_domains": [auth_domain.pk], + # Default workspace data for formset. + "workspacedata-TOTAL_FORMS": 1, + "workspacedata-INITIAL_FORMS": 0, + "workspacedata-MIN_NUM_FORMS": 1, + "workspacedata-MAX_NUM_FORMS": 1, + }, + ) + self.assertEqual(response.status_code, 302) + new_object = models.Workspace.objects.latest("pk") + self.assertIsInstance(new_object, models.Workspace) + # Has an auth domain. + self.assertEqual(new_object.authorization_domains.count(), 1) + self.assertIn(auth_domain, new_object.authorization_domains.all()) + responses.assert_call_count(self.api_url, 1) + # History is added. + self.assertEqual(new_object.history.count(), 1) + self.assertEqual(new_object.history.latest().history_type, "+") + + def test_can_add_an_auth_domains(self): + """Posting valid data to the form creates an object.""" + auth_domain = factories.ManagedGroupFactory.create() + self.workspace_to_clone.authorization_domains.add(auth_domain) + new_auth_domain = factories.ManagedGroupFactory.create() + billing_project = factories.BillingProjectFactory.create( + name="test-billing-project" + ) + json_data = { + "namespace": "test-billing-project", + "name": "test-workspace", + "attributes": {}, + "authorizationDomain": [ + {"membersGroupName": auth_domain.name}, + {"membersGroupName": new_auth_domain.name}, + ], + } + responses.add( + responses.POST, + self.api_url, + status=self.api_success_code, + match=[responses.matchers.json_params_matcher(json_data)], + ) + self.client.force_login(self.user) + response = self.client.post( + self.get_url( + self.workspace_to_clone.billing_project.name, + self.workspace_to_clone.name, + self.workspace_type, + ), + { + "billing_project": billing_project.pk, + "name": "test-workspace", + "authorization_domains": [auth_domain.pk, new_auth_domain.pk], + # Default workspace data for formset. + "workspacedata-TOTAL_FORMS": 1, + "workspacedata-INITIAL_FORMS": 0, + "workspacedata-MIN_NUM_FORMS": 1, + "workspacedata-MAX_NUM_FORMS": 1, + }, + ) + self.assertEqual(response.status_code, 302) + new_object = models.Workspace.objects.latest("pk") + self.assertIsInstance(new_object, models.Workspace) + # Has an auth domain. + self.assertEqual(new_object.authorization_domains.count(), 2) + self.assertIn(auth_domain, new_object.authorization_domains.all()) + self.assertIn(new_auth_domain, new_object.authorization_domains.all()) + responses.assert_call_count(self.api_url, 1) + + def test_creates_default_workspace_data(self): + """Posting valid data to the form creates the default workspace data object.""" + billing_project = factories.BillingProjectFactory.create( + name="test-billing-project" + ) + json_data = { + "namespace": "test-billing-project", + "name": "test-workspace", + "attributes": {}, + } + responses.add( + responses.POST, + self.api_url, + status=self.api_success_code, + match=[responses.matchers.json_params_matcher(json_data)], + ) + self.client.force_login(self.user) + response = self.client.post( + self.get_url( + self.workspace_to_clone.billing_project.name, + self.workspace_to_clone.name, + self.workspace_type, + ), + { + "billing_project": billing_project.pk, + "name": "test-workspace", + # Default workspace data for formset. + "workspacedata-TOTAL_FORMS": 1, + "workspacedata-INITIAL_FORMS": 0, + "workspacedata-MIN_NUM_FORMS": 1, + "workspacedata-MAX_NUM_FORMS": 1, + }, + ) + self.assertEqual(response.status_code, 302) + new_workspace = models.Workspace.objects.latest("pk") + # Also creates a workspace data object. + self.assertEqual(models.DefaultWorkspaceData.objects.count(), 1) + self.assertIsInstance( + new_workspace.defaultworkspacedata, models.DefaultWorkspaceData + ) + + def test_success_message(self): + """Response includes a success message if successful.""" + billing_project = factories.BillingProjectFactory.create() + json_data = { + "namespace": billing_project.name, + "name": "test-workspace", + "attributes": {}, + } + responses.add( + responses.POST, + self.api_url, + status=self.api_success_code, + match=[responses.matchers.json_params_matcher(json_data)], + ) + self.client.force_login(self.user) + response = self.client.post( + self.get_url( + self.workspace_to_clone.billing_project.name, + self.workspace_to_clone.name, + self.workspace_type, + ), + { + "billing_project": billing_project.pk, + "name": "test-workspace", + # Default workspace data for formset. + "workspacedata-TOTAL_FORMS": 1, + "workspacedata-INITIAL_FORMS": 0, + "workspacedata-MIN_NUM_FORMS": 1, + "workspacedata-MAX_NUM_FORMS": 1, + }, + follow=True, + ) + self.assertIn("messages", response.context) + messages = list(response.context["messages"]) + self.assertEqual(len(messages), 1) + self.assertEqual(views.WorkspaceCreate.success_msg, str(messages[0])) + + def test_redirects_to_new_object_detail(self): + """After successfully creating an object, view redirects to the object's detail page.""" + # This needs to use the client because the RequestFactory doesn't handle redirects. + billing_project = factories.BillingProjectFactory.create() + json_data = { + "namespace": billing_project.name, + "name": "test-workspace", + "attributes": {}, + } + responses.add( + responses.POST, + self.api_url, + status=self.api_success_code, + match=[responses.matchers.json_params_matcher(json_data)], + ) + self.client.force_login(self.user) + response = self.client.post( + self.get_url( + self.workspace_to_clone.billing_project.name, + self.workspace_to_clone.name, + self.workspace_type, + ), + { + "billing_project": billing_project.pk, + "name": "test-workspace", + # Default workspace data for formset. + "workspacedata-TOTAL_FORMS": 1, + "workspacedata-INITIAL_FORMS": 0, + "workspacedata-MIN_NUM_FORMS": 1, + "workspacedata-MAX_NUM_FORMS": 1, + }, + ) + new_object = models.Workspace.objects.latest("pk") + self.assertRedirects(response, new_object.get_absolute_url()) + responses.assert_call_count(self.api_url, 1) + + def test_cannot_create_duplicate_object(self): + """Cannot create two workspaces with the same billing project and name.""" + obj = factories.WorkspaceFactory.create() + self.client.force_login(self.user) + response = self.client.post( + self.get_url( + self.workspace_to_clone.billing_project.name, + self.workspace_to_clone.name, + self.workspace_type, + ), + { + "billing_project": obj.billing_project.pk, + "name": obj.name, + # Default workspace data for formset. + "workspacedata-TOTAL_FORMS": 1, + "workspacedata-INITIAL_FORMS": 0, + "workspacedata-MIN_NUM_FORMS": 1, + "workspacedata-MAX_NUM_FORMS": 1, + }, + ) + self.assertEqual(response.status_code, 200) + form = response.context_data["form"] + self.assertFalse(form.is_valid()) + self.assertIn("already exists", form.non_field_errors()[0]) + self.assertEqual(models.Workspace.objects.count(), 2) + self.assertIn(self.workspace_to_clone, models.Workspace.objects.all()) + self.assertIn(obj, models.Workspace.objects.all()) + + def test_can_create_workspace_with_same_billing_project_different_name(self): + """Can create a workspace with a different name in the same billing project.""" + billing_project = factories.BillingProjectFactory.create() + factories.WorkspaceFactory.create( + billing_project=billing_project, name="test-name-1" + ) + json_data = { + "namespace": billing_project.name, + "name": "test-name-2", + "attributes": {}, + } + responses.add( + responses.POST, + self.api_url, + status=self.api_success_code, + match=[responses.matchers.json_params_matcher(json_data)], + ) + self.client.force_login(self.user) + response = self.client.post( + self.get_url( + self.workspace_to_clone.billing_project.name, + self.workspace_to_clone.name, + self.workspace_type, + ), + { + "billing_project": billing_project.pk, + "name": "test-name-2", + # Default workspace data for formset. + "workspacedata-TOTAL_FORMS": 1, + "workspacedata-INITIAL_FORMS": 0, + "workspacedata-MIN_NUM_FORMS": 1, + "workspacedata-MAX_NUM_FORMS": 1, + }, + ) + self.assertEqual(response.status_code, 302) + self.assertEqual(models.Workspace.objects.count(), 3) + # Make sure you can get the new object. + models.Workspace.objects.get( + billing_project=billing_project, name="test-name-2" + ) + responses.assert_call_count(self.api_url, 1) + + def test_can_create_workspace_with_same_name_different_billing_project(self): + """Can create a workspace with the same name in a different billing project.""" + billing_project_1 = factories.BillingProjectFactory.create(name="project-1") + billing_project_2 = factories.BillingProjectFactory.create(name="project-2") + workspace_name = "test-name" + factories.WorkspaceFactory.create( + billing_project=billing_project_1, name=workspace_name + ) + json_data = { + "namespace": billing_project_2.name, + "name": "test-name", + "attributes": {}, + } + responses.add( + responses.POST, + self.api_url, + status=self.api_success_code, + match=[responses.matchers.json_params_matcher(json_data)], + ) + self.client.force_login(self.user) + response = self.client.post( + self.get_url( + self.workspace_to_clone.billing_project.name, + self.workspace_to_clone.name, + self.workspace_type, + ), + { + "billing_project": billing_project_2.pk, + "name": workspace_name, + # Default workspace data for formset. + "workspacedata-TOTAL_FORMS": 1, + "workspacedata-INITIAL_FORMS": 0, + "workspacedata-MIN_NUM_FORMS": 1, + "workspacedata-MAX_NUM_FORMS": 1, + }, + ) + self.assertEqual(response.status_code, 302) + self.assertEqual(models.Workspace.objects.count(), 3) + # Make sure you can get the new object. + models.Workspace.objects.get( + billing_project=billing_project_2, name=workspace_name + ) + responses.assert_call_count(self.api_url, 1) + + def test_invalid_input_name(self): + """Posting invalid data to name field does not create an object.""" + billing_project = factories.BillingProjectFactory.create() + self.client.force_login(self.user) + response = self.client.post( + self.get_url( + self.workspace_to_clone.billing_project.name, + self.workspace_to_clone.name, + self.workspace_type, + ), + { + "billing_project": billing_project.pk, + "name": "invalid name", + # Default workspace data for formset. + "workspacedata-TOTAL_FORMS": 1, + "workspacedata-INITIAL_FORMS": 0, + "workspacedata-MIN_NUM_FORMS": 1, + "workspacedata-MAX_NUM_FORMS": 1, + }, + ) + self.assertEqual(response.status_code, 200) + form = response.context_data["form"] + self.assertFalse(form.is_valid()) + self.assertIn("name", form.errors.keys()) + self.assertIn("slug", form.errors["name"][0]) + self.assertEqual(models.Workspace.objects.count(), 1) + self.assertIn(self.workspace_to_clone, models.Workspace.objects.all()) + self.assertEqual(len(responses.calls), 0) + + def test_invalid_input_billing_project(self): + """Posting invalid data to billing_project field does not create an object.""" + self.client.force_login(self.user) + response = self.client.post( + self.get_url( + self.workspace_to_clone.billing_project.name, + self.workspace_to_clone.name, + self.workspace_type, + ), + { + "billing_project": 100, + "name": "test-name", + # Default workspace data for formset. + "workspacedata-TOTAL_FORMS": 1, + "workspacedata-INITIAL_FORMS": 0, + "workspacedata-MIN_NUM_FORMS": 1, + "workspacedata-MAX_NUM_FORMS": 1, + }, + ) + self.assertEqual(response.status_code, 200) + form = response.context_data["form"] + self.assertFalse(form.is_valid()) + self.assertIn("billing_project", form.errors.keys()) + self.assertIn("valid choice", form.errors["billing_project"][0]) + self.assertEqual(models.Workspace.objects.count(), 1) + self.assertIn(self.workspace_to_clone, models.Workspace.objects.all()) + self.assertEqual(len(responses.calls), 0) + + def test_post_invalid_name_billing_project(self): + """Posting blank data does not create an object.""" + self.client.force_login(self.user) + response = self.client.post( + self.get_url( + self.workspace_to_clone.billing_project.name, + self.workspace_to_clone.name, + self.workspace_type, + ), + {}, + ) + self.assertEqual(response.status_code, 200) + form = response.context_data["form"] + self.assertFalse(form.is_valid()) + self.assertIn("billing_project", form.errors.keys()) + self.assertIn("required", form.errors["billing_project"][0]) + self.assertIn("name", form.errors.keys()) + self.assertIn("required", form.errors["name"][0]) + self.assertEqual(models.Workspace.objects.count(), 1) + self.assertIn(self.workspace_to_clone, models.Workspace.objects.all()) + self.assertEqual(len(responses.calls), 0) + + def test_post_blank_data(self): + """Posting blank data does not create an object.""" + self.client.force_login(self.user) + response = self.client.post( + self.get_url( + self.workspace_to_clone.billing_project.name, + self.workspace_to_clone.name, + self.workspace_type, + ), + {}, + ) + self.assertEqual(response.status_code, 200) + form = response.context_data["form"] + self.assertFalse(form.is_valid()) + self.assertIn("billing_project", form.errors.keys()) + self.assertIn("required", form.errors["billing_project"][0]) + self.assertIn("name", form.errors.keys()) + self.assertIn("required", form.errors["name"][0]) + self.assertEqual(models.Workspace.objects.count(), 1) + self.assertIn(self.workspace_to_clone, models.Workspace.objects.all()) + self.assertEqual(len(responses.calls), 0) + + def test_api_error_message(self): + """Shows a method if an AnVIL API error occurs.""" + billing_project = factories.BillingProjectFactory.create() + json_data = { + "namespace": billing_project.name, + "name": "test-workspace", + "attributes": {}, + } + responses.add( + responses.POST, + self.api_url, + status=500, + match=[responses.matchers.json_params_matcher(json_data)], + json={"message": "workspace create test error"}, + ) + # Need a client to check messages. + self.client.force_login(self.user) + response = self.client.post( + self.get_url( + self.workspace_to_clone.billing_project.name, + self.workspace_to_clone.name, + self.workspace_type, + ), + { + "billing_project": billing_project.pk, + "name": "test-workspace", + # Default workspace data for formset. + "workspacedata-TOTAL_FORMS": 1, + "workspacedata-INITIAL_FORMS": 0, + "workspacedata-MIN_NUM_FORMS": 1, + "workspacedata-MAX_NUM_FORMS": 1, + }, + ) + self.assertEqual(response.status_code, 200) + self.assertIn("form", response.context) + self.assertIn("messages", response.context) + messages = list(response.context["messages"]) + self.assertEqual(len(messages), 1) + self.assertIn("AnVIL API Error: workspace create test error", str(messages[0])) + responses.assert_call_count(self.api_url, 1) + # Make sure that no object is created. + self.assertEqual(models.Workspace.objects.count(), 1) + self.assertIn(self.workspace_to_clone, models.Workspace.objects.all()) + + def test_invalid_auth_domain(self): + """Does not create a workspace when an invalid authorization domain is specified.""" + billing_project = factories.BillingProjectFactory.create( + name="test-billing-project" + ) + self.client.force_login(self.user) + response = self.client.post( + self.get_url( + self.workspace_to_clone.billing_project.name, + self.workspace_to_clone.name, + self.workspace_type, + ), + { + "billing_project": billing_project.pk, + "name": "test-workspace", + "authorization_domains": [1], + # Default workspace data for formset. + "workspacedata-TOTAL_FORMS": 1, + "workspacedata-INITIAL_FORMS": 0, + "workspacedata-MIN_NUM_FORMS": 1, + "workspacedata-MAX_NUM_FORMS": 1, + }, + ) + self.assertEqual(response.status_code, 200) + self.assertIn("form", response.context_data) + form = response.context_data["form"] + self.assertFalse(form.is_valid()) + self.assertIn("authorization_domains", form.errors.keys()) + self.assertIn("valid choice", form.errors["authorization_domains"][0]) + # No object was created. + self.assertEqual(models.Workspace.objects.count(), 1) + self.assertIn(self.workspace_to_clone, models.Workspace.objects.all()) + # No API calls made. + responses.assert_call_count(self.api_url, 0) + + def test_one_valid_one_invalid_auth_domain(self): + billing_project = factories.BillingProjectFactory.create( + name="test-billing-project" + ) + auth_domain = factories.ManagedGroupFactory.create() + self.client.force_login(self.user) + response = self.client.post( + self.get_url( + self.workspace_to_clone.billing_project.name, + self.workspace_to_clone.name, + self.workspace_type, + ), + { + "billing_project": billing_project.pk, + "name": "test-workspace", + "authorization_domains": [auth_domain.pk, auth_domain.pk + 1], + # Default workspace data for formset. + "workspacedata-TOTAL_FORMS": 1, + "workspacedata-INITIAL_FORMS": 0, + "workspacedata-MIN_NUM_FORMS": 1, + "workspacedata-MAX_NUM_FORMS": 1, + }, + ) + self.assertEqual(response.status_code, 200) + self.assertIn("form", response.context_data) + form = response.context_data["form"] + self.assertFalse(form.is_valid()) + self.assertIn("authorization_domains", form.errors.keys()) + self.assertIn("valid choice", form.errors["authorization_domains"][0]) + # No object was created. + self.assertEqual(len(models.Workspace.objects.all()), 1) + self.assertIn(self.workspace_to_clone, models.Workspace.objects.all()) + # No API calls made. + responses.assert_call_count(self.api_url, 0) + + def test_auth_domain_does_not_exist_on_anvil(self): + """No workspace is displayed if the auth domain group doesn't exist on AnVIL.""" + billing_project = factories.BillingProjectFactory.create( + name="test-billing-project" + ) + auth_domain = factories.ManagedGroupFactory.create() + json_data = { + "namespace": "test-billing-project", + "name": "test-workspace", + "attributes": {}, + "authorizationDomain": [ + {"membersGroupName": auth_domain.name}, + ], + } + responses.add( + responses.POST, + self.api_url, + status=400, + json={"message": "api error"}, + match=[responses.matchers.json_params_matcher(json_data)], + ) + # Need a client to check messages. + self.client.force_login(self.user) + response = self.client.post( + self.get_url( + self.workspace_to_clone.billing_project.name, + self.workspace_to_clone.name, + self.workspace_type, + ), + { + "billing_project": billing_project.pk, + "name": "test-workspace", + "authorization_domains": [auth_domain.pk], + # Default workspace data for formset. + "workspacedata-TOTAL_FORMS": 1, + "workspacedata-INITIAL_FORMS": 0, + "workspacedata-MIN_NUM_FORMS": 1, + "workspacedata-MAX_NUM_FORMS": 1, + }, + ) + self.assertEqual(response.status_code, 200) + # The form is valid but there was an API error. + form = response.context_data["form"] + self.assertTrue(form.is_valid()) + # Check messages. + self.assertIn("messages", response.context) + messages = list(response.context["messages"]) + self.assertEqual(len(messages), 1) + self.assertEqual("AnVIL API Error: api error", str(messages[0])) + # Did not create any new Workspaces. + self.assertEqual(models.Workspace.objects.count(), 1) + self.assertIn(self.workspace_to_clone, models.Workspace.objects.all()) + responses.assert_call_count(self.api_url, 1) + + def test_not_admin_of_auth_domain_on_anvil(self): + """No workspace is displayed if we are not the admins of the auth domain on AnVIL.""" + billing_project = factories.BillingProjectFactory.create( + name="test-billing-project" + ) + auth_domain = factories.ManagedGroupFactory.create() + json_data = { + "namespace": "test-billing-project", + "name": "test-workspace", + "attributes": {}, + "authorizationDomain": [ + {"membersGroupName": auth_domain.name}, + ], + } + responses.add( + responses.POST, + self.api_url, + status=400, + json={"message": "api error"}, + match=[responses.matchers.json_params_matcher(json_data)], + ) + # Need a client to check messages. + self.client.force_login(self.user) + response = self.client.post( + self.get_url( + self.workspace_to_clone.billing_project.name, + self.workspace_to_clone.name, + self.workspace_type, + ), + { + "billing_project": billing_project.pk, + "name": "test-workspace", + "authorization_domains": [auth_domain.pk], + # Default workspace data for formset. + "workspacedata-TOTAL_FORMS": 1, + "workspacedata-INITIAL_FORMS": 0, + "workspacedata-MIN_NUM_FORMS": 1, + "workspacedata-MAX_NUM_FORMS": 1, + }, + ) + self.assertEqual(response.status_code, 200) + # The form is valid but there was an API error. + form = response.context_data["form"] + self.assertTrue(form.is_valid()) + # Check messages. + self.assertIn("messages", response.context) + messages = list(response.context["messages"]) + self.assertEqual(len(messages), 1) + self.assertEqual("AnVIL API Error: api error", str(messages[0])) + # Did not create any new Workspaces. + self.assertEqual(models.Workspace.objects.count(), 1) + self.assertIn(self.workspace_to_clone, models.Workspace.objects.all()) + responses.assert_call_count(self.api_url, 1) + + def test_not_user_of_billing_project(self): + """Posting a billing project where we are not users does not create an object.""" + billing_project = factories.BillingProjectFactory.create( + name="test-billing-project", has_app_as_user=False + ) + self.client.force_login(self.user) + response = self.client.post( + self.get_url( + self.workspace_to_clone.billing_project.name, + self.workspace_to_clone.name, + self.workspace_type, + ), + { + "billing_project": billing_project.pk, + "name": "test-workspace", + # Default workspace data for formset. + "workspacedata-TOTAL_FORMS": 1, + "workspacedata-INITIAL_FORMS": 0, + "workspacedata-MIN_NUM_FORMS": 1, + "workspacedata-MAX_NUM_FORMS": 1, + }, + ) + self.assertEqual(response.status_code, 200) + self.assertIn("form", response.context_data) + form = response.context_data["form"] + self.assertFalse(form.is_valid()) + self.assertIn("billing_project", form.errors.keys()) + self.assertIn("valid choice", form.errors["billing_project"][0]) + # No workspace was created. + self.assertEqual(models.Workspace.objects.count(), 1) + self.assertIn(self.workspace_to_clone, models.Workspace.objects.all()) + + def test_adapter_includes_workspace_data_formset(self): + """Response includes the workspace data formset if specified.""" + # Overriding settings doesn't work, because appconfig.ready has already run and + # registered the default adapter. Instead, unregister the default and register the + # new adapter here. + workspace_adapter_registry.unregister(DefaultWorkspaceAdapter) + workspace_adapter_registry.register(TestWorkspaceAdapter) + self.workspace_type = "test" + self.client.force_login(self.user) + response = self.client.get( + self.get_url( + self.workspace_to_clone.billing_project.name, + self.workspace_to_clone.name, + self.workspace_type, + ) + ) + self.assertTrue("workspace_data_formset" in response.context_data) + formset = response.context_data["workspace_data_formset"] + self.assertIsInstance(formset, BaseInlineFormSet) + self.assertEqual(len(formset.forms), 1) + self.assertIsInstance(formset.forms[0], app_forms.TestWorkspaceDataForm) + + def test_adapter_creates_workspace_data(self): + """Posting valid data to the form creates a workspace data object when using a custom adapter.""" + # Overriding settings doesn't work, because appconfig.ready has already run and + # registered the default adapter. Instead, unregister the default and register the + # new adapter here. + workspace_adapter_registry.unregister(DefaultWorkspaceAdapter) + workspace_adapter_registry.register(TestWorkspaceAdapter) + self.workspace_type = "test" + billing_project = factories.BillingProjectFactory.create( + name="test-billing-project" + ) + json_data = { + "namespace": "test-billing-project", + "name": "test-workspace", + "attributes": {}, + } + responses.add( + responses.POST, + self.api_url, + status=self.api_success_code, + match=[responses.matchers.json_params_matcher(json_data)], + ) + self.client.force_login(self.user) + response = self.client.post( + self.get_url( + self.workspace_to_clone.billing_project.name, + self.workspace_to_clone.name, + self.workspace_type, + ), + { + "billing_project": billing_project.pk, + "name": "test-workspace", + # Default workspace data for formset. + "workspacedata-TOTAL_FORMS": 1, + "workspacedata-INITIAL_FORMS": 0, + "workspacedata-MIN_NUM_FORMS": 1, + "workspacedata-MAX_NUM_FORMS": 1, + "workspacedata-0-study_name": "test study", + }, + ) + self.assertEqual(response.status_code, 302) + # The workspace is created. + new_workspace = models.Workspace.objects.latest("pk") + # workspace_type is set properly. + self.assertEqual( + new_workspace.workspace_type, + TestWorkspaceAdapter().get_type(), + ) + # Workspace data is added. + self.assertEqual(app_models.TestWorkspaceData.objects.count(), 1) + new_workspace_data = app_models.TestWorkspaceData.objects.latest("pk") + self.assertEqual(new_workspace_data.workspace, new_workspace) + self.assertEqual(new_workspace_data.study_name, "test study") + responses.assert_call_count(self.api_url, 1) + + def test_adapter_does_not_create_objects_if_workspace_data_form_invalid(self): + """Posting invalid data to the workspace_data_form form does not create a workspace when using an adapter.""" + # Overriding settings doesn't work, because appconfig.ready has already run and + # registered the default adapter. Instead, unregister the default and register the + # new adapter here. + workspace_adapter_registry.unregister(DefaultWorkspaceAdapter) + workspace_adapter_registry.register(TestWorkspaceAdapter) + self.workspace_type = "test" + billing_project = factories.BillingProjectFactory.create() + url = self.entry_point + "/api/workspaces" + json_data = { + "namespace": "test-billing-project", + "name": "test-workspace", + "attributes": {}, + } + responses.add( + responses.POST, + url, + status=self.api_success_code, + match=[responses.matchers.json_params_matcher(json_data)], + ) + self.client.force_login(self.user) + response = self.client.post( + self.get_url( + self.workspace_to_clone.billing_project.name, + self.workspace_to_clone.name, + self.workspace_type, + ), + { + "billing_project": billing_project.pk, + "name": "test-workspace", + # Default workspace data for formset. + "workspacedata-TOTAL_FORMS": 1, + "workspacedata-INITIAL_FORMS": 0, + "workspacedata-MIN_NUM_FORMS": 1, + "workspacedata-MAX_NUM_FORMS": 1, + "workspacedata-0-study_name": "", + }, + ) + self.assertEqual(response.status_code, 200) + # Workspace form is valid. + form = response.context_data["form"] + self.assertTrue(form.is_valid()) + # workspace_data_form is not valid. + workspace_data_formset = response.context_data["workspace_data_formset"] + self.assertEqual(workspace_data_formset.is_valid(), False) + workspace_data_form = workspace_data_formset.forms[0] + self.assertEqual(workspace_data_form.is_valid(), False) + self.assertEqual(len(workspace_data_form.errors), 1) + self.assertIn("study_name", workspace_data_form.errors) + self.assertEqual(len(workspace_data_form.errors["study_name"]), 1) + self.assertIn("required", workspace_data_form.errors["study_name"][0]) + self.assertEqual(models.Workspace.objects.count(), 1) + self.assertIn(self.workspace_to_clone, models.Workspace.objects.all()) + self.assertEqual(app_models.TestWorkspaceData.objects.count(), 0) + self.assertEqual(len(responses.calls), 0) + + def test_workspace_to_clone_does_not_exist_on_anvil(self): + """Shows a method if an AnVIL API 404 error occurs.""" + billing_project = factories.BillingProjectFactory.create() + json_data = { + "namespace": billing_project.name, + "name": "test-workspace", + "attributes": {}, + } + responses.add( + responses.POST, + self.api_url, + status=404, + match=[responses.matchers.json_params_matcher(json_data)], + json={"message": "workspace create test error"}, + ) + # Need a client to check messages. + self.client.force_login(self.user) + response = self.client.post( + self.get_url( + self.workspace_to_clone.billing_project.name, + self.workspace_to_clone.name, + self.workspace_type, + ), + { + "billing_project": billing_project.pk, + "name": "test-workspace", + # Default workspace data for formset. + "workspacedata-TOTAL_FORMS": 1, + "workspacedata-INITIAL_FORMS": 0, + "workspacedata-MIN_NUM_FORMS": 1, + "workspacedata-MAX_NUM_FORMS": 1, + }, + ) + self.assertEqual(response.status_code, 200) + self.assertIn("form", response.context) + self.assertIn("messages", response.context) + messages = list(response.context["messages"]) + self.assertEqual(len(messages), 1) + self.assertIn("AnVIL API Error: workspace create test error", str(messages[0])) + responses.assert_call_count(self.api_url, 1) + # Make sure that no object is created. + self.assertEqual(models.Workspace.objects.count(), 1) + self.assertIn(self.workspace_to_clone, models.Workspace.objects.all()) + + class WorkspaceUpdateTest(TestCase): def setUp(self): """Set up test class.""" diff --git a/anvil_consortium_manager/urls.py b/anvil_consortium_manager/urls.py index 7110a28f..03acad5b 100644 --- a/anvil_consortium_manager/urls.py +++ b/anvil_consortium_manager/urls.py @@ -208,6 +208,11 @@ views.WorkspaceImport.as_view(), name="import", ), + # path( + # "types//clone/", + # views.WorkspaceClone.as_view(), + # name="clone", + # ), path("audit/", views.WorkspaceAudit.as_view(), name="audit"), path( "//delete/", @@ -233,6 +238,11 @@ views.WorkspaceUpdate.as_view(), name="update", ), + path( + "//clone//", + views.WorkspaceClone.as_view(), + name="clone", + ), ], "workspaces", ) diff --git a/anvil_consortium_manager/views.py b/anvil_consortium_manager/views.py index a92b7b34..361efc8d 100644 --- a/anvil_consortium_manager/views.py +++ b/anvil_consortium_manager/views.py @@ -1313,6 +1313,158 @@ def forms_invalid(self, form, workspace_data_formset): ) +class WorkspaceClone( + auth.AnVILConsortiumManagerEditRequired, + SuccessMessageMixin, + WorkspaceAdapterMixin, + SingleObjectMixin, + FormView, +): + model = models.Workspace + form_class = forms.WorkspaceCloneForm + success_msg = "Successfully created Workspace on AnVIL." + template_name = "anvil_consortium_manager/workspace_clone.html" + + def get_object(self, queryset=None): + """Return the workspace to clone.""" + if queryset is None: + queryset = self.get_queryset() + # Filter the queryset based on kwargs. + billing_project_slug = self.kwargs.get("billing_project_slug") + workspace_slug = self.kwargs.get("workspace_slug") + queryset = queryset.filter( + billing_project__name=billing_project_slug, name=workspace_slug + ) + try: + # Get the single item from the filtered queryset + obj = queryset.get() + except queryset.model.DoesNotExist: + raise Http404( + _("No %(verbose_name)s found matching the query") + % {"verbose_name": queryset.model._meta.verbose_name} + ) + return obj + + def get(self, request, *args, **kwargs): + self.object = self.get_object() + return super().get(request, *args, **kwargs) + + def get_initial(self): + """Add the authorization domains of the workspace to be cloned to the form.""" + initial = super().get_initial() + initial["authorization_domains"] = self.object.authorization_domains.all() + return initial + + def get_form(self, form_class=None): + """Return an instance of the form to be used in this view.""" + if form_class is None: + form_class = self.get_form_class() + return form_class(self.object, **self.get_form_kwargs()) + + def get_workspace_data_formset(self): + """Return an instance of the workspace data form to be used in this view.""" + formset_prefix = "workspacedata" + form_class = self.adapter.get_workspace_data_form_class() + model = self.adapter.get_workspace_data_model() + formset_factory = inlineformset_factory( + models.Workspace, + model, + form=form_class, + can_delete=False, + can_delete_extra=False, + absolute_max=1, + max_num=1, + min_num=1, + ) + if self.request.method in ("POST"): + formset = formset_factory( + self.request.POST, + instance=self.new_workspace, + prefix=formset_prefix, + initial=[{"workspace": self.new_workspace}], + ) + else: + formset = formset_factory(prefix=formset_prefix, initial=[{}]) + return formset + + def get_context_data(self, **kwargs): + """Insert the workspace data formset into the context dict.""" + if "workspace_data_formset" not in kwargs: + kwargs["workspace_data_formset"] = self.get_workspace_data_formset() + return super().get_context_data(**kwargs) + + def post(self, request, *args, **kwargs): + """ + Handle POST requests: instantiate the forms instances with the passed + POST variables and then check if they are valid. + """ + self.adapter = self.get_adapter() + self.object = self.get_object() + self.new_workspace = None + form = self.get_form() + # First, check if the workspace form is valid. + # If it is, we'll save the model and then check the workspace data formset in the post method. + if form.is_valid(): + return self.form_valid(form) + else: + workspace_data_formset = self.get_workspace_data_formset() + return self.forms_invalid(form, workspace_data_formset) + + def form_valid(self, form): + """If the form(s) are valid, save the associated model(s) and create the workspace on AnVIL.""" + # Need to use a transaction because the object needs to be saved to access the many-to-many field. + try: + with transaction.atomic(): + # Calling form.save() does not create the history for the authorization domain many to many field. + # Instead, save the workspace first and then create the auth domain relationships one by one. + # Add the workspace data type from the adapter to the instance. + form.instance.workspace_type = self.adapter.get_type() + self.new_workspace = form.save(commit=False) + self.new_workspace.save() + # Now check the workspace_data_formset. + workspace_data_formset = self.get_workspace_data_formset() + if not workspace_data_formset.is_valid(): + # Tell the transaction to roll back, since we are not raising an exception. + transaction.set_rollback(True) + return self.forms_invalid(form, workspace_data_formset) + # Now save the auth domains and the workspace_data_form. + for auth_domain in form.cleaned_data["authorization_domains"]: + models.WorkspaceAuthorizationDomain.objects.create( + workspace=self.new_workspace, group=auth_domain + ) + workspace_data_formset.forms[0].save() + # Then create the workspace on AnVIL. + authorization_domains = self.new_workspace.authorization_domains.all() + print(authorization_domains) + self.object.anvil_clone( + self.new_workspace.billing_project, + self.new_workspace.name, + authorization_domains=authorization_domains, + ) + except AnVILAPIError as e: + # If the API call failed, rerender the page with the responses and show a message. + messages.add_message( + self.request, messages.ERROR, "AnVIL API Error: " + str(e) + ) + return self.render_to_response( + self.get_context_data( + form=form, workspace_data_formset=workspace_data_formset + ) + ) + return super().form_valid(form) + + def forms_invalid(self, form, workspace_data_formset): + """If the form(s) are invalid, render the invalid form.""" + return self.render_to_response( + self.get_context_data( + form=form, workspace_data_formset=workspace_data_formset + ) + ) + + def get_success_url(self): + return self.new_workspace.get_absolute_url() + + class WorkspaceUpdate( auth.AnVILConsortiumManagerEditRequired, SuccessMessageMixin, From 5ed88aa769d45f5404b1edd49e2b71136d55e441 Mon Sep 17 00:00:00 2001 From: Adrienne Stilp Date: Tue, 6 Dec 2022 12:34:17 -0800 Subject: [PATCH 4/7] Verify that authorization_domains is a list In anvil_api.create_workspace and clone_workspace, check that the authorization_domains argument is a list and arise a ValueError if it is not. --- anvil_consortium_manager/anvil_api.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/anvil_consortium_manager/anvil_api.py b/anvil_consortium_manager/anvil_api.py index 051181b1..65f2caf3 100644 --- a/anvil_consortium_manager/anvil_api.py +++ b/anvil_consortium_manager/anvil_api.py @@ -241,11 +241,11 @@ def create_workspace( "name": workspace_name, "attributes": {}, } + if not isinstance(authorization_domains, list): + raise ValueError("authorization_domains must be a list.") # Add authorization domains. if authorization_domains: - if not isinstance(authorization_domains, list): - authorization_domains = [authorization_domains] auth_domain = [{"membersGroupName": g} for g in authorization_domains] body["authorizationDomain"] = auth_domain @@ -286,11 +286,11 @@ def clone_workspace( "name": cloned_workspace_name, "attributes": {}, } + if not isinstance(authorization_domains, list): + raise ValueError("authorization_domains must be a list.") # Add authorization domains. if authorization_domains: - if not isinstance(authorization_domains, list): - authorization_domains = [authorization_domains] auth_domain = [{"membersGroupName": g} for g in authorization_domains] body["authorizationDomain"] = auth_domain From 40973608c1dde50865ce47e3cae2fd330e44ddc4 Mon Sep 17 00:00:00 2001 From: Adrienne Stilp Date: Wed, 7 Dec 2022 15:54:41 -0800 Subject: [PATCH 5/7] Add test for WorkspaceClone if workspace does not exist Add a test to check that WorkspaceClone raises a 404 error when the workspace does not exist. --- anvil_consortium_manager/tests/test_views.py | 15 ++++++++++++++- 1 file changed, 14 insertions(+), 1 deletion(-) diff --git a/anvil_consortium_manager/tests/test_views.py b/anvil_consortium_manager/tests/test_views.py index edc7c996..9a70011a 100644 --- a/anvil_consortium_manager/tests/test_views.py +++ b/anvil_consortium_manager/tests/test_views.py @@ -8116,7 +8116,7 @@ def get_url(self, *args): def get_view(self): """Return the view being tested.""" - return views.WorkspaceCreate.as_view() + return views.WorkspaceClone.as_view() def test_view_redirect_not_logged_in(self): "View redirects to login view when user is not logged in." @@ -8234,6 +8234,19 @@ def test_post_workspace_type_not_registered(self): workspace_type="foo", ) + def test_get_workspace_not_found(self): + """Raises a 404 error when workspace does not exist.""" + print(self.get_url("foo", "bar", self.workspace_type)) + request = self.factory.get(self.get_url("foo", "bar", self.workspace_type)) + request.user = self.user + with self.assertRaises(Http404): + self.get_view()( + request, + billing_project_slug="foo", + workspace_slug="bar", + workspace_type=self.workspace_type, + ) + def test_has_form_in_context(self): """Response includes a form.""" self.client.force_login(self.user) From fccc322e3f7ade876978b2223cd11b02174027fe Mon Sep 17 00:00:00 2001 From: Adrienne Stilp Date: Wed, 7 Dec 2022 16:01:07 -0800 Subject: [PATCH 6/7] Add info about cloning a workspace to docs --- docs/user_guide.rst | 25 +++++++++++++++++++++++++ 1 file changed, 25 insertions(+) diff --git a/docs/user_guide.rst b/docs/user_guide.rst index fa91ceb8..ddb5cc62 100644 --- a/docs/user_guide.rst +++ b/docs/user_guide.rst @@ -55,6 +55,31 @@ Creating a new workspace If successful, you will be shown a success message and information about the workspace that you just created. The examples would create a workspace named "primed-cc/test-workspace". +Cloning an existing workspace +----------------------------- + +1. Navigate to the workspace detail page that you would like to clone. + +2. Click the "Clone workspace" button and then choose the workspace type that you would like to use for the new workspace. + +3. In the "Billing project" field, select the billing project in which to create the workspace (e.g., primed-cc). If it doesn’t exist, you may need to import a billing project (see instructions). + +4. In the "name" field, type the name of the workspace that you would like to create within the selected billing project (e.g., test-workspace). + +5. If applicable, in the "authorization domains" box, select one or more authorization domains for this workspace. + + * You can select multiple authorization domains using Command-click. + * To de-select an authorization domain, click on it while holding the Command key. + * You must include all the authorization domains from the workspace you are cloning. The form will be autopopulated with these groups. + +6. If desired, add any notes about the Workspace in the "note" field. + +7. Fill out all other required fields for the type of workspace you are creating. + +8. Click on the "Save workspace" button. This can take some time due to delays from AnVIL. + +If successful, you will be shown a success message and information about the workspace that you just created. The examples would create a workspace named "primed-cc/test-workspace". + Importing an existing workspace ------------------------------- From d0ae33f9bd627d7acf980a330eb6e25b06f4af4c Mon Sep 17 00:00:00 2001 From: Adrienne Stilp Date: Wed, 7 Dec 2022 16:01:45 -0800 Subject: [PATCH 7/7] Bump version number and update CHANGELOG --- CHANGELOG.md | 4 ++++ anvil_consortium_manager/__init__.py | 2 +- 2 files changed, 5 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index f29b39da..cf95383e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,9 @@ # Change log +## Devel + +- Add ability to clone a workspace on AnVIL. + ## 0.6 (2022-11-29) - Add ability to update `BillingProject`, `Account`, `ManagedGroup`, and `Workspace`, and workspace data models. diff --git a/anvil_consortium_manager/__init__.py b/anvil_consortium_manager/__init__.py index 4f20502e..1d811ce5 100644 --- a/anvil_consortium_manager/__init__.py +++ b/anvil_consortium_manager/__init__.py @@ -1 +1 @@ -__version__ = "0.6" +__version__ = "0.7dev1"