Feature/cls2 1231 one list permission changes (#5942)

Allow Account manager to change one list tier and core team members for "their" companies.

---------

Co-authored-by: Claudia Gonzalez-Casales <[email protected]>

datahub/company/permissions.py

@@ -0,0 +1,17 @@
+from rest_framework.permissions import BasePermission
+
+from datahub.company.models import CompanyPermission
+
+
+class IsAccountManagerOnCompany(BasePermission):
+    """
+    Allows users:
+    - that have change_company and change_one_list_tier_and_global_account_manager permissions
+    - or that are one_list_account_owners (account managers/ITA Leads) for the current record.
+    """
+
+    def has_object_permission(self, request, view, obj):
+        return request.user.has_perms([
+            f'company.{CompanyPermission.change_company}',
+            f'company.{CompanyPermission.change_one_list_tier_and_global_account_manager}',
+        ]) or (obj.one_list_account_owner == request.user)

datahub/company/test/test_company_views_core_team.py

@@ -219,6 +219,53 @@ def _get_url(company):
             },
         )
 
+    def _assert_update_core_team_members(
+        self,
+        one_list_company,
+        existing_team_count,
+        new_team_count,
+        api_client,
+    ):
+        url = self._get_url(one_list_company)
+
+        if existing_team_count:
+            team_member_advisers = AdviserFactory.create_batch(existing_team_count)
+            OneListCoreTeamMemberFactory.create_batch(
+                len(team_member_advisers),
+                company=one_list_company,
+                adviser=factory.Iterator(team_member_advisers),
+            )
+
+        old_core_team_members = [
+            core_team_member.adviser.id
+            for core_team_member in one_list_company.one_list_core_team_members.all()
+        ]
+
+        new_core_team_members = [
+            adviser.id for adviser in AdviserFactory.create_batch(2)
+        ] if new_team_count else []
+
+        response = api_client.patch(
+            url,
+            {
+                'core_team_members':
+                [
+                    {
+                        'adviser': adviser_id,
+                    } for adviser_id in new_core_team_members
+                ],
+            },
+        )
+        assert response.status_code == status.HTTP_204_NO_CONTENT
+
+        core_team_members = [
+            core_team_member.adviser.id
+            for core_team_member in one_list_company.one_list_core_team_members.all()
+        ]
+
+        assert core_team_members != old_core_team_members
+        assert core_team_members == new_core_team_members
+
     def test_returns_401_if_unauthenticated(self, api_client):
         """Test that a 401 is returned if no credentials are provided."""
         company = CompanyFactory()
@@ -264,45 +311,51 @@ def test_can_update_core_team_members(
     ):
         """Test that core team members can be updated."""
         api_client = self.create_api_client(user=one_list_editor)
-        url = self._get_url(one_list_company)
 
-        if existing_team_count:
-            team_member_advisers = AdviserFactory.create_batch(existing_team_count)
-            OneListCoreTeamMemberFactory.create_batch(
-                len(team_member_advisers),
-                company=one_list_company,
-                adviser=factory.Iterator(team_member_advisers),
-            )
+        self._assert_update_core_team_members(
+            one_list_company, existing_team_count, new_team_count, api_client)
 
-        old_core_team_members = [
-            core_team_member.adviser.id
-            for core_team_member in one_list_company.one_list_core_team_members.all()
-        ]
+    @pytest.mark.parametrize(
+        'existing_team_count,new_team_count',
+        (
+            (0, 2),
+            (2, 2),
+            (2, 0),
+        ),
+    )
+    def test_account_manage_can_update_core_team_members(
+        self,
+        existing_team_count,
+        new_team_count,
+    ):
+        """
+        Test that an account manager can update core team members.
+        """
+        company = CompanyFactory(
+            one_list_account_owner=AdviserFactory(),
+            one_list_tier=random_non_ita_one_list_tier(),
+        )
+        api_client = self.create_api_client(user=company.one_list_account_owner)
 
-        new_core_team_members = [
-            adviser.id for adviser in AdviserFactory.create_batch(2)
-        ] if new_team_count else []
+        self._assert_update_core_team_members(
+            company, existing_team_count, new_team_count, api_client)
 
-        response = api_client.patch(
-            url,
-            {
-                'core_team_members':
-                [
-                    {
-                        'adviser': adviser_id,
-                    } for adviser_id in new_core_team_members
-                ],
-            },
+    def test_returns_403_if_account_manager_updates_other_company(self):
+        """
+        Test that a 403 is returned if an account manager tries to update the core team from
+        a company they are not the account manage for.
+        """
+        account_managers_company = CompanyFactory(
+            one_list_account_owner=AdviserFactory(),
+            one_list_tier=random_non_ita_one_list_tier(),
         )
-        assert response.status_code == status.HTTP_204_NO_CONTENT
+        api_client = self.create_api_client(user=account_managers_company.one_list_account_owner)
 
-        core_team_members = [
-            core_team_member.adviser.id
-            for core_team_member in one_list_company.one_list_core_team_members.all()
-        ]
+        company = CompanyFactory()
+        url = self._get_url(company)
 
-        assert core_team_members != old_core_team_members
-        assert core_team_members == new_core_team_members
+        response = api_client.patch(url)
+        assert response.status_code == status.HTTP_403_FORBIDDEN
 
     def test_cannot_update_duplicate_core_team_members(self, one_list_company, one_list_editor):
         """Test that duplicate team members cannot be updated."""

datahub/company/test/test_company_views_one_list.py

@@ -74,13 +74,13 @@ def test_returns_403_if_without_permission(self, permission_codenames):
         ),
     )
     @pytest.mark.django_db
-    def test_assigns_one_list_tier_and_global_account_manager(
+    def test_assigns_one_list_tier_and_account_manager(
         self,
         company_factory,
         one_list_editor,
     ):
         """
-        Test that a One List tier and global account manager can be assigned to:
+        Test that a One List tier and account manager can be assigned to:
 
         - a company not on the One List
         - a company on random One List tier except 'Tier D - International Trade Adviser Accounts'
@@ -113,6 +113,76 @@ def test_assigns_one_list_tier_and_global_account_manager(
         assert versions[0].field_dict['one_list_tier_id'] == new_one_list_tier.id
         assert versions[0].field_dict['one_list_account_owner_id'] == global_account_manager.id
 
+    @pytest.mark.django_db
+    def test_assigns_one_list_tier_by_account_manager(
+        self,
+    ):
+        """
+        Test that an account manager:
+        - can update the One List tier of the company they are managing
+        """
+        company = CompanyFactory(
+            one_list_account_owner=AdviserFactory(),
+            one_list_tier=random_non_ita_one_list_tier(),
+        )
+        api_client = self.create_api_client(user=company.one_list_account_owner)
+
+        url = self._get_url(company)
+
+        new_one_list_tier = random_non_ita_one_list_tier(
+            exclude=company.one_list_tier,
+        )
+
+        response = api_client.post(
+            url,
+            {
+                'one_list_tier': new_one_list_tier.id,
+                'global_account_manager': company.one_list_account_owner.id,
+            },
+        )
+        assert response.status_code == status.HTTP_204_NO_CONTENT
+
+        company.refresh_from_db()
+        assert company.one_list_account_owner == company.one_list_account_owner
+        assert company.one_list_tier_id == new_one_list_tier.pk
+        assert company.modified_by == company.one_list_account_owner
+
+        # Check that object version is stored correctly
+        versions = Version.objects.get_for_object(company)
+        assert versions.count() == 1
+        assert versions[0].field_dict['one_list_tier_id'] == new_one_list_tier.id
+        assert versions[0].field_dict['one_list_account_owner_id'] == \
+            company.one_list_account_owner.id
+
+    @pytest.mark.django_db
+    def test_returns_403_on_editing_one_list_tier_by_other_account_manager(
+        self,
+    ):
+        """
+        Test that an account manager:
+        - can not update the One List tier of a company they are not managing.
+        """
+        company = CompanyFactory(
+            one_list_account_owner=AdviserFactory(),
+        )
+
+        api_client = self.create_api_client()
+
+        url = self._get_url(company)
+
+        new_one_list_tier = random_non_ita_one_list_tier(
+            exclude=company.one_list_tier,
+        )
+
+        response = api_client.post(
+            url,
+            {
+                'one_list_tier': new_one_list_tier.id,
+                'global_account_manager': company.one_list_account_owner.id,
+            },
+        )
+        assert response.status_code == status.HTTP_403_FORBIDDEN
+
     @pytest.mark.parametrize(
         'company_factory,adviser_id_fn,new_one_list_tier_id_fn,expected_errors',
         (

datahub/company/test/utils.py

@@ -5,11 +5,16 @@
 )
 
 
-def random_non_ita_one_list_tier():
-    """Returns random non ITA One List tier."""
+def random_non_ita_one_list_tier(exclude=None):
+    """
+    Returns random non ITA One List tier.
+    :param exclude: OneListTier object to also exclude
+    """
     queryset = OneListTier.objects.exclude(
         pk=OneListTierID.tier_d_international_trade_advisers.value,
     )
+    if exclude is not None:
+        queryset.exclude(pk=exclude.id)
     return random_obj_for_queryset(queryset)
 
 

datahub/company/views.py

@@ -31,6 +31,7 @@
     Objective,
 )
 from datahub.company.pagination import ContactPageSize
+from datahub.company.permissions import IsAccountManagerOnCompany
 from datahub.company.queryset import (
     get_contact_queryset,
     get_export_country_queryset,
@@ -221,10 +222,7 @@ def remove_account_manager(self, request, *args, **kwargs):
         methods=['post'],
         detail=True,
         permission_classes=[
-            HasPermissions(
-                f'company.{CompanyPermission.change_company}',
-                f'company.{CompanyPermission.change_one_list_tier_and_global_account_manager}',
-            ),
+            IsAccountManagerOnCompany,
         ],
         schema=StubSchema(),
     )
@@ -275,10 +273,7 @@ def remove_from_one_list(self, request, *args, **kwargs):
         methods=['patch'],
         detail=True,
         permission_classes=[
-            HasPermissions(
-                f'company.{CompanyPermission.change_company}',
-                f'company.{CompanyPermission.change_one_list_core_team_member}',
-            ),
+            IsAccountManagerOnCompany,
         ],
         schema=StubSchema(),
     )