cubetiq/pyztk
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

added some unittests (and moved basic test)

Browse Source
This commit is contained in:
Arturo Hernandez 2018-06-13 15:23:21 -04:00
parent 4cedf3ee5b
commit b69471f45c
4 changed files with 164 additions and 36 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

36
basic_test.py Normal file
View File

@ -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()

155
test.py
View File

@ -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 ...')
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()
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:
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()

4
test_machine.py
View File

@ -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

5
zk/base.py
View File

@ -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: