test(python): make sure integration tests wait for services to start

.github/workflows/dev_pr.yml

@@ -23,7 +23,7 @@ jobs:
           github.event_name == 'pull_request_target' &&
             (github.event.action == 'opened' ||
              github.event.action == 'synchronize')
-        uses: actions/labeler@4.0.2
+        uses: actions/labeler@v4.0.2
         with:
           repo-token: ${{ secrets.GITHUB_TOKEN }}
           configuration-path: .github/workflows/dev_pr/labeler.yml

docker-compose.yml

@@ -1,7 +1,7 @@
 version: "3.9"
 services:
   localstack:
-    image: localstack/localstack:0.14.4
+    image: localstack/localstack:0.14
     ports:
       - 4566:4566
       - 8080:8080

python/pyproject.toml

@@ -38,7 +38,8 @@ devel = [
 ]
 pyspark = [
     "pyspark",
-    "delta-spark"
+    "delta-spark",
+    "numpy==1.22.2" # pyspark is no compatible with latest numpy
 ]
 
 [project.urls]

python/tests/conftest.py

@@ -3,13 +3,28 @@
 import subprocess
 from datetime import date, datetime, timedelta
 from decimal import Decimal
+from time import sleep
 
 import pyarrow as pa
 import pytest
 
 from deltalake import DeltaTable, write_deltalake
 
 
+def wait_till_host_is_available(host: str, timeout_sec: int = 30):
+    spacing = 2
+    attempts = timeout_sec / spacing
+    while True:
+        try:
+            subprocess.run(["curl", host], timeout=500, check=True)
+        except:
+            pass
+        else:
+            break
+
+        sleep(spacing)
+
+
 @pytest.fixture(scope="session")
 def s3_localstack_creds():
     endpoint_url = "http://localhost:4566"
@@ -46,6 +61,8 @@ def s3_localstack_creds():
         ],
     ]
 
+    wait_till_host_is_available(endpoint_url)
+
     try:
         for args in setup_commands:
             subprocess.run(args, env=env)
@@ -109,7 +126,7 @@ def azurite_creds():
         f"AccountKey={config['AZURE_STORAGE_ACCOUNT_KEY']};"
         f"BlobEndpoint={endpoint_url};"
     )
