Cluster test3 (#2286)

### What problem does this PR solve?

Add test
Refactor cluster test logger.
Refactor cluster clear with __exit__

### Type of change

- [x] Refactoring
- [x] Test cases

python/infinity_http.py

@@ -86,8 +86,6 @@ def request(self, url, method, header={}, data={}):
                         pass
                     else:
                         raise e
-                except Exception as e:
-                    raise e
                 else:
                     break
                 logging.debug(f"retry {i} times")
@@ -867,12 +865,6 @@ def fusion(self, method: str, topn: int, fusion_params: Optional[dict] = None):
         return self
 
     def to_result(self):
-        self.select()
-
-    def to_pl(self):
-        return pl.from_pandas(self.to_df())
-
-    def to_df(self):
         if self.output_res == []:
             self.select()
 
@@ -914,7 +906,7 @@ def to_df(self):
                     new_tup = tup + (res[k],)
                 df_dict[k] = new_tup
         # print(self.output_res)
-        print(df_dict)
+        # print(df_dict)
 
         df_type = {}
         for k in df_dict:
@@ -949,6 +941,13 @@ def to_df(self):
                     if (function_name in bool_functions):
                         df_type[k] = dtype('bool')
                         break
+        return df_dict, df_type
+
+    def to_pl(self):
+        return pl.from_pandas(self.to_df())
+
+    def to_df(self):
+        df_dict, df_type = self.to_result()
         return pd.DataFrame(df_dict).astype(df_type)
 
     def to_arrow(self):

python/parallel_test/conftest.py

@@ -87,7 +87,7 @@ def pytest_addoption(parser):
 def pytest_configure(config):
     logger = logging.getLogger("run_parallel_test")
     logger.setLevel(logging.INFO)
-    handler = logging.FileHandler(log_output_file)
+    handler = logging.FileHandler(log_output_file, delay=True)
     logger.addHandler(handler)
     logger.addHandler(logging.StreamHandler())
     formatter = logging.Formatter('%(asctime)s - %(threadName)s - %(levelname)s - %(message)s')

python/restart_test/infinity_runner.py

@@ -36,7 +36,7 @@ def __init__(self, infinity_path: str, *, logger=None):
             handler.setFormatter(formatter)
             self.logger.addHandler(handler)
 
-            handler = logging.FileHandler(PYTEST_LOG_FILE)
+            handler = logging.FileHandler(PYTEST_LOG_FILE, delay=True)
             handler.setLevel(logging.DEBUG)
             handler.setFormatter(formatter)
             self.logger.addHandler(handler)

python/test_cluster/conftest.py

@@ -1,9 +1,8 @@
-import os
 import pytest
 from infinity_cluster import InfinityCluster, MinioParams
 from docker_infinity_cluster import DockerInfinityCluster
 from mocked_infinity_cluster import MockInfinityCluster
-import logging
+
 
 def pytest_addoption(parser):
     parser.addoption(
@@ -30,18 +29,10 @@ def pytest_addoption(parser):
     parser.addoption("--docker", action="store_true", default=False)
 
 
-log_output_file = "run_cluster_test.log"
 def pytest_configure(config):
     config.addinivalue_line(
         "markers", "docker: mark test to run only when --docker option is provided"
     )
-    logger = logging.getLogger("run_parallel_test")
-    logger.setLevel(logging.INFO)
-    handler = logging.FileHandler(log_output_file)
-    logger.addHandler(handler)
-    logger.addHandler(logging.StreamHandler())
-    formatter = logging.Formatter('%(asctime)s - %(threadName)s - %(levelname)s - %(message)s')
-    handler.setFormatter(formatter)
 
 
 def pytest_collection_modifyitems(config, items):
@@ -67,24 +58,19 @@ def pytest_generate_tests(metafunc):
     print("infinity_dir: ", infinity_dir)
     # print(metafunc.fixturenames)
 
+    test_name = metafunc.function.__name__
     if "docker_cluster" in metafunc.fixturenames:
-        # skip if docker is in option and the testcase is marked with docker
-        if (
-            not metafunc.config.getoption("--docker")
-            and "docker" in metafunc.definition.keywords
-        ):
-            return
-
-        print("Init DockerInfinityCluster")
         docker_infinity_cluster = DockerInfinityCluster(
             infinity_path, minio_params=minio_params, infinity_dir=infinity_dir
         )
         metafunc.parametrize("docker_cluster", [docker_infinity_cluster])
     elif "cluster" in metafunc.fixturenames:
-        infinity_cluster = InfinityCluster(infinity_path, minio_params=minio_params)
+        infinity_cluster = InfinityCluster(
+            infinity_path, minio_params=minio_params, test_name=test_name
+        )
         metafunc.parametrize("cluster", [infinity_cluster])
     elif "mock_cluster" in metafunc.fixturenames:
         mock_infinity_cluster = MockInfinityCluster(
-            infinity_path, minio_params=minio_params
+            infinity_path, minio_params=minio_params, test_name=test_name
         )
         metafunc.parametrize("mock_cluster", [mock_infinity_cluster])

python/test_cluster/database_operations.py

@@ -1,15 +1,9 @@
 import time
 
-import pytest
 import infinity_http
-from infinity_cluster import InfinityCluster
-from mocked_infinity_cluster import MockInfinityCluster
-from docker_infinity_cluster import DockerInfinityCluster, MinioParams
 from numpy import dtype
 import pandas as pd
 import time
-from infinity.errors import ErrorCode
-from infinity.common import InfinityException
 from infinity.common import ConflictType
 
 def do_some_operations(client : infinity_http.infinity_http) :