test: make dbt and warehouse tests hermetic

.github/workflows/dagster-ci.yml

@@ -52,6 +52,30 @@ jobs:
           cd macro_agents
           uv sync --extra dev
 
+      - name: Cache dbt packages
+        id: cache-dbt-packages
+        uses: actions/cache@27d5ce7f107fe9357f9df03efb73ab90386fccae  # v5
+        with:
+          path: dbt_project/dbt_packages
+          key: ${{ runner.os }}-dbt-packages-${{ hashFiles('dbt_project/package-lock.yml') }}
+
+      - name: Install dbt packages
+        if: steps.cache-dbt-packages.outputs.cache-hit != 'true'
+        env:
+          BIGQUERY_PROJECT: econ-data-project-478800
+          DBT_SEND_ANONYMOUS_USAGE_STATS: "false"
+        run: |
+          cd macro_agents
+          uv run dbt deps --project-dir ../dbt_project --profiles-dir ../dbt_project --target dev --no-send-anonymous-usage-stats
+
+      - name: Prepare dbt manifest offline
+        env:
+          BIGQUERY_PROJECT: econ-data-project-478800
+          DBT_SEND_ANONYMOUS_USAGE_STATS: "false"
+        run: |
+          cd macro_agents
+          uv run dbt parse --project-dir ../dbt_project --profiles-dir ../dbt_project --target dev --no-send-anonymous-usage-stats
+
       - name: Lint with ruff
         run: |
           cd macro_agents
@@ -63,6 +87,12 @@ jobs:
           cd macro_agents
           uv run ty check --extra-search-path src src/macro_agents
 
+      - name: Lint dbt SQL with SQLFluff
+        run: |
+          cd macro_agents
+          # The preceding dbt parse validates dbt templating; Jinja mode keeps style lint credential-free.
+          uv run sqlfluff lint ../dbt_project/models --templater jinja --ignore templating,parsing --disable-progress-bar --processes 4
+
       - name: Test Dagster definitions with dbt Platform
         if: ${{ env.DBT_PLATFORM_TOKEN != '' }}
         env:
@@ -85,4 +115,9 @@ jobs:
       - name: Test with pytest
         run: |
           cd macro_agents
-          uv run pytest tests/ -v -m "not skip_ci"
+          uv run pytest tests/ -v -m "not skip_ci and not network"
+
+      - name: Enforce critical-path coverage floor
+        run: |
+          cd macro_agents
+          uv run pytest tests/test_bigquery_warehouse.py tests/test_asset_check_failure_sensor.py tests/test_sec_search.py --cov --cov-report=term-missing

AGENTS.md

@@ -12,6 +12,7 @@
 - `make ruff`: lint/format Python with Ruff.
 - `make lint` / `make fix`: lint or fix dbt SQL with SQLFluff.
 - `make typecheck-dagster`: run `ty` type-checker over `macro_agents`.
+- `make dbt-deps`: explicitly install locked dbt packages (networked setup step).
 - `make dbt-manifest`: parse dbt project.
 
 ## Worktrunk Hooks (wt)
@@ -54,11 +55,13 @@
 ## Testing Guidelines
 - Framework: `pytest` for Python (`macro_agents`).
 - Run Dagster tests: `cd macro_agents && uv run pytest tests/ -v`.
+- Install dbt packages once with `make dbt-deps`; pytest performs one offline dbt parse per session.
+- Network integration tests are opt-in: `RUN_NETWORK_TESTS=1 uv run pytest -m network`.
 - Test standards:
   - Prefer behavior-focused assertions (status, metadata, materialized values) over snapshots.
   - Cover loading, empty, error, and populated states.
   - Use deterministic fixtures and local resources (temp DuckDB, ephemeral Dagster instances); avoid network calls.
-  - For Dagster assets, use `DagsterInstance.ephemeral()` and `build_op_context` with mocked resources.
+  - Use `DagsterInstance.ephemeral()` and `build_op_context` as context managers so their database resources are disposed deterministically.
   - Keep tests isolated and fast; clean up temp files and avoid shared mutable state.
 
 ## Commit & Pull Request Guidelines

Readme.md

