feat: Add tokenless mutation and field in owner

codecov_auth/commands/owner/interactors/set_tokens_required.py

@@ -0,0 +1,43 @@
+from dataclasses import dataclass
+
+from codecov.commands.base import BaseInteractor
+from codecov.commands.exceptions import Unauthenticated, Unauthorized, ValidationError
+from codecov.db import sync_to_async
+from codecov_auth.helpers import current_user_part_of_org
+from codecov_auth.models import Owner
+
+
+@dataclass
+class SetTokensRequiredInput:
+    tokens_required: bool
+    org_username: str
+
+
+class SetTokensRequiredInteractor(BaseInteractor):
+    def validate(self, owner_obj):
+        if not self.current_user.is_authenticated:
+            raise Unauthenticated()
+        if not owner_obj:
+            raise ValidationError("Owner not found")
+        if not current_user_part_of_org(self.current_owner, owner_obj):
+            raise Unauthorized()
+        if not owner_obj.is_admin(self.current_owner):
+            raise Unauthorized("Admin authorization required")
+
+    @sync_to_async
+    def execute(self, input: dict):
+        typed_input = SetTokensRequiredInput(
+            tokens_required=input.get("tokens_required"),
+            org_username=input.get("org_username"),
+        )
+
+        owner_obj = Owner.objects.filter(
+            username=typed_input.org_username, service=self.service
+        ).first()
+
+        self.validate(owner_obj)
+
+        owner_obj.tokens_required = typed_input.tokens_required
+        owner_obj.save()
+
+        return typed_input.tokens_required

codecov_auth/commands/owner/interactors/tests/test_set_tokesn_required.py

@@ -0,0 +1,111 @@
+import pytest
+from asgiref.sync import async_to_sync
+from django.contrib.auth.models import AnonymousUser
+from django.test import TransactionTestCase
+
+from codecov.commands.exceptions import Unauthenticated, Unauthorized, ValidationError
+from codecov_auth.tests.factories import OwnerFactory
+
+from ..set_tokens_required import SetTokensRequiredInteractor
+
+
+class SetTokensRequiredInteractorTest(TransactionTestCase):
+    def setUp(self):
+        self.current_user = OwnerFactory(username="codecov-user")
+        self.service = "github"
+        self.owner = OwnerFactory(
+            username="codecov-owner",
+            service=self.service,
+        )
+
+        self.owner_with_admins = OwnerFactory(
+            username="codecov-admin-owner",
+            service=self.service,
+            admins=[self.current_user.ownerid],
+        )
+
+        self.interactor = SetTokensRequiredInteractor(
+            current_owner=self.owner,
+            service=self.service,
+            current_user=self.current_user,
+        )
+
+    @async_to_sync
+    async def execute(
+        self,
+        interactor: SetTokensRequiredInteractor | None = None,
+        input: dict | None = None,
+    ):
+        if not interactor:
+            interactor = self.interactor
+        return await interactor.execute(input)
+
+    @pytest.mark.asyncio
+    async def test_user_is_not_authenticated(self):
+        with pytest.raises(Unauthenticated):
+            await self.execute(
+                interactor=SetTokensRequiredInteractor(
+                    current_owner=None,
+                    service=self.service,
+                    current_user=AnonymousUser(),
+                ),
+                input={
+                    "tokens_required": True,
+                    "org_username": self.owner.username,
+                },
+            )
+
+    @pytest.mark.asyncio
+    async def test_validation_error_when_owner_not_found(self):
+        with pytest.raises(ValidationError):
+            await self.execute(
+                input={
+                    "tokens_required": True,
+                    "org_username": "non-existent-user",
+                },
+            )
+
+    @pytest.mark.asyncio
+    async def test_unauthorized_error_when_user_is_not_admin(self):
+        with pytest.raises(Unauthorized):
+            await self.execute(
+                input={
+                    "tokens_required": True,
+                    "org_username": self.owner.username,
+                },
+            )
+
+    @pytest.mark.asyncio
+    async def test_set_tokens_required_when_user_is_admin(self):
+        input_data = {
+            "tokens_required": True,
+            "org_username": self.owner_with_admins.username,
+        }
+
+        interactor = SetTokensRequiredInteractor(
+            current_owner=self.current_user, service=self.service
+        )
+        result = await self.execute(interactor=interactor, input=input_data)
+
+        assert result == True
+        self.owner_with_admins.refresh_from_db()
+        assert self.owner_with_admins.tokens_required == True
+
+    @pytest.mark.asyncio
+    async def test_set_tokens_required_to_false(self):
+        self.owner_with_admins.tokens_required = True
+        self.owner_with_admins.save()
+
+        input_data = {
+            "tokens_required": False,
+            "org_username": self.owner_with_admins.username,
+        }
+
+        interactor = SetTokensRequiredInteractor(
+            current_owner=self.current_user, service=self.service
+        )
+        result = await self.execute(interactor=interactor, input=input_data)
+
+        assert result == False
+        self.owner_with_admins.refresh_from_db()
+        assert self.owner_with_admins.tokens_required == False

