From bc9ba0236169931cd505724ffa0d53391d37492a Mon Sep 17 00:00:00 2001 From: "Mark W. Eichin" Date: Mon, 6 Apr 2009 05:51:25 +0000 Subject: unit tests for libzephyr. uses python ctypes to build structs and call functions; minimal current case is an empty ZFormatNotice/ZParseNotice round trip. --- lib/zephyr_tests.py | 310 ++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 310 insertions(+) create mode 100755 lib/zephyr_tests.py (limited to 'lib/zephyr_tests.py') diff --git a/lib/zephyr_tests.py b/lib/zephyr_tests.py new file mode 100755 index 0000000..4b865a3 --- /dev/null +++ b/lib/zephyr_tests.py @@ -0,0 +1,310 @@ +#!/usr/bin/python + +# zephyr test suite +# +# Operates on libraries in the build tree. To test purely internal +# interfaces, stuff them in a testing library... + + +"""Test Suite for libzephyr""" + +import optparse +import os +import ctypes +from ctypes import c_int, c_uint, c_ushort, c_char, c_ubyte +from ctypes import c_uint16, c_uint32 +from ctypes import POINTER, c_void_p, c_char_p +from ctypes import Structure, Union, sizeof + +__revision__ = "$Id 0 0 0 $" +__version__ = "%s/%s" % (__revision__.split()[3], __revision__.split()[2]) + +# TODO: pick some real framework later, we're just poking around for now +class TestSuite(object): + """test collection and runner""" + def __init__(self, **kwargs): + for k, v in kwargs.items(): + setattr(self, k, v) + +class TestFailure(Exception): + pass + +# POSIX socket types... +class in_addr(Structure): + _fields_ = [ + ("s_addr", c_uint32), + ] + +class _U_in6_u(Union): + _fields_ = [ + ("u6_addr8", c_ubyte * 16), + ("u6_addr16", c_uint16 * 8), + ("u6_addr32", c_uint32 * 4), + ] + +class in6_addr(Structure): + _fields_ = [ + ("in6_u", _U_in6_u), + ] + +class sockaddr(Structure): + _fields_ = [ + ("sa_family", c_uint16), + ("sa_data", c_char * 14), + ] + +class sockaddr_in(Structure): + _fields_ = [ + ("sin_family", c_uint16), + ("sin_port", c_uint16), + ("sin_addr", in_addr), + # hack from linux - do we actually need it? + ("sin_zero", c_ubyte * (sizeof(sockaddr)-sizeof(c_uint16)-sizeof(c_uint16)-sizeof(in_addr))), + ] + +# RFC2553... +class sockaddr_in6(Structure): + _fields_ = [ + ("sin6_family", c_uint16), + ("sin6_port", c_uint16), + ("sin6_flowinfo", c_uint32), + ("sin6_addr", in6_addr), + ("sin6_scope_id", c_uint32), + ] + +# zephyr/zephyr.h +#define Z_MAXOTHERFIELDS 10 /* Max unknown fields in ZNotice_t */ +Z_MAXOTHERFIELDS = 10 +#define ZAUTH (ZMakeAuthentication) +#define ZCAUTH (ZMakeZcodeAuthentication) +#define ZNOAUTH ((Z_AuthProc)0) +ZNOAUTH = 0 + +# struct _ZTimeval { +class _ZTimeval(Structure): + _fields_ = [ +# int tv_sec; + ("tv_sec", c_int), +# int tv_usec; + ("tv_usec", c_int), +# }; + ] + +# typedef struct _ZUnique_Id_t { +class ZUnique_Id_t(Structure): + _fields_ = [ + # struct in_addr zuid_addr; + ("zuid_addr", in_addr), + # struct _ZTimeval tv; + ("tv", _ZTimeval), + # } ZUnique_Id_t; + ] + +# union { +class _U_z_sender_sockaddr(Union): + _fields_ = [ + # struct sockaddr sa; + ("sa", sockaddr), + # struct sockaddr_in ip4; + ("ip4", sockaddr_in), + # struct sockaddr_in6 ip6; + ("ip6", sockaddr_in6), + # } z_sender_sockaddr; + ] + +# typedef struct _ZNotice_t { +class ZNotice_t(Structure): + _fields_ = [ + # char *z_packet; + ("z_packet", c_char_p), + # char *z_version; + ("z_version", c_char_p), + # ZNotice_Kind_t z_kind; + ("z_kind", c_int), # no enums yet + # ZUnique_Id_t z_uid; + ("z_uid", ZUnique_Id_t), + # union { + # struct sockaddr sa; + # struct sockaddr_in ip4; + # struct sockaddr_in6 ip6; + # } z_sender_sockaddr; + ("z_sender_sockaddr", _U_z_sender_sockaddr), + + # /* heavily deprecated: */ + # #define z_sender_addr z_sender_sockaddr.ip4.sin_addr + # /* probably a bad idea?: */ + # struct _ZTimeval z_time; + ("z_time", _ZTimeval), + # unsigned short z_port; + ("z_port", c_ushort), + # unsigned short z_charset; + ("z_charset", c_ushort), + # int z_auth; + ("z_auth", c_int), + # int z_checked_auth; + ("z_checked_auth", c_int), + # int z_authent_len; + ("z_authent_len", c_int), + # char *z_ascii_authent; + ("z_ascii_authent", c_char_p), + # char *z_class; + ("z_class", c_char_p), + # char *z_class_inst; + ("z_class_inst", c_char_p), + # char *z_opcode; + ("z_opcode", c_char_p), + # char *z_sender; + ("z_sender", c_char_p), + # char *z_recipient; + ("z_recipient", c_char_p), + # char *z_default_format; + ("z_default_format", c_char_p), + # char *z_multinotice; + ("z_multinotice", c_char_p), + # ZUnique_Id_t z_multiuid; + ("z_multiuid", ZUnique_Id_t), + # ZChecksum_t z_checksum; + ("z_checksum", c_uint), + # char *z_ascii_checksum; + ("z_ascii_checksum", c_char_p), + # int z_num_other_fields; + ("z_num_other_fields", c_int), + # char *z_other_fields[Z_MAXOTHERFIELDS]; + ("z_other_fields", c_char_p * Z_MAXOTHERFIELDS), + # caddr_t z_message; + ("z_message", c_char_p), # not 1980 + # int z_message_len; + ("z_message_len", c_int), + # int z_num_hdr_fields; + ("z_num_hdr_fields", c_int), + # char **z_hdr_fields; + ("z_hdr_fields", POINTER(c_char_p)), + # } ZNotice_t; + ] + + +class ZephyrTestSuite(TestSuite): + """Tests for libzephyr""" + testable_funcs = [ + "ZInitialize", + "ZGetFD", + "ZGetRealm", + "ZGetSender", + "Z_FormatRawHeader", + "ZParseNotice", + "ZFormatNotice", + ] + + def setup(self): + # find the library + self._libzephyr_path = os.path.join(self.builddir, "libzephyr.so.4.0.0") + # check for libtool... + if not os.path.exists(self._libzephyr_path): + self._libzephyr_path = os.path.join(self.builddir, ".libs", "libzephyr.so.4.0.0") + self._libzephyr = ctypes.cdll.LoadLibrary(self._libzephyr_path) + # generic bindings? + for funcname in self.testable_funcs: + setattr(self, funcname, getattr(self._libzephyr, funcname)) + + # TODO: fix return types, caller types in a more generic way later + # (perhaps by parsing the headers or code) + self.ZGetRealm.restype = ctypes.c_char_p + self.ZGetSender.restype = ctypes.c_char_p + + # Code_t + # Z_FormatRawHeader(ZNotice_t *notice, + # char *buffer, + # int buffer_len, + # int *len, + # char **cstart, + # char **cend) + # This stuffs a notice into a buffer; cstart/cend point into the checksum in buffer + self.Z_FormatRawHeader.argtypes = [ + c_void_p, # *notice + c_char_p, # *buffer + c_int, # buffer_len + POINTER(c_int), # *len + POINTER(c_char_p), # **cstart + POINTER(c_char_p), # **cend + ] + + # Code_t + # ZParseNotice(char *buffer, + # int len, + # ZNotice_t *notice) + self.ZParseNotice.argtypes = [ + c_char_p, # *buffer + c_int, # len + POINTER(ZNotice_t), # *notice + ] + + # Code_t + # ZFormatNotice(register ZNotice_t *notice, + # char **buffer, + # int *ret_len, + # Z_AuthProc cert_routine) + self.ZFormatNotice.argtypes = [ + POINTER(ZNotice_t), # *notice + POINTER(c_char_p), # **buffer + POINTER(c_int), # *ret_len + c_void_p, # cert_routine + ] + + # library-specific setup... + self.ZInitialize() + + def run(self): + tests = sorted([testname for testname in dir(self) + if testname.startswith("test_")]) + failures = [] + for test in tests: + try: + getattr(self, test)() + except TestFailure, tf: + failures.append([test, tf]) + + return failures + + def cleanup(self): + # no cleanup needed yet + pass + + def test_zinit(self): + """test that ZInitialize did something""" + print "fd", self.ZGetFD() + realm = self.ZGetRealm() + print "realm", realm + if not realm or realm == "local-realm": + raise TestFailure("useless realm %s" % realm) + print self.ZGetSender() + + def test_notices(self): + """test notice construct/destruct""" + notice = ZNotice_t() + print "sizeof ZNotice_t", sizeof(notice) + zbuf = c_char_p(0) + zbuflen = c_int(0) + st = self.ZFormatNotice(notice, zbuf, zbuflen, ZNOAUTH) + print "ZFormatNotice:", "retval", st + print "\tzbuflen", zbuflen + print "\tzbuf", repr(zbuf.value) + new_notice = ZNotice_t() + st = self.ZParseNotice(zbuf, zbuflen, new_notice) + print "ZParseNotice:", "retval", st + print "\tz_version", new_notice.z_version + + +if __name__ == "__main__": + parser = optparse.OptionParser(usage=__doc__, + version = "%%prog %s" % __version__) + parser.add_option("--builddir", default="..", + help="where to find the top of the build tree") + opts, args = parser.parse_args() + assert not args, "no args yet" + + tester = ZephyrTestSuite(builddir=os.path.join(opts.builddir, "lib")) + tester.setup() + failures = tester.run() + tester.cleanup() + for failure, exc in failures: + print "FAIL:", failure, str(exc) -- cgit v1.2.3