From e78bce1793a4645d3cddf10b1eb06640bf4a202a Mon Sep 17 00:00:00 2001 From: Arturo Hernandez Date: Thu, 14 Jun 2018 21:12:40 -0400 Subject: [PATCH] added some test --- test.py | 125 ++++++++++++++++++++++++++++++++++++++++++++++++----- zk/base.py | 2 +- 2 files changed, 116 insertions(+), 11 deletions(-) diff --git a/test.py b/test.py index 1887407..b4f8a37 100644 --- a/test.py +++ b/test.py @@ -45,15 +45,17 @@ class PYZKTest(unittest.TestCase): @patch('zk.base.ZK_helper') def test_no_ping(self,helper): """ what if ping doesn't response """ + helper.return_value.test_ping.return_value = False #no ping simulated + #begin zk = ZK('192.168.1.201') helper.assert_called_with('192.168.1.201', 4370) # called correctly - helper.return_value.test_ping.return_value = False #no ping simulated self.assertRaisesRegexp(ZKNetworkError, "can't reach device", zk.connect) @patch('zk.base.ZK_helper') def test_correct_ping(self,helper): """ what if ping is ok """ helper.return_value.test_ping.return_value = True # ping simulated + #begin zk = ZK('192.168.1.201') helper.assert_called_with('192.168.1.201', 4370) # called correctly self.assertRaisesRegexp(ZKNetworkError, "unpack requires", zk.connect) # no data...? @@ -65,7 +67,6 @@ class PYZKTest(unittest.TestCase): helper.return_value.test_ping.return_value = True # ping simulated helper.return_value.test_tcp.return_value = 0 # helper tcp ok socket.return_value.recv.return_value = 'Invalid tcp data' - #socket.return_value.connect_ex.return_value = 0 # socket tcp ok #begin zk = ZK('192.168.1.201') helper.assert_called_with('192.168.1.201', 4370) # called correctly @@ -93,7 +94,7 @@ class PYZKTest(unittest.TestCase): helper.return_value.test_tcp.return_value = 0 # helper tcp ok socket.return_value.recv.return_value = codecs.decode('d007fffc2ffb0000','hex') # tcp CMD_ACK_OK #begin - zk = ZK('192.168.1.201', force_udp=True) # already tested + zk = ZK('192.168.1.201', force_udp=True) conn = zk.connect() socket.return_value.sendto.assert_called_with(codecs.decode('e80317fc00000000', 'hex'), ('192.168.1.201', 4370)) conn.disconnect() @@ -107,23 +108,127 @@ class PYZKTest(unittest.TestCase): helper.return_value.test_tcp.return_value = 1 # helper tcp nope socket.return_value.recv.return_value = codecs.decode('d007fffc2ffb0000','hex') # tcp CMD_ACK_OK #begin - zk = ZK('192.168.1.201', verbose=True) # already tested + zk = ZK('192.168.1.201') conn = zk.connect() + socket.return_value.sendto.assert_called_with(codecs.decode('e80317fc00000000', 'hex'), ('192.168.1.201', 4370)) + conn.disconnect() + socket.return_value.sendto.assert_called_with(codecs.decode('e903e6002ffb0100', 'hex'), ('192.168.1.201', 4370)) + + @patch('zk.base.socket') + @patch('zk.base.ZK_helper') + def test_tcp_unauth(self, helper, socket): + """ Basic auth test """ + helper.return_value.test_ping.return_value = True # ping simulated + helper.return_value.test_tcp.return_value = 0 # helper tcp ok + socket.return_value.recv.side_effect = [ + codecs.decode('5050827d08000000d5075bb2cf450000', 'hex'), # tcp CMD_UNAUTH + codecs.decode('5050827d08000000d5075ab2cf450100', 'hex') # tcp CMD_UNAUTH + ] + #begin + zk = ZK('192.168.1.201', password=12) + self.assertRaisesRegexp(ZKErrorResponse, "Unauthenticated", zk.connect) + socket.return_value.send.assert_called_with(codecs.decode('5050827d0c0000004e044e2ccf450100614d323c', 'hex')) # try with password 12 + + @patch('zk.base.socket') + @patch('zk.base.ZK_helper') + def test_tcp_auth(self, helper, socket): + """ Basic auth test """ + helper.return_value.test_ping.return_value = True # ping simulated + helper.return_value.test_tcp.return_value = 0 # helper tcp ok + socket.return_value.recv.side_effect = [ + codecs.decode('5050827d08000000d5075bb2cf450000', 'hex'), # tcp CMD_UNAUTH + codecs.decode('5050827d08000000d0075fb2cf450100', 'hex'), # tcp CMD_ACK_OK + codecs.decode('5050827d08000000d00745b2cf451b00', 'hex') # tcp random CMD_ACK_OK TODO: generate proper sequenced response + + ] + #begin + zk = ZK('192.168.1.201', password=45) + conn = zk.connect() + socket.return_value.send.assert_called_with(codecs.decode('5050827d0c0000004e044db0cf45010061c9323c', 'hex')) #auth with pass 45 + conn.disconnect() + socket.return_value.send.assert_called_with(codecs.decode('5050827d08000000e90345b6cf450200', 'hex')) #exit + + @patch('zk.base.socket') + @patch('zk.base.ZK_helper') + def test_tcp_get_size(self, helper, socket): + """ can read sizes? """ + helper.return_value.test_ping.return_value = True # ping simulated + helper.return_value.test_tcp.return_value = 0 # helper tcp ok + socket.return_value.recv.side_effect = [ + codecs.decode('5050827d08000000d0075fb2cf450100', 'hex'), # tcp CMD_ACK_OK + codecs.decode('5050827d64000000d007a3159663130000000000000000000000000000000000070000000000000006000000000000005d020000000000000f0c0000000000000100000000000000b80b000010270000a0860100b20b00000927000043840100000000000000', 'hex'), #sizes + codecs.decode('5050827d08000000d00745b2cf451b00', 'hex'), # tcp random CMD_ACK_OK TODO: generate proper sequenced response + ] + #begin + zk = ZK('192.168.1.201') # already tested + conn = zk.connect() + socket.return_value.send.assert_called_with(codecs.decode('5050827d08000000e80317fc00000000', 'hex')) + conn.read_sizes() + socket.return_value.send.assert_called_with(codecs.decode('5050827d080000003200fcb9cf450200', 'hex')) + conn.disconnect() + self.assertEqual(conn.users, 7, "missed user data %s" % conn.users) + self.assertEqual(conn.fingers, 6, "missed finger data %s" % conn.fingers) + self.assertEqual(conn.records, 605, "missed record data %s" % conn.records) + self.assertEqual(conn.users_cap, 10000, "missed user cap %s" % conn.users_cap) + self.assertEqual(conn.fingers_cap, 3000, "missed finger cap %s" % conn.fingers_cap) + self.assertEqual(conn.rec_cap, 100000, "missed record cap %s" % conn.rec_cap) + + @patch('zk.base.socket') + @patch('zk.base.ZK_helper') + def test_tcp_get_users_small(self, helper, socket): + """ can get empty? """ + helper.return_value.test_ping.return_value = True # ping simulated + helper.return_value.test_tcp.return_value = 0 # helper tcp ok + socket.return_value.recv.side_effect = [ + codecs.decode('5050827d08000000d0075fb2cf450100', 'hex'), # tcp CMD_ACK_OK + codecs.decode('5050827d64000000d007a3159663130000000000000000000000000000000000070000000000000006000000000000005d020000000000000f0c0000000000000100000000000000b80b000010270000a0860100b20b00000927000043840100000000000000', 'hex'), #sizes + codecs.decode('5050827d04020000dd05942c96631500f801000001000e0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000003830380000000000000000000000000000000000000000000200000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000003832310000000000000000000000000000000000000000000300000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000003833350000000000000000000000000000000000000000000400000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000003833310000000000000000000000000000000000000000000500000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000003833320000000000000000000000000000000000000000000600000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000003836000000000000000000000000000000000000000000000c0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000383432000000000000000000000000000000000000000000','hex'), #DATA directly(not ok) + codecs.decode('5050827d08000000d00745b2cf451b00', 'hex'), # tcp random CMD_ACK_OK TODO: generate proper sequenced response + #codecs.decode('5050827d08000000d00745b2cf451b00', 'hex') # tcp random CMD_ACK_OK TODO: generate proper sequenced response + ] + #begin + zk = ZK('192.168.1.201' ) + conn = zk.connect() + socket.return_value.send.assert_called_with(codecs.decode('5050827d08000000e80317fc00000000', 'hex')) + users = conn.get_users() + socket.return_value.send.assert_called_with(codecs.decode('5050827d13000000df053ca6cf4514000109000500000000000000', 'hex')) #get users + self.assertEqual(len(users), 7, "incorrect size %s" % len(users)) + #assert one user + usu = users[3] + self.assertIsInstance(usu.uid, int, "uid should be int() %s" % type(usu.uid)) + self.assertIsInstance(usu.user_id, (str, unicode), "user_id should be str() or unicode() %s" % type(usu.user_id)) + self.assertEqual(usu.uid, 4, "incorrect uid %s" % usu.uid) + self.assertEqual(usu.user_id, "831", "incorrect user_id %s" % usu.user_id) + self.assertEqual(usu.name, "NN-831", "incorrect uid %s" % usu.name) # generated conn.disconnect() @patch('zk.base.socket') @patch('zk.base.ZK_helper') - def test_tcp_(self, helper, socket): - """ Basic connection test """ + def _test_tcp_get_template(self, helper, socket): + """ can get empty? """ helper.return_value.test_ping.return_value = True # ping simulated helper.return_value.test_tcp.return_value = 0 # helper tcp ok - socket.return_value.recv.return_value = codecs.decode('5050827d08000000d007fffc2ffb0000','hex') # tcp CMD_ACK_OK + socket.return_value.recv.side_effect = [ + codecs.decode('5050827d08000000d0075fb2cf450100', 'hex'), # tcp CMD_ACK_OK + codecs.decode('5050827d15000000d007acf93064160000941d0000941d0000b400be00', 'hex'), # ack ok with size 7572 + codecs.decode('5050827d10000000dc05477830641700941d000000000100', 'hex'), #prepare data + codecs.decode('5050827d08000000d00745b2cf451b00', 'hex'), # tcp random CMD_ACK_OK TODO: generate proper sequenced response + #codecs.decode('5050827d08000000d00745b2cf451b00', 'hex') # tcp random CMD_ACK_OK TODO: generate proper sequenced response + ] #begin - zk = ZK('192.168.1.201') # already tested + zk = ZK('192.168.1.201', verbose=True) conn = zk.connect() + socket.return_value.send.assert_called_with(codecs.decode('5050827d08000000e80317fc00000000', 'hex')) + templates = conn.get_templates() + self.assertEqual(len(templates), 6, "incorrect size %s" % len(templates)) + #assert one user + usu = users[3] + self.assertIsInstance(usu.uid, int, "uid should be int() %s" % type(usu.uid)) + self.assertIsInstance(usu.user_id, (str, unicode), "user_id should be str() or unicode() %s" % type(usu.user_id)) + self.assertEqual(usu.uid, 4, "incorrect uid %s" % usu.uid) + self.assertEqual(usu.user_id, "831", "incorrect user_id %s" % usu.user_id) + self.assertEqual(usu.name, "NN-831", "incorrect uid %s" % usu.name) # generated conn.disconnect() - - if __name__ == '__main__': unittest.main() \ No newline at end of file diff --git a/zk/base.py b/zk/base.py index a86f25d..29baa14 100644 --- a/zk/base.py +++ b/zk/base.py @@ -1341,7 +1341,7 @@ class ZK(object): def __recieve_chunk(self): """ recieve a chunk """ if self.__response == const.CMD_DATA: # less than 1024!!! - if self.verbose: print ("size was {} len is {}".format(size, len(self.__data))) + if self.verbose: print ("_rc len is {}".format(len(self.__data))) return self.__data #without headers elif self.__response == const.CMD_PREPARE_DATA: data = []