[apache#2292] improvement(PyClient): Fix unstable Python client integ…

…ration test in Github Action

.github/workflows/backend-integration-test.yml

@@ -95,7 +95,7 @@ jobs:
       - name: Backend Integration Test
         id: integrationTest
         run: |
-          ./gradlew test --rerun-tasks -PskipTests -PtestMode=${{ matrix.test-mode }} -PjdkVersion=${{ matrix.java-version }} -PskipWebITs -P${{ matrix.backend }}
+          ./gradlew test --rerun-tasks -PskipTests -PtestMode=${{ matrix.test-mode }} -PjdkVersion=${{ matrix.java-version }} -PskipWebITs -P${{ matrix.backend }} -PskipPyClientITs
 
       - name: Upload integrate tests reports
         uses: actions/upload-artifact@v3

.github/workflows/python-integration-test.yml

@@ -67,7 +67,7 @@ jobs:
           ./gradlew compileDistribution -x test -PjdkVersion=${{ matrix.java-version }}
           for pythonVersion in "3.8" "3.9" "3.10" "3.11"
           do
-            ./gradlew -PjdkVersion=${{ matrix.java-version }} -PpythonVersion=${pythonVersion} :client:client-python:integrationTest
+            ./gradlew -PjdkVersion=${{ matrix.java-version }} -PpythonVersion=${pythonVersion} :client:client-python:test
           done
 
       - name: Upload integrate tests reports

clients/client-python/build.gradle.kts

@@ -26,9 +26,9 @@ fun gravitinoServer(operation: String) {
     val exitCode = process.waitFor()
     if (exitCode == 0) {
       val currentContext = process.inputStream.bufferedReader().readText()
-      println("Current docker context is: $currentContext")
+      println("Gravitino server status: $currentContext")
     } else {
-      println("checkOrbStackStatus Command execution failed with exit code $exitCode")
+      println("Gravitino server execution failed with exit code $exitCode")
     }
 }
 
@@ -39,26 +39,25 @@ tasks {
   }
 
   val test by registering(VenvTask::class) {
-    dependsOn(pipInstall)
-    venvExec = "python"
-    args = listOf("-m", "unittest")
-    workingDir = projectDir.resolve(".")
-  }
-
-  val integrationTest by registering(VenvTask::class) {
-    doFirst() {
-      gravitinoServer("start")
-    }
+    val skipPyClientITs = project.hasProperty("skipPyClientITs")
+    if (!skipPyClientITs) {
+      doFirst {
+        gravitinoServer("start")
+      }
 
-    dependsOn(pipInstall)
-    venvExec = "python"
-    args = listOf("-m", "unittest")
-    workingDir = projectDir.resolve(".")
-    environment = mapOf("PROJECT_VERSION" to project.version,
-      "GRADLE_START_GRAVITINO" to "True")
+      dependsOn(pipInstall)
+      venvExec = "python"
+      args = listOf("-m", "unittest")
+      workingDir = projectDir.resolve(".")
+      environment = mapOf(
+        "PROJECT_VERSION" to project.version,
+        "GRAVITINO_HOME" to project.rootDir.path + "/distribution/package",
+        "EXTERNAL_START_GRAVITINO" to "true"
+      )
 
-    doLast {
-      gravitinoServer("stop")
+      doLast {
+        gravitinoServer("stop")
+      }
     }
   }
 

clients/client-python/tests/integration/integration_test_env.py

@@ -20,11 +20,11 @@ def get_gravitino_server_version():
         response.close()
         return True
     except requests.exceptions.RequestException as e:
-        logger.warning("Failed to access the server: {}", e)
+        logger.warning("Failed to access the Gravitino server")
         return False
 
 
-def check_gravitino_server_status():
+def check_gravitino_server_status() -> bool:
     gravitino_server_running = False
     for i in range(5):
         logger.info("Monitoring Gravitino server status. Attempt %s", i + 1)
@@ -53,15 +53,20 @@ class IntegrationTestEnv(unittest.TestCase):
     def setUpClass(cls):
         _init_logging()
 
-        if os.environ.get('GRADLE_START_GRAVITINO') is not None:
-            logger.info('Manual start gravitino server [%s].', check_gravitino_server_status())
+        if os.environ.get('EXTERNAL_START_GRAVITINO') is not None:
+            """Maybe Gravitino server already startup by Gradle test command or developer manual startup."""
+            if not check_gravitino_server_status():
+                logger.error("ERROR: Can't find online Gravitino server!")
             return
 
-        current_path = os.getcwd()
-        cls.gravitino_startup_script = os.path.join(current_path, '../../../distribution/package/bin/gravitino.sh')
+        GravitinoHome = os.environ.get('GRAVITINO_HOME')
+        if GravitinoHome is None:
+            logger.error('Gravitino Python client integration test must configure `GRAVITINO_HOME`')
+
+        cls.gravitino_startup_script = os.path.join(GravitinoHome, 'bin/gravitino.sh')
         if not os.path.exists(cls.gravitino_startup_script):
             logger.error("Can't find Gravitino startup script: %s, "
-                         "Please execute `./gradlew compileDistribution -x test` in the gravitino project root "
+                         "Please execute `./gradlew compileDistribution -x test` in the Gravitino project root "
                          "directory.", cls.gravitino_startup_script)
             quit(0)
 
@@ -78,11 +83,9 @@ def setUpClass(cls):
             logger.error("ERROR: Can't start Gravitino server!")
             quit(0)
 
-        cls.clean_test_date()
-
     @classmethod
     def tearDownClass(cls):
-        if os.environ.get('GRADLE_START_GRAVITINO') is not None:
+        if os.environ.get('EXTERNAL_START_GRAVITINO') is not None:
             return
 
         logger.info("Stop integration test environment...")

clients/client-python/tests/integration/test_fileset_catalog.py

@@ -3,13 +3,12 @@
 This software is licensed under the Apache License version 2.
 """
 import logging
+from typing import Dict, List
 
-from gravitino.api.catalog import Catalog
 from gravitino.api.fileset import Fileset
 from gravitino.api.fileset_change import FilesetChange
 from gravitino.client.gravitino_admin_client import GravitinoAdminClient
 from gravitino.client.gravitino_client import GravitinoClient
-from gravitino.client.gravitino_metalake import GravitinoMetalake
 from gravitino.dto.catalog_dto import CatalogDTO
 from gravitino.name_identifier import NameIdentifier
 from tests.integration.integration_test_env import IntegrationTestEnv
@@ -18,110 +17,89 @@
 
 
 class TestFilesetCatalog(IntegrationTestEnv):
-    catalog: Catalog = None
-    metalake: GravitinoMetalake = None
-    metalake_name: str = "testMetalake"
+    metalake_name: str = "testMetalake-TestFilesetCatalog"
     catalog_name: str = "testCatalog"
     schema_name: str = "testSchema"
+
     fileset_name: str = "testFileset1"
-    fileset_alter_name: str = "testFilesetAlter"
+    fileset_comment: str = "fileset_comment"
+    fileset_location: str = "/tmp/TestFilesetCatalog location"
+    fileset_propertie_key1: str = "fileset_propertie_key1"
+    fileset_propertie_value1: str = "fileset_propertie_value1"
+    fileset_propertie_key2: str = "fileset_propertie_key2"
+    fileset_propertie_value2: str = "fileset_propertie_value2"
+    fileset_properties: Dict[str, str] = {fileset_propertie_key1: fileset_propertie_value1,
+                                          fileset_propertie_key2: fileset_propertie_value2}
+    fileset_new_name = fileset_name + "_new"
     provider: str = "hadoop"
 
     metalake_ident: NameIdentifier = NameIdentifier.of(metalake_name)
     catalog_ident: NameIdentifier = NameIdentifier.of_catalog(metalake_name, catalog_name)
     schema_ident: NameIdentifier = NameIdentifier.of_schema(metalake_name, catalog_name, schema_name)
     fileset_ident: NameIdentifier = NameIdentifier.of_fileset(metalake_name, catalog_name, schema_name, fileset_name)
-    fileset_alter_ident: NameIdentifier = NameIdentifier.of_fileset(metalake_name, catalog_name, schema_name,
-                                                                    fileset_alter_name)
+    fileset_new_ident: NameIdentifier = NameIdentifier.of_fileset(metalake_name, catalog_name, schema_name,
+                                                                  fileset_new_name)
 
-    gravitino_admin_client: GravitinoAdminClient = None
+    gravitino_admin_client: GravitinoAdminClient = GravitinoAdminClient(uri="http://localhost:8090")
     gravitino_client: GravitinoClient = None
 
-    @classmethod
-    def setUpClass(cls):
-        super().setUpClass()
-        cls.clean_test_data()
-
-        cls.gravitino_admin_client = GravitinoAdminClient(uri="http://localhost:8090")
-        cls.metalake = cls.gravitino_admin_client.create_metalake(ident=cls.metalake_ident,
-                                                                  comment="test comment", properties={})
-        cls.gravitino_client = GravitinoClient(uri="http://localhost:8090", metalake_name=cls.metalake_name)
-
-        cls.catalog = cls.gravitino_client.create_catalog(
-            ident=cls.catalog_ident,
-            type=CatalogDTO.Type.FILESET,
-            provider=cls.provider,
-            comment="comment",
-            properties={"k1": "v1"}
-        )
+    def setUp(self):
+        self.clean_test_data()
+        self.init_test_env()
 
-        cls.catalog.as_schemas().create_schema(ident=cls.schema_ident, comment="comment", properties={"k1": "v1"})
-
-    @classmethod
-    def tearDownClass(cls):
-        """Clean test data"""
-        cls.clean_test_data()
-        super().tearDownClass()
+    def tearDown(self):
+        self.clean_test_data()
 
     @classmethod
     def clean_test_data(cls):
         try:
-            cls.gravitino_admin_client = GravitinoAdminClient(uri="http://localhost:8090")
-            gravitino_metalake = cls.gravitino_admin_client.load_metalake(ident=cls.metalake_ident)
-            cls.catalog = gravitino_metalake.load_catalog(ident=cls.catalog_ident)
-            cls.catalog.as_fileset_catalog().drop_fileset(ident=cls.fileset_ident)
-            cls.catalog.as_fileset_catalog().drop_fileset(ident=cls.fileset_alter_ident)
-            cls.catalog.as_schemas().drop_schema(ident=cls.schema_ident, cascade=True)
-            gravitino_metalake.drop_catalog(ident=cls.catalog_ident)
+            cls.gravitino_client = GravitinoClient(uri="http://localhost:8090", metalake_name=cls.metalake_name)
+            catalog = cls.gravitino_client.load_catalog(ident=cls.catalog_ident)
+            catalog.as_fileset_catalog().drop_fileset(ident=cls.fileset_ident)
+            catalog.as_fileset_catalog().drop_fileset(ident=cls.fileset_new_ident)
+            catalog.as_schemas().drop_schema(ident=cls.schema_ident, cascade=True)
+            cls.gravitino_client.drop_catalog(ident=cls.catalog_ident)
             cls.gravitino_admin_client.drop_metalake(cls.metalake_ident)
         except Exception as e:
-            logger.debug(e)
-
-    def create_catalog(self):
-        self.catalog = self.gravitino_client.create_catalog(
-            ident=self.catalog_ident,
-            type=CatalogDTO.Type.FILESET,
-            provider=self.provider,
-            comment="comment",
-            properties={"k1": "v1"})
-
-        assert self.catalog.name == self.catalog_name
-        assert self.catalog.type == CatalogDTO.Type.FILESET
-        assert self.catalog.provider == self.provider
-
-    def create_schema(self):
-        self.catalog.as_schemas().create_schema(
-            ident=self.schema_ident,
-            comment="comment",
-            properties={"k1": "v1"})
+            logger.debug("Clean test data")
+
+    def init_test_env(self):
+        self.gravitino_admin_client.create_metalake(ident=self.metalake_ident, comment="", properties={})
+        self.gravitino_client = GravitinoClient(uri="http://localhost:8090", metalake_name=self.metalake_name)
+        catalog = self.gravitino_client.create_catalog(
+            ident=self.catalog_ident, type=CatalogDTO.Type.FILESET,
+            provider=self.provider, comment="", properties={})
+        catalog.as_schemas().create_schema(ident=self.schema_ident, comment="", properties={})
+
+    def create_fileset(self) -> Fileset:
+        catalog = self.gravitino_client.load_catalog(ident=self.catalog_ident)
+        return (catalog.as_fileset_catalog().create_fileset(ident=self.fileset_ident,
+                                                            type=Fileset.Type.MANAGED,
+                                                            comment=self.fileset_comment,
+                                                            storage_location=self.fileset_location,
+                                                            properties=self.fileset_properties))
 
     def test_create_fileset(self):
-        fileset = self.catalog.as_fileset_catalog().create_fileset(ident=self.fileset_ident,
-                                                                   type=Fileset.Type.MANAGED,
-                                                                   comment="mock comment",
-                                                                   storage_location="mock location",
-                                                                   properties={"k1": "v1"})
-        assert fileset is not None
+        fileset = self.create_fileset()
+        self.assertIsNotNone(fileset)
+        self.assertEqual(fileset.type(), Fileset.Type.MANAGED)
+        self.assertEqual(fileset.comment(), self.fileset_comment)
+        self.assertEqual(fileset.properties(), self.fileset_properties)
 
-        fileset_list = self.catalog.as_fileset_catalog().list_filesets(self.fileset_ident.namespace())
-        assert fileset_list is not None and len(fileset_list) == 1
+    def test_drop_fileset(self):
+        self.create_fileset()
+        self.assertTrue(self.gravitino_client.load_catalog(ident=self.catalog_ident)
+                        .as_fileset_catalog().drop_fileset(ident=self.fileset_ident))
 
-        fileset = self.catalog.as_fileset_catalog().load_fileset(self.fileset_ident)
-        assert fileset is not None
-        assert fileset.name() == self.fileset_ident.name()
+    def test_alter_fileset(self):
+        self.create_fileset()
+        fileset_propertie_new_value = self.fileset_propertie_value2 + "_new"
 
-        # Alter fileset
         changes = (
-            FilesetChange.rename(self.fileset_alter_name),
-            FilesetChange.update_comment("new fileset comment"),
-            FilesetChange.set_property("key1", "value1"),
-            FilesetChange.remove_property("k1"),
+            FilesetChange.remove_property(self.fileset_propertie_key1),
+            FilesetChange.set_property(self.fileset_propertie_key2, fileset_propertie_new_value),
         )
-        fileset_alter = self.catalog.as_fileset_catalog().alter_fileset(self.fileset_ident, *changes)
-        assert fileset_alter is not None
-        assert fileset_alter.name() == self.fileset_alter_name
-        assert fileset_alter.comment() == "new fileset comment"
-        assert fileset_alter.properties().get("key1") == "value1"
-
-        # Clean test data
-        self.catalog.as_fileset_catalog().drop_fileset(ident=self.fileset_ident)
+        catalog = self.gravitino_client.load_catalog(ident=self.catalog_ident)
+        fileset_new = catalog.as_fileset_catalog().alter_fileset(self.fileset_ident, *changes)
+        self.assertEqual(fileset_new.properties().get(self.fileset_propertie_key2), fileset_propertie_new_value)
+        self.assertTrue(self.fileset_propertie_key1 not in fileset_new.properties())