utils keep going and unit tests

binding/python/util_test.py

@@ -0,0 +1,68 @@
+# Copyright 2022 The Ip2Region Authors. All rights reserved.
+# Use of this source code is governed by a Apache2.0-style
+# license that can be found in the LICENSE file.
+
+# util test script on 2025/10/29
+# Author Leon<chenxin619315@gmail.com>
+
+import os
+import sys
+import time
+from xdb import util
+
+script_dir = os.path.dirname(__file__)
+data_dir = os.path.join(script_dir, '../../data/')
+xdb_v4_path = os.path.join(data_dir, "ip2region_v4.xdb")
+xdb_v6_path = os.path.join(data_dir, "ip2region_v6.xdb")
+
+# print(script_dir, data_dir, xdb_v4_path, xdb_v6_path)
+def test_version():
+    print("1, version contants: ")
+    print("IPv4 -> ", util.IPv4)
+    print("IPv6 -> ", util.IPv6)
+
+    # version from name
+    print("2, version from name: ")
+    for name in ["v4", "IPv4", "v4x", "v6", "IPv6", "v6x"]:
+        print("version_from_name({}) -> ".format(name), util.version_from_name(name))
+
+    # version from header
+    print("3, version from header: ")
+    v4_header = util.load_header_from_file(xdb_v4_path)
+    v6_header = util.load_header_from_file(xdb_v6_path)
+    print("version_from_header(v4_header) -> ", util.version_from_header(v4_header))
+    print("version_from_header(v6_header) -> ", util.version_from_header(v6_header))
+
+def test_load_header():
+    v4_header = util.load_header_from_file(xdb_v4_path)
+    v6_header = util.load_header_from_file(xdb_v6_path)
+    print("v4_header -> ", v4_header)
+    print("v6_header -> ", v6_header)
+
+def test_load_vector_index():
+    v4_v_index = util.load_vector_index_from_file(xdb_v4_path)
+    v6_v_index = util.load_vector_index_from_file(xdb_v6_path)
+    print("v4_v_index.length={}".format(len(v4_v_index)))
+    print("v6_v_index.length={}".format(len(v6_v_index)))
+
+def test_load_content():
+    v4_content = util.load_content_from_file(xdb_v4_path)
+    v6_content = util.load_content_from_file(xdb_v6_path)
+    print("v4_content.length={}".format(len(v4_content)))
+    print("v6_content.length={}".format(len(v6_content)))
+
+if __name__ == "__main__":
+    # check and call the specified function
+    if len(sys.argv) < 2:
+        sys.exit("please specified the function to test")
+    
+    func = sys.argv[1]
+    all_ids = globals()
+    if func in all_ids and callable(all_ids[func]):
+        print("+---calling test function {} ...".format(func))
+        s_time = time.time()
+        all_ids[func]()
+        c_time = time.time() - s_time
+        print(f"|---Done, elapsed {c_time:.6f}s")
+    else:
+        sys.exit("unable to call function {}".format(func))

binding/python/xdb/util.py

@@ -6,7 +6,6 @@
 # Author Leon<chenxin619315@gmail.com>
 
 import io
-import struct
 
 # global constants
 XdbStructure20 = 2
@@ -18,21 +17,24 @@ HeaderInfoLength = 256
 VectorIndexRows  = 256
 VectorIndexCols  = 256
 VectorIndexSize  = 8
+# cache of VectorIndexCols × VectorIndexRows × VectorIndexSize
+VectorIndexLength = 524288
 
 class Header(object):
     '''
     header class
     '''
     def __init__(self, buff):
-        self.version = struct.unpack_from("<H", buff, 0)[0]
-        self.indexPolicy = struct.unpack_from("<H", buff, 2)[0]
-        self.createdAt = struct.unpack_from("<I", buff, 4)[0]
-        self.startIndexPtr = struct.unpack_from("<I", buff, 8)[0]
-        self.endIndexPtr = struct.unpack_from("<I", buff, 12)[0]
+        self.version = le_get_uint16(buff, 0)
+        self.indexPolicy = le_get_uint16(buff, 2)
+        self.createdAt = le_get_uint32(buff, 4)
+        self.startIndexPtr = le_get_uint32(buff, 8)
+        self.endIndexPtr = le_get_uint32(buff, 12)
 
         # since IPv6 supporting
-        self.ipVersion = struct.unpack_from("<H", buff, 16)[0]
-        self.runtimePtrBytes = struct.unpack_from("<H", buff, 18)[0]
+        self.ipVersion = le_get_uint16(buff, 16)
+        self.runtimePtrBytes = le_get_uint16(buff, 18)
+
 
         # keep the raw data
         self.buff = buff
@@ -98,6 +100,48 @@ def version_from_header(header):
         return None
 
 
+# ---
+# ip parse and convert functions
+
+def parse_ip(ip_string):
+    pass
+
+def ip_to_string(ip_bytes):
+    pass
+
+def ip_compare(ip1, ip2):
+    pass
+
+def ip_sub_compare(ip1, ip2, offset):
+    pass
+
+
+# ---
+# buffer decode functions
+
+def le_get_uint32(buff, offset):
+    '''
+    decode an unsinged 4-bytes int from a buffer started from offset
+    with little byte endian
+    '''
+    return (
+        ((buff[offset  ]) & 0x000000FF) |
+        ((buff[offset+1] <<  8) & 0x0000FF00) | 
+        ((buff[offset+2] << 16) & 0x00FF0000) |
+        ((buff[offset+3] << 24) & 0xFF000000)
+    )
+
+def le_get_uint16(buff, offset):
+    '''
+    decode an unsinged 2-bytes short from a buffer started from offset
+    with little byte endian
+    '''
+    return (
+        ((buff[offset  ]) & 0x000000FF) |
+        ((buff[offset+1] <<  8) & 0x0000FF00)
+    )
+
+
 # ---
 # xdb buffer load functions
 
@@ -106,8 +150,7 @@ def load_header(handle):
     load xdb header from a specified file handle
     '''
     handle.seek(0)
-    buff = handle.read(HeaderInfoLength)
-    return Header(buff)
+    return Header(handle.read(HeaderInfoLength))
 
 def load_header_from_file(db_file):
     handle = io.open(db_file, "rb")
@@ -115,15 +158,28 @@ def load_header_from_file(db_file):
     handle.close()
     return header
 
+def load_vector_index(handle):
+    '''
+    load xdb vector index from a specified file handle
+    '''
+    handle.seek(HeaderInfoLength)
+    return handle.read(VectorIndexLength)
 
-if __name__ == "__main__":
-    # header class test
-    header = load_header_from_file("../../../data/ip2region_v4.xdb")
-    print(header)
+def load_vector_index_from_file(db_file):
+    handle = io.open(db_file, "rb")
+    v_index = load_vector_index(handle)
+    handle.close()
+    return v_index
 
-    # verison class test
-    print("IPv4 ->", IPv4)
-    print("IPv6 ->", IPv6)
-    print("version_from_name(v4) ->", version_from_name("v4"))
-    print("version_from_name(v6) ->", version_from_name("v4"))
-    print("version_from_header() ->", version_from_header(header))
+def load_content(handle):
+    '''
+    load the whole xdb content from a specified file handle
+    '''
+    handle.seek(0)
+    return handle.read()
+
+def load_content_from_file(db_file):
+    handle = io.open(db_file, "rb")
+    c_buff = load_content(handle)
+    handle.close()
+    return c_buff