codecov_auth/commands/owner/owner.py

@@ -14,6 +14,7 @@
 from .interactors.revoke_user_token import RevokeUserTokenInteractor
 from .interactors.save_okta_config import SaveOktaConfigInteractor
 from .interactors.save_terms_agreement import SaveTermsAgreementInteractor
+from .interactors.set_tokens_required import SetTokensRequiredInteractor
 from .interactors.set_yaml_on_owner import SetYamlOnOwnerInteractor
 from .interactors.start_trial import StartTrialInteractor
 from .interactors.store_codecov_metric import StoreCodecovMetricInteractor
@@ -98,3 +99,6 @@ def store_codecov_metric(
 
     def save_okta_config(self, input) -> None:
         return self.get_interactor(SaveOktaConfigInteractor).execute(input)
+
+    def set_tokens_required(self, input) -> None:
+        return self.get_interactor(SetTokensRequiredInteractor).execute(input)

codecov_auth/commands/owner/tests/test_owner.py

@@ -134,3 +134,12 @@ def test_save_okta_config_delegate_to_interactor(self, interactor_mock):
         }
         self.command.save_okta_config(input_dict)
         interactor_mock.assert_called_once_with(input_dict)
+
+    @patch("codecov_auth.commands.owner.owner.SetTokensRequiredInteractor.execute")
+    def test_set_tokens_required_delegate_to_interactor(self, interactor_mock):
+        input_dict = {
+            "tokens_required": True,
+            "org_username": "codecov-user",
+        }
+        self.command.set_tokens_required(input_dict)
+        interactor_mock.assert_called_once_with(input_dict)

graphql_api/tests/mutation/test_set_tokens_required.py

@@ -0,0 +1,91 @@
+from django.test import TransactionTestCase
+
+from codecov_auth.tests.factories import OwnerFactory
+from graphql_api.tests.helper import GraphQLTestHelper
+
+query = """
+mutation($input: SetTokensRequiredInput!) {
+  setTokensRequired(input: $input) {
+    tokensRequired
+    error {
+      __typename
+      ... on ResolverError {
+        message
+      }
+    }
+  }
+}
+"""
+
+
+class SetTokensRequiredTests(GraphQLTestHelper, TransactionTestCase):
+    def setUp(self):
+        self.org = OwnerFactory(username="codecov")
+
+    def test_when_authenticated_updates_tokens_required(self):
+        user = OwnerFactory(
+            organizations=[self.org.ownerid],
+            permission=[self.org.ownerid],
+            is_admin=True,
+        )
+
+        data = self.gql_request(
+            query,
+            owner=user,
+            variables={"input": {"org_username": "codecov", "tokensRequired": True}},
+        )
+
+        assert data["setTokensRequired"]["tokensRequired"] == True
+
+    def test_when_validation_error_org_not_found(self):
+        data = self.gql_request(
+            query,
+            owner=self.org,
+            variables={
+                "input": {
+                    "org_username": "non_existent_org",
+                    "tokensRequired": True,
+                }
+            },
+        )
+        assert data["setTokensRequired"]["error"]["__typename"] == "ValidationError"
+
+    def test_when_unauthorized_non_admin(self):
+        non_admin_user = OwnerFactory(
+            organizations=[self.org.ownerid],
+            permission=[self.org.ownerid],
+            is_admin=False,
+        )
+
+        data = self.gql_request(
+            query,
+            owner=non_admin_user,
+            variables={"input": {"org_username": "codecov", "tokensRequired": True}},
+        )
+
+        assert data["setTokensRequired"]["error"]["__typename"] == "UnauthorizedError"
+
+    def test_when_unauthenticated(self):
+        data = self.gql_request(
+            query,
+            variables={"input": {"org_username": "codecov", "tokensRequired": True}},
+        )
+
+        assert (
+            data["setTokensRequired"]["error"]["__typename"] == "UnauthenticatedError"
+        )
+
+    def test_when_not_part_of_org(self):
+        non_part_of_org_user = OwnerFactory(
+            organizations=[self.org.ownerid],
+            permission=[self.org.ownerid],
+            is_admin=False,
+        )
+
+        data = self.gql_request(
+            query,
+            owner=non_part_of_org_user,
+            variables={"input": {"org_username": "codecov", "tokensRequired": True}},
+        )
+
+        assert data["setTokensRequired"]["error"]["__typename"] == "UnauthorizedError"

graphql_api/tests/test_owner.py

