#
# Copyright (c) SAS Institute Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#


from testrunner import testhelp
from conary_test import rephelp

from conary_test import dbstoretest

from conary.repository import errors
from conary.repository.netrepos import netauth
from conary.repository.netrepos.trovestore import TroveStore
from conary.server import schema
from conary import versions


class NetAuthTest(dbstoretest.DBStoreTestBase):

    def _setupDB(self):
        db = self.getDB()
        schema.createSchema(db)
        schema.setupTempTables(db)

        return db

    def _addUserRole(self, na, username, password):
        na.addRole(username)
        na.addUser(username, password)
        na.addRoleMember(username, username)
        
    def testManageAcls(self):
        db = self._setupDB()

        ts = TroveStore(db)
        na = netauth.NetworkAuthorization(db, "conary.rpath.com")

        authToken = ("testuser", "testpass", [ (None, None) ], None )
        self._addUserRole(na, "testuser", "testpass")
        na.addAcl("testuser", None, None, write = True)

        ## TODO Test the trove/label aspects of ACL management
        na.deleteAcl("testuser", None, None)

        #If the delete above failed, this will throw an exception
        na.addAcl("testuser", None, None)
        assert(na.authCheck(authToken, admin=False) == True)
        assert(na.authCheck(authToken, admin=True) == False)

        #Now give the user has admin rights
        na.setAdmin("testuser", True)

        assert(na.authCheck(authToken, admin=True) == True)

    def testNetAuth(self):
        db = self._setupDB()
        ts = TroveStore(db)
        na = netauth.NetworkAuthorization(db, "conary.rpath.com")

        self._addUserRole(na, "testuser", "testpass")
        na.addAcl("testuser", None, None, write = True, remove = True)
        self._addUserRole(na, "luser", "luserpass")
        na.addAcl("luser", None, None)
        self._addUserRole(na, "root", "rootpass")
        na.addAcl("root", None, None, write = True, remove = True)
        na.setAdmin("root", True)

        authToken = ("testuser", "testpass", [ (None, None) ], None )
        badAuthToken = ("testuser", "testPass", [ (None, None) ], None )
        luserToken = ("luser", "luserpass", [ (None, None) ], None )
        badLuserToken = ("luser", "luserfoo", [ (None, None) ], None )
        rootToken = ("root", "rootpass", [ (None, None) ], None )

        assert(na.check(authToken, write=False) != False)
        assert(na.check(authToken, write=True) != False)
        assert(na.check(badAuthToken, write=False) != True)
        assert(na.check(badAuthToken, write=True) != True)

        assert(na.check(luserToken, write=False) != False)
        assert(na.check(luserToken, write=True) != True)

        assert(na.authCheck(rootToken, admin=True) != False)
        assert(na.authCheck(luserToken, admin=True) != True)

        assert(na.check(rootToken, remove=True) != False)
        assert(na.check(luserToken, remove=True) != True)
        assert(na.check(authToken, remove=True) != False)

        assert(na.check(authToken) != False)
        assert(na.check(badAuthToken) != True)

        # Shim clients, with a ValidPasswordToken
        entitlements = [(None, None)]
        authShim = ("testuser", netauth.ValidPasswordToken, entitlements, None)
        luserShim = ("luser", netauth.ValidPasswordToken, entitlements, None)
        badShim = ("nobody", netauth.ValidPasswordToken, entitlements, None)

        assert na.check(authShim, write=False)
        assert na.check(authShim, write=True)
        assert na.check(authShim, remove=True)

        assert na.check(luserShim, write=False)
        assert not na.check(luserShim, write=True)
        assert not na.check(luserShim, remove=True)

        assert not na.check(badShim, write=False)

        # Shim clients, with a ValidUser()
        authRoleShim = (netauth.ValidUser('testuser'), None, entitlements, None)
        luserRoleShim = (netauth.ValidUser('luser'), None, entitlements, None)
        badRoleShim = (netauth.ValidUser('nobody'), None, entitlements, None)

        assert na.check(authRoleShim, write=False)
        assert na.check(authRoleShim, write=True)
        assert na.check(authRoleShim, remove=True)

        assert na.check(luserRoleShim, write=False)
        assert not na.check(luserRoleShim, write=True)
        assert not na.check(luserRoleShim, remove=True)

        assert not na.check(badRoleShim, write=False)

        try:
            na.addAcl("testuser", None, None, write = True)
        except errors.PermissionAlreadyExists:
            pass
        else:
            self.fail("PermissionAlreadyExists exception expected")

        try:
            na.addUser("teStUseR", 'TeStPaSs')
        except (errors.UserAlreadyExists, errors.RoleAlreadyExists):
            pass
        else:
            self.fail("UserAlreadyExists or RoleAlreadyExists exception expected")
        try:
            na.addRole("lUseR")
        except errors.RoleAlreadyExists:
            pass
        else:
            self.fail("RoleAlreadyExists exception expected")

    def testDeleteAuth(self):
        db = self._setupDB()
        ts = TroveStore(db)
        na = netauth.NetworkAuthorization(db, "conary.rpath.com")

        # schema creation creates these for us - we need to get rid of them for this test
        for user in na.userAuth.getUserList():
            na.deleteUserByName(user)
        for group in na.getRoleList():
            na.deleteRoleByName(group)

        self._addUserRole(na, "deluser1", "delpass")
        na.addAcl("deluser1", None, None)

        self._addUserRole(na, "deluser2", "delpass")
        na.addAcl("deluser2", None, None, write = True)
        na.addRoleMember("deluser2", "deluser1")

        # Create another group and add deluser1 to said group
        na.addRole('delgroup1')
        na.addAcl('delgroup1', None, None, write = True)
        na.setAdmin('delgroup1', True)
        na.addRoleMember("delgroup1", "deluser1")
        na.addRole('delgroup2')
        na.addAcl('delgroup2', None, None, write = True)
        na.setAdmin('delgroup2', True)
        na.addRoleMember("delgroup2", "deluser1")
        na.addRoleMember("delgroup2", "deluser2")

        self.assertEqual(na.getRoles('deluser1'),
                             ['deluser1', 'deluser2', 'delgroup1', 'delgroup2'])
        self.assertEqual(na.getRoles('deluser2'),
                             ['deluser2', 'delgroup2'])

        # Delete user2 and see if group2 still lists user2
        na.deleteUserByName('deluser2')
        self.assertEqual(list(na.getRoleMembers('delgroup2')),
                             ['deluser1'] )
        self.assertEqual(list(na.userAuth.getRolesByUser('deluser1')),
                             [ 'deluser1', 'deluser2', 'delgroup1', 'delgroup2' ])
        na.deleteRole('delgroup1')

        self.assertEqual(list(na.userAuth.getRolesByUser('deluser1')),
                             [ 'deluser1', 'deluser2', 'delgroup2' ])
        # because deluser1 will have no acl, it should go too
        na.deleteAcl("deluser1", None, None)
        na.deleteUserByName('deluser1')
        self.assertEqual(list(na.getRoleMembers('delgroup2')), [])