modify storage class name
Change-Id: I1727937514933e8ccf663897dca6725e3a162e9a
diff --git a/tests/test_device_storage.py b/tests/test_device_storage.py
new file mode 100644
index 0000000..d04449e
--- /dev/null
+++ b/tests/test_device_storage.py
@@ -0,0 +1,338 @@
+# -*- 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)
+
+