Add search tests

data/model/legacy.py

@@ -681,8 +681,12 @@ def get_user_or_org_by_customer_id(customer_id):
 
 def get_matching_user_entities(entity_prefix, user):
   matching_user_orgs = ((User.username ** (entity_prefix + '%')) & (User.robot == False))
+
+  if user is not None:
     matching_robots = ((User.username ** (user.username + '+%' + entity_prefix + '%')) &
                        (User.robot == True))
+  else:
+    matching_robots = False
 
   query = (User.select()
                .where(matching_user_orgs | matching_robots)

endpoints/api/search.py

@@ -113,7 +113,10 @@ class ConductSearch(ApiResource):
   @nickname('conductSearch')
   def get(self, args):
     """ Get a list of entities and resources that match the specified query. """
-    prefix = args['query']
+    query = args['query']
+    if not query:
+      return {'results': []}
+
     username = None
     results = []
 
@@ -145,7 +148,7 @@ class ConductSearch(ApiResource):
       # Find the matching teams where the user is a member.
       encountered_teams = set()
 
-      matching_teams = model.get_matching_user_teams(prefix, get_authenticated_user())
+      matching_teams = model.get_matching_user_teams(query, get_authenticated_user())
       for team in matching_teams:
         if team.id in encountered_teams:
           continue
@@ -162,7 +165,7 @@ class ConductSearch(ApiResource):
         })
 
       # Find matching teams in orgs admined by the user.
-      matching_teams = model.get_matching_admined_teams(prefix, get_authenticated_user())
+      matching_teams = model.get_matching_admined_teams(query, get_authenticated_user())
       for team in matching_teams:
         if team.id in encountered_teams:
           continue
@@ -180,7 +183,7 @@ class ConductSearch(ApiResource):
 
 
     # Find the matching repositories.
-    matching_repos = model.get_matching_repositories(prefix, username)
+    matching_repos = model.get_matching_repositories(query, username)
     matching_repo_counts = {t[0]: t[1] for t in model.get_repository_pull_counts(matching_repos)}
 
     for repo in matching_repos:
@@ -195,12 +198,12 @@ class ConductSearch(ApiResource):
       })
 
     # Find the matching users, robots and organizations.
-    matching_entities = model.get_matching_user_entities(prefix, get_authenticated_user())
+    matching_entities = model.get_matching_user_entities(query, get_authenticated_user())
     for entity in matching_entities:
       results.append(entity_view(entity))
 
     for result in results:
-      result['score'] = result['score'] * liquidmetal.score(result['name'], prefix)
+      result['score'] = result['score'] * liquidmetal.score(result['name'], query)
 
     return {'results': sorted(results, key=itemgetter('score'), reverse=True)}
 

test/test_api_usage.py

@@ -15,7 +15,7 @@ from data import model, database
 
 from endpoints.api.team import TeamMember, TeamMemberList, TeamMemberInvite, OrganizationTeam
 from endpoints.api.tag import RepositoryTagImages, RepositoryTag
-from endpoints.api.search import FindRepositories, EntitySearch
+from endpoints.api.search import FindRepositories, EntitySearch, ConductSearch
 from endpoints.api.image import RepositoryImage, RepositoryImageList
 from endpoints.api.build import (RepositoryBuildStatus, RepositoryBuildLogs, RepositoryBuildList,
                                  RepositoryBuildResource)
@@ -472,6 +472,82 @@ class TestSignout(ApiTestCase):
     self.getJsonResponse(User, expected_code=401)
 
 
+class TestConductSearch(ApiTestCase):
+  def test_noaccess(self):
+    self.login(NO_ACCESS_USER)
+
+    json = self.getJsonResponse(ConductSearch,
+                                params=dict(query='read'))
+
+    self.assertEquals(1, len(json['results']))
+    self.assertEquals(json['results'][0]['kind'], 'user')
+    self.assertEquals(json['results'][0]['name'], 'reader')
+
+    json = self.getJsonResponse(ConductSearch,
+                                params=dict(query='owners'))
+
+    self.assertEquals(0, len(json['results']))
+
+
+  def test_nouser(self):
+    json = self.getJsonResponse(ConductSearch,
+                                params=dict(query='read'))
+
+    self.assertEquals(1, len(json['results']))
+    self.assertEquals(json['results'][0]['kind'], 'user')
+    self.assertEquals(json['results'][0]['name'], 'reader')
+
+    json = self.getJsonResponse(ConductSearch,
+                                params=dict(query='public'))
+
+    self.assertEquals(2, len(json['results']))
+    self.assertEquals(json['results'][0]['kind'], 'user')
+    self.assertEquals(json['results'][0]['name'], 'public')
+
+    self.assertEquals(json['results'][1]['kind'], 'repository')
+    self.assertEquals(json['results'][1]['name'], 'publicrepo')
+
+    json = self.getJsonResponse(ConductSearch,
+                                params=dict(query='owners'))
+
+    self.assertEquals(0, len(json['results']))
+
+
+  def test_orgmember(self):
+    self.login(READ_ACCESS_USER)
+
+    json = self.getJsonResponse(ConductSearch,
+                                params=dict(query='owners'))
+
+    self.assertEquals(0, len(json['results']))
+
+    json = self.getJsonResponse(ConductSearch,
+                                params=dict(query='readers'))
+
+    self.assertEquals(1, len(json['results']))
+    self.assertEquals(json['results'][0]['kind'], 'team')
+    self.assertEquals(json['results'][0]['name'], 'readers')
+
+
+  def test_orgadmin(self):
+    self.login(ADMIN_ACCESS_USER)
+
+    json = self.getJsonResponse(ConductSearch,
+                                params=dict(query='owners'))
+
+    self.assertEquals(1, len(json['results']))
+    self.assertEquals(json['results'][0]['kind'], 'team')
+    self.assertEquals(json['results'][0]['name'], 'owners')
+
+    json = self.getJsonResponse(ConductSearch,
+                                params=dict(query='readers'))
+
+    self.assertEquals(1, len(json['results']))
+    self.assertEquals(json['results'][0]['kind'], 'team')
+    self.assertEquals(json['results'][0]['name'], 'readers')
+
+
+
 class TestGetMatchingEntities(ApiTestCase):
   def test_notinorg(self):
     self.login(NO_ACCESS_USER)