#! /usr/bin/env python

import os
import sys
import unittest
import time
import calendar

import irods.test.helpers as helpers
import tempfile
from irods.session import iRODSSession
import irods.exception as ex
import irods.keywords as kw
from irods.ticket import Ticket
from irods.models import TicketQuery, DataObject, Collection


# As with most of the modules in this test suite, session objects created via
# make_session() are implicitly agents of a rodsadmin unless otherwise indicated.
# Counterexamples within this module shall be obvious as they are instantiated by
# the login() method, and always tied to one of the traditional rodsuser names
# widely used in iRODS test suites, ie. 'alice' or 'bob'.


def gmtime_to_timestamp(gmt_struct):
    return (
        "{0.tm_year:04d}-{0.tm_mon:02d}-{0.tm_mday:02d}."
        "{0.tm_hour:02d}:{0.tm_min:02d}:{0.tm_sec:02d}".format(gmt_struct)
    )


def delete_my_tickets(session):
    my_userid = session.users.get(session.username).id
    my_tickets = session.query(TicketQuery.Ticket).filter(
        TicketQuery.Ticket.user_id == my_userid
    )
    for res in my_tickets:
        Ticket(session, result=res).delete()


class TestRodsUserTicketOps(unittest.TestCase):

    def login(self, user):
        return iRODSSession(
            port=self.port,
            zone=self.zone,
            host=self.host,
            user=user.name,
            password=self.users[user.name],
        )

    @staticmethod
    def irods_homedir(sess, path_only=False):
        path = f"/{sess.zone}/home/{sess.username}"
        if path_only:
            return path
        return sess.collections.get(path)

    @staticmethod
    def list_objects(sess):
        return [
            "{}/{}".format(o[Collection.name], o[DataObject.name])
            for o in sess.query(Collection.name, DataObject.name)
        ]

    users = {"alice": "apass", "bob": "bpass"}

    def setUp(self):

        self.alice = self.bob = None

        with helpers.make_session() as ses:
            u = ses.users.get(ses.username)
            if u.type != "rodsadmin":
                self.skipTest("""Test runnable only by rodsadmin.""")
            self.host = ses.host
            self.port = ses.port
            self.zone = ses.zone
            for newuser, passwd in self.users.items():
                u = ses.users.create(newuser, "rodsuser")
                setattr(self, newuser, u)
                u.modify("password", passwd)

    def tearDown(self):
        with helpers.make_session() as ses:
            for u in self.users:
                ses.users.remove(u)

    def test_session_clone_initialization_of_tickets_applied__issue_619(self):
        ticket_string = ""
        try:
            # Make a ticket with target collection being a rodsuser's home:
            with self.login(self.alice) as alice:
                alice_home_path = helpers.home_collection(alice)
                ticket_string = Ticket(alice).issue("read", alice_home_path).string

            # Create an original session. Cause a connection to be generated and used.
            old_sess = helpers.make_session()
            Ticket(old_sess, ticket_string).supply()
            old_sess.collections.get(alice_home_path)

            # Clone the session.
            new_sess = old_sess.clone()

            # Original session object should have had the ticket applied to a connection.
            self.assertLess(0, len(old_sess.ticket_applied))
            # Cloned session object should be starting out with a zero-length ticket/connection map.
            self.assertEqual(0, len(new_sess.ticket_applied))

            # The cloned session should inherit the original's associated ticket. We therefore test
            # that the new session can avail itself of the permissions granted by that ticket.
            new_sess.collections.get(alice_home_path)

        finally:
            # Delete the rodsuser ticket.
            if ticket_string:
                Ticket(new_sess, ticket_string).delete(**{kw.ADMIN_KW: ""})

    def test_admin_keyword_for_tickets(self):

        if helpers.make_session().server_version < (4, 2, 11):
            self.skipTest("ADMIN_KW not valid for Tickets API before iRODS 4.2.11")

        N_TICKETS = 3

        # Create some tickets as alice.

        with self.login(self.alice) as alice:
            alice_home_path = self.irods_homedir(alice, path_only=True)
            ticket_strings = [
                Ticket(alice).issue("read", alice_home_path).string
                for _ in range(N_TICKETS)
            ]

        # As rodsadmin, use the ADMIN_KW flag to delete alice's tickets.

        with helpers.make_session() as ses:
            alices_tickets = [
                t[TicketQuery.Ticket.string]
                for t in ses.query(TicketQuery.Ticket).filter(
                    TicketQuery.Owner.name == "alice"
                )
            ]
            self.assertEqual(len(alices_tickets), N_TICKETS)
            for s in alices_tickets:
                Ticket(ses, s).delete(**{kw.ADMIN_KW: ""})
            alices_tickets = [
                t[TicketQuery.Ticket.string]
                for t in ses.query(TicketQuery.Ticket).filter(
                    TicketQuery.Owner.name == "alice"
                )
            ]
            self.assertEqual(len(alices_tickets), 0)

    def test_ticket_expiry(self):
        with helpers.make_session() as ses:
            t1 = t2 = dobj = None
            try:
                gm_now = time.gmtime()
                gm_later = time.gmtime(calendar.timegm(gm_now) + 10)
                home = self.irods_homedir(ses)
                dobj = helpers.make_object(ses, home.path + "/dummy", content="abcxyz")

                later_ts = gmtime_to_timestamp(gm_later)
                later_epoch = calendar.timegm(gm_later)

                t1 = Ticket(ses)
                t2 = Ticket(ses)

                tickets = [
                    _.issue("read", dobj.path).string
                    for _ in (
                        t1,
                        t2,
                    )
                ]
                t1.modify(
                    "expire", later_ts
                )  # - Specify expiry with the human readable timestamp.
                t2.modify(
                    "expire", later_epoch
                )  # - Specify expiry formatted as epoch seconds.

                # Check normal access succeeds prior to expiration
                for ticket_string in tickets:
                    with self.login(self.alice) as alice:
                        Ticket(alice, ticket_string).supply()
                        alice.data_objects.get(dobj.path)

                # Check that both time formats have effected the same expiry time (The catalog returns epoch secs.)
                timestamps = []
                for ticket_string in tickets:
                    t = (
                        ses.query(TicketQuery.Ticket)
                        .filter(TicketQuery.Ticket.string == ticket_string)
                        .one()
                    )
                    timestamps.append(t[TicketQuery.Ticket.expiry_ts])
                self.assertEqual(len(timestamps), 2)
                self.assertEqual(timestamps[0], timestamps[1])

                # Wait for tickets to expire.
                epoch = int(time.time())
                while epoch <= later_epoch:
                    time.sleep(later_epoch - epoch + 1)
                    epoch = int(time.time())

                Expected_Exception = (
                    ex.CAT_TICKET_EXPIRED
                    if ses.server_version >= (4, 2, 9)
                    else ex.SYS_FILE_DESC_OUT_OF_RANGE
                )

                # Check tickets no longer allow access.
                for ticket_string in tickets:
                    with self.login(
                        self.alice
                    ) as alice, tempfile.NamedTemporaryFile() as f:
                        Ticket(alice, ticket_string).supply()
                        with self.assertRaises(Expected_Exception):
                            alice.data_objects.get(
                                dobj.path, f.name, **{kw.FORCE_FLAG_KW: ""}
                            )

            finally:
                if t1:
                    t1.delete()
                if t2:
                    t2.delete()
                if dobj:
                    dobj.unlink(force=True)

    def test_object_read_and_write_tickets(self):
        if self.alice is None or self.bob is None:
            self.skipTest("A rodsuser (alice and/or bob) could not be created.")
        t = None
        data_objs = []
        tmpfiles = []
        tickets = {}
        try:
            # Create ticket for read access to alice's home collection.
            alice = self.login(self.alice)
            home = self.irods_homedir(alice)

            # Create 'R' and 'W' in alice's home collection.
            data_objs = [
                helpers.make_object(alice, home.path + "/" + name, content="abcxyz")
                for name in ("R", "W")
            ]
            tickets = {
                "R": Ticket(alice).issue("read", home.path + "/R"),
                "W": Ticket(alice).issue("write", home.path + "/W"),
            }
            # Test only write ticket allows upload.
            with self.login(self.bob) as bob:
                rw_names = {}
                for name in ("R", "W"):
                    Ticket(bob, tickets[name].string).supply()
                    with tempfile.NamedTemporaryFile(delete=False) as tmpf:
                        tmpfiles += [tmpf]
                        rw_names[name] = tmpf.name
                        tmpf.write(b"hello")
                    if name == "W":
                        bob.data_objects.put(tmpf.name, home.path + "/" + name)
                    else:
                        try:
                            bob.data_objects.put(tmpf.name, home.path + "/" + name)
                        except ex.CAT_NO_ACCESS_PERMISSION:
                            pass
                        else:
                            raise AssertionError(
                                "A read ticket allowed a data object write operation to happen without error."
                            )

            # Test upload was successful, by getting and confirming contents.
            with self.login(
                self.bob
            ) as bob:  # This check must be in a new session or we get CollectionDoesNotExist. - Possibly a new issue [ ]
                for name in ("R", "W"):
                    bob.cleanup()  # clear out existing connections
                    Ticket(bob, tickets[name].string).supply()
                    bob.data_objects.get(
                        home.path + "/" + name, rw_names[name], **{kw.FORCE_FLAG_KW: ""}
                    )
                    with open(rw_names[name], "r") as tmpread:
                        self.assertEqual(
                            tmpread.read(), "abcxyz" if name == "R" else "hello"
                        )
        finally:
            for t in tickets.values():
                t.delete()
            for d in data_objs:
                d.unlink(force=True)
            for file_ in tmpfiles:
                os.unlink(file_.name)
            alice.cleanup()

    def test_coll_read_ticket_between_rodsusers(self):
        t = None
        data_objs = []
        tmpfiles = []
        try:
            # Create ticket for read access to alice's home collection.
            alice = self.login(self.alice)
            tc = Ticket(alice)
            home = self.irods_homedir(alice)
            tc.issue("read", home.path)

            # Create 'x' and 'y' in alice's home collection
            data_objs = [
                helpers.make_object(alice, home.path + "/" + name, content="abcxyz")
                for name in ("x", "y")
            ]

            with self.login(self.bob) as bob:
                ts = Ticket(bob, tc.string)
                ts.supply()
                # Check collection access ticket allows bob to list both subobjects
                self.assertEqual(len(self.list_objects(bob)), 2)
                # and that we can get (and read) them properly.
                for name in ("x", "y"):
                    with tempfile.NamedTemporaryFile(delete=False) as tmpf:
                        tmpfiles += [tmpf]
                    bob.data_objects.get(
                        home.path + "/" + name, tmpf.name, **{kw.FORCE_FLAG_KW: ""}
                    )
                    with open(tmpf.name, "r") as tmpread:
                        self.assertEqual(tmpread.read(), "abcxyz")

            td = Ticket(alice)
            td.issue("read", home.path + "/x")

            with self.login(self.bob) as bob:
                ts = Ticket(bob, td.string)
                ts.supply()

                # Check data access ticket allows bob to list only one data object
                self.assertEqual(len(self.list_objects(bob)), 1)

                # ... and fetch that object (verifying content)
                with tempfile.NamedTemporaryFile(delete=False) as tmpf:
                    tmpfiles += [tmpf]
                bob.data_objects.get(
                    home.path + "/x", tmpf.name, **{kw.FORCE_FLAG_KW: ""}
                )
                with open(tmpf.name, "r") as tmpread:
                    self.assertEqual(tmpread.read(), "abcxyz")

                # ... but not fetch the other data object owned by alice.
                with self.assertRaises(ex.DataObjectDoesNotExist):
                    bob.data_objects.get(home.path + "/y")
        finally:
            if t:
                t.delete()
            for d in data_objs:
                d.unlink(force=True)
            for file_ in tmpfiles:
                os.unlink(file_.name)
            alice.cleanup()


class TestTicketOps(unittest.TestCase):

    def setUp(self):
        """Create objects for test"""
        self.sess = helpers.make_session()
        user = self.sess.users.get(self.sess.username)
        if user.type != "rodsadmin":
            self.skipTest("""Test runnable only by rodsadmin.""")

        admin = self.sess
        delete_my_tickets(admin)

        # Create test collection

        self.coll_path = f"/{admin.zone}/home/{admin.username}/ticket_test_dir"
        self.coll = helpers.make_collection(admin, self.coll_path)

        # Create anonymous test user
        self.user = admin.users.create("anonymous", "rodsuser")
        self.rodsuser_params = {
            "host": admin.host,
            "port": admin.port,
            "user": "anonymous",
            "password": "",
            "zone": admin.zone,
        }

        # make new data object in the test collection with some initialized content

        self.INITIALIZED_DATA = b"1" * 16
        self.data_path = f"{self.coll_path}/ticketed_data"
        helpers.make_object(admin, self.data_path, content=self.INITIALIZED_DATA)

        self.MODIFIED_DATA = b"2" * 16

        # make new tickets for the various combinations

        self.tickets = {"coll": {}, "data": {}}
        for obj_type in ("coll", "data"):
            for access in ("read", "write"):
                ticket = Ticket(admin)
                self.tickets[obj_type][access] = ticket.string
                ticket.issue(access, getattr(self, obj_type + "_path"))

    def tearDown(self):
        """Clean up tickets , collections and data objects used for test."""
        admin = self.sess
        delete_my_tickets(admin)
        if getattr(self, "coll", None):
            self.coll.remove(recurse=True, force=True)
        if getattr(self, "user", None):
            self.user.remove()
        admin.cleanup()

    def _ticket_read_helper(self, obj_type, download=False):
        with iRODSSession(**self.rodsuser_params) as user_sess:
            temp_file = []
            if download:
                temp_file += [tempfile.mktemp()]
            try:
                Ticket(user_sess, self.tickets[obj_type]["read"]).supply()
                data = user_sess.data_objects.get(self.data_path, *temp_file)
                self.assertEqual(data.open("r").read(), self.INITIALIZED_DATA)
                if temp_file:
                    with open(temp_file[0], "rb") as local_file:
                        self.assertEqual(local_file.read(), self.INITIALIZED_DATA)
            finally:
                if temp_file and os.path.exists(temp_file[0]):
                    os.unlink(temp_file[0])

    def test_data_ticket_read(self):
        self._ticket_read_helper(obj_type="data")

    def test_coll_ticket_read(self):
        self._ticket_read_helper(obj_type="coll")

    def test_data_ticket_read_with_download(self):
        self._ticket_read_helper(obj_type="data", download=True)

    def test_coll_ticket_read_with_download(self):
        self._ticket_read_helper(obj_type="coll", download=True)

    def _ticket_write_helper(self, obj_type):
        with iRODSSession(**self.rodsuser_params) as user_sess:
            Ticket(user_sess, self.tickets[obj_type]["write"]).supply()
            data = user_sess.data_objects.get(self.data_path)
            with data.open("w") as obj:
                obj.write(self.MODIFIED_DATA)
            self.assertEqual(data.open("r").read(), self.MODIFIED_DATA)

    def test_data_ticket_write(self):
        self._ticket_write_helper(obj_type="data")

    def test_coll_ticket_write(self):
        self._ticket_write_helper(obj_type="coll")


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