Update tests for musl linux build.

Author: niyue

tests/musl.py

@@ -0,0 +1,19 @@
+import platform
+import subprocess
+
+
+def is_musl_linux():
+    """Return True when running on a musl-based Linux distribution."""
+    if platform.system() != "Linux":
+        return False
+    try:
+        result = subprocess.run(
+            ["ldd", "--version"], capture_output=True, text=True
+        )
+        print(f"stdout: {result.stdout.lower()}")
+        print(f"stderr: {result.stderr.lower()}")
+        output_text = (result.stdout + result.stderr).lower()
+        return "musl" in output_text
+    except Exception as e:
+        print(f"Exception in is_musl_linux: {e}")
+        return False

tests/test_delta_lake.py

@@ -2,25 +2,9 @@
 
 import unittest
 import sys
-import platform
-import subprocess
 import chdb
 from chdb import session
-
-def is_musl_linux():
-    """Check if running on musl Linux"""
-    if platform.system() != "Linux":
-        return False
-    try:
-        result = subprocess.run(['ldd', '--version'], capture_output=True, text=True)
-        print(f"stdout: {result.stdout.lower()}")
-        print(f"stderr: {result.stderr.lower()}")
-        # Check both stdout and stderr for musl
-        output_text = (result.stdout + result.stderr).lower()
-        return 'musl' in output_text
-    except Exception as e:
-        print(f"Exception in is_musl_linux: {e}")
-        return False
+from tests.musl import is_musl_linux
 
 @unittest.skipUnless(sys.platform.startswith("linux") and not is_musl_linux(), "Runs only on Linux platforms")
 class TestDeltaLake(unittest.TestCase):

tests/test_query_params.py

@@ -5,9 +5,17 @@
 import chdb
 from chdb import session
 from chdb.state import connect
+from tests.musl import is_musl_linux
 
 
 class TestQueryParams(unittest.TestCase):
+    def assert_exception_message(self, exc, expected):
+        message = str(exc.exception)
+        if is_musl_linux():
+            self.assertIn("Caught an unknown exception", message)
+        else:
+            self.assertIn(expected, message)
+
     def test_connection_query_with_params(self):
         df = chdb.query(
             "SELECT toDate({base_date:String}) + number AS d "
@@ -114,7 +122,7 @@ def test_query_param_with_equals_sign(self):
     def test_query_param_key_with_equals_sign(self):
         with self.assertRaises(RuntimeError) as exc:
             chdb.query("SELECT {a=b:UInt64}", "DataFrame", params={"a=b": 1})
-        self.assertIn("SYNTAX_ERROR", str(exc.exception))
+        self.assert_exception_message(exc, "SYNTAX_ERROR")
 
     def test_query_with_array_param(self):
         df = chdb.query(
@@ -135,12 +143,12 @@ def test_query_with_tuple_param_string_encoded(self):
     def test_query_with_params_missing_value(self):
         with self.assertRaises(RuntimeError) as exc:
             chdb.query("SELECT {x:UInt64} AS v")
-        self.assertIn("Substitution `x` is not set", str(exc.exception))
+        self.assert_exception_message(exc, "Substitution `x` is not set")
 
     def test_query_with_params_invalid_type(self):
         with self.assertRaises(RuntimeError) as exc:
             chdb.query("SELECT {x:UInt64} AS v", params={"x": "not-a-number"})
-        self.assertIn("cannot be parsed as UInt64", str(exc.exception))
+        self.assert_exception_message(exc, "cannot be parsed as UInt64")
 
     def test_session_send_query_with_params_missing_value_stream(self):
         sess = session.Session(":memory:")
@@ -149,7 +157,7 @@ def test_session_send_query_with_params_missing_value_stream(self):
             with stream:
                 with self.assertRaises(RuntimeError) as exc:
                     stream.fetch()
-            self.assertIn("Substitution `n` is not set", str(exc.exception))
+            self.assert_exception_message(exc, "Substitution `n` is not set")
         finally:
             sess.close()
 
@@ -160,7 +168,7 @@ def test_session_send_query_with_params_invalid_type_stream(self):
             with stream:
                 with self.assertRaises(RuntimeError) as exc:
                     stream.fetch()
-            self.assertIn("cannot be parsed as UInt64", str(exc.exception))
+            self.assert_exception_message(exc, "cannot be parsed as UInt64")
         finally:
             sess.close()
 

tests/test_session_concurrency.py

@@ -4,29 +4,13 @@
 import shutil
 import os
 import threading
-import platform
-import subprocess
 from chdb import session
+from tests.musl import is_musl_linux
 
 
 test_concurrent_dir = ".tmp_test_session_concurrency"
 
 
-def is_musl_linux():
-    if platform.system() != "Linux":
-        return False
-    try:
-        result = subprocess.run(['ldd', '--version'], capture_output=True, text=True)
-        print(f"stdout: {result.stdout.lower()}")
-        print(f"stderr: {result.stderr.lower()}")
-        # Check both stdout and stderr for musl
-        output_text = (result.stdout + result.stderr).lower()
-        return 'musl' in output_text
-    except Exception as e:
-        print(f"Exception in is_musl_linux: {e}")
-        return False
-
-
 class TestSessionConcurrency(unittest.TestCase):
     def setUp(self) -> None:
         shutil.rmtree(test_concurrent_dir, ignore_errors=True)

tests/test_unsupported_arrow_types.py

@@ -1,28 +1,11 @@
 #!/usr/bin/env python3
 
 import unittest
-import platform
-import subprocess
 import pyarrow as pa
 import pyarrow.compute as pc
 import chdb
 from chdb import ChdbError
-
-
-def is_musl_linux():
-    """Check if running on musl Linux"""
-    if platform.system() != "Linux":
-        return False
-    try:
-        result = subprocess.run(['ldd', '--version'], capture_output=True, text=True)
-        print(f"stdout: {result.stdout.lower()}")
-        print(f"stderr: {result.stderr.lower()}")
-        # Check both stdout and stderr for musl
-        output_text = (result.stdout + result.stderr).lower()
-        return 'musl' in output_text
-    except Exception as e:
-        print(f"Exception in is_musl_linux: {e}")
-        return False
+from tests.musl import is_musl_linux
 
 
 class TestUnsupportedArrowTypes(unittest.TestCase):