tests/test_device_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 device_storage import DeviceStorage
from device_profile import DeviceProfile
from hmac_key import HMACKey

class TestDeviceStorageMethods(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 = DeviceStorage()

    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("Device"), "Device table doesn't exist")  
        self.assertTrue(self.storage.doesTableExist("Command"), "Command table doesn't exist")
        self.assertTrue(self.storage.doesTableExist("ServiceProfile"), "ServiceProfile 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 = DeviceStorage(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 device (prefix = /home/sensor/LED/1)
        prefixStr = '/home/sensor/LED/1'
        name = Name(prefixStr)        
        profile = DeviceProfile(prefix = name)
        keyContent = 'this is key content'
        seed =  HMACKey( 0, 0 ,keyContent, 'led1')
        configurationToken =  HMACKey(0, 0, keyContent)
        result = self.storage.doesDeviceExist(name)
        self.assertFalse(result, "device with prefix '" + prefixStr +  "' shouldn't exist")
        
        #test existence of an existed device 
        self.storage.addDevice(profile, seed, configurationToken)
        result = self.storage.doesDeviceExist(name)
        self.assertTrue(result, "device with prefix '" + prefixStr +"' should exist")
        
        #test existence of a non-existed command
        deviceId = 1 
        commandName = 'turn_on'
        result = self.storage.doesCommandExist(deviceId, commandName)
        self.assertFalse(result, "Command : 'device_id =" + str(deviceId) + ", name = " + commandName + "' shouldn't exist" ) 
       
        #test existence of a existed command
        deviceId = self.storage.getDeviceId(name)
        commandName = 'turn_on'
        commandToken = self.create_a_default_key('turn_on')
        self.storage.addCommand(deviceId, commandName, commandToken)
        result = self.storage.doesCommandExist(deviceId, commandName)
        self.assertTrue(result, "Command : 'device_id =" + str(deviceId) + ", name = " + commandName + "' should exist" )
           
        #test existence of a non-existed service profile
        deviceId = 1
        serviceProfileName = '/standard/sensor/simple-LED-control/v0'
        
        result = self.storage.doesServiceProfileExist(deviceId, serviceProfileName)
        self.assertFalse(result, "service profile : 'device_id =" + str(deviceId) + ", name = " + serviceProfileName + "' shouldn't exist" )

        #test existence of a existed service profile
        deviceId = self.storage.getDeviceId(name)
        serviceProfileName = '/standard/sensor/simple-LED-control/v0'

        self.storage.addServiceProfile(deviceId, serviceProfileName)
        result = self.storage.doesServiceProfileExist(deviceId, serviceProfileName)
        self.assertTrue(result, "service profile : 'device_id =" + str(deviceId) + ", name = " + serviceProfileName + "' should exist" )
 

    def test_03_add_device(self):
        prefixStr = '/home/sensor/LED/1'
        name = Name(prefixStr)
        profile = DeviceProfile(prefix = name)
        keyContent = 'this is key content'
        seed =  HMACKey( 0, 0 ,keyContent, 'led1')
        configurationToken =  HMACKey(0, 0, keyContent)
        
        result = self.storage.addDevice(profile, seed, configurationToken)

        self.assertTrue(result > 0, 'fail to add device entry')             
        self.assertTrue(self.storage.doesDeviceExist(name), "device doesn't exist in table")        
        row = self.storage.getDeviceEntry(name)
        self.assertTrue(row[1] == prefixStr, "column prefix has incorrect value")
        #print row[2:7]
        self.assertTrue(row[2:8] == (None, None, None, None, None, None), "column 2-7 have incorrect values ")
        self.assertTrue(row[8] == 'led1', 'colum seed_name has incorrect values')
        self.assertTrue(row[9:12] == (0,0,keyContent), 'colum seed_sequence or seed_counter has incorrect value')
        self.assertTrue(row[12:] == (keyContent,0,0), 'one or more configuration token columns have incoorect value')
        #try to insert the same device
        result = self.storage.addDevice(profile, seed, configurationToken)
        self.assertTrue(result == 0, 'unexpected result when trying to insert an already existed device')

    def test_04_delete_device(self):
        name = Name('/home/sensor/LED/1')
        profile = DeviceProfile(prefix = name)
        seed =  HMACKey( 0, 0 ,'this is key content', 'led1')
        configurationToken =  HMACKey(0, 0, 'this is key content')
        self.storage.addDevice(profile, seed, configurationToken)
        self.assertTrue(self.storage.doesDeviceExist(name), "device doesn't exist in table")
        result = self.storage.deleteDevice(name)
        #print 'result : %d' %(result)
        self.assertTrue(result == 1, 'fail to delete device')

    def test_05_get_deviceid(self):
        name = Name('/home/sensor/LED/1')
        profile = DeviceProfile(prefix = name)
        seed =  HMACKey( 0, 0 ,'this is key content', 'led1')
        configurationToken =  HMACKey(0, 0, 'this is key content')
        self.storage.addDevice(profile, seed, configurationToken)
        deviceId = self.storage.getDeviceId(name)
        self.assertTrue(deviceId==1, "get a wrong device id")
        
        name2 = Name('/home/sensor/LED/2')
        profile = DeviceProfile(prefix = name2)
        seed =  HMACKey( 0, 0 ,'this is key content', 'led1')
        configurationToken =  HMACKey(0, 0, 'this is key content')
        self.storage.addDevice(profile, seed, configurationToken)
        deviceId2 = self.storage.getDeviceId(name2)
        self.assertTrue(deviceId2 == 2, "get a wrong device id")
     
    def test_06_update_device(self):
        prefixStr = '/home/sensor/LED/1' 
        name = Name(prefixStr)
        self.add_a_default_device(prefixStr)

        #update column prefix of device 
        newPrefixStr = '/home/sensor/LED/2'
        newSeedName = 'led2'
        self.storage.updateOneColumnOfDevice(name, 'seed_name', newSeedName)
        row = self.storage.getDeviceEntry(name)
        self.assertTrue(row[8] == newSeedName, "fail to update column: seed_name")

        self.storage.updateOneColumnOfDevice(name, 'prefix', newPrefixStr) 
        row = self.storage.getDeviceEntry(Name(newPrefixStr))
       
        self.assertTrue(row[1] == newPrefixStr, "fail to update coumn: prefix")     
  
    def test_07_add_command(self):
        prefixStr = '/home/sensor/LED/2'
        name = Name(prefixStr)
        self.add_a_default_device(prefixStr)
        deviceId = self.storage.getDeviceId(name)
        commandName = 'turn_on'
        commandToken = self.create_a_default_key('turn_on')
        self.storage.addCommand(deviceId, commandName, commandToken)        
        result = self.storage.doesCommandExist(deviceId, commandName)
        self.assertTrue(result == True, "fail to add command")

    def test_08_delete_command(self):
        prefixStr1 = 'home/sensor/LED/1'
        name1 = Name(prefixStr1)
        self.add_a_default_device(prefixStr1)     
        deviceId = self.storage.getDeviceId(name1)
        commandName = 'turn_on'
        commandToken = self.create_a_default_key(commandName)
        self.storage.addCommand(deviceId, commandName, commandToken)
        self.assertTrue(self.storage.doesCommandExist(deviceId, commandName), 'before delete, the command should exist')
        self.storage.deleteCommand(deviceId, commandName)
        self.assertFalse(self.storage.doesCommandExist(deviceId, commandName), "after delete, the command shouldn't exist")
        
    def test_09_get_commands_of_device(self):
        #test with an empty table
        deviceId = 2
        result = self.storage.getCommandsOfDevice(deviceId) 
        print result
        self.assertTrue(not result, "there should be no command found")
      
        #test with a device has two commands
        prefixStr = 'home/sensor/LED/1' 
        name = Name(prefixStr)
        self.add_a_default_device(prefixStr)
        deviceId = self.storage.getDeviceId(name)
        commandName = 'turn_on'
        commandName2 = 'turn_off'
        commandToken = self.create_a_default_key(commandName)
        commandToken2 = self.create_a_default_key(commandName2)
        self.storage.addCommand(deviceId, commandName, commandToken)
        self.storage.addCommand(deviceId, commandName2, commandToken)
        result = self.storage.getCommandsOfDevice(deviceId)
        print 'result %s' %(result)
       
    def test_10_add_service_profile(self):
        prefixStr = '/home/sensor/LED/2'
        name = Name(prefixStr)
        self.add_a_default_device(prefixStr)
        deviceId = self.storage.getDeviceId(name)

        serviceProfileName = '/standard/sensor/simple-LED-control/v0'
        self.storage.addServiceProfile(deviceId, serviceProfileName)
        result = self.storage.doesServiceProfileExist(deviceId, serviceProfileName)
        self.assertTrue(result == True, "fail to add service profile")

    def test_11_delete_service_profile(self):
        prefixStr1 = 'home/sensor/LED/1'
        name1 = Name(prefixStr1)
        self.add_a_default_device(prefixStr1)
        deviceId = self.storage.getDeviceId(name1)
        serviceProfileName = '/standard/sensor/simple-LED-control/v0'        

        self.storage.addServiceProfile(deviceId, serviceProfileName)
        self.assertTrue(self.storage.doesServiceProfileExist(deviceId, serviceProfileName), 'before delete, the service profile should exist')
        self.storage.deleteServiceProfile(deviceId, serviceProfileName)
        self.assertFalse(self.storage.doesServiceProfileExist(deviceId, serviceProfileName), "after delete, the service profile shouldn't exist")

    def test_12_get_service_profiles_of_device(self):
        #test with an empty table
        deviceId = 2
        result = self.storage.getServiceProfilesOfDevice(deviceId)
        print result
        self.assertTrue(not result, "there should be no command found")

        #test with a device has two commands
        prefixStr = 'home/sensor/LED/1'
        name = Name(prefixStr)
        self.add_a_default_device(prefixStr)
        deviceId = self.storage.getDeviceId(name)
        serviceProfileName = '/standard/sensor/simple-LED-control/v0'
        serviceProfileName2 = '/standard/sensor/simple-LED-control/v2'
        self.storage.addServiceProfile(deviceId, serviceProfileName)
        self.storage.addServiceProfile(deviceId, serviceProfileName2)
        result = self.storage.getServiceProfilesOfDevice(deviceId)
        print 'result %s' %(result)

    def test_13_get_device_profile_from_device(self):
        #test with a non exisited device prefix
        prefixStr = '/home/sensor/LED/2'
        name = Name(prefixStr)
        deviceProfile = self.storage.getDeviceProfileFromDevice(name)
        self.assertTrue(deviceProfile == None, "no device profile should be return with an non-existed prefix ")
       
        #test with a existed device prefix
        profile = DeviceProfile(prefix = name, location = 'bedroom', category = 'sensors')
        seed = self.create_a_default_key('led1')
        configurationToken = self.create_a_default_key()
        self.storage.addDevice(profile, seed, configurationToken)

        deviceProfile  = self.storage.getDeviceProfileFromDevice(name)
        
        self.assertTrue(deviceProfile.getLocation() == 'bedroom', 'wrong field: location')
        self.assertTrue(deviceProfile.getCategory() == 'sensors', 'wrong field: category')

    def test_14_get_seed_and_get_configuration_token(self):
        #test with a non existed device prefix
        prefixStr = '/home/sensor/LED/2'
        name = Name(prefixStr)
        seed = self.storage.getSeed(name)
        configurationToken = self.storage.getConfigurationToken(name)
        self.assertTrue(seed == None, "no seed should be returned with an non-existed prefix ")
        self.assertTrue(configurationToken == None, "no configuration token should be returned with an non-existed prefix ")
        
        #test with an existed device prefix
        profile= DeviceProfile(prefix = name)
        seed = self.create_a_default_key('led2')
        configurationToken = self.create_a_default_key()
        self.storage.addDevice(profile, seed, configurationToken)
        seed = self.storage.getSeed(name)
        configurationToken = self.storage.getConfigurationToken(name)
        keyContent = 'this is key content'
        self.assertTrue(seed.getKey() == keyContent, 'key content of seed  is incorrect')
        self.assertTrue(configurationToken.getKey() == keyContent, 'key content of configuration token is incorrect')
        self.assertTrue(seed.getName() == 'led2', 'name of seed is incorrect')
        self.assertTrue(configurationToken.getName() == None, 'name of configuration token is incorrect')
  
    def test_15_get_Command_Token(self):
        #test with a non existed command
        deviceId = 5
        commandName = "turn_on"
        commandToken =  self.storage.getCommandToken(deviceId, commandName)
        self.assertTrue(commandToken == None,  'no commandToken should be returned with an non-existed command')
        
        #test with an existed command
        prefixStr = 'home/sensor/LED/1'
        name = Name(prefixStr)
        self.add_a_default_device(prefixStr)
        deviceId = self.storage.getDeviceId(name)
        commandToken = self.create_a_default_key(commandName)
        self.storage.addCommand(deviceId, commandName, commandToken)

        commandTokenReturned =  self.storage.getCommandToken(deviceId, commandName)
        #print commandTokenReturned.getName()
        self.assertTrue(commandTokenReturned.getName() == commandName,  'wrong command token')

    def add_a_default_device(self, prefixStr):
        name = Name(prefixStr)
        profile = DeviceProfile(prefix = name)
        seed = self.create_a_default_key('led1')
        configurationToken = self.create_a_default_key()
        self.storage.addDevice(profile, seed, configurationToken)
    
    def create_a_default_key(self, keyName = None):
        keyContent = 'this is key content'
        seed = HMACKey(0,0, keyContent, keyName)
        return seed
         
if __name__ == '__main__':
    ut.main(verbosity=2)