Ele 1942 package tests separate schema per process (#599)

* tests: separate schemas for separate processes

* tests: fixes to places that use target.schema

* test_data source: fix

* simplify usage of schema_name_suffix

* docker-compose: increase postgres max connections

* tests: bugfix

integration_tests/dbt_project/macros/clear_env.sql

@@ -1,5 +1,5 @@
 {% macro clear_env() %}
-    {% do elementary_tests.edr_drop_schema(elementary.target_database(), target.schema) %}
+    {% do elementary_tests.edr_drop_schema(elementary.target_database(), generate_schema_name()) %}
     {% set database_name, schema_name = elementary.get_package_database_and_schema('elementary') %}
     {% do elementary_tests.edr_drop_schema(database_name, schema_name) %}
 {% endmacro %}

integration_tests/dbt_project/macros/generate_schema_name.sql

@@ -0,0 +1,13 @@
+{% macro generate_schema_name(custom_schema_name, node) -%}
+    {%- set schema_name = target.schema -%}
+    {% if custom_schema_name %}
+        {% set schema_name = "{}_{}".format(schema_name, custom_schema_name) %}
+    {% endif %}
+
+    {% set schema_name_suffix_by_var = var('schema_name_suffix', '') %}
+    {% if schema_name_suffix_by_var %}
+        {% set schema_name = schema_name + schema_name_suffix_by_var %}
+    {% endif %}
+
+    {% do return(schema_name) %}
+{%- endmacro %}

integration_tests/dbt_project/models/test_data.yaml

@@ -2,7 +2,7 @@ version: 2
 
 sources:
   - name: test_data
-    schema: "{{ target.schema }}"
+    schema: "{{ target.schema + var('schema_name_suffix', '') }}"
     tables:
       - name: metrics_seed1
       - name: metrics_seed2

integration_tests/docker-compose.yml

@@ -5,6 +5,7 @@ services:
     image: postgres
     ports:
       - "127.0.0.1:5432:5432"
+    command: postgres -c max_connections=500
     environment:
       POSTGRES_USER: admin
       POSTGRES_PASSWORD: admin

integration_tests/tests/conftest.py

@@ -6,7 +6,6 @@
 import pytest
 from dbt.version import __version__ as dbt_version
 from dbt_project import DbtProject
-from filelock import FileLock
 from packaging import version
 
 DBT_PROJECT_PATH = Path(__file__).parent.parent / "dbt_project"
@@ -32,21 +31,8 @@ def project_dir_copy():
 
 
 @pytest.fixture(scope="session", autouse=True)
-def init_tests_env(target, tmp_path_factory, worker_id: str, project_dir_copy: str):
-    # Tests are not multi-threaded.
-    if worker_id == "master":
-        env.init(target, project_dir_copy)
-        return
-
-    # Temp dir shared by all workers.
-    tmp_dir = tmp_path_factory.getbasetemp().parent
-    env_ready_indicator_path = tmp_dir / ".wait_env_ready"
-    with FileLock(str(env_ready_indicator_path) + ".lock"):
-        if env_ready_indicator_path.is_file():
-            return
-        else:
-            env.init(target, project_dir_copy)
-            env_ready_indicator_path.touch()
+def init_tests_env(target, project_dir_copy: str):
+    env.init(target, project_dir_copy)
 
 
 @pytest.fixture(autouse=True)

integration_tests/tests/dbt_project.py

@@ -1,4 +1,5 @@
 import json
+import os
 from contextlib import contextmanager, nullcontext
 from pathlib import Path
 from tempfile import NamedTemporaryFile
@@ -9,12 +10,16 @@
 from logger import get_logger
 from ruamel.yaml import YAML
 
+PYTEST_XDIST_WORKER = os.environ.get("PYTEST_XDIST_WORKER", None)
+SCHEMA_NAME_SUFFIX = f"_{PYTEST_XDIST_WORKER}" if PYTEST_XDIST_WORKER else ""
+
 _DEFAULT_VARS = {
     "disable_dbt_invocation_autoupload": True,
     "disable_dbt_artifacts_autoupload": True,
     "disable_run_results": True,
     "debug_logs": True,
     "collect_metrics": False,
+    "schema_name_suffix": SCHEMA_NAME_SUFFIX,
 }
 
 DEFAULT_DUMMY_CODE = "SELECT 1 AS col"
@@ -174,7 +179,7 @@ def test(
                 "sources": [
                     {
                         "name": "test_data",
-                        "schema": "{{ target.schema }}",
+                        "schema": f"{{{{ target.schema }}}}{SCHEMA_NAME_SUFFIX}",
                         "tables": [table_yaml],
                     }
                 ],

integration_tests/tests/test_string_monitors.py

@@ -8,6 +8,7 @@ def test_missing_count(dbt_project: DbtProject, test_id: str):
     data = [{COLUMN_NAME: value} for value in ["a", "b", "c", " a "] + missing_values]
     dbt_project.seed(data, test_id)
     result = dbt_project.run_query(
-        f"select {{{{ elementary.missing_count('{COLUMN_NAME}') }}}} as missing_count from {{{{ target.schema }}}}.{test_id}"
+        f"select {{{{ elementary.missing_count('{COLUMN_NAME}') }}}} "
+        f"as missing_count from {{{{ generate_schema_name() }}}}.{test_id}"
     )[0]
     assert result["missing_count"] == len(missing_values)