#! /usr/bin/env python

import os
import sys
import unittest
from irods.models import User
from irods.exception import UserDoesNotExist, ResourceDoesNotExist, SYS_NO_API_PRIV
from irods.session import iRODSSession
from irods.resource import iRODSResource
import irods.test.helpers as helpers
import irods.keywords as kw


class TestAdmin(unittest.TestCase):
    """Suite of tests on admin operations"""

    # test data
    new_user_name = "bobby"
    new_user_type = "rodsuser"

    def setUp(self):
        self.sess = helpers.make_session()

    def tearDown(self):
        """Close connections"""
        self.sess.cleanup()

    def test_create_delete_local_user(self):
        # user should not be already present
        with self.assertRaises(UserDoesNotExist):
            self.sess.users.get(self.new_user_name)

        # create user
        user = self.sess.users.create(self.new_user_name, self.new_user_type)

        # assertions
        self.assertEqual(user.name, self.new_user_name)
        self.assertEqual(user.zone, self.sess.zone)
        self.assertEqual(
            repr(user), "<iRODSUser {id} {name} {type} {zone}>".format(**vars(user))
        )

        # delete user
        user.remove()

        # user should be gone
        with self.assertRaises(UserDoesNotExist):
            self.sess.users.get(self.new_user_name)

    def test_create_delete_user_zone(self):
        # user should not be already present
        with self.assertRaises(UserDoesNotExist):
            self.sess.users.get(self.new_user_name, self.sess.zone)

        # create user
        user = self.sess.users.create(
            self.new_user_name, self.new_user_type, self.sess.zone
        )

        # assertions
        self.assertEqual(user.name, self.new_user_name)
        self.assertEqual(user.zone, self.sess.zone)

        # delete user
        user.remove()

        # user should be gone
        with self.assertRaises(UserDoesNotExist):
            self.sess.users.get(self.new_user_name, self.sess.zone)

    def test_groupadmin_creates_group_and_unable_to_delete_group__374(self):
        # Have user with groupadmin
        user = self.sess.users.create(self.new_user_name, "groupadmin")
        self.assertEqual(user.name, self.new_user_name)

        # Have a password for login
        user_password = "bpass"
        user.modify("password", user_password)

        group_name = "coolgroup"

        # Create session as user
        with iRODSSession(
            port=self.sess.port,
            zone=self.sess.zone,
            host=self.sess.host,
            user=self.new_user_name,
            password=user_password,
        ) as bobby:
            # Create group
            group = bobby.groups.create(group_name, group_admin=True)
            self.assertEqual(group.name, group_name)

            # groupadmin cannot remove groups!
            with self.assertRaises(SYS_NO_API_PRIV):
                group.remove()

        # Only an admin can remove groups!
        self.sess.groups.get(group_name).remove()
        user.remove()

    def test_admin_creates_and_deletes_group__374(self):
        group_name = "evencoolergroup"

        # Create and remove group
        group = self.sess.groups.create(group_name)
        self.assertEqual(group.name, group_name)

        group.remove()

    def test_modify_user_type(self):
        # make new regular user
        self.sess.users.create(self.new_user_name, self.new_user_type)

        # check type
        row = self.sess.query(User.type).filter(User.name == self.new_user_name).one()
        self.assertEqual(row[User.type], self.new_user_type)

        # change type to rodsadmin
        self.sess.users.modify(self.new_user_name, "type", "rodsadmin")

        # check type again
        row = self.sess.query(User.type).filter(User.name == self.new_user_name).one()
        self.assertEqual(row[User.type], "rodsadmin")

        # delete user
        self.sess.users.remove(self.new_user_name)

        # user should be gone
        with self.assertRaises(UserDoesNotExist):
            self.sess.users.get(self.new_user_name)

    def test_modify_user_type_with_zone(self):
        # make new regular user
        self.sess.users.create(self.new_user_name, self.new_user_type)

        # check type
        row = self.sess.query(User.type).filter(User.name == self.new_user_name).one()
        self.assertEqual(row[User.type], self.new_user_type)

        # change type to rodsadmin
        self.sess.users.modify(
            "{}#{}".format(self.new_user_name, self.sess.zone), "type", "rodsadmin"
        )

        # check type again
        row = self.sess.query(User.type).filter(User.name == self.new_user_name).one()
        self.assertEqual(row[User.type], "rodsadmin")

        # delete user
        self.sess.users.remove(self.new_user_name)

        # user should be gone
        with self.assertRaises(UserDoesNotExist):
            self.sess.users.get(self.new_user_name)

    def test_make_compound_resource(self):
        if self.sess.server_version < (4, 0, 0):
            self.skipTest("For iRODS 4+")

        session = self.sess
        zone = self.sess.zone
        username = self.sess.username
        obj_path = "/{zone}/home/{username}/foo.txt".format(**locals())
        dummy_str = b"blah"

        # make compound resource
        comp = session.resources.create("comp_resc", "compound")

        # make 1st ufs resource
        resc_name = "ufs1"
        resc_type = "unixfilesystem"
        resc_host = self.sess.host
        resc_path = "/tmp/" + resc_name
        ufs1 = session.resources.create(resc_name, resc_type, resc_host, resc_path)

        # make 2nd ufs resource
        resc_name = "ufs2"
        resc_path = "/tmp/" + resc_name
        ufs2 = session.resources.create(resc_name, resc_type, resc_host, resc_path)

        # add children to compound resource
        session.resources.add_child(comp.name, ufs1.name, "archive")
        session.resources.add_child(comp.name, ufs2.name, "cache")

        obj = None

        try:
            # create object on compound resource
            obj = session.data_objects.create(obj_path, resource=comp.name)

            # write to object
            with obj.open("w+", **{kw.DEST_RESC_NAME_KW: comp.name}) as obj_desc:
                obj_desc.write(dummy_str)

            # refresh object
            obj = session.data_objects.get(obj_path)

            # check that we have 2 replicas
            self.assertEqual(len(obj.replicas), 2)
        finally:
            # remove object
            if obj:
                obj.unlink(force=True)

            # remove children from compound resource
            session.resources.remove_child(comp.name, ufs1.name)
            session.resources.remove_child(comp.name, ufs2.name)

            # remove resources
            ufs1.remove()
            ufs2.remove()
            comp.remove()

    def test_get_resource_children(self):
        if self.sess.server_version < (4, 0, 0):
            self.skipTest("For iRODS 4+")

        session = self.sess
        username = self.sess.username

        # make compound resource
        compound_resource = session.resources.create("comp_resc", "compound")

        # make 1st ufs resource
        resc_name = "ufs1"
        resc_type = "unixfilesystem"
        resc_host = self.sess.host
        resc_path = "/tmp/" + resc_name
        ufs1 = session.resources.create(resc_name, resc_type, resc_host, resc_path)

        # make 2nd ufs resource
        resc_name2 = "ufs2"
        resc_path = "/tmp/" + resc_name2
        ufs2 = session.resources.create(resc_name2, resc_type, resc_host, resc_path)

        # add children to compound resource
        session.resources.add_child(compound_resource.name, ufs1.name, "archive")
        session.resources.add_child(compound_resource.name, ufs2.name, "cache")

        # confirm number of children
        self.assertEqual(len(compound_resource.children), 2)

        # assertions on children
        for child in compound_resource.children:
            self.assertIsInstance(child, iRODSResource)
            self.assertIn(child.name, [resc_name, resc_name2])
            self.assertEqual(child.type, resc_type)
            self.assertEqual(child.location, resc_host)

            if session.server_version >= (4, 2, 0):
                self.assertIn(child.parent_context, ["archive", "cache"])

        # remove children from compound resource
        session.resources.remove_child(compound_resource.name, ufs1.name)
        session.resources.remove_child(compound_resource.name, ufs2.name)

        # remove resources
        ufs1.remove()
        ufs2.remove()
        compound_resource.remove()

    def test_resource_context_string(self):
        if self.sess.server_version < (4, 0, 0):
            self.skipTest("For iRODS 4+")

        session = self.sess
        zone = self.sess.zone
        username = self.sess.username
        context = {
            "S3_DEFAULT_HOSTNAME": "storage.example.com",
            "S3_AUTH_FILE": "/path/to/auth/file",
            "S3_STSDATE": "date",
            "obj_bucket": "my_bucket",
            "arch_bucket": "test_archive",
            "S3_WAIT_TIME_SEC": "1",
            "S3_PROTO": "HTTPS",
            "S3_RETRY_COUNT": "3",
        }

        # make a resource
        resc_name = "s3archive"
        resc_type = "s3"
        resc_host = self.sess.host
        resc_path = "/nobucket"
        s3 = session.resources.create(
            resc_name, resc_type, resc_host, resc_path, context
        )

        # verify context fields
        self.assertEqual(context, s3.context_fields)

        # modify resource context
        context["S3_PROTO"] = "HTTP"
        s3 = session.resources.modify(s3.name, "context", context)

        # verify context fields again
        self.assertEqual(context, s3.context_fields)

        # remove resource
        s3.remove()

    def test_make_ufs_resource(self):
        RESC_PATH_BASE = helpers.irods_shared_tmp_dir()
        if not (RESC_PATH_BASE) and not helpers.irods_session_host_local(self.sess):
            self.skipTest("for non-local server with shared tmp dir missing")
        # test data
        resc_name = "temporary_test_resource"
        if self.sess.server_version < (4, 0, 0):
            resc_type = "unix file system"
            resc_class = "cache"
        else:
            resc_type = "unixfilesystem"
            resc_class = ""
        resc_host = self.sess.host
        resc_path = "/tmp/" + resc_name
        dummy_str = b"blah"
        zone = self.sess.zone
        username = self.sess.username

        coll_path = "/{zone}/home/{username}/test_dir".format(**locals())
        obj_name = "test1"
        obj_path = "{coll_path}/{obj_name}".format(**locals())

        # make new resource
        self.sess.resources.create(
            resc_name, resc_type, resc_host, resc_path, resource_class=resc_class
        )

        # try invalid params
        with self.assertRaises(ResourceDoesNotExist):
            resource = self.sess.resources.get(resc_name, zone="invalid_zone")

        # retrieve resource
        resource = self.sess.resources.get(resc_name)

        # assertions
        self.assertEqual(resource.name, resc_name)
        self.assertEqual(
            repr(resource),
            "<iRODSResource {id} {name} {type}>".format(**vars(resource)),
        )

        # make test collection
        coll = self.sess.collections.create(coll_path)

        # create file on new resource
        obj = self.sess.data_objects.create(obj_path, resc_name)

        # write something to the file
        # (can omit use of DEST_RESC_NAME_KW on resolution of
        #  https://github.com/irods/irods/issues/5548 )
        with obj.open("w+", **{kw.DEST_RESC_NAME_KW: resc_name}) as obj_desc:
            obj_desc.write(dummy_str)

        # refresh object (size has changed)
        obj = self.sess.data_objects.get(obj_path)

        # checks on file
        self.assertEqual(obj.name, obj_name)
        self.assertEqual(obj.size, len(dummy_str))

        # delete test collection
        coll.remove(recurse=True, force=True)

        # test delete resource
        self.sess.resources.remove(resc_name, test=True)

        # delete resource for good
        self.sess.resources.remove(resc_name)

    def test_set_user_password(self):
        # make a new user
        username = self.new_user_name
        zone = self.sess.zone
        self.sess.users.create(self.new_user_name, self.new_user_type)

        # make a really horrible password
        new_password = """abc123!"#$%&'()*+,-./:;<=>?@[\]^_`{|}~Z"""
        self.sess.users.modify(username, "password", new_password)

        # open a session as the new user
        with iRODSSession(
            host=self.sess.host,
            port=self.sess.port,
            user=username,
            password=new_password,
            zone=self.sess.zone,
        ) as session:

            # do something that connects to the server
            session.users.get(username)

        # delete new user
        self.sess.users.remove(self.new_user_name)

        # user should be gone
        with self.assertRaises(UserDoesNotExist):
            self.sess.users.get(self.new_user_name)

    def test_set_user_comment(self):
        # make a new user
        self.sess.users.create(self.new_user_name, self.new_user_type)

        # modify user comment
        new_comment = """comment-abc123!"#$%&'()*+,-./:;<=>?@[\]^_{|}~Z"""  # omitting backtick due to #170
        self.sess.users.modify(self.new_user_name, "comment", new_comment)

        # check comment was modified
        new_user = self.sess.users.get(self.new_user_name)
        self.assertEqual(new_user.comment, new_comment)

        # delete new user
        self.sess.users.remove(self.new_user_name)

        # user should be gone
        with self.assertRaises(UserDoesNotExist):
            self.sess.users.get(self.new_user_name)

    def test_set_user_info(self):
        # make a new user
        self.sess.users.create(self.new_user_name, self.new_user_type)

        # modify user info
        new_info = """info-abc123!"#$%&'()*+,-./:;<=>?@[\]^_{|}~Z"""  # omitting backtick due to #170
        self.sess.users.modify(self.new_user_name, "info", new_info)

        # check info was modified
        new_user = self.sess.users.get(self.new_user_name)
        self.assertEqual(new_user.info, new_info)

        # delete new user
        self.sess.users.remove(self.new_user_name)

        # user should be gone
        with self.assertRaises(UserDoesNotExist):
            self.sess.users.get(self.new_user_name)


if __name__ == "__main__":
    # let the tests find the parent irods lib
    sys.path.insert(0, os.path.abspath("../.."))
    unittest.main()