From b69471f45cab3e26788812d465c013288f319066 Mon Sep 17 00:00:00 2001 From: Arturo Hernandez Date: Wed, 13 Jun 2018 15:23:21 -0400 Subject: [PATCH] added some unittests (and moved basic test) --- basic_test.py | 36 +++++++++++ test.py | 155 ++++++++++++++++++++++++++++++++++++++---------- test_machine.py | 4 +- zk/base.py | 5 +- 4 files changed, 164 insertions(+), 36 deletions(-) create mode 100644 basic_test.py diff --git a/basic_test.py b/basic_test.py new file mode 100644 index 0000000..388b9d0 --- /dev/null +++ b/basic_test.py @@ -0,0 +1,36 @@ +# -*- coding: utf-8 -*- +import sys +sys.path.append("zk") + +from zk import ZK, const + +conn = None +zk = ZK('192.168.1.201', port=4370, timeout=5, password=0, force_udp=False, ommit_ping=False) +try: + print ('Connecting to device ...') + conn = zk.connect() + print ('Disabling device ...') + conn.disable_device() + print ('Firmware Version: : {}'.format(conn.get_firmware_version())) + # print '--- Get User ---' + users = conn.get_users() + for user in users: + privilege = 'User' + if user.privilege == const.USER_ADMIN: + privilege = 'Admin' + print ('- UID #{}'.format(user.uid)) + print (' Name : {}'.format(user.name)) + print (' Privilege : {}'.format(privilege)) + print (' Password : {}'.format(user.password)) + print (' Group ID : {}'.format(user.group_id)) + print (' User ID : {}'.format(user.user_id)) + + print ("Voice Test ...") + conn.test_voice() + print ('Enabling device ...') + conn.enable_device() +except Exception as e: + print ("Process terminate : {}".format(e)) +finally: + if conn: + conn.disconnect() diff --git a/test.py b/test.py index 388b9d0..1887407 100644 --- a/test.py +++ b/test.py @@ -1,36 +1,129 @@ -# -*- coding: utf-8 -*- import sys -sys.path.append("zk") - +import os +import unittest +import codecs +from mock import patch, Mock, MagicMock +mock_socket = MagicMock(name='zk.socket') +sys.modules['zk.socket'] = mock_socket from zk import ZK, const +from zk.base import ZK_helper +from zk.user import User +from zk.finger import Finger +from zk.attendance import Attendance +from zk.exception import ZKErrorResponse, ZKNetworkError +def dump(obj, nested_level=0, output=sys.stdout): + spacing = ' ' + if type(obj) == dict: + print >> output, '%s{' % ((nested_level) * spacing) + for k, v in obj.items(): + if hasattr(v, '__iter__'): + print >> output, '%s%s:' % ((nested_level + 1) * spacing, k) + dump(v, nested_level + 1, output) + else: + print >> output, '%s%s: %s' % ((nested_level + 1) * spacing, k, v) + print >> output, '%s}' % (nested_level * spacing) + elif type(obj) == list: + print >> output, '%s[' % ((nested_level) * spacing) + for v in obj: + if hasattr(v, '__iter__'): + dump(v, nested_level + 1, output) + else: + print >> output, '%s%s' % ((nested_level + 1) * spacing, v) + print >> output, '%s]' % ((nested_level) * spacing) + else: + print >> output, '%s%s' % (nested_level * spacing, obj) -conn = None -zk = ZK('192.168.1.201', port=4370, timeout=5, password=0, force_udp=False, ommit_ping=False) -try: - print ('Connecting to device ...') - conn = zk.connect() - print ('Disabling device ...') - conn.disable_device() - print ('Firmware Version: : {}'.format(conn.get_firmware_version())) - # print '--- Get User ---' - users = conn.get_users() - for user in users: - privilege = 'User' - if user.privilege == const.USER_ADMIN: - privilege = 'Admin' - print ('- UID #{}'.format(user.uid)) - print (' Name : {}'.format(user.name)) - print (' Privilege : {}'.format(privilege)) - print (' Password : {}'.format(user.password)) - print (' Group ID : {}'.format(user.group_id)) - print (' User ID : {}'.format(user.user_id)) - print ("Voice Test ...") - conn.test_voice() - print ('Enabling device ...') - conn.enable_device() -except Exception as e: - print ("Process terminate : {}".format(e)) -finally: - if conn: +class PYZKTest(unittest.TestCase): + def setup(self): + + pass + + def tearDown(self): + pass + + @patch('zk.base.ZK_helper') + def test_no_ping(self,helper): + """ what if ping doesn't response """ + 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 + 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...? + + @patch('zk.base.socket') + @patch('zk.base.ZK_helper') + def test_tcp_invalid(self, helper, socket): + """ Basic connection 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.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 + self.assertRaisesRegexp(ZKNetworkError, "TCP packet invalid", zk.connect) + + @patch('zk.base.socket') + @patch('zk.base.ZK_helper') + def test_tcp_connect(self, helper, socket): + """ Basic connection 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.return_value = codecs.decode('5050827d08000000d007fffc2ffb0000','hex') # tcp CMD_ACK_OK + #begin + zk = ZK('192.168.1.201') # already tested + conn = zk.connect() + socket.return_value.send.assert_called_with(codecs.decode('5050827d08000000e80317fc00000000', 'hex')) conn.disconnect() + socket.return_value.send.assert_called_with(codecs.decode('5050827d08000000e903e6002ffb0100', 'hex')) + + @patch('zk.base.socket') + @patch('zk.base.ZK_helper') + def test_force_udp_connect(self, helper, socket): + """ Force UDP connection 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.return_value = codecs.decode('d007fffc2ffb0000','hex') # tcp CMD_ACK_OK + #begin + zk = ZK('192.168.1.201', force_udp=True) # already tested + 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_udp_connect(self, helper, socket): + """ Basic auto UDP connection test """ + helper.return_value.test_ping.return_value = True # ping simulated + 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 + conn = zk.connect() + conn.disconnect() + + @patch('zk.base.socket') + @patch('zk.base.ZK_helper') + def test_tcp_(self, helper, socket): + """ Basic connection 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.return_value = codecs.decode('5050827d08000000d007fffc2ffb0000','hex') # tcp CMD_ACK_OK + #begin + zk = ZK('192.168.1.201') # already tested + conn = zk.connect() + conn.disconnect() + + + +if __name__ == '__main__': + unittest.main() \ No newline at end of file diff --git a/test_machine.py b/test_machine.py index 4f37979..4c907ad 100755 --- a/test_machine.py +++ b/test_machine.py @@ -10,10 +10,10 @@ from builtins import input sys.path.append("zk") from zk import ZK, const -from zk.attendance import Attendance -from zk.exception import ZKErrorResponse, ZKNetworkError from zk.user import User from zk.finger import Finger +from zk.attendance import Attendance +from zk.exception import ZKErrorResponse, ZKNetworkError conn = None diff --git a/zk/base.py b/zk/base.py index 76d6920..b27f166 100644 --- a/zk/base.py +++ b/zk/base.py @@ -70,8 +70,8 @@ class ZK_helper(object): """ Returns True if host responds to a ping request """ + print "ping" import subprocess, platform - # Ping parameters as function of OS ping_str = "-n 1" if platform.system().lower()=="windows" else "-c 1 -W 5" args = "ping " + " " + ping_str + " " + self.ip @@ -202,7 +202,6 @@ class ZK(object): ''' send command to the terminal ''' - buf = self.__create_header(command, command_string, self.__session_id, self.__reply_id) try: if self.tcp: @@ -211,7 +210,7 @@ class ZK(object): self.__tcp_data_recv = self.__sock.recv(response_size + 8) self.__tcp_length = self.__test_tcp_top(self.__tcp_data_recv) if self.__tcp_length == 0: - raise ZKNetworkError("TCP Packet invalid") + raise ZKNetworkError("TCP packet invalid") self.__header = unpack('<4H', self.__tcp_data_recv[8:16]) self.__data_recv = self.__tcp_data_recv[8:] # dirty hack else: