Enable new CI pipeline for standard & distributed tests (#1481)

* debug

* Minor - Add autoimport pandas env var to ray init to fix failing pytest tests

* Marking distributed tests

* Register marker in pyproject

* Adding global mark

* removing unecessary assert for wr.config.distributed

* setting environment var in codebuild

* Adding environment variable assembly

* Testing env var with tox

* reverting to comprehenstion of ray runtime env vars

Co-authored-by: Abdel Jaidi <[email protected]>

Author: malachi-constant

awswrangler/distributed/_distributed.py

@@ -107,6 +107,10 @@ def initialize_ray(
                 )
                 os.environ["RAY_ENABLE_MAC_LARGE_OBJECT_STORE"] = "1"
 
+            ray_runtime_env_vars = [
+                "__MODIN_AUTOIMPORT_PANDAS__",
+            ]
+
             ray_init_kwargs = {
                 "num_cpus": cpu_count or multiprocessing.cpu_count(),
                 "num_gpus": gpu_count,
@@ -115,6 +119,9 @@ def initialize_ray(
                 "object_store_memory": object_store_memory,
                 "_redis_password": redis_password,
                 "_memory": object_store_memory,
+                "runtime_env": {
+                    "env_vars": {var: os.environ.get(var) for var in ray_runtime_env_vars if os.environ.get(var)}
+                },
             }
             ray.init(**ray_init_kwargs)
 

pyproject.toml

@@ -129,3 +129,6 @@ warn_unused_ignores = false
 log_cli = false
 filterwarnings = "ignore::DeprecationWarning"
 addopts = "--log-cli-format \"[%(name)s][%(funcName)s] %(message)s\" --verbose --capture=sys"
+markers = [
+    "distributed: tests againsts methods with distributed functionality",
+]

tests/test_lakeformation.py

@@ -3,13 +3,16 @@
 import time
 
 import pandas as pd
+import pytest
 
 import awswrangler as wr
 
 from ._utils import ensure_data_types, ensure_data_types_csv, get_df, get_df_csv
 
 logging.getLogger("awswrangler").setLevel(logging.DEBUG)
 
+pytestmark = pytest.mark.distributed
+
 
 def test_lakeformation(path, path2, glue_database, glue_table, glue_table2, use_threads=False):
     wr.catalog.delete_table_if_exists(database=glue_database, table=glue_table)

tests/test_s3_select.py

@@ -11,6 +11,8 @@
 
 logging.getLogger("awswrangler").setLevel(logging.DEBUG)
 
+pytestmark = pytest.mark.distributed
+
 
 @pytest.mark.parametrize("use_threads", [True, False, 2])
 def test_full_table(path, use_threads):