@@ -11,7 +11,11 @@
 from shared.django_apps.reports.models import ReportType
 from shared.upload.utils import UploaderType, insert_coverage_measurement
 
-from codecov.commands.exceptions import MissingService, UnauthorizedGuestAccess
+from codecov.commands.exceptions import (
+    MissingService,
+    Unauthenticated,
+    UnauthorizedGuestAccess,
+)
 from codecov_auth.models import OwnerProfile
 from codecov_auth.tests.factories import (
     AccountFactory,
@@ -833,3 +837,28 @@ def test_fetch_is_github_rate_limited_not_on_gh_service(self):
         """ % (current_org.username)
         data = self.gql_request(query, owner=current_org, provider="bb")
         assert data["owner"]["isGithubRateLimited"] == False
+
+    def test_set_tokens_required(self):
+        owner = OwnerFactory(username="sample-owner", service="github")
+        query = """
+        mutation {
+            setTokensRequired(input: { org_username: "sample-owner", tokensRequired: true }) {
+                tokensRequired
+            }
+        }
+        """
+        data = self.gql_request(query, owner=owner)
+        assert data["setTokensRequired"]["tokensRequired"] == True
+
+    def test_set_tokens_required_unauthenticated_returns_None(self):
+        OwnerFactory(username="sample-owner", service="github")
+        query = """
+        mutation {
+            setTokensRequired(input: { org_username: "sample-owner", tokensRequired: true }) {
+                tokensRequired
+            }
+        }
+        """
+        data = self.gql_request(query, with_errors=True)
+        assert data["errors"][0]["message"] == Unauthenticated.message
+        assert data["setTokensRequired"] is None

graphql_api/types/mutation/__init__.py

@@ -18,6 +18,7 @@
 from .save_okta_config import gql_save_okta_config
 from .save_sentry_state import gql_save_sentry_state
 from .save_terms_agreement import gql_save_terms_agreement
+from .set_tokens_required import gql_set_tokens_required
 from .set_yaml_on_owner import gql_set_yaml_on_owner
 from .start_trial import gql_start_trial
 from .store_event_metrics import gql_store_event_metrics
@@ -53,3 +54,4 @@
 mutation = mutation + gql_encode_secret_string
 mutation = mutation + gql_store_event_metrics
 mutation = mutation + gql_save_okta_config
+mutation = mutation + gql_set_tokens_required

graphql_api/types/mutation/mutation.graphql

@@ -38,4 +38,5 @@ type Mutation {
   encodeSecretString(input: EncodeSecretStringInput!): EncodeSecretStringPayload
   storeEventMetric(input: StoreEventMetricsInput!): StoreEventMetricsPayload
   saveOktaConfig(input: SaveOktaConfigInput!): SaveOktaConfigPayload
+  setTokensRequired(input: SetTokensRequiredInput!): SetTokensRequiredPayload
 }

graphql_api/types/mutation/mutation.py

@@ -38,6 +38,7 @@
     error_save_terms_agreement,
     resolve_save_terms_agreement,
 )
+from .set_tokens_required import error_set_tokens_required, resolve_set_tokens_required
 from .set_yaml_on_owner import error_set_yaml_error, resolve_set_yaml_on_owner
 from .start_trial import error_start_trial, resolve_start_trial
 from .store_event_metrics import error_store_event_metrics, resolve_store_event_metrics
@@ -94,6 +95,7 @@
 mutation_bindable.field("storeEventMetric")(resolve_store_event_metrics)
 
 mutation_bindable.field("saveOktaConfig")(resolve_save_okta_config)
+mutation_bindable.field("setTokensRequired")(resolve_set_tokens_required)
 
 mutation_resolvers = [
     mutation_bindable,
@@ -122,4 +124,5 @@
     error_encode_secret_string,
     error_store_event_metrics,
     error_save_okta_config,
+    error_set_tokens_required,
 ]

graphql_api/types/mutation/set_tokens_required/__init__.py

@@ -0,0 +1,12 @@
+from graphql_api.helpers.ariadne import ariadne_load_local_graphql
+
+from .set_tokens_required import (
+    error_set_tokens_required,
+    resolve_set_tokens_required,
+)
+
+gql_set_tokens_required = ariadne_load_local_graphql(
+    __file__, "set_tokens_required.graphql"
+)
+
+__all__ = ["error_set_tokens_required", "resolve_set_tokens_required"]

graphql_api/types/mutation/set_tokens_required/set_tokens_required.graphql

@@ -0,0 +1,14 @@
+union SetTokensRequiredError =
+    UnauthenticatedError
+  | UnauthorizedError
+  | ValidationError
+
+type SetTokensRequiredPayload {
+  error: SetTokensRequiredError
+  tokensRequired: Boolean!
+}
+
+input SetTokensRequiredInput {
+  org_username: String!
+  tokensRequired: Boolean!
+}