Allow erasing a foreign Repo

codecov/commands/base.py

@@ -1,7 +1,16 @@
+from django.conf import settings
 from django.contrib.auth.models import AnonymousUser
 
-from codecov.commands.exceptions import MissingService
+import services.self_hosted as self_hosted
+from codecov.commands.exceptions import (
+    MissingService,
+    Unauthenticated,
+    Unauthorized,
+    ValidationError,
+)
+from codecov_auth.helpers import current_user_part_of_org
 from codecov_auth.models import Owner, User
+from core.models import Repository
 
 
 class BaseCommand:
@@ -44,3 +53,59 @@ def __init__(self, current_owner: Owner, service: str, current_user: User = None
 
         if self.current_owner:
             self.current_user = self.current_owner.user
+
+    def ensure_is_admin(self, owner: Owner) -> None:
+        """
+        Ensures that the `current_owner` is an admin of `owner`,
+        or raise `Unauthorized` otherwise.
+        """
+
+        if not current_user_part_of_org(self.current_owner, owner):
+            raise Unauthorized()
+
+        if settings.IS_ENTERPRISE:
+            if not self_hosted.is_admin_owner(self.current_owner):
+                raise Unauthorized()
+        else:
+            if not owner.is_admin(self.current_owner):
+                raise Unauthorized()
+
+    def resolve_owner_and_repo(
+        self,
+        owner_username: str,
+        repo_name: str,
+        ensure_is_admin: bool = False,
+        only_viewable: bool = False,
+        only_active: bool = False,
+    ) -> tuple[Owner, Repository]:
+        """
+        Resolves the `Owner` and `Repository` based on the passed `owner_username`
+        and `repo_name` respectively.
+
+        If `ensure_is_admin` is set, this will also ensure that the `current_owner` is an
+        admin on the resolved `Owner`.
+        """
+        if ensure_is_admin and not self.current_user.is_authenticated:
+            raise Unauthenticated()
+
+        owner = Owner.objects.filter(
+            service=self.service, username=owner_username
+        ).first()
+
+        if not owner:
+            raise ValidationError("Owner not found")
+
+        if ensure_is_admin:
+            self.ensure_is_admin(owner)
+
+        repo_query = Repository.objects
+        if only_viewable:
+            repo_query = repo_query.viewable_repos(self.current_owner)
+        if only_active:
+            repo_query = repo_query.filter(active=True)
+
+        repo = repo_query.filter(author=owner, name=repo_name).first()
+        if not repo:
+            raise ValidationError("Repo not found")
+
+        return (owner, repo)

core/commands/component/interactors/delete_component_measurements.py

@@ -1,45 +1,12 @@
-from django.conf import settings
-
-import services.self_hosted as self_hosted
 from codecov.commands.base import BaseInteractor
-from codecov.commands.exceptions import (
-    Unauthenticated,
-    Unauthorized,
-    ValidationError,
-)
-from codecov_auth.models import Owner
-from core.models import Repository
 from services.task import TaskService
 
 
 class DeleteComponentMeasurementsInteractor(BaseInteractor):
-    def validate(self, owner: Owner, repo: Repository):
-        if not self.current_user.is_authenticated:
-            raise Unauthenticated()
-
-        if not owner:
-            raise ValidationError("Owner not found")
-
-        if not repo:
-            raise ValidationError("Repo not found")
-
-        if settings.IS_ENTERPRISE:
-            if not self_hosted.is_admin_owner(self.current_owner):
-                raise Unauthorized()
-        else:
-            if not owner.is_admin(self.current_owner):
-                raise Unauthorized()
-
     def execute(self, owner_username: str, repo_name: str, component_id: str):
-        owner = Owner.objects.filter(
-            service=self.service, username=owner_username
-        ).first()
-
-        repo = None
-        if owner:
-            repo = Repository.objects.filter(author_id=owner.pk, name=repo_name).first()
-
-        self.validate(owner, repo)
+        _owner, repo = self.resolve_owner_and_repo(
+            owner_username, repo_name, ensure_is_admin=True
+        )
 
         TaskService().delete_component_measurements(
             repo.repoid,

core/commands/component/tests/test_component.py

@@ -17,6 +17,7 @@ class ComponentCommandsTest(TransactionTestCase):
     def setUp(self):
         self.owner = OwnerFactory(username="test-user")
         self.org = OwnerFactory(username="test-org", admins=[self.owner.pk])
+        self.owner.organizations = [self.org.pk]
         self.repo = RepositoryFactory(author=self.org)
         self.command = ComponentCommands(self.owner, "github")
 

core/commands/flag/interactors/delete_flag.py

@@ -1,46 +1,15 @@
-from django.conf import settings
-
-import services.self_hosted as self_hosted
 from codecov.commands.base import BaseInteractor
 from codecov.commands.exceptions import (
     NotFound,
-    Unauthenticated,
-    Unauthorized,
-    ValidationError,
 )
-from codecov_auth.models import Owner
-from core.models import Repository
 from reports.models import RepositoryFlag
 
 
 class DeleteFlagInteractor(BaseInteractor):
-    def validate(self, owner: Owner, repo: Repository):
-        if not self.current_user.is_authenticated:
-            raise Unauthenticated()
-
-        if not owner:
-            raise ValidationError("Owner not found")
-
-        if not repo:
-            raise ValidationError("Repo not found")
-
-        if settings.IS_ENTERPRISE:
-            if not self_hosted.is_admin_owner(self.current_owner):
-                raise Unauthorized()
-        else:
-            if not owner.is_admin(self.current_owner):
-                raise Unauthorized()
-
     def execute(self, owner_username: str, repo_name: str, flag_name: str):
-        owner = Owner.objects.filter(
-            service=self.service, username=owner_username
-        ).first()
-
-        repo = None
-        if owner:
-            repo = Repository.objects.filter(author_id=owner.pk, name=repo_name).first()
-
-        self.validate(owner, repo)
+        _owner, repo = self.resolve_owner_and_repo(
+            owner_username, repo_name, ensure_is_admin=True
+        )
 
         flag = RepositoryFlag.objects.filter(
             repository_id=repo.pk, flag_name=flag_name

core/commands/flag/tests/test_flag.py

@@ -18,6 +18,7 @@ class FlagCommandsTest(TransactionTestCase):
     def setUp(self):
         self.owner = OwnerFactory(username="test-user")
         self.org = OwnerFactory(username="test-org", admins=[self.owner.pk])
+        self.owner.organizations = [self.org.pk]
         self.repo = RepositoryFactory(author=self.org)
         self.command = FlagCommands(self.owner, "github")
         self.flag = RepositoryFlagFactory(repository=self.repo, flag_name="test-flag")

core/commands/repository/interactors/activate_measurements.py

@@ -3,30 +3,21 @@
 from codecov.commands.base import BaseInteractor
 from codecov.commands.exceptions import ValidationError
 from codecov.db import sync_to_async
-from codecov_auth.models import Owner
-from core.models import Repository
 from timeseries.helpers import trigger_backfill
 from timeseries.models import Dataset, MeasurementName
 
 
 class ActivateMeasurementsInteractor(BaseInteractor):
-    def validate(self, repo):
-        if not repo:
-            raise ValidationError("Repo not found")
-        if not settings.TIMESERIES_ENABLED:
-            raise ValidationError("Timeseries storage not enabled")
-
     @sync_to_async
     def execute(
         self, repo_name: str, owner_name: str, measurement_type: MeasurementName
-    ):
-        author = Owner.objects.filter(username=owner_name, service=self.service).first()
-        repo = (
-            Repository.objects.viewable_repos(self.current_owner)
-            .filter(author=author, name=repo_name, active=True)
-            .first()
+    ) -> Dataset:
+        if not settings.TIMESERIES_ENABLED:
+            raise ValidationError("Timeseries storage not enabled")
+
+        _owner, repo = self.resolve_owner_and_repo(
+            owner_name, repo_name, only_viewable=True, only_active=True
         )
-        self.validate(repo)
 
         dataset, created = Dataset.objects.get_or_create(
             name=measurement_type.value,

core/commands/repository/interactors/erase_repository.py

@@ -1,32 +1,14 @@
-from django.conf import settings
-
-import services.self_hosted as self_hosted
 from codecov.commands.base import BaseInteractor
-from codecov.commands.exceptions import 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
-from core.models import Repository
 from services.task.task import TaskService
 
 
 class EraseRepositoryInteractor(BaseInteractor):
-    def validate_owner(self, owner: Owner) -> None:
-        if not current_user_part_of_org(self.current_owner, owner):
-            raise Unauthorized()
-
-        if settings.IS_ENTERPRISE:
-            if not self_hosted.is_admin_owner(self.current_owner):
-                raise Unauthorized()
-        else:
-            if not owner.is_admin(self.current_owner):
-                raise Unauthorized()
-
     @sync_to_async
-    def execute(self, repo_name: str, owner: Owner) -> None:
-        self.validate_owner(owner)
-        repo = Repository.objects.filter(author_id=owner.pk, name=repo_name).first()
-        if not repo:
-            raise ValidationError("Repo not found")
+    def execute(self, owner_username: str, repo_name: str) -> None:
+        _owner, repo = self.resolve_owner_and_repo(
+            owner_username, repo_name, ensure_is_admin=True
+        )
+
         TaskService().delete_timeseries(repository_id=repo.repoid)
         TaskService().flush_repo(repository_id=repo.repoid)

core/commands/repository/interactors/regenerate_repository_token.py

@@ -1,23 +1,14 @@
 from codecov.commands.base import BaseInteractor
-from codecov.commands.exceptions import ValidationError
 from codecov.db import sync_to_async
-from codecov_auth.models import Owner, RepositoryToken
-from core.models import Repository
+from codecov_auth.models import RepositoryToken
 
 
 class RegenerateRepositoryTokenInteractor(BaseInteractor):
     @sync_to_async
     def execute(self, repo_name: str, owner_username: str, token_type: str):
-        author = Owner.objects.filter(
-            username=owner_username, service=self.service
-        ).first()
-        repo = (
-            Repository.objects.viewable_repos(self.current_owner)
-            .filter(author=author, name=repo_name, active=True)
-            .first()
+        _owner, repo = self.resolve_owner_and_repo(
+            owner_username, repo_name, only_viewable=True, only_active=True
         )
-        if not repo:
-            raise ValidationError("Repo not found")
 
         token, created = RepositoryToken.objects.get_or_create(
             repository_id=repo.repoid, token_type=token_type

core/commands/repository/interactors/regenerate_repository_upload_token.py

@@ -1,25 +1,16 @@
 import uuid
 
 from codecov.commands.base import BaseInteractor
-from codecov.commands.exceptions import ValidationError
 from codecov.db import sync_to_async
-from codecov_auth.models import Owner
-from core.models import Repository
 
 
 class RegenerateRepositoryUploadTokenInteractor(BaseInteractor):
     @sync_to_async
     def execute(self, repo_name: str, owner_username: str) -> uuid.UUID:
-        author = Owner.objects.filter(
-            username=owner_username, service=self.service
-        ).first()
-        repo = (
-            Repository.objects.viewable_repos(self.current_owner)
-            .filter(author=author, name=repo_name)
-            .first()
+        _owner, repo = self.resolve_owner_and_repo(
+            owner_username, repo_name, only_viewable=True
         )
-        if not repo:
-            raise ValidationError("Repo not found")
+
         repo.upload_token = uuid.uuid4()
         repo.save()
         return repo.upload_token

core/commands/repository/interactors/tests/test_erase_repository.py

@@ -15,14 +15,12 @@ def setUp(self):
 
     def execute_unauthorized_owner(self):
         return EraseRepositoryInteractor(self.owner, "github").execute(
-            repo_name="repo-1",
-            owner=self.random_user,
+            self.random_user.username, "repo-1"
         )
 
     def execute_user_not_admin(self):
         return EraseRepositoryInteractor(self.non_admin_user, "github").execute(
-            repo_name="repo-1",
-            owner=self.owner,
+            self.owner.username, "repo-1"
         )
 
     async def test_when_validation_error_unauthorized_owner_not_part_of_org(self):

core/commands/repository/interactors/update_bundle_cache_config.py

@@ -8,17 +8,13 @@
 from codecov.commands.base import BaseInteractor
 from codecov.commands.exceptions import ValidationError
 from codecov.db import sync_to_async
-from codecov_auth.models import Owner
 from core.models import Repository
 
 
 class UpdateBundleCacheConfigInteractor(BaseInteractor):
     def validate(
         self, repo: Repository, cache_config: List[Dict[str, str | bool]]
     ) -> None:
-        if not repo:
-            raise ValidationError("Repo not found")
-
         # Find any missing bundle names
         bundle_names = [
             bundle["bundle_name"]
@@ -44,13 +40,8 @@ def execute(
         repo_name: str,
         cache_config: List[Dict[str, str | bool]],
     ) -> List[Dict[str, str | bool]]:
-        author = Owner.objects.filter(
-            username=owner_username, service=self.service
-        ).first()
-        repo = (
-            Repository.objects.viewable_repos(self.current_owner)
-            .filter(author=author, name=repo_name)
-            .first()
+        _owner, repo = self.resolve_owner_and_repo(
+            owner_username, repo_name, only_viewable=True
         )
 
         self.validate(repo, cache_config)

core/commands/repository/repository.py

@@ -79,8 +79,10 @@ def activate_measurements(
             repo_name, owner_name, measurement_type
         )
 
-    def erase_repository(self, repo_name: str, owner: Owner) -> None:
-        return self.get_interactor(EraseRepositoryInteractor).execute(repo_name, owner)
+    def erase_repository(self, owner_username: str, repo_name: str) -> None:
+        return self.get_interactor(EraseRepositoryInteractor).execute(
+            owner_username, repo_name
+        )
 
     def encode_secret_string(self, owner: Owner, repo_name: str, value: str) -> str:
         return self.get_interactor(EncodeSecretStringInteractor).execute(

graphql_api/types/mutation/erase_repository/erase_repository.py

@@ -18,7 +18,9 @@ async def resolve_erase_repository(
     command = info.context["executor"].get_command("repository")
     current_owner = info.context["request"].current_owner
     repo_name = input.get("repo_name")
-    await command.erase_repository(repo_name=repo_name, owner=current_owner)
+    # TODO: change the graphql mutation to allow working on other owners
+    owner_username = current_owner.username
+    await command.erase_repository(owner_username, repo_name)
     return None