local tests

python/test/src/testipaaca.py

@@ -7,7 +7,8 @@ import sys
 import unittest
 	
 def handle_iu_event(iu, event_type, local):
-	print('(IU event '+event_type+' '+str(iu.uid)+')')
+	#print('(IU event '+event_type+' '+str(iu.uid)+')')
+	pass
 
 class IpaacaIUStoreTestCase(unittest.TestCase):
 	def setUp(self):
@@ -65,6 +66,56 @@ class IpaacaLinksTestCase(unittest.TestCase):
 		grinlinks = received_iu.get_links('grin')
 		self.assertEqual(len(grinlinks), 0)
 
+class IpaacaCommitTestCase(unittest.TestCase):
+	def setUp(self):
+		self.ib = ipaaca.InputBuffer('TestIn', ['sensorcategory'])
+		self.ib.register_handler(handle_iu_event)
+		self.ob = ipaaca.OutputBuffer('TestOut')
+		self.sensor_iu = ipaaca.IU('sensorcategory')
+		self.sensor_iu.payload = {'data': 'sensordata'}
+		time.sleep(0.1)
+		self.ob.add(self.sensor_iu)
+		time.sleep(0.1)
+	def tearDown(self):
+		pass
+	def testCommitAndLocalWrite(self):
+		self.sensor_iu.commit()
+		with self.assertRaises(ipaaca.IUCommittedError):
+			self.sensor_iu.payload['data'] = 'updatedData'
+	def testCommitAndRemoteWrite(self):
+		self.sensor_iu.commit()
+		received_iu = self.ib.iu_store[self.sensor_iu.uid]
+		with self.assertRaises(ipaaca.IUCommittedError):
+			received_iu.payload['data'] = 'updatedData'
+
+class IpaacaRemoteWriteTestCase(unittest.TestCase):
+	def setUp(self):
+		self.ib = ipaaca.InputBuffer('TestIn', ['sensorcategory'])
+		self.ib.register_handler(handle_iu_event)
+		self.ob = ipaaca.OutputBuffer('TestOut')
+		self.sensor_iu = ipaaca.IU('sensorcategory')
+		self.sensor_iu.payload = {'data': 'sensordata'}
+		time.sleep(0.1)
+		self.ob.add(self.sensor_iu)
+		time.sleep(0.1)
+	def tearDown(self):
+		pass
+	def testRemotePayloadChange(self):
+		self.assertIn(self.sensor_iu.uid, self.ib.iu_store)
+		received_iu = self.ib.iu_store[self.sensor_iu.uid]
+		received_iu.payload['data'] = 'updatedData'
+		time.sleep(0.1)
+		self.assertEqual(self.sensor_iu.payload['data'], 'updatedData')
+	def testRemotePayloadReplace(self):
+		self.assertIn(self.sensor_iu.uid, self.ib.iu_store)
+		received_iu = self.ib.iu_store[self.sensor_iu.uid]
+		received_iu.payload = { 'key1': 'value1', 'key2': 'value2' }
+		time.sleep(0.1)
+		self.assertEqual(len(self.sensor_iu.payload), 2)
+		self.assertEqual(self.sensor_iu.payload['key1'], 'value1')
+		self.assertEqual(self.sensor_iu.payload['key2'], 'value2')
+
+
 if __name__ == '__main__':
 	unittest.main()