-
+    wait_till_host_is_available(endpoint_url)
     try:
         subprocess.run(
             [

python/tests/test_fs.py

@@ -12,7 +12,7 @@
 
 @pytest.mark.s3
 @pytest.mark.integration
-@pytest.mark.timeout(timeout=5, method="thread")
+@pytest.mark.timeout(timeout=15, method="thread")
 def test_read_files(s3_localstack):
     table_path = "s3://deltars/simple"
     handler = DeltaStorageHandler(table_path)
@@ -29,7 +29,7 @@ def test_read_files(s3_localstack):
 
 @pytest.mark.s3
 @pytest.mark.integration
-@pytest.mark.timeout(timeout=4, method="thread")
+@pytest.mark.timeout(timeout=15, method="thread")
 def test_s3_authenticated_read_write(s3_localstack_creds):
     # Create unauthenticated handler
     storage_handler = DeltaStorageHandler(
@@ -54,7 +54,7 @@ def test_s3_authenticated_read_write(s3_localstack_creds):
 
 @pytest.mark.s3
 @pytest.mark.integration
-@pytest.mark.timeout(timeout=5, method="thread")
+@pytest.mark.timeout(timeout=15, method="thread")
 def test_read_simple_table_from_remote(s3_localstack):
     table_path = "s3://deltars/simple"
     dt = DeltaTable(table_path)
@@ -63,7 +63,7 @@ def test_read_simple_table_from_remote(s3_localstack):
 
 @pytest.mark.s3
 @pytest.mark.integration
-@pytest.mark.timeout(timeout=5, method="thread")
+@pytest.mark.timeout(timeout=15, method="thread")
 def test_roundtrip_s3_env(s3_localstack, sample_data: pa.Table, monkeypatch):
     table_path = "s3://deltars/roundtrip"
 
@@ -91,7 +91,7 @@ def test_roundtrip_s3_env(s3_localstack, sample_data: pa.Table, monkeypatch):
 
 @pytest.mark.s3
 @pytest.mark.integration
-@pytest.mark.timeout(timeout=5, method="thread")
+@pytest.mark.timeout(timeout=15, method="thread")
 def test_roundtrip_s3_direct(s3_localstack_creds, sample_data: pa.Table):
     table_path = "s3://deltars/roundtrip2"
 
@@ -146,7 +146,7 @@ def test_roundtrip_s3_direct(s3_localstack_creds, sample_data: pa.Table):
 
 @pytest.mark.azure
 @pytest.mark.integration
-@pytest.mark.timeout(timeout=5, method="thread")
+@pytest.mark.timeout(timeout=15, method="thread")
 def test_roundtrip_azure_env(azurite_env_vars, sample_data: pa.Table):
     table_path = "az://deltars/roundtrip"
 
@@ -168,7 +168,7 @@ def test_roundtrip_azure_env(azurite_env_vars, sample_data: pa.Table):
 
 @pytest.mark.azure
 @pytest.mark.integration
-@pytest.mark.timeout(timeout=5, method="thread")
+@pytest.mark.timeout(timeout=15, method="thread")
 def test_roundtrip_azure_direct(azurite_creds, sample_data: pa.Table):
     table_path = "az://deltars/roundtrip2"
 
@@ -197,7 +197,7 @@ def test_roundtrip_azure_direct(azurite_creds, sample_data: pa.Table):
 
 @pytest.mark.azure
 @pytest.mark.integration
-@pytest.mark.timeout(timeout=5, method="thread")
+@pytest.mark.timeout(timeout=15, method="thread")
 def test_roundtrip_azure_sas(azurite_sas_creds, sample_data: pa.Table):
     table_path = "az://deltars/roundtrip3"
 

rust/src/storage/utils.rs

@@ -15,12 +15,15 @@ pub async fn copy_table(
     from_options: Option<HashMap<String, String>>,
     to: impl AsRef<str>,
     to_options: Option<HashMap<String, String>>,
+    allow_http: bool,
 ) -> Result<(), DeltaTableError> {
     let from_store = DeltaTableBuilder::from_uri(from)
         .with_storage_options(from_options.unwrap_or_default())
+        .with_allow_http(allow_http)
         .build_storage()?;
     let to_store = DeltaTableBuilder::from_uri(to)
         .with_storage_options(to_options.unwrap_or_default())
+        .with_allow_http(allow_http)
         .build_storage()?;
     sync_stores(from_store, to_store).await
 }

rust/src/test_utils.rs

@@ -46,7 +46,7 @@ impl IntegrationContext {
                 account_path.as_path().to_str().unwrap(),
             );
         }
-        integration.crate_bucket(&bucket)?;
+        integration.create_bucket(&bucket)?;
         let store_uri = match integration {
             StorageIntegration::Amazon => format!("s3://{}", &bucket),
             StorageIntegration::Microsoft => format!("az://{}", &bucket),
@@ -109,12 +109,12 @@ impl IntegrationContext {
                 options.content_only = true;
                 let dest_path = self.tmp_dir.path().join(name.as_ref());
                 std::fs::create_dir_all(&dest_path)?;
-                copy(&table.as_path(), &dest_path, &options)?;
+                copy(table.as_path(), &dest_path, &options)?;
             }
             _ => {
                 let from = table.as_path().as_str().to_owned();
                 let to = format!("{}/{}", self.root_uri(), name.as_ref());
-                copy_table(from, None, to, None).await?;
+                copy_table(from, None, to, None, true).await?;
             }
         };
         Ok(())
@@ -125,7 +125,7 @@ impl Drop for IntegrationContext {
     fn drop(&mut self) {
         match self.integration {
             StorageIntegration::Amazon => {
-                s3_cli::delete_bucket(&self.root_uri()).unwrap();
+                s3_cli::delete_bucket(self.root_uri()).unwrap();
                 s3_cli::delete_lock_table().unwrap();
             }
             StorageIntegration::Microsoft => {
@@ -157,14 +157,14 @@ impl StorageIntegration {
         }
     }
 
-    fn crate_bucket(&self, name: impl AsRef<str>) -> std::io::Result<()> {
+    fn create_bucket(&self, name: impl AsRef<str>) -> std::io::Result<()> {
         match self {
             Self::Microsoft => {
                 az_cli::create_container(name)?;
                 Ok(())
             }
             Self::Amazon => {
-                s3_cli::create_bucket(&format!("s3://{}", name.as_ref()))?;
+                s3_cli::create_bucket(format!("s3://{}", name.as_ref()))?;
                 set_env_if_not_set(
                     "DYNAMO_LOCK_PARTITION_KEY_VALUE",
                     format!("s3://{}", name.as_ref()),