From 55a1130373215488cdafa3ee9b6d2fdeeb383253 Mon Sep 17 00:00:00 2001
From: Ramin Yaghoubzadeh <ryaghoub@techfak.uni-bielefeld.de>
Date: Mon, 5 Mar 2012 13:48:19 +0100
Subject: [PATCH] local tests
---
python/test/src/testipaaca.py | 53 ++++++++++++++++++++++++++++++++++-
1 file changed, 52 insertions(+), 1 deletion(-)
diff --git a/python/test/src/testipaaca.py b/python/test/src/testipaaca.py
index 9aa9313..a2a3c93 100755
--- a/python/test/src/testipaaca.py
+++ b/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()
--
GitLab