add some test fixtures for running in LocalStack

Author: whummer

.github/workflows/build_test.yml

@@ -145,9 +145,11 @@ jobs:
         env:
           LOCALSTACK_API_KEY: ${{ secrets.LOCALSTACK_API_KEY }}
         run: |
-          docker pull localstack/snowflake &
+          image_name=localstack/snowflake
+          # image_name=whummer/snowflake-tmp
+          docker pull $image_name &
           pip install localstack
-          IMAGE_NAME=localstack/snowflake DEBUG=1 localstack start -d
+          IMAGE_NAME=$image_name DEBUG=1 localstack start -d
           localstack wait
 
 #      - name: Setup parameters file
@@ -178,6 +180,7 @@ jobs:
           PYTEST_ADDOPTS: --color=yes --tb=short
           TOX_PARALLEL_NO_SPINNER: 1
         shell: bash
+        timeout-minutes: 22
       - name: Combine coverages
         run: python -m tox run -e coverage --skip-missing-interpreters false
         shell: bash

setup.cfg

@@ -79,6 +79,7 @@ console_scripts =
 
 [options.extras_require]
 development =
+    boto3
     Cython
     coverage
     more-itertools

test/integ/conftest.py

@@ -83,6 +83,35 @@ def print_help() -> None:
     )
 
 
+# START changes for LocalStack
+
+
+@pytest.fixture(scope="session", autouse=True)
+def create_external_volume():
+    import boto3
+
+    # create S3 bucket in LocalStack
+    s3_client = boto3.client(
+        "s3",
+        endpoint_url="http://localhost:4566",
+        aws_access_key_id="test",
+        aws_secret_access_key="test",
+    )
+    s3_client.create_bucket(Bucket="test")
+
+    # create a local external volume which is required by some of the tests
+    query = (
+        "CREATE EXTERNAL VOLUME IF NOT EXISTS python_connector_iceberg_exvol STORAGE_LOCATIONS = "
+        "((NAME = 'sl1' STORAGE_PROVIDER='S3' STORAGE_BASE_URL='s3://test' "
+        "STORAGE_AWS_ROLE_ARN='arn:aws:iam::000000000000:role/test'))"
+    )
+    connection = create_connection("default")
+    connection.cursor().execute(query)
+
+
+# END changes for LocalStack
+
+
 @pytest.fixture(scope="session")
 def is_public_test() -> bool:
     return is_public_testaccount()