Add additional tests for repo visibility and further simplify the query for perf

data/model/repository.py

@@ -14,6 +14,10 @@ from data.database import (Repository, Namespace, RepositoryTag, Star, Image, Im
 logger = logging.getLogger(__name__)
 
 
+def get_public_repo_visibility():
+  return _basequery.get_public_repo_visibility()
+
+
 def create_repository(namespace, name, creating_user, visibility='private'):
   private = Visibility.get(name=visibility)
   namespace_user = User.get(username=namespace)
@@ -241,25 +245,9 @@ def get_visible_repositories(username, namespace=None, page=None, limit=None, in
   if not include_public and not username:
     return []
 
-  fields = [Repository.name, Repository.id, Repository.description, Visibility.name,
-            Namespace.username]
-
-  query = _visible_repository_query(username=username, page=page,
-                                    limit=limit, namespace=namespace, include_public=include_public,
-                                    select_models=fields)
-
-  if limit:
-    query = query.limit(limit)
-
-  return query
-
-
-def _visible_repository_query(username=None, include_public=True, limit=None,
-                              page=None, namespace=None, select_models=[]):
   query = (Repository
-           .select(*select_models)  # MySQL/RDS complains is there are selected models for counts.
+           .select(Repository.name, Repository.id, Repository.description, Namespace.username)
            .distinct()
-           .join(Visibility)
            .switch(Repository)
            .join(Namespace, on=(Repository.namespace_user == Namespace.id))
            .switch(Repository)

endpoints/api/repository.py

@@ -203,7 +203,7 @@ class RepositoryList(ApiResource):
         'namespace': repo_obj.namespace_user.username,
         'name': repo_obj.name,
         'description': repo_obj.description,
-        'is_public': repo_obj.visibility.name == 'public'
+        'is_public': repo_obj.visibility_id == model.repository.get_public_repo_visibility(),
       }
 
       repo_id = repo_obj.id

test/test_visible_repos.py

@@ -0,0 +1,98 @@
+import unittest
+
+from app import app
+from initdb import setup_database_for_testing, finished_database_for_testing
+from data import model
+
+NO_ACCESS_USER = 'freshuser'
+READ_ACCESS_USER = 'reader'
+ADMIN_ACCESS_USER = 'devtable'
+PUBLIC_USER = 'public'
+RANDOM_USER = 'randomuser'
+OUTSIDE_ORG_USER = 'outsideorg'
+
+ADMIN_ROBOT_USER = 'devtable+dtrobot'
+
+ORGANIZATION = 'buynlarge'
+
+SIMPLE_REPO = 'simple'
+PUBLIC_REPO = 'publicrepo'
+RANDOM_REPO = 'randomrepo'
+
+OUTSIDE_ORG_REPO = 'coolrepo'
+
+ORG_REPO = 'orgrepo'
+ANOTHER_ORG_REPO = 'anotherorgrepo'
+
+# Note: The shared repo has devtable as admin, public as a writer and reader as a reader.
+SHARED_REPO = 'shared'
+
+class TestVisibleRepositories(unittest.TestCase):
+  def setUp(self):
+    setup_database_for_testing(self)
+    self.app = app.test_client()
+    self.ctx = app.test_request_context()
+    self.ctx.__enter__()
+
+  def tearDown(self):
+    finished_database_for_testing(self)
+    self.ctx.__exit__(True, None, None)
+
+  def assertDoesNotHaveRepo(self, username, name):
+    repos = list(model.repository.get_visible_repositories(username))
+    names = [repo.name for repo in repos]
+    self.assertNotIn(name, names)
+
+  def assertHasRepo(self, username, name):
+    repos = list(model.repository.get_visible_repositories(username))
+    names = [repo.name for repo in repos]
+    self.assertIn(name, names)
+
+  def test_noaccess(self):
+    repos = list(model.repository.get_visible_repositories(NO_ACCESS_USER))
+    names = [repo.name for repo in repos]
+    self.assertEquals(0, len(names))
+
+    # Try retrieving public repos now.
+    repos = list(model.repository.get_visible_repositories(NO_ACCESS_USER, include_public=True))
+    names = [repo.name for repo in repos]
+    self.assertIn(PUBLIC_REPO, names)
+
+
+  def test_public(self):
+    self.assertHasRepo(PUBLIC_USER, PUBLIC_REPO)
+    self.assertHasRepo(PUBLIC_USER, SHARED_REPO)
+
+    self.assertDoesNotHaveRepo(PUBLIC_USER, SIMPLE_REPO)
+    self.assertDoesNotHaveRepo(PUBLIC_USER, RANDOM_REPO)
+    self.assertDoesNotHaveRepo(PUBLIC_USER, OUTSIDE_ORG_REPO)
+
+  def test_reader(self):
+    self.assertHasRepo(READ_ACCESS_USER, SHARED_REPO)
+    self.assertHasRepo(READ_ACCESS_USER, ORG_REPO)
+
+    self.assertDoesNotHaveRepo(READ_ACCESS_USER, SIMPLE_REPO)
+    self.assertDoesNotHaveRepo(READ_ACCESS_USER, RANDOM_REPO)
+    self.assertDoesNotHaveRepo(READ_ACCESS_USER, OUTSIDE_ORG_REPO)
+    self.assertDoesNotHaveRepo(READ_ACCESS_USER, PUBLIC_REPO)
+
+  def test_random(self):
+    self.assertHasRepo(RANDOM_USER, RANDOM_REPO)
+
+    self.assertDoesNotHaveRepo(RANDOM_USER, SIMPLE_REPO)
+    self.assertDoesNotHaveRepo(RANDOM_USER, SHARED_REPO)
+    self.assertDoesNotHaveRepo(RANDOM_USER, ORG_REPO)
+    self.assertDoesNotHaveRepo(RANDOM_USER, ANOTHER_ORG_REPO)
+    self.assertDoesNotHaveRepo(RANDOM_USER, PUBLIC_REPO)
+
+  def test_admin(self):
+    self.assertHasRepo(ADMIN_ACCESS_USER, SIMPLE_REPO)
+    self.assertHasRepo(ADMIN_ACCESS_USER, SHARED_REPO)
+
+    self.assertHasRepo(ADMIN_ACCESS_USER, ORG_REPO)
+    self.assertHasRepo(ADMIN_ACCESS_USER, ANOTHER_ORG_REPO)
+
+    self.assertDoesNotHaveRepo(ADMIN_ACCESS_USER, OUTSIDE_ORG_REPO)
+
+if __name__ == '__main__':
+  unittest.main()