@@ -81,8 +81,11 @@ Dagster UI: `http://localhost:3000`
 ### Development (without Docker)
 
 ```bash
+# From the repository root, install Python and locked dbt dependencies
+make setup-dagster
+make dbt-manifest
+
 cd macro_agents
-uv sync --extra dev
 
 # Validate definitions
 uv run dg check defs
@@ -152,6 +155,9 @@ docker compose --env-file .env build       # Rebuild images
 docker compose logs -f dagster_user_code   # Tail user code logs
 
 # Development
+make dbt-deps                            # Install locked dbt packages
+make dbt-manifest                        # Generate the local manifest
+cd macro_agents
 uv run pytest tests/ -v                    # Run tests
 uv run ruff check . && ruff format .       # Lint + format
 uv run dg check defs                       # Validate Dagster definitions

dbt_project/models/signals/int_cross_asset_commodity_signals.sql

@@ -74,7 +74,7 @@ gold_real_zscore AS (
             STDDEV_SAMP(CASE WHEN beta IS NOT NULL AND alpha IS NOT NULL THEN gold_price - (alpha + beta * real_yield_10y) END) OVER w AS residual_std
         FROM gold_real_residual
         WINDOW w AS (ORDER BY date ROWS BETWEEN 251 PRECEDING AND CURRENT ROW)
-    )
+    ) AS rolling_residuals
 ),
 
 copper_gold_yield_corr AS (

macro_agents/pyproject.toml

@@ -100,6 +100,7 @@ root_module = "macro_agents"
 [tool.pytest.ini_options]
 markers = [
     "essential: Essential tests that must pass in CI",
+    "network: Tests that require external network access and are opt-in",
     "skip_ci: Tests to skip in CI (brittle or non-essential)",
 ]
 filterwarnings = [
@@ -109,3 +110,16 @@ filterwarnings = [
     "ignore::DeprecationWarning:dagster._core.definitions.sensor_definition:",
     "ignore:Non-compliant tag keys like .*:DeprecationWarning:dagster._core.definitions.schedule_definition",
 ]
+
+[tool.coverage.run]
+source = ["macro_agents"]
+
+[tool.coverage.report]
+fail_under = 50
+include = [
+    "src/macro_agents/defs/asset_failure_sensor.py",
+    "src/macro_agents/defs/domains/sec/search.py",
+    "src/macro_agents/defs/resources/bigquery_query.py",
+    "src/macro_agents/defs/resources/bigquery_warehouse.py",
+]
+show_missing = true

macro_agents/src/macro_agents/defs/transformation/dbt.py

@@ -30,6 +30,7 @@
     "nasdaq_companies_summary "
     "source:staging.nasdaq_companies_prices_raw"
 )
+REQUIRED_DBT_PACKAGES = ("dbt_project_evaluator", "dbt_utils")
 
 
 class CustomizedDagsterDbtTranslator(DagsterDbtTranslator):
@@ -91,6 +92,21 @@ def _required_env_var(*names: str) -> dg.EnvVar:
         raise RuntimeError(f"Set {joined_names} to enable dbt Platform orchestration.")
 
 
+def _require_dbt_packages(project_dir: Path) -> None:
+    """Require explicitly installed dbt packages without network side effects."""
+    packages_dir = project_dir / "dbt_packages"
+    missing_packages = [
+        package
+        for package in REQUIRED_DBT_PACKAGES
+        if not (packages_dir / package).exists()
+    ]
+    if missing_packages:
+        raise RuntimeError(
+            "dbt packages are not installed: "
+            f"{', '.join(missing_packages)}. Run `make dbt-deps`."
+        )
+
+
 dbt_platform_mode = orchestration_mode in {
     "dbt_platform",
     "dbt_cloud",
@@ -156,11 +172,12 @@ def full_dbt_assets(context: dg.AssetExecutionContext, dbt: DbtCloudWorkspace):
 else:
     dbt_project_dir = os.getenv("DBT_PROJECT_DIR")
     if dbt_project_dir:
-        dbt_project_dir = Path(dbt_project_dir).resolve()
-        if not dbt_project_dir.exists():
+        configured_project_dir = Path(dbt_project_dir)
+        if not configured_project_dir.exists():
             raise FileNotFoundError(
                 "DBT_PROJECT_DIR environment variable points to a non-existent path."
             )
+        dbt_project_dir = configured_project_dir.resolve()
     else:
         current_file = Path(__file__).resolve()
         macro_agents_root = current_file.parent.parent.parent
@@ -184,16 +201,14 @@ def full_dbt_assets(context: dg.AssetExecutionContext, dbt: DbtCloudWorkspace):
 
         dbt_project_dir = None
         for path in possible_dbt_project_paths:
-            try:
-                abs_path = path.resolve()
-                if abs_path.exists() and abs_path.is_dir():
-                    if (abs_path / "dbt_project.yml").exists() or (
-                        abs_path / "dbt_project.yaml"
-                    ).exists():
-                        dbt_project_dir = abs_path
-                        break
-            except (OSError, RuntimeError):
+            if not path.exists() or not path.is_dir():
                 continue
+            abs_path = path.resolve()
+            if (abs_path / "dbt_project.yml").exists() or (
+                abs_path / "dbt_project.yaml"
+            ).exists():
+                dbt_project_dir = abs_path
+                break
 
         if dbt_project_dir is None:
             raise FileNotFoundError(
@@ -207,83 +222,28 @@ def full_dbt_assets(context: dg.AssetExecutionContext, dbt: DbtCloudWorkspace):
 
     dbt_project_dir_path = dbt_project_dir
 
-    dbt_packages_dir = dbt_project_dir_path / "dbt_packages"
-    dbt_utils_dir = (
-        dbt_packages_dir / "dbt_utils" if dbt_packages_dir.exists() else None
-    )
-
-    if not dbt_packages_dir.exists() or not dbt_utils_dir or not dbt_utils_dir.exists():
-        import subprocess
-
-        logger.warning("dbt packages not found. Running 'dbt deps' (15s timeout)...")
-        try:
-            result = subprocess.run(
-                ["dbt", "deps", "--target", environment],
-                cwd=dbt_project_dir_path,
-                capture_output=True,
-                text=True,
-                timeout=15,
-            )
-            if result.returncode != 0:
-                logger.warning(
-                    f"dbt deps failed (non-fatal): {result.stderr or result.stdout}"
-                )
-            else:
-                logger.info("dbt packages installed successfully")
-        except subprocess.TimeoutExpired:
-            logger.warning(
-                "dbt deps timed out (network may be unavailable), continuing without packages"
-            )
-        except Exception as e:
-            logger.warning(f"Could not install dbt packages (non-fatal): {e}")
+    _require_dbt_packages(dbt_project_dir_path)
 
     dbt_project = DbtProject(
         project_dir=dbt_project_dir_path,
         target=environment,
     )
 
-    dbt_project.prepare_if_dev()
-
     manifest_path = dbt_project.manifest_path
-    if manifest_path and manifest_path.exists():
-        logger.info("Using dbt manifest")
+    if not manifest_path.exists():
+        raise FileNotFoundError(
+            f"dbt manifest not found at {manifest_path}. Run `make dbt-manifest`."
+        )
+    logger.info("Using dbt manifest at %s", manifest_path)
 
     dbt_resource = DbtCliResource(project_dir=dbt_project_dir_path)
     dbt_cloud_polling_sensor = None
 
     @dbt_assets(
-        manifest=dbt_project.manifest_path,
+        manifest=manifest_path,
         exclude=DBT_NASDAQ_EXCLUDE,
         dagster_dbt_translator=CustomizedDagsterDbtTranslator(),
     )
     def full_dbt_assets(context: dg.AssetExecutionContext, dbt: DbtCliResource):
-        dbt_packages_dir = dbt_project_dir_path / "dbt_packages"
-        dbt_utils_dir = (
-            dbt_packages_dir / "dbt_utils" if dbt_packages_dir.exists() else None
-        )
-
-        if (
-            not dbt_packages_dir.exists()
-            or not dbt_utils_dir
-            or not dbt_utils_dir.exists()
-        ):
-            context.log.info(
-                "dbt packages not found. Running 'dbt deps' before build..."
-            )
-            try:
-                deps_result = dbt.cli(["deps"], context=context).wait()
-                return_code = getattr(deps_result, "return_code", None)
-                if return_code not in (None, 0):
-                    context.log.error("Failed to install dbt packages.")
-                    raise RuntimeError(
-                        "dbt packages installation failed. Please run 'dbt deps' manually"
-                    )
-                else:
-                    context.log.info("dbt packages installed successfully")
-            except Exception as e:
-                context.log.error(f"Could not install dbt packages: {e}")
-                raise RuntimeError(
-                    "dbt packages are required but could not be installed. Please run 'dbt deps' manually"
-                ) from e
-
+        _require_dbt_packages(dbt_project_dir_path)
         yield from dbt.cli(["build"], context=context).stream()