harbor/tests/apitests/python/test_assign_role_to_ldap_group.py

# coding: utf-8

"""
    Harbor API

    These APIs provide services for manipulating Harbor project.

    OpenAPI spec version: 1.4.0

    Generated by: https://github.com/swagger-api/swagger-codegen.git
"""


from __future__ import absolute_import
import os
import sys
sys.path.append(os.environ["SWAGGER_CLIENT_PATH"])

import unittest
import testutils
import docker

from testutils import ADMIN_CLIENT
from swagger_client.models.project_member import ProjectMember
from swagger_client.models.user_group import UserGroup
from swagger_client.models.configurations import Configurations
from library.project import Project
from library.base import _assert_status_code
from library.base import _random_name
from v2_swagger_client.rest import ApiException
from pprint import pprint

#Testcase
#3-07-LDAP usergroup manage project group members
class TestAssignRoleToLdapGroup(unittest.TestCase):
    harbor_host = os.environ["HARBOR_HOST"]
    """AssignRoleToLdapGroup unit test stubs"""
    product_api = testutils.GetProductApi("admin", "Harbor12345")
    repository_api = testutils.GetRepositoryApi("admin", "Harbor12345")
    project_id = 0
    docker_client = docker.from_env()
    _project_name = _random_name("test-ldap-group")

    def setUp(self):
        self.project = Project()

        #login with admin, create a project and assign role to ldap group
        result = self.product_api.configurations_put(configurations=Configurations(ldap_filter="", ldap_group_attribute_name="cn", ldap_group_base_dn="ou=groups,dc=example,dc=com", ldap_group_search_filter="objectclass=groupOfNames", ldap_group_search_scope=2))
        pprint(result)
        cfgs = self.product_api.configurations_get()
        pprint(cfgs)
        result = self.project.create_project(self._project_name, dict(public="false"))
        pprint(result)

        projs = self.project.get_projects(dict(name = self._project_name))
        if len(projs)>0 :
            project = projs[0]
            self.project_id = project.project_id

        # asign role to project with dn
        group_dn = "cn=harbor_admin,ou=groups,dc=example,dc=com"
        projectmember = ProjectMember()
        projectmember.role_id = 1
        projectmember.member_group = UserGroup(ldap_group_dn=group_dn)

        result = self.product_api.projects_project_id_members_post( project_id=self.project_id, project_member=projectmember )
        pprint(result)

        group_dn = "cn=harbor_dev,ou=groups,dc=example,dc=com"
        projectmember = ProjectMember()
        projectmember.role_id = 2
        projectmember.member_group = UserGroup(ldap_group_dn=group_dn)

        result = self.product_api.projects_project_id_members_post( project_id=self.project_id, project_member=projectmember )
        pprint(result)

        group_dn = "cn=harbor_guest,ou=groups,dc=example,dc=com"
        projectmember = ProjectMember()
        projectmember.role_id = 3
        projectmember.member_group = UserGroup(ldap_group_dn=group_dn)

        result = self.product_api.projects_project_id_members_post( project_id=self.project_id, project_member=projectmember )
        pprint(result)

    def tearDown(self):
        if self.project_id > 0 :
            # delete images in project
            result = self.repository_api.delete_repository(self._project_name, "busybox")
            pprint(result)
            result = self.repository_api.delete_repository(self._project_name, "busyboxdev")
            pprint(result)
            self.project.delete_project(self.project_id)

    def testAssignRoleToLdapGroup(self):
        """Test AssignRoleToLdapGroup"""
        admin_product_api = Project("admin_user", "zhu88jie")
        projects = admin_product_api.get_projects(dict(name=self._project_name))
        self.assertTrue(len(projects) == 1)
        self.assertEqual(1, projects[0].current_user_role_id)

        dev_product_api = Project("dev_user", "zhu88jie")
        projects = dev_product_api.get_projects(dict(name=self._project_name))
        self.assertTrue(len(projects) == 1)
        self.assertEqual(2, projects[0].current_user_role_id)

        guest_product_api = Project("guest_user", "zhu88jie")
        projects = guest_product_api.get_projects(dict(name=self._project_name))
        self.assertTrue(len(projects) == 1)
        self.assertEqual(3, projects[0].current_user_role_id)

        self.dockerCmdLoginAdmin(username="admin_user", password="zhu88jie")
        self.dockerCmdLoginDev(username="dev_user", password="zhu88jie")
        self.dockerCmdLoginGuest(username="guest_user", password="zhu88jie")

        self.assertTrue(self.queryUserLogs(username="admin_user", password="zhu88jie")>0, "admin user can see logs")
        self.assertTrue(self.queryUserLogs(username="dev_user", password="zhu88jie")>0, "dev user can see logs")
        self.assertTrue(self.queryUserLogs(username="guest_user", password="zhu88jie")>0, "guest user can see logs")
        self.assertTrue(self.queryUserLogs(username="test", password="123456", status_code=403)==0, "test user can not see any logs")

    # admin user can push, pull images
    def dockerCmdLoginAdmin(self, username, password):
        pprint(self.docker_client.info())
        self.docker_client.login(username=username, password=password, registry=self.harbor_host)
        self.docker_client.images.pull("busybox:latest")
        image = self.docker_client.images.get("busybox:latest")
        image.tag(repository=self.harbor_host+"/"+self._project_name+"/busybox", tag="latest")
        output = self.docker_client.images.push(repository=self.harbor_host+"/"+self._project_name+"/busybox", tag="latest")
        if output.find("error")>0 :
            self.fail("Should not fail to push image for admin_user")
        self.docker_client.images.pull(repository=self.harbor_host+"/"+self._project_name+"/busybox", tag="latest")

    # dev user can push, pull images
    def dockerCmdLoginDev(self, username, password, harbor_server=harbor_host):
        self.docker_client.login(username=username, password=password, registry=self.harbor_host)
        self.docker_client.images.pull("busybox:latest")
        image = self.docker_client.images.get("busybox:latest")
        image.tag(repository=self.harbor_host+"/"+self._project_name+"/busyboxdev", tag="latest")
        output = self.docker_client.images.push(repository=self.harbor_host+"/"+self._project_name+"/busyboxdev", tag="latest")
        if output.find("error") >0 :
            self.fail("Should not fail to push images for dev_user")

    # guest user can pull images
    def dockerCmdLoginGuest(self, username, password, harbor_server=harbor_host):
        self.docker_client.login(username=username, password=password, registry=self.harbor_host)
        self.docker_client.images.pull("busybox:latest")
        image = self.docker_client.images.get("busybox:latest")
        image.tag(repository=self.harbor_host+"/"+self._project_name+"/busyboxguest", tag="latest")
        output = self.docker_client.images.push(repository=self.harbor_host+"1/"+self._project_name+"/busyboxguest", tag="latest")
        if output.find("error")<0 :
            self.fail("Should failed to push image for guest user")
        self.docker_client.images.pull(repository=self.harbor_host+"/"+self._project_name+"/busybox", tag="latest")

    # check can see his log in current project
    def queryUserLogs(self, username, password, status_code=200):
        client=dict(endpoint = ADMIN_CLIENT["endpoint"], username = username, password = password)
        try:
            logs = self.project.get_project_log(self._project_name, status_code, **client)
            count = 0
            for log in list(logs):
                count = count + 1
            return count
        except ApiException as e:
            _assert_status_code(status_code, e.status)
            return 0

if __name__ == '__main__':
    unittest.main()