Merge pull request devopshq#286 from zhan9san/feature/project-list

Add get_projects

artifactory.py

@@ -2087,7 +2087,10 @@ def _get_all(self, lazy: bool, url=None, key="name", cls=None):
         :param cls: Create objects of this class
         "return: A list of found objects
         """
-        request_url = self.drive + url
+        if cls is Project:
+            request_url = self.drive.rstrip("/artifactory") + url
+        else:
+            request_url = self.drive + url
         r = self.session.get(request_url, auth=self.auth)
         r.raise_for_status()
         response = r.json()
@@ -2140,6 +2143,16 @@ def get_permissions(self, lazy=False):
             url="/api/security/permissions", key="name", cls=PermissionTarget, lazy=lazy
         )
 
+    def get_projects(self, lazy=False):
+        """
+        Get all projects
+
+        :param lazy: `True` if we don't need anything except object's name
+        """
+        return self._get_all(
+            url="/access/api/v1/projects", key="project_key", cls=Project, lazy=lazy
+        )
+
 
 class ArtifactorySaaSPath(ArtifactoryPath):
     """Class for SaaS Artifactory"""

dohq_artifactory/admin.py

@@ -427,7 +427,7 @@ def revoke_for_all_users(self):
 
 
 class Group(AdminObject):
-    _uri = "groups"
+    _uri = "security/groups"
     _uri_deletion = "security/groups"
 
     def __init__(self, artifactory, name):

tests/unit/test_artifactory_path.py

@@ -11,6 +11,9 @@
 from artifactory import ArtifactoryPath
 from artifactory import quote_url
 from dohq_artifactory import ArtifactoryException
+from dohq_artifactory.admin import Group
+from dohq_artifactory.admin import Project
+from dohq_artifactory.admin import User
 
 
 class UtilTest(unittest.TestCase):
@@ -1158,5 +1161,229 @@ def test_from_aql_file(self):
         assert artifact.root == "/reponame/"
 
 
+class TestArtifactoryPathGetAll(unittest.TestCase):
+    # TODO: test repositories and permissions
+    def setUp(self):
+        self.arti = ArtifactoryPath("http://b.com/artifactory")
+        self.users_request_url = f"{self.arti.drive}/api/security/users"
+        self.users = [
+            {
+                "name": "user_1",
+                "uri": "http://b.com/artifactory/api/security/users/user_1",
+                "realm": "internal",
+            },
+            {
+                "name": "user_2",
+                "uri": "http://b.com/artifactory/api/security/users/user_2",
+                "realm": "internal",
+            },
+        ]
+        self.user_1 = {"name": "user_1", "email": "[email protected]"}
+        self.user_2 = {"name": "user_2", "email": "[email protected]"}
+
+        self.groups_request_url = f"{self.arti.drive}/api/security/groups"
+        self.groups = [
+            {
+                "name": "group_1",
+                "uri": "http://b.com/artifactory/api/security/groups/group_1",
+            },
+            {
+                "name": "group_2",
+                "uri": "http://b.com/artifactory/api/security/groups/group_2",
+            },
+        ]
+        self.group_1 = {
+            "name": "group_1",
+            "realm": "internal",
+        }
+        self.group_2 = {
+            "name": "group_2",
+            "realm": "internal",
+        }
+
+        self.projects_request_url = (
+            f"{self.arti.drive.rstrip('/artifactory')}/access/api/v1/projects"
+        )
+        self.projects = [
+            {
+                "project_key": "project_key_1",
+                "description": "description_1",
+            },
+            {
+                "project_key": "project_key_2",
+                "description": "description_2",
+            },
+        ]
+        self.project_1 = {
+            "project_key": "project_key_1",
+            "description": "description_1",
+            "admin_privileges": {},
+        }
+        self.project_2 = {
+            "project_key": "project_key_2",
+            "description": "description_2",
+            "admin_privileges": {},
+        }
+
+    def test_get_users(self):
+        with responses.RequestsMock() as rsps:
+            rsps.add(responses.GET, self.users_request_url, json=self.users, status=200)
+            rsps.add(
+                responses.GET,
+                f"{self.users_request_url}/user_1",
+                json=self.user_1,
+                status=200,
+            )
+            rsps.add(
+                responses.GET,
+                f"{self.users_request_url}/user_2",
+                json=self.user_2,
+                status=200,
+            )
+
+            results = self.arti.get_users(lazy=False)
+
+            for user in results:
+                self.assertIsInstance(user, User)
+            self.assertEqual(results[0].name, "user_1")
+            self.assertEqual(results[0].email, "[email protected]")
+            self.assertEqual(results[1].name, "user_2")
+            self.assertEqual(results[1].email, "[email protected]")
+
+            self.assertEqual(len(rsps.calls), 3)
+            self.assertEqual(rsps.calls[0].request.url, self.users_request_url)
+            self.assertEqual(
+                rsps.calls[1].request.url, f"{self.users_request_url}/user_1"
+            )
+            self.assertEqual(
+                rsps.calls[2].request.url, f"{self.users_request_url}/user_2"
+            )
+
+    def test_get_users_lazy(self):
+        with responses.RequestsMock() as rsps:
+            rsps.add(responses.GET, self.users_request_url, json=self.users, status=200)
+
+            results = self.arti.get_users(lazy=True)
+
+            for user in results:
+                self.assertIsInstance(user, User)
+            self.assertEqual(results[0].name, "user_1")
+            self.assertIsNone(results[0].email)
+            self.assertEqual(results[1].name, "user_2")
+            self.assertIsNone(results[1].email)
+
+            self.assertEqual(len(rsps.calls), 1)
+            self.assertEqual(rsps.calls[0].request.url, self.users_request_url)
+
+    def test_get_groups(self):
+        with responses.RequestsMock() as rsps:
+            rsps.add(
+                responses.GET, self.groups_request_url, json=self.groups, status=200
+            )
+            rsps.add(
+                responses.GET,
+                f"{self.groups_request_url}/group_1",
+                json=self.group_1,
+                status=200,
+            )
+            rsps.add(
+                responses.GET,
+                f"{self.groups_request_url}/group_2",
+                json=self.group_2,
+                status=200,
+            )
+
+            results = self.arti.get_groups(lazy=False)
+
+            for group in results:
+                self.assertIsInstance(group, Group)
+            self.assertEqual(results[0].name, "group_1")
+            self.assertEqual(results[0].realm, "internal")
+            self.assertEqual(results[1].name, "group_2")
+            self.assertEqual(results[1].realm, "internal")
+
+            self.assertEqual(len(rsps.calls), 3)
+            self.assertEqual(rsps.calls[0].request.url, self.groups_request_url)
+            self.assertEqual(
+                rsps.calls[1].request.url, f"{self.groups_request_url}/group_1"
+            )
+            self.assertEqual(
+                rsps.calls[2].request.url, f"{self.groups_request_url}/group_2"
+            )
+
+    def test_get_groups_lazy(self):
+        with responses.RequestsMock() as rsps:
+            rsps.add(
+                responses.GET, self.groups_request_url, json=self.groups, status=200
+            )
+
+            results = self.arti.get_groups(lazy=True)
+
+            for group in results:
+                self.assertIsInstance(group, Group)
+            self.assertEqual(results[0].name, "group_1")
+            self.assertEqual(results[0].realm, "artifactory")
+            self.assertEqual(results[1].name, "group_2")
+            self.assertEqual(results[1].realm, "artifactory")
+
+            self.assertEqual(len(rsps.calls), 1)
+            self.assertEqual(rsps.calls[0].request.url, self.groups_request_url)
+
+    def test_get_projects(self):
+        with responses.RequestsMock() as rsps:
+            rsps.add(
+                responses.GET, self.projects_request_url, json=self.projects, status=200
+            )
+
+            rsps.add(
+                responses.GET,
+                f"{self.projects_request_url}/project_key_1",
+                json=self.project_1,
+                status=200,
+            )
+            rsps.add(
+                responses.GET,
+                f"{self.projects_request_url}/project_key_2",
+                json=self.project_2,
+                status=200,
+            )
+
+            results = self.arti.get_projects(lazy=False)
+
+            for project in results:
+                self.assertIsInstance(project, Project)
+            self.assertEqual(results[0].project_key, "project_key_1")
+            self.assertEqual(results[0].description, "description_1")
+            self.assertEqual(results[1].project_key, "project_key_2")
+            self.assertEqual(results[1].description, "description_2")
+
+            self.assertEqual(len(rsps.calls), 3)
+            self.assertEqual(rsps.calls[0].request.url, self.projects_request_url)
+            self.assertEqual(
+                rsps.calls[1].request.url, f"{self.projects_request_url}/project_key_1"
+            )
+            self.assertEqual(
+                rsps.calls[2].request.url, f"{self.projects_request_url}/project_key_2"
+            )
+
+    def test_get_projects_lazy(self):
+        with responses.RequestsMock() as rsps:
+            rsps.add(
+                responses.GET, self.projects_request_url, json=self.projects, status=200
+            )
+
+            results = self.arti.get_projects(lazy=True)
+
+            for project in results:
+                self.assertIsInstance(project, Project)
+            self.assertEqual(results[0].project_key, "project_key_1")
+            self.assertEqual(results[0].description, "")
+            self.assertEqual(results[1].project_key, "project_key_2")
+            self.assertEqual(results[1].description, "")
+
+            self.assertEqual(len(rsps.calls), 1)
+            self.assertEqual(rsps.calls[0].request.url, self.projects_request_url)
+
+
 if __name__ == "__main__":
     unittest.main()