Merge pull request #2583 from AlanCoding/2530_job_cancel_access

Allow job self-canceling and update other cancel logic

awx/main/tests/functional/test_rbac_job.py

@@ -1,7 +1,18 @@
 import pytest
 
-from awx.main.access import JobAccess
-from awx.main.models import Job
+from awx.main.access import (
+    JobAccess,
+    AdHocCommandAccess,
+    InventoryUpdateAccess,
+    ProjectUpdateAccess
+)
+from awx.main.models import (
+    Job,
+    AdHocCommand,
+    InventoryUpdate,
+    InventorySource,
+    ProjectUpdate
+)
 
 
 @pytest.fixture
@@ -12,6 +23,28 @@ def normal_job(deploy_jobtemplate):
         inventory=deploy_jobtemplate.inventory
     )
 
+@pytest.fixture
+def jt_user(deploy_jobtemplate, rando):
+    deploy_jobtemplate.execute_role.members.add(rando)
+    return rando
+
+@pytest.fixture
+def inv_updater(inventory, rando):
+    inventory.update_role.members.add(rando)
+    return rando
+
+@pytest.fixture
+def host_adhoc(host, machine_credential, rando):
+    host.inventory.adhoc_role.members.add(rando)
+    machine_credential.use_role.members.add(rando)
+    return rando
+
+@pytest.fixture
+def proj_updater(project, rando):
+    project.update_role.members.add(rando)
+    return rando
+
+
 # Read permissions testing
 @pytest.mark.django_db
 def test_superuser_sees_orphans(normal_job, admin_user):
@@ -70,3 +103,59 @@ def test_project_org_admin_delete_allowed(normal_job, org_admin):
     normal_job.inventory = None # do this so we test job->project->org->admin connection
     access = JobAccess(org_admin)
     assert access.can_delete(normal_job)
+
+@pytest.mark.django_db
+class TestJobAndUpdateCancels:
+
+    # used in view: job_template_launch
+    def test_jt_self_cancel(self, deploy_jobtemplate, jt_user):
+        job = Job(job_template=deploy_jobtemplate, created_by=jt_user)
+        access = JobAccess(jt_user)
+        assert access.can_cancel(job)
+
+    def test_jt_friend_cancel(self, deploy_jobtemplate, admin_user, jt_user):
+        job = Job(job_template=deploy_jobtemplate, created_by=admin_user)
+        access = JobAccess(jt_user)
+        assert not access.can_cancel(job)
+
+    def test_jt_org_admin_cancel(self, deploy_jobtemplate, org_admin, jt_user):
+        job = Job(job_template=deploy_jobtemplate, created_by=jt_user)
+        access = JobAccess(org_admin)
+        assert access.can_cancel(job)
+
+    # used in view: host_ad_hoc_commands_list
+    def test_host_self_cancel(self, host, host_adhoc):
+        adhoc_command = AdHocCommand(inventory=host.inventory, created_by=host_adhoc)
+        access = AdHocCommandAccess(host_adhoc)
+        assert access.can_cancel(adhoc_command)
+
+    def test_host_friend_cancel(self, host, admin_user, host_adhoc):
+        adhoc_command = AdHocCommand(inventory=host.inventory, created_by=admin_user)
+        access = AdHocCommandAccess(host_adhoc)
+        assert not access.can_cancel(adhoc_command)
+
+    # used in view: inventory_source_update_view
+    def test_inventory_self_cancel(self, inventory, inv_updater):
+        inventory_update = InventoryUpdate(inventory_source=InventorySource(
+            name=inventory.name, inventory=inventory, source='gce'
+        ), created_by=inv_updater)
+        access = InventoryUpdateAccess(inv_updater)
+        assert access.can_cancel(inventory_update)
+
+    def test_inventory_friend_cancel(self, inventory, admin_user, inv_updater):
+        inventory_update = InventoryUpdate(inventory_source=InventorySource(
+            name=inventory.name, inventory=inventory, source='gce'
+        ), created_by=admin_user)
+        access = InventoryUpdateAccess(inv_updater)
+        assert not access.can_cancel(inventory_update)
+
+    # used in view: project_update_view
+    def test_project_self_cancel(self, project, proj_updater):
+        project_update = ProjectUpdate(project=project, created_by=proj_updater)
+        access = ProjectUpdateAccess(proj_updater)
+        assert access.can_cancel(project_update)
+
+    def test_project_friend_cancel(self, project, admin_user, proj_updater):
+        project_update = ProjectUpdate(project=project, created_by=admin_user)
+        access = ProjectUpdateAccess(proj_updater)
+        assert not access.can_cancel(project_update)