zeromq集成测试

Administrator

2024-02-27 0f1dab84c23d8d435074a3ca3c8320f5088b9ab1

zeromq集成测试

 test/test_pyzmq.py

@@ -89,7 +89,6 @@
    # 分配HOST
    def distribute_upload_host(self, code):
        self.rlock.acquire()
        print("锁住")
        try:
            host_info = self.__get_available_ipchost()
            if host_info:
@@ -119,5 +118,7 @@
        ipclist.append(f"ipc://l2order{i}.ipc")
    manager = L2DataUploadProtocolManager(ipclist)
    code = "000333"
    for i in range(0, 70):
        manager.distribute_upload_host(f"{i}" * 6)
    manager.distribute_upload_host(code)
    manager.upload_data_as_json(code, {"test": "test"})