Tests

Author: rohitvinnakota-codecov

graphql_api/actions/repository.py

@@ -6,53 +6,44 @@
 from shared.django_apps.codecov_auth.models import GithubAppInstallation, Owner
 from shared.django_apps.core.models import Repository
 
-from graphql_api.types.owner.owner import AI_FEATURES_GH_APP_ID
+from utils.config import get_config
 
 log = logging.getLogger(__name__)
+AI_FEATURES_GH_APP_ID = get_config("github", "ai_features_app_id")
 
 
-def apply_filters_to_queryset(
-    queryset: QuerySet, filters: dict[str, Any] | None, owner: Owner
-) -> QuerySet:
+def basic_filters(queryset: QuerySet, filters: dict[str, Any]) -> QuerySet:
     filters = filters or {}
-    term = filters.get("term")
-    active = filters.get("active")
-    activated = filters.get("activated")
-    repo_names = filters.get("repo_names")
-    is_public = filters.get("is_public")
-    ai_enabled = filters.get("ai_enabled")
-
-    if repo_names:
+    if repo_names := filters.get("repo_names"):
         queryset = queryset.filter(name__in=repo_names)
-    if term:
+    if term := filters.get("term"):
         queryset = queryset.filter(name__contains=term)
-    if activated is not None:
-        queryset = queryset.filter(activated=activated)
-    if active is not None:
-        queryset = queryset.filter(active=active)
-    if is_public is not None:
-        queryset = queryset.filter(private=not is_public)
-    if is_public is not None:
-        queryset = queryset.filter(private=not is_public)
-    if ai_enabled is not None:
-        queryset = filter_queryset_by_ai_enabled_repos(queryset, owner)
+    for field in ("activated", "active"):
+        if filters.get(field) is not None:
+            queryset = queryset.filter(**{field: filters[field]})
+    if filters.get("is_public") is not None:
+        queryset = queryset.filter(private=not filters["is_public"])
     return queryset
 
 
 def filter_queryset_by_ai_enabled_repos(queryset: QuerySet, owner: Owner) -> QuerySet:
-    ai_features_app_install = GithubAppInstallation.objects.filter(
+    install = GithubAppInstallation.objects.filter(
         app_id=AI_FEATURES_GH_APP_ID, owner=owner
     ).first()
-
-    if not ai_features_app_install:
+    if not install:
         return Repository.objects.none()
+    if install.repository_service_ids:
+        queryset = queryset.filter(service_id__in=install.repository_service_ids)
+    return queryset
 
-    if ai_features_app_install.repository_service_ids:
-        queryset = queryset.filter(
-            service_id__in=ai_features_app_install.repository_service_ids
-        )
 
-    return queryset
+def apply_filters(
+    queryset: QuerySet, filters: dict[str, Any] | None, owner: Owner
+) -> QuerySet:
+    filters = filters or {}
+    if filters.get("ai_enabled"):
+        return filter_queryset_by_ai_enabled_repos(queryset, owner)
+    return basic_filters(queryset, filters)
 
 
 @sentry_sdk.trace
@@ -63,23 +54,14 @@ def list_repository_for_owner(
     okta_account_auths: list[int],
     exclude_okta_enforced_repos: bool = True,
 ) -> QuerySet:
-    queryset = Repository.objects.viewable_repos(current_owner)
-
-    ai_enabled_filter = filters.get("ai_enabled")
-
-    if ai_enabled_filter:
-        return filter_queryset_by_ai_enabled_repos(queryset, owner)
-
+    filters = filters or {}
+    qs = Repository.objects.viewable_repos(current_owner)
+    if filters.get("ai_enabled"):
+        return filter_queryset_by_ai_enabled_repos(qs, owner)
     if exclude_okta_enforced_repos:
-        queryset = queryset.exclude_accounts_enforced_okta(okta_account_auths)
-
-    if not ai_enabled_filter:
-        queryset = (
-            queryset.with_recent_coverage().with_latest_commit_at().filter(author=owner)
-        )
-
-    queryset = apply_filters_to_queryset(queryset, filters, owner)
-    return queryset
+        qs = qs.exclude_accounts_enforced_okta(okta_account_auths)
+    qs = qs.with_recent_coverage().with_latest_commit_at().filter(author=owner)
+    return basic_filters(qs, filters)
 
 
 @sentry_sdk.trace
@@ -89,16 +71,14 @@ def search_repos(
     okta_account_auths: list[int],
     exclude_okta_enforced_repos: bool = True,
 ) -> QuerySet:
-    authors_from = [current_owner.ownerid] + (current_owner.organizations or [])
-    queryset = Repository.objects.viewable_repos(current_owner)
-
+    filters = filters or {}
+    authors = [current_owner.ownerid] + (current_owner.organizations or [])
+    qs = Repository.objects.viewable_repos(current_owner)
     if exclude_okta_enforced_repos:
-        queryset = queryset.exclude_accounts_enforced_okta(okta_account_auths)
-
-    queryset = (
-        queryset.with_recent_coverage()
+        qs = qs.exclude_accounts_enforced_okta(okta_account_auths)
+    qs = (
+        qs.with_recent_coverage()
         .with_latest_commit_at()
-        .filter(author__ownerid__in=authors_from)
+        .filter(author__ownerid__in=authors)
     )
-    queryset = apply_filters_to_queryset(queryset, filters)
-    return queryset
+    return apply_filters(qs, filters, current_owner)

graphql_api/tests/test_owner.py

@@ -1209,3 +1209,32 @@ def test_fetch_owner_with_no_service(self):
         """ % (current_org.username)
         data = self.gql_request(query, owner=current_org, provider="", with_errors=True)
         assert data == {"data": {"owner": None}}
+
+    def test_fetch_repositories_ai_features_enabled(self):
+        ai_app_installation = GithubAppInstallation(
+            name="ai-features",
+            owner=self.owner,
+            repository_service_ids=[],
+            installation_id=12345,
+        )
+
+        ai_app_installation.save()
+        query = query_repositories % (
+            self.owner.username,
+            "(filters: { aiEnabled: true })",
+            "",
+        )
+
+        data = self.gql_request(query, owner=self.owner)
+        repos = paginate_connection(data["owner"]["repositories"])
+        assert repos == [{"name": "a"}, {"name": "b"}]
+
+    def test_fetch_repositories_ai_features_enabled_no_app_install(self):
+        query = query_repositories % (
+            self.owner.username,
+            "(filters: { aiEnabled: true })",
+            "",
+        )
+        data = self.gql_request(query, owner=self.owner)
+        repos = paginate_connection(data["owner"]["repositories"])
+        assert repos == []

utils/test_utils.py

@@ -68,4 +68,4 @@ def setUp(self) -> None:
         self.apps = executor.loader.project_state(self.migrate_to).apps
 
     def setUpBeforeMigration(self, apps: Any) -> None:
-        pass
+        pass