Merge pull request #420 from dimagi/hy/limit-credential-visiblity

Limit Visible Credentials to Organization-Issued Credentials in User Invites

commcare_connect/connect_id_client/main.py

@@ -54,8 +54,11 @@ def add_credential(organization: Organization, credential: str, users: list[str]
     return
 
 
-def fetch_credentials():
-    response = _make_request(GET, "/users/fetch_credentials")
+def fetch_credentials(org_slug=None):
+    params = {}
+    if org_slug:
+        params["org_slug"] = org_slug
+    response = _make_request(GET, "/users/fetch_credentials", params=params)
     data = response.json()
     return [Credential(**c) for c in data["credentials"]]
 

commcare_connect/opportunity/forms.py

@@ -33,7 +33,8 @@
 
 class OpportunityUserInviteForm(forms.Form):
     def __init__(self, *args, **kwargs):
-        credentials = connect_id_client.fetch_credentials()
+        org_slug = kwargs.pop("org_slug", None)
+        credentials = connect_id_client.fetch_credentials(org_slug)
         super().__init__(*args, **kwargs)
 
         self.helper = FormHelper(self)
@@ -73,7 +74,10 @@ def clean_users(self):
         return split_users
 
 
-class OpportunityChangeForm(forms.ModelForm, OpportunityUserInviteForm):
+class OpportunityChangeForm(
+    OpportunityUserInviteForm,
+    forms.ModelForm,
+):
     class Meta:
         model = Opportunity
         fields = [

commcare_connect/opportunity/tests/test_forms.py

@@ -5,8 +5,8 @@
 import pytest
 from factory.fuzzy import FuzzyDate, FuzzyText
 
-from commcare_connect.opportunity.forms import OpportunityCreationForm
-from commcare_connect.opportunity.tests.factories import ApplicationFactory
+from commcare_connect.opportunity.forms import OpportunityChangeForm, OpportunityCreationForm
+from commcare_connect.opportunity.tests.factories import ApplicationFactory, CommCareAppFactory, OpportunityFactory
 
 
 class TestOpportunityCreationForm:
@@ -111,3 +111,197 @@ def test_save(self, user, organization):
         )
         form.is_valid()
         form.save()
+
+
+@pytest.mark.django_db
+class TestOpportunityChangeForm:
+    @pytest.fixture(autouse=True)
+    def setup_credentials_mock(self, monkeypatch):
+        self.mock_credentials = [
+            type("Credential", (), {"slug": "cert1", "name": "Work for test"}),
+            type("Credential", (), {"slug": "cert2", "name": "Work for test"}),
+        ]
+        monkeypatch.setattr(
+            "commcare_connect.connect_id_client.fetch_credentials", lambda org_slug: self.mock_credentials
+        )
+
+    @pytest.fixture
+    def valid_opportunity(self, organization):
+        return OpportunityFactory(
+            organization=organization,
+            active=True,
+            learn_app=CommCareAppFactory(cc_app_id="test_learn_app"),
+            deliver_app=CommCareAppFactory(cc_app_id="test_deliver_app"),
+            name="Test Opportunity",
+            description="Test Description",
+            short_description="Short Description",
+            currency="USD",
+            is_test=False,
+            end_date=datetime.date.today() + datetime.timedelta(days=30),
+        )
+
+    @pytest.fixture
+    def base_form_data(self, valid_opportunity):
+        return {
+            "name": "Updated Opportunity",
+            "description": "Updated Description",
+            "short_description": "Updated Short Description",
+            "active": True,
+            "currency": "EUR",
+            "is_test": False,
+            "delivery_type": valid_opportunity.delivery_type.id,
+            "additional_users": 5,
+            "end_date": (datetime.date.today() + datetime.timedelta(days=60)).isoformat(),
+            "users": "+1234567890\n+9876543210",
+            "filter_country": "US",
+            "filter_credential": "cert1",
+        }
+
+    def test_form_initialization(self, valid_opportunity, organization):
+        form = OpportunityChangeForm(instance=valid_opportunity, org_slug=organization.slug)
+        expected_fields = {
+            "name",
+            "description",
+            "short_description",
+            "active",
+            "currency",
+            "is_test",
+            "delivery_type",
+            "additional_users",
+            "end_date",
+            "users",
+            "filter_country",
+            "filter_credential",
+        }
+        assert all(field in form.fields for field in expected_fields)
+
+        expected_initial = {
+            "name": valid_opportunity.name,
+            "description": valid_opportunity.description,
+            "short_description": valid_opportunity.short_description,
+            "active": valid_opportunity.active,
+            "currency": valid_opportunity.currency,
+            "is_test": valid_opportunity.is_test,
+            "delivery_type": valid_opportunity.delivery_type.id,
+            "end_date": valid_opportunity.end_date.isoformat(),
+            "filter_country": [""],
+            "filter_credential": [""],
+        }
+        assert all(form.initial.get(key) == value for key, value in expected_initial.items())
+
+    @pytest.mark.parametrize(
+        "field",
+        [
+            "name",
+            "description",
+            "short_description",
+            "currency",
+        ],
+    )
+    def test_required_fields(self, valid_opportunity, organization, field, base_form_data):
+        data = base_form_data.copy()
+        data[field] = ""
+        form = OpportunityChangeForm(data=data, instance=valid_opportunity, org_slug=organization.slug)
+        assert not form.is_valid()
+        assert field in form.errors
+
+    @pytest.mark.parametrize(
+        "test_data",
+        [
+            pytest.param(
+                {
+                    "field": "additional_users",
+                    "value": "invalid",
+                    "error_expected": True,
+                    "error_message": "Enter a whole number.",
+                },
+                id="invalid_additional_users",
+            ),
+            pytest.param(
+                {
+                    "field": "end_date",
+                    "value": "invalid-date",
+                    "error_expected": True,
+                    "error_message": "Enter a valid date.",
+                },
+                id="invalid_end_date",
+            ),
+            pytest.param(
+                {
+                    "field": "users",
+                    "value": "  +1234567890  \n  +9876543210  ",
+                    "error_expected": False,
+                    "expected_clean": ["+1234567890", "+9876543210"],
+                },
+                id="valid_users_with_whitespace",
+            ),
+        ],
+    )
+    def test_field_validation(self, valid_opportunity, organization, base_form_data, test_data):
+        data = base_form_data.copy()
+        data[test_data["field"]] = test_data["value"]
+        form = OpportunityChangeForm(data=data, instance=valid_opportunity, org_slug=organization.slug)
+        if test_data["error_expected"]:
+            assert not form.is_valid()
+            assert test_data["error_message"] in str(form.errors[test_data["field"]])
+        else:
+            assert form.is_valid()
+            if "expected_clean" in test_data:
+                assert form.cleaned_data[test_data["field"]] == test_data["expected_clean"]
+
+    @pytest.mark.parametrize(
+        "app_scenario",
+        [
+            pytest.param(
+                {
+                    "active_app_ids": ("unique_app1", "unique_app2"),
+                    "new_app_ids": ("different_app1", "different_app2"),
+                    "expected_valid": True,
+                },
+                id="unique_apps",
+            ),
+            pytest.param(
+                {
+                    "active_app_ids": ("shared_app1", "shared_app2"),
+                    "new_app_ids": ("shared_app1", "shared_app2"),
+                    "expected_valid": False,
+                },
+                id="reused_apps",
+            ),
+        ],
+    )
+    def test_app_reuse_validation(self, organization, base_form_data, app_scenario):
+        OpportunityFactory(
+            organization=organization,
+            active=True,
+            learn_app=CommCareAppFactory(cc_app_id=app_scenario["active_app_ids"][0]),
+            deliver_app=CommCareAppFactory(cc_app_id=app_scenario["active_app_ids"][1]),
+        )
+
+        inactive_opp = OpportunityFactory(
+            organization=organization,
+            active=False,
+            learn_app=CommCareAppFactory(cc_app_id=app_scenario["new_app_ids"][0]),
+            deliver_app=CommCareAppFactory(cc_app_id=app_scenario["new_app_ids"][1]),
+        )
+
+        form = OpportunityChangeForm(data=base_form_data, instance=inactive_opp, org_slug=organization.slug)
+
+        assert form.is_valid() == app_scenario["expected_valid"]
+        if not app_scenario["expected_valid"]:
+            assert "Cannot reactivate opportunity with reused applications" in str(form.errors["active"])
+
+    @pytest.mark.parametrize(
+        "data_updates,expected_valid",
+        [
+            ({"currency": "USD", "additional_users": 5}, True),
+            ({"currency": "EUR", "additional_users": 10}, True),
+            ({"currency": "INVALID", "additional_users": 5}, False),
+            ({"currency": "USD", "additional_users": -5}, True),
+        ],
+    )
+    def test_valid_combinations(self, valid_opportunity, organization, base_form_data, data_updates, expected_valid):
+        data = base_form_data.copy()
+        data.update(data_updates)
+        form = OpportunityChangeForm(data=data, instance=valid_opportunity, org_slug=organization.slug)
+        assert form.is_valid() == expected_valid

commcare_connect/opportunity/views.py

@@ -212,6 +212,11 @@ class OpportunityEdit(OrganizationUserMemberRoleMixin, UpdateView):
     def get_success_url(self):
         return reverse("opportunity:detail", args=(self.request.org.slug, self.object.id))
 
+    def get_form_kwargs(self):
+        kwargs = super().get_form_kwargs()
+        kwargs["org_slug"] = self.request.org.slug
+        return kwargs
+
     def form_valid(self, form):
         opportunity = form.instance
         opportunity.modified_by = self.request.user.email
@@ -1079,7 +1084,7 @@ def import_catchment_area(request, org_slug=None, pk=None):
 @org_member_required
 def opportunity_user_invite(request, org_slug=None, pk=None):
     opportunity = get_object_or_404(Opportunity, organization=request.org, id=pk)
-    form = OpportunityUserInviteForm(data=request.POST or None)
+    form = OpportunityUserInviteForm(data=request.POST or None, org_slug=request.org.slug)
     if form.is_valid():
         users = form.cleaned_data["users"]
         filter_country = form.cleaned_data["filter_country"]

commcare_connect/organization/views.py

@@ -47,7 +47,7 @@ def organization_home(request, org_slug):
     if not form:
         form = OrganizationChangeForm(instance=org)
 
-    credentials = connect_id_client.fetch_credentials()
+    credentials = connect_id_client.fetch_credentials(org_slug=request.org.slug)
     credential_name = f"Worked for {org.name}"
     if not any(c.name == credential_name for c in credentials):
         credentials.append(Credential(name=credential_name, slug=slugify(credential_name)))
@@ -96,7 +96,7 @@ def accept_invite(request, org_slug, invite_id):
 @require_POST
 def add_credential_view(request, org_slug):
     org = get_object_or_404(Organization, slug=org_slug)
-    credentials = connect_id_client.fetch_credentials()
+    credentials = connect_id_client.fetch_credentials(org_slug=request.org.slug)
     credential_name = f"Worked for {org.name}"
     if not any(c.name == credential_name for c in credentials):
         credentials.append(Credential(name=credential_name, slug=slugify(credential_name)))