tests/test_user_access_storage.py

# -*- Mode:python; c-file-style:"gnu"; indent-tabs-mode:nil -*- */
#
# Copyright (C) 2014 Regents of the University of California.
# Author: Weiwei Liu <summerwing10@gmail.com>
# 
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.
# A copy of the GNU General Public License is in the file COPYING.

import unittest as ut
import os.path
from pyndn import Name
from user_access_storage import UserAccessStorage
from device_profile import DeviceProfile
from hmac_key import HMACKey

class TestUserAccessStorageMethods(ut.TestCase):
    def setUp(self):
        if not "HOME" in os.environ:
            home = '.'
        else:
            home = os.environ["HOME"]

        dbDirectory = os.path.join(home, '.ndn')
        self.databaseFilePath = os.path.join(dbDirectory, 'ndnhome-controller.db')

        if os.path.isfile(self.databaseFilePath):
            os.remove(self.databaseFilePath)
        
        self.storage = UserAccessStorage()

    def tearDown(self):
        pass
    
    def test_01_storage_constructor(self):
        
        self.assertTrue(os.path.isfile(self.databaseFilePath), 'fail to create database file')        
        self.assertTrue(self.storage.doesTableExist("User"), "Device table doesn't exist")  
        self.assertTrue(self.storage.doesTableExist("Access"), "Command table doesn't exist")
        
        #test constructor with file path HOME/.ndn/homedb 
        home = os.environ["HOME"]
        dir = os.path.join(home, '.ndn')
        dbdir = os.path.join(dir, 'homedb')
        
        if not os.path.exists(dbdir):
                os.makedirs(dbdir)
       
        filePath = os.path.join(dbdir, 'ndnhome-controller.db')
        storage2 = UserAccessStorage(filePath)
        self.assertTrue(os.path.isfile(filePath), 'fail to create database file' )
        os.remove(filePath)
   
    def test_02_entry_existence_methods(self):
        # test the existence of non-existed user (prefix = /home/weiwei)
        username = 'weiwei'
        prefixBase = '/home'
        prefixStr =prefixBase +'/' + username
        prefixName = Name(prefixStr)                
        result1 = self.storage.doesUserExistByPrefix(prefixName)
        result2 = self.storage.doesUserExistByUsername(username)
        self.assertFalse(result1, "user with prefix '" + prefixStr +  "' shouldn't exist")
        self.assertFalse(result2, "user with username '" + username + "' shouldn't exist")

        #test existence of an existed user
        self.add_a_default_user(username)  
        result1 = self.storage.doesUserExistByPrefix(prefixName)
        result2 = self.storage.doesUserExistByUsername(username)

        self.assertTrue(result1, "user with prefix '" + prefixStr +  "' should exist")
        self.assertTrue(result2, "user with username '" + username + "' should exist")

        #test the existence of a non-existed access 
        commandId = 1
        userId = 1 
        result = self.storage.doesAccessExist(commandId, userId)
        self.assertFalse(result, "access shouldn't exist with a empty Access table")

        #test the exitence of an existed access
        accessToken = self.create_a_default_access_token('user1_command1')
        self.storage.addAccess(1,1, 'laptop', accessToken)
        result = self.storage.doesAccessExist(commandId, userId)
        self.assertTrue(result, "access should exist")

    def test_03_add_user(self):
        username = 'weiwei'
        prefixBase = '/home'
        prefixStr =prefixBase +'/' + username
        prefixName = Name(prefixStr)
        hash_ = 'EEADFADSFAGASLGALS'
        salt = 'adfafdwekldsfljcdc'
        type_ = 'guest'

        result = self.storage.addUser(prefixName, username, hash_, salt, type_)

        self.assertTrue(result > 0, 'fail to add user')             
        self.assertTrue(self.storage.doesUserExistByPrefix(prefixName), "user doesn't exist in table")        
        row = self.storage.getUserEntry(prefixName)

        self.assertTrue(row[1] == prefixStr, "column prefix has incorrect value")
        self.assertTrue(row[2] == username, "column username has incorrect value ")
        self.assertTrue(row[3] == hash_, 'colum hash has incorrect value')
        self.assertTrue(row[4] == salt, 'colum salt has incorrect value')
        self.assertTrue(row[5] == type_, 'column type_ has incoorect value')
        
        #try to insert the same device
        result = self.storage.addUser(prefixName, username, hash_, salt, type_)
        self.assertTrue(result == 0, 'unexpected result when trying to insert an already existed user')

    def test_04_delete_user(self):
        username = 'user1'
        self.add_a_default_user(username)
        prefixBase = '/home'
        prefixStr =prefixBase +'/' + username
        prefixName = Name(prefixStr)
       
        self.assertTrue(self.storage.doesUserExistByPrefix(prefixName), "user doesn't exist in table")
        result = self.storage.deleteUser(prefixName)       
        self.assertTrue(result == 1, 'fail to delete device')

    def test_05_get_userid(self):
        username = 'user1'
        self.add_a_default_user(username) 
        prefixBase = '/home'
        prefixStr =prefixBase +'/' + username
        prefixName = Name(prefixStr)

        userId = self.storage.getUserId(prefixName)
        self.assertTrue(userId==1, "get a wrong user id")
        
        username = 'user2'
        prefixStr =prefixBase +'/' + username
        prefixName = Name(prefixStr)
        self.add_a_default_user(username)

        userId = self.storage.getUserId(prefixName)
        self.assertTrue(userId==2, "get a wrong user id")

    def test_06_update_user(self):
        username = 'user1'
        self.add_a_default_user(username)
        prefixBase = '/home'
        prefixStr =prefixBase +'/' + username
        prefixName = Name(prefixStr)

        #update hash_ salt, type_ value of user 
        newHash = 'NNNMMHGFGKUOIPRT'
        newSalt = 'kjkjfioewprwerke'
        newType = 'user'
        self.storage.updateUser(prefixName, newHash, newSalt, newType)
        row = self.storage.getUserEntry(prefixName)
        self.assertTrue(row[3] == newHash, "fail to update column: hash")
        self.assertTrue(row[4] == newSalt, "fail to update column: salt")
        self.assertTrue(row[5] == newType, "fail to update column: type")

    def test_07_add_access(self):
        username = 'user1'
        self.add_a_default_user(username)

        accessToken = self.create_a_default_access_token('user1_command1')
        self.storage.addAccess(1,1, 'laptop', accessToken)        
        result = self.storage.doesAccessExist(1, 1)
        self.assertTrue(result == True, "fail to add command")

    def test_08_delete_access(self):
        username = 'user1'
        self.add_a_default_user(username)

        accessToken = self.create_a_default_access_token('user1_command1')
        self.storage.addAccess(1,1, 'laptop', accessToken) 

        self.assertTrue(self.storage.doesAccessExist(1, 1), 'before delete, the command should exist')
        self.storage.deleteAccess(1, 1)
        self.assertFalse(self.storage.doesAccessExist(1, 1), "after delete, the command shouldn't exist")

    def test_09_get_commands_of_user(self):        
        accessToken = self.create_a_default_access_token('token1')
        self.storage.addAccess(1,1, 'laptop', accessToken)
        self.storage.addAccess(1,1, 'desktop', accessToken)
        self.storage.addAccess(2,1,'laptop', accessToken)
        self.storage.addAccess(2,1, 'desktop', accessToken)
        result = self.storage.getCommandsOfUser(1)
        self.assertTrue(result == [1,2], 'wrong command list returned')

    def test_10_get_access_token(self):
        accessTokenName1 =  'accessToken1'
        accessTokenName2 = 'accessToken2'
        accessToken1 = self.create_a_default_access_token(accessTokenName1)
        accessToken2 = self.create_a_default_access_token(accessTokenName2)
        userDevice1 = 'laptop'
        userDevice2 = 'desktop'
        accessId1 = self.storage.addAccess(1,1, userDevice1, accessToken1)
        accessId2 = self.storage.addAccess(1,1, userDevice2, accessToken2)
        
        if accessId1 <= 0 or accessId2 <= 0:
            raise RuntimeError('fail to add access') 
        result1 = self.storage.getAccessToken(1,1, userDevice1)
        result2 = self.storage.getAccessToken(1,1, userDevice2)
        if result1 == None or result2 == None:
            self.assertTrue(False, 'no access returned')
        self.assertTrue(result1.getName() == accessTokenName1, 'wrong access token returned')
        self.assertTrue(result2.getName() == accessTokenName2, 'wrong access token returned')
    
    def add_a_default_user(self, username):
        prefixBase = '/home'
        prefixStr =prefixBase +'/' + username
        prefixName = Name(prefixStr)
        hash_ = 'EEADFADSFAGASLGALS'
        salt = 'adfafdwekldsfljcdc'
        type_ = 'guest'

        self.storage.addUser(prefixName, username, hash_, salt, type_)

    def create_a_default_access_token(self, keyName = None):
        keyContent = 'this is key content'
        accessToken = HMACKey(sequence = 0, key = keyContent, name = keyName)
        return accessToken
         
if __name__ == '__main__':
    ut.main(verbosity=2)