From a347e6f2f51904bdea43ffe8414bdd59f1be0945 Mon Sep 17 00:00:00 2001 From: Alex Noonan <48368867+C00ldudeNoonan@users.noreply.github.com> Date: Sat, 11 Jul 2026 14:31:10 -0400 Subject: [PATCH 1/2] test: make dbt and warehouse tests hermetic --- .github/workflows/dagster-ci.yml | 36 +++- AGENTS.md | 5 +- Readme.md | 8 +- .../int_cross_asset_commodity_signals.sql | 2 +- macro_agents/pyproject.toml | 14 ++ .../macro_agents/defs/transformation/dbt.py | 108 ++++------- macro_agents/tests/conftest.py | 131 +++++++++++++ .../tests/test_asset_check_failure_sensor.py | 117 ++++++------ macro_agents/tests/test_dbt_validation.py | 179 +++++------------- macro_agents/tests/test_integration.py | 14 +- macro_agents/tests/test_network_isolation.py | 16 ++ macro_agents/tests/test_schedules.py | 49 +++-- .../tests/test_sec_search_functional.py | 10 + macro_agents/tests/test_social_assets.py | 87 +++++---- makefile | 16 +- 15 files changed, 447 insertions(+), 345 deletions(-) create mode 100644 macro_agents/tests/test_network_isolation.py diff --git a/.github/workflows/dagster-ci.yml b/.github/workflows/dagster-ci.yml index 43a3fb8..0c2b878 100644 --- a/.github/workflows/dagster-ci.yml +++ b/.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,11 @@ 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 + uv run sqlfluff lint ../dbt_project/models --disable-progress-bar --processes 4 + - name: Test Dagster definitions with dbt Platform if: ${{ env.DBT_PLATFORM_TOKEN != '' }} env: @@ -85,4 +114,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 diff --git a/AGENTS.md b/AGENTS.md index cd83845..5b7f121 100644 --- a/AGENTS.md +++ b/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 diff --git a/Readme.md b/Readme.md index b7a52f8..fd85f59 100644 --- a/Readme.md +++ b/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 diff --git a/dbt_project/models/signals/int_cross_asset_commodity_signals.sql b/dbt_project/models/signals/int_cross_asset_commodity_signals.sql index 08911d4..e9f7ad4 100644 --- a/dbt_project/models/signals/int_cross_asset_commodity_signals.sql +++ b/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 ( diff --git a/macro_agents/pyproject.toml b/macro_agents/pyproject.toml index 647d989..f484629 100644 --- a/macro_agents/pyproject.toml +++ b/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 diff --git a/macro_agents/src/macro_agents/defs/transformation/dbt.py b/macro_agents/src/macro_agents/defs/transformation/dbt.py index de0516d..a990d31 100644 --- a/macro_agents/src/macro_agents/defs/transformation/dbt.py +++ b/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() diff --git a/macro_agents/tests/conftest.py b/macro_agents/tests/conftest.py index 30cf385..d92a5af 100644 --- a/macro_agents/tests/conftest.py +++ b/macro_agents/tests/conftest.py @@ -2,7 +2,15 @@ Pytest configuration and fixtures. """ +import json +import os import re +import socket +import subprocess +import sys +from pathlib import Path +from types import TracebackType +from typing import NoReturn, Self from unittest.mock import Mock import duckdb @@ -20,6 +28,118 @@ _NAMED_PARAMETER_RE = re.compile(r"@([A-Za-z_][A-Za-z0-9_]*)") +class NetworkAccessBlocked(RuntimeError): + """Raised when a default test attempts external network access.""" + + +def _block_network(*args: object, **kwargs: object) -> NoReturn: + raise NetworkAccessBlocked( + "Network access is disabled in the default test suite. " + "Mark the test with @pytest.mark.network and opt in explicitly." + ) + + +_network_monkeypatch: pytest.MonkeyPatch | None = None + + +def pytest_configure(config: pytest.Config) -> None: + """Block Python network access before collection and session fixtures run.""" + if os.getenv("RUN_NETWORK_TESTS") == "1": + return + global _network_monkeypatch + _network_monkeypatch = pytest.MonkeyPatch() + _network_monkeypatch.setattr(socket, "create_connection", _block_network) + _network_monkeypatch.setattr(socket.socket, "connect", _block_network) + _network_monkeypatch.setattr(socket.socket, "connect_ex", _block_network) + + +def pytest_unconfigure(config: pytest.Config) -> None: + """Restore socket functions after the pytest session.""" + global _network_monkeypatch + if _network_monkeypatch is not None: + _network_monkeypatch.undo() + _network_monkeypatch = None + + +@pytest.fixture(scope="session") +def dbt_project_dir() -> Path: + """Return the repository dbt project used by integration tests.""" + project_dir = Path(__file__).resolve().parents[2] / "dbt_project" + if not project_dir.exists(): + pytest.fail(f"dbt project directory not found at {project_dir}") + return project_dir + + +@pytest.fixture(scope="session") +def dbt_manifest(dbt_project_dir: Path) -> dict[str, object]: + """Parse dbt once, offline, and return the generated manifest.""" + packages_dir = dbt_project_dir / "dbt_packages" + missing_packages = [ + package + for package in ("dbt_project_evaluator", "dbt_utils") + if not (packages_dir / package).exists() + ] + if missing_packages: + pytest.fail( + "dbt packages are not installed: " + f"{', '.join(missing_packages)}. Run `make dbt-deps` before pytest." + ) + + dbt_executable = Path(sys.executable).with_name("dbt") + if not dbt_executable.exists(): + pytest.fail(f"dbt executable not found next to Python at {dbt_executable}") + + command = [ + str(dbt_executable), + "parse", + "--project-dir", + str(dbt_project_dir), + "--profiles-dir", + str(dbt_project_dir), + "--target", + "dev", + "--no-send-anonymous-usage-stats", + ] + environment = { + **os.environ, + "BIGQUERY_PROJECT": "econ-data-project-478800", + "DBT_SEND_ANONYMOUS_USAGE_STATS": "false", + } + try: + subprocess.run( + command, + check=True, + capture_output=True, + text=True, + timeout=60, + env=environment, + ) + except subprocess.CalledProcessError as error: + pytest.fail( + f"dbt parse failed.\nSTDOUT:\n{error.stdout}\n\nSTDERR:\n{error.stderr}", + pytrace=False, + ) + except subprocess.TimeoutExpired as error: + pytest.fail(f"dbt parse timed out after {error.timeout} seconds", pytrace=False) + + manifest_path = dbt_project_dir / "target" / "manifest.json" + if not manifest_path.exists(): + pytest.fail(f"dbt parse did not create {manifest_path}") + manifest = json.loads(manifest_path.read_text(encoding="utf-8")) + if not isinstance(manifest, dict): + pytest.fail("dbt manifest must contain a JSON object") + return manifest + + +@pytest.fixture(scope="session") +def dagster_defs(dbt_manifest: dict[str, object]): + """Load Dagster definitions only after the offline dbt parse succeeds.""" + # Deferred intentionally: definitions read the generated manifest at import time. + from macro_agents.definitions import defs + + return defs + + def _bq_to_duckdb(sql: str) -> str: """Translate BigQuery-specific SQL constructs to DuckDB equivalents.""" sql = sql.replace("`", "") @@ -64,6 +184,17 @@ class DuckDBWarehouseStub: def __init__(self): self._conn = duckdb.connect() + def __enter__(self) -> Self: + return self + + def __exit__( + self, + exc_type: type[BaseException] | None, + exc_value: BaseException | None, + traceback: TracebackType | None, + ) -> None: + self.close() + def execute_query( self, query: str, diff --git a/macro_agents/tests/test_asset_check_failure_sensor.py b/macro_agents/tests/test_asset_check_failure_sensor.py index f2bc571..180d493 100644 --- a/macro_agents/tests/test_asset_check_failure_sensor.py +++ b/macro_agents/tests/test_asset_check_failure_sensor.py @@ -215,29 +215,29 @@ def test_asset_failure_monitor_materializes_with_metadata(self): mock_github = Mock() mock_github.setup_for_execution = Mock() - instance = DagsterInstance.ephemeral() - instance.event_log_storage.get_asset_check_summary_records = Mock( - return_value={} - ) - instance.get_latest_materialization_event = Mock(return_value=None) - mock_context = build_op_context( - resources={"bq": mock_md, "github": mock_github}, instance=instance - ) - - with ( - patch( - "macro_agents.defs.asset_failure_sensor._initialize_failure_tracking_table" - ) as mock_initialize, - patch( - "macro_agents.defs.asset_failure_sensor._get_all_asset_check_keys", - return_value=[], - ), - patch( - "macro_agents.defs.asset_failure_sensor._get_all_asset_keys", - return_value=[], - ), - ): - result = asset_failure_monitor(mock_context) + with DagsterInstance.ephemeral() as instance: + instance.event_log_storage.get_asset_check_summary_records = Mock( + return_value={} + ) + instance.get_latest_materialization_event = Mock(return_value=None) + with ( + build_op_context( + resources={"bq": mock_md, "github": mock_github}, + instance=instance, + ) as mock_context, + patch( + "macro_agents.defs.asset_failure_sensor._initialize_failure_tracking_table" + ) as mock_initialize, + patch( + "macro_agents.defs.asset_failure_sensor._get_all_asset_check_keys", + return_value=[], + ), + patch( + "macro_agents.defs.asset_failure_sensor._get_all_asset_keys", + return_value=[], + ), + ): + result = asset_failure_monitor(mock_context) mock_initialize.assert_called_once() assert result.metadata["status"] == "ok" @@ -253,16 +253,18 @@ def test_asset_failure_monitor_handles_errors(self): mock_github = Mock() mock_github.setup_for_execution = Mock() - instance = DagsterInstance.ephemeral() - mock_context = build_op_context( - resources={"bq": mock_md, "github": mock_github}, instance=instance - ) - - with patch( - "macro_agents.defs.asset_failure_sensor._initialize_failure_tracking_table", - side_effect=RuntimeError("boom"), - ): - result = asset_failure_monitor(mock_context) + with DagsterInstance.ephemeral() as instance: + with ( + build_op_context( + resources={"bq": mock_md, "github": mock_github}, + instance=instance, + ) as mock_context, + patch( + "macro_agents.defs.asset_failure_sensor._initialize_failure_tracking_table", + side_effect=RuntimeError("boom"), + ), + ): + result = asset_failure_monitor(mock_context) assert result.metadata["status"] == "error" assert "boom" in result.metadata["error"] @@ -276,31 +278,30 @@ def test_asset_failure_monitor_creates_tracking_table(self, tmp_path): from macro_agents.defs.asset_failure_sensor import asset_failure_monitor from tests.conftest import DuckDBWarehouseStub - bq_resource = DuckDBWarehouseStub() - mock_github = Mock() mock_github.setup_for_execution = Mock() - instance = DagsterInstance.ephemeral() - instance.event_log_storage.get_asset_check_summary_records = Mock( - return_value={} - ) - - context = build_op_context( - resources={"bq": bq_resource, "github": mock_github}, instance=instance - ) - - with ( - patch( - "macro_agents.defs.asset_failure_sensor._get_all_asset_check_keys", - return_value=[], - ), - patch( - "macro_agents.defs.asset_failure_sensor._get_all_asset_keys", - return_value=[], - ), - ): - result = asset_failure_monitor(context) - - assert result.metadata["status"] == "ok" - assert bq_resource.table_exists("asset_check_failures") + with DuckDBWarehouseStub() as bq_resource: + with DagsterInstance.ephemeral() as instance: + instance.event_log_storage.get_asset_check_summary_records = Mock( + return_value={} + ) + + with ( + build_op_context( + resources={"bq": bq_resource, "github": mock_github}, + instance=instance, + ) as context, + patch( + "macro_agents.defs.asset_failure_sensor._get_all_asset_check_keys", + return_value=[], + ), + patch( + "macro_agents.defs.asset_failure_sensor._get_all_asset_keys", + return_value=[], + ), + ): + result = asset_failure_monitor(context) + + assert result.metadata["status"] == "ok" + assert bq_resource.table_exists("asset_check_failures") diff --git a/macro_agents/tests/test_dbt_validation.py b/macro_agents/tests/test_dbt_validation.py index 81b1331..6fc7880 100644 --- a/macro_agents/tests/test_dbt_validation.py +++ b/macro_agents/tests/test_dbt_validation.py @@ -2,138 +2,61 @@ Tests for dbt project validation and compilation. """ -import os -import subprocess from pathlib import Path -import pytest - class TestDbtProjectValidation: """Test dbt project configuration and compilation.""" - @pytest.fixture - def dbt_project_dir(self): - """Get the dbt project directory.""" - current_file = Path(__file__).resolve() - macro_agents_dir = current_file.parent.parent - repo_root = macro_agents_dir.parent - dbt_project_dir = repo_root / "dbt_project" - - if not dbt_project_dir.exists(): - pytest.skip(f"dbt_project directory not found at {dbt_project_dir}") - - return dbt_project_dir - - def test_dbt_packages_installed(self, dbt_project_dir): - """Test that dbt packages are installed (dbt_packages directory exists).""" + def test_dbt_packages_installed(self, dbt_project_dir: Path) -> None: + """Test that all locked dbt packages are installed.""" dbt_packages_dir = dbt_project_dir / "dbt_packages" - - if not dbt_packages_dir.exists(): - pytest.skip( - f"dbt_packages directory not found at {dbt_packages_dir}. " - f"Run 'dbt deps' in the dbt_project directory to install packages. " - f"This is a setup requirement, not a code issue." - ) - - dbt_utils_dir = dbt_packages_dir / "dbt_utils" - if not dbt_utils_dir.exists(): - pytest.skip( - "dbt_utils package not found. " - "Run 'dbt deps' in the dbt_project directory to install packages. " - "This is a setup requirement, not a code issue." - ) - - def test_dbt_parse_succeeds(self, dbt_project_dir): - """Test that dbt can parse the project without errors.""" - original_cwd = os.getcwd() - try: - os.chdir(dbt_project_dir) - - result = subprocess.run( - ["dbt", "parse", "--target", "dev"], - capture_output=True, - text=True, - timeout=60, - env={**os.environ, "BIGQUERY_PROJECT": "econ-data-project-478800"}, - ) - - if result.returncode != 0: - pytest.fail( - f"dbt parse failed with exit code {result.returncode}.\n" - f"STDOUT:\n{result.stdout}\n\n" - f"STDERR:\n{result.stderr}" - ) - finally: - os.chdir(original_cwd) - - def test_dbt_list_models_succeeds(self, dbt_project_dir): - """Test that dbt can list all models (validates project structure).""" - original_cwd = os.getcwd() - try: - os.chdir(dbt_project_dir) - - result = subprocess.run( - ["dbt", "list", "--target", "dev", "--resource-type", "model"], - capture_output=True, - text=True, - timeout=60, - env={**os.environ, "BIGQUERY_PROJECT": "econ-data-project-478800"}, - ) - - if result.returncode != 0: - pytest.fail( - f"dbt list models failed with exit code {result.returncode}.\n" - f"STDOUT:\n{result.stdout}\n\n" - f"STDERR:\n{result.stderr}" - ) - - models = [ - line.strip() for line in result.stdout.split("\n") if line.strip() - ] - if len(models) == 0: - pytest.fail("No models found in dbt project") - finally: - os.chdir(original_cwd) - - def test_dbt_test_syntax_valid(self, dbt_project_dir): - """Test that dbt can validate test syntax without running tests.""" - original_cwd = os.getcwd() - try: - os.chdir(dbt_project_dir) - - result = subprocess.run( - ["dbt", "parse", "--target", "dev"], - capture_output=True, - text=True, - timeout=60, - env={**os.environ, "BIGQUERY_PROJECT": "econ-data-project-478800"}, - ) - - if result.returncode != 0: - error_output = result.stderr or result.stdout - if "syntax error" in error_output.lower(): - pytest.fail(f"dbt test syntax errors detected:\n{error_output}") - pytest.fail( - f"dbt parse failed (may indicate test syntax issues):\n{error_output}" - ) - finally: - os.chdir(original_cwd) - - def test_dbt_project_yml_exists(self, dbt_project_dir): + assert dbt_packages_dir.is_dir() + assert (dbt_packages_dir / "dbt_utils").is_dir() + assert (dbt_packages_dir / "dbt_project_evaluator").is_dir() + + def test_dbt_parse_succeeds(self, dbt_manifest: dict[str, object]) -> None: + """Test that the session-level offline parse produced a manifest.""" + assert dbt_manifest["metadata"] + + def test_dbt_manifest_contains_models( + self, dbt_manifest: dict[str, object] + ) -> None: + """Test that the parsed project contains model nodes.""" + nodes = dbt_manifest["nodes"] + assert isinstance(nodes, dict) + models = [ + node + for node in nodes.values() + if isinstance(node, dict) and node.get("resource_type") == "model" + ] + assert models + + def test_dbt_manifest_contains_tests(self, dbt_manifest: dict[str, object]) -> None: + """Test that dbt test definitions parsed successfully.""" + nodes = dbt_manifest["nodes"] + assert isinstance(nodes, dict) + tests = [ + node + for node in nodes.values() + if isinstance(node, dict) and node.get("resource_type") == "test" + ] + assert tests + + def test_dbt_project_yml_exists(self, dbt_project_dir: Path) -> None: """Test that dbt_project.yml exists and is valid.""" dbt_project_yml = dbt_project_dir / "dbt_project.yml" - if not dbt_project_yml.exists(): - pytest.fail(f"dbt_project.yml not found at {dbt_project_yml}") + assert dbt_project_yml.exists(), ( + f"dbt_project.yml not found at {dbt_project_yml}" + ) dbt_project_yaml = dbt_project_dir / "dbt_project.yaml" - if not dbt_project_yaml.exists() and not dbt_project_yml.exists(): - pytest.fail( - f"Neither dbt_project.yml nor dbt_project.yaml found in {dbt_project_dir}" - ) + assert dbt_project_yaml.exists() or dbt_project_yml.exists(), ( + f"Neither dbt_project.yml nor dbt_project.yaml found in {dbt_project_dir}" + ) - def test_dbt_packages_yml_or_in_project_yml(self, dbt_project_dir): + def test_dbt_packages_yml_or_in_project_yml(self, dbt_project_dir: Path) -> None: """Test that packages are defined (either in packages.yml or dbt_project.yml).""" packages_yml = dbt_project_dir / "packages.yml" dbt_project_yml = dbt_project_dir / "dbt_project.yml" @@ -142,13 +65,11 @@ def test_dbt_packages_yml_or_in_project_yml(self, dbt_project_dir): has_packages_in_project = False if dbt_project_yml.exists(): - with open(dbt_project_yml, "r") as f: - content = f.read() - if "packages:" in content or "package:" in content: - has_packages_in_project = True - - if not has_packages_yml and not has_packages_in_project: - pytest.fail( - "No packages definition found. " - "Either create packages.yml or define packages in dbt_project.yml" - ) + content = dbt_project_yml.read_text(encoding="utf-8") + if "packages:" in content or "package:" in content: + has_packages_in_project = True + + assert has_packages_yml or has_packages_in_project, ( + "No packages definition found. Either create packages.yml or define " + "packages in dbt_project.yml" + ) diff --git a/macro_agents/tests/test_integration.py b/macro_agents/tests/test_integration.py index 12d9a40..7b42ded 100644 --- a/macro_agents/tests/test_integration.py +++ b/macro_agents/tests/test_integration.py @@ -2,20 +2,18 @@ Integration tests for the macro_agents project. """ -from macro_agents.definitions import defs - class TestDagsterDefinitions: """Test Dagster definitions integration.""" - def test_definitions_load(self): + def test_definitions_load(self, dagster_defs): """Test that definitions load without errors.""" - assert defs is not None - assert len(defs.assets) > 0 - assert len(defs.resources) > 0 + assert dagster_defs is not None + assert len(dagster_defs.assets) > 0 + assert len(dagster_defs.resources) > 0 - def test_all_resources_are_configurable(self): + def test_all_resources_are_configurable(self, dagster_defs): """Test that all resources are configurable.""" - for resource_key, resource in defs.resources.items(): + for resource_key, resource in dagster_defs.resources.items(): assert hasattr(resource, "model_config") assert hasattr(type(resource), "model_fields") diff --git a/macro_agents/tests/test_network_isolation.py b/macro_agents/tests/test_network_isolation.py new file mode 100644 index 0000000..e107b14 --- /dev/null +++ b/macro_agents/tests/test_network_isolation.py @@ -0,0 +1,16 @@ +"""Regression coverage for the default no-network test contract.""" + +import os +import socket + +import pytest + +pytestmark = pytest.mark.skipif( + os.getenv("RUN_NETWORK_TESTS") == "1", + reason="Network isolation is intentionally disabled for opt-in network tests", +) + + +def test_default_suite_blocks_socket_connections() -> None: + with pytest.raises(RuntimeError, match="Network access is disabled"): + socket.create_connection(("127.0.0.1", 9)) diff --git a/macro_agents/tests/test_schedules.py b/macro_agents/tests/test_schedules.py index 419465d..8f8a789 100644 --- a/macro_agents/tests/test_schedules.py +++ b/macro_agents/tests/test_schedules.py @@ -3,7 +3,6 @@ """ from dagster import DefaultSensorStatus -from macro_agents.definitions import defs from macro_agents.defs.analysis.schedules import ( daily_reddit_summary_schedule, weekly_economic_analysis_schedule_bullish, @@ -111,9 +110,9 @@ def test_telemetry_schedules_do_not_overlap(self): class TestScheduledJobs: """Test cases for scheduled job definitions.""" - def test_job_creation(self): + def test_job_creation(self, dagster_defs): """Test that scheduled jobs are created correctly.""" - job_names = [job.name for job in defs.jobs] + job_names = [job.name for job in dagster_defs.jobs] assert "weekly_economic_analysis_job_skeptical" in job_names assert "weekly_economic_analysis_job_neutral" in job_names @@ -156,9 +155,9 @@ def test_job_creation(self): assert "entropy_complexity_job" in job_names assert "network_correlation_job" in job_names - def test_job_asset_selection(self): + def test_job_asset_selection(self, dagster_defs): """Test that jobs select the correct assets.""" - job_names = {job.name for job in defs.jobs} + job_names = {job.name for job in dagster_defs.jobs} assert "weekly_economic_analysis_job_skeptical" in job_names assert "weekly_economic_analysis_job_neutral" in job_names @@ -219,12 +218,12 @@ def test_company_list_schedule_creation(self): class TestDefinitionsIntegration: """Test integration of schedules with definitions.""" - def test_definitions_include_schedules(self): + def test_definitions_include_schedules(self, dagster_defs): """Test that definitions include all schedules.""" - assert defs is not None - assert len(defs.schedules) > 0 + assert dagster_defs is not None + assert len(dagster_defs.schedules) > 0 - schedule_names = [schedule.name for schedule in defs.schedules] + schedule_names = [schedule.name for schedule in dagster_defs.schedules] expected_schedules = [ "weekly_economic_analysis_schedule_skeptical", "weekly_economic_analysis_schedule_neutral", @@ -259,12 +258,12 @@ def test_definitions_include_schedules(self): for expected_schedule in expected_schedules: assert expected_schedule in schedule_names - def test_definitions_include_sensors(self): + def test_definitions_include_sensors(self, dagster_defs): """Test that definitions include ingestion sensors.""" - assert defs is not None - assert len(defs.sensors) > 0 + assert dagster_defs is not None + assert len(dagster_defs.sensors) > 0 - sensor_names = [sensor.name for sensor in defs.sensors] + sensor_names = [sensor.name for sensor in dagster_defs.sensors] expected_sensors = [ "treasury_yields_ingestion_schedule", "realtor_gdrive_file_monitor", @@ -273,12 +272,12 @@ def test_definitions_include_sensors(self): for expected_sensor in expected_sensors: assert expected_sensor in sensor_names - def test_definitions_include_jobs(self): + def test_definitions_include_jobs(self, dagster_defs): """Test that definitions include scheduled jobs.""" - assert defs is not None - assert len(defs.jobs) > 0 + assert dagster_defs is not None + assert len(dagster_defs.jobs) > 0 - job_names = [job.name for job in defs.jobs] + job_names = [job.name for job in dagster_defs.jobs] expected_jobs = [ "weekly_economic_analysis_job_skeptical", "weekly_economic_analysis_job_neutral", @@ -324,9 +323,9 @@ class TestAssetScheduling: """Test cases for asset scheduling metadata.""" @staticmethod - def _get_asset_key_strings(): + def _get_asset_key_strings(dagster_defs): asset_key_strings = [] - for asset_def in defs.assets: + for asset_def in dagster_defs.assets: if hasattr(asset_def, "keys"): asset_key_strings.extend( [key.to_user_string() for key in asset_def.keys] @@ -335,7 +334,7 @@ def _get_asset_key_strings(): asset_key_strings.append(asset_def.key.to_user_string()) return asset_key_strings - def test_economic_analysis_assets_exist(self): + def test_economic_analysis_assets_exist(self, dagster_defs): """Test that economic analysis assets exist.""" economic_assets = [ "analyze_economy_state", @@ -343,15 +342,15 @@ def test_economic_analysis_assets_exist(self): "generate_investment_recommendations", ] - asset_key_strings = self._get_asset_key_strings() + asset_key_strings = self._get_asset_key_strings(dagster_defs) for asset_name in economic_assets: assert asset_name in asset_key_strings - def test_asset_failure_monitor_has_cron_automation(self): + def test_asset_failure_monitor_has_cron_automation(self, dagster_defs): """Test that asset failure monitoring is configured with a cron automation condition.""" asset_def = None - for asset in defs.assets: + for asset in dagster_defs.assets: if hasattr(asset, "keys"): for key in asset.keys: if key.to_user_string() == "asset_failure_monitor": @@ -478,13 +477,13 @@ def test_sensor_configuration(self): class TestScheduleDependencies: """Test cases for schedule dependencies and ordering.""" - def test_economic_analysis_pipeline_dependencies(self): + def test_economic_analysis_pipeline_dependencies(self, dagster_defs): """Test that economic analysis pipeline dependencies are correct.""" # Verify the pipeline order: economy_state -> relationships -> recommendations asset_key_strings = [] asset_defs_by_key_string = {} - for asset_def in defs.assets: + for asset_def in dagster_defs.assets: if hasattr(asset_def, "keys"): for key in asset_def.keys: key_str = key.to_user_string() diff --git a/macro_agents/tests/test_sec_search_functional.py b/macro_agents/tests/test_sec_search_functional.py index d344b5c..452fe5c 100644 --- a/macro_agents/tests/test_sec_search_functional.py +++ b/macro_agents/tests/test_sec_search_functional.py @@ -4,6 +4,8 @@ against an in-memory DuckDB with realistic SEC filing content. """ +import os + import duckdb import polars as pl import pytest @@ -17,6 +19,14 @@ keyword_search, ) +pytestmark = [ + pytest.mark.network, + pytest.mark.skipif( + os.getenv("RUN_NETWORK_TESTS") != "1", + reason=("DuckDB downloads the FTS extension; run with RUN_NETWORK_TESTS=1"), + ), +] + def vector_search_3d( conn, query_embedding, *, symbol=None, section_name=None, top_k=10 diff --git a/macro_agents/tests/test_social_assets.py b/macro_agents/tests/test_social_assets.py index c5d7108..8751a52 100644 --- a/macro_agents/tests/test_social_assets.py +++ b/macro_agents/tests/test_social_assets.py @@ -5,6 +5,7 @@ import dagster as dg import polars as pl +import pytest from macro_agents.defs.domains.social import ( reddit_comments_raw, reddit_content_embeddings, @@ -16,10 +17,12 @@ def _partition_key(subreddit: str = "investing", date: str = "2024-01-15"): return dg.MultiPartitionKey({"subreddit": subreddit, "date": date}) -def _build_context(subreddit: str = "investing", date: str = "2024-01-15"): - return dg.build_asset_context( - partition_key=_partition_key(subreddit, date), - ) +@pytest.fixture() +def asset_context(): + with dg.build_asset_context( + partition_key=_partition_key(), + ) as context: + yield context def _meta_val(metadata: dict, key: str): @@ -35,8 +38,8 @@ def _meta_val(metadata: dict, key: str): class TestRedditPostContentRaw: """Tests for reddit_post_content_raw asset.""" - def test_fetches_content_for_posts(self): - context = _build_context() + def test_fetches_content_for_posts(self, asset_context): + context = asset_context reddit = Mock() md = Mock() @@ -80,8 +83,8 @@ def test_fetches_content_for_posts(self): assert _meta_val(result.metadata, "skipped") == 0 md.upsert_data.assert_called_once() - def test_handles_empty_posts(self): - context = _build_context() + def test_handles_empty_posts(self, asset_context): + context = asset_context reddit = Mock() md = Mock() @@ -94,8 +97,8 @@ def test_handles_empty_posts(self): assert _meta_val(result.metadata, "num_posts") == 0 reddit.get_post_content.assert_not_called() - def test_handles_fetch_failure(self): - context = _build_context() + def test_handles_fetch_failure(self, asset_context): + context = asset_context reddit = Mock() md = Mock() @@ -112,8 +115,8 @@ def test_handles_fetch_failure(self): assert _meta_val(result.metadata, "failed") == 1 assert _meta_val(result.metadata, "num_posts") == 0 - def test_skips_empty_permalink(self): - context = _build_context() + def test_skips_empty_permalink(self, asset_context): + context = asset_context reddit = Mock() md = Mock() @@ -130,8 +133,8 @@ def test_skips_empty_permalink(self): class TestRedditCommentsRaw: """Tests for reddit_comments_raw asset.""" - def test_fetches_comments_for_posts(self): - context = _build_context() + def test_fetches_comments_for_posts(self, asset_context): + context = asset_context reddit = Mock() md = Mock() @@ -164,8 +167,8 @@ def test_fetches_comments_for_posts(self): assert _meta_val(result.metadata, "failed") == 0 md.upsert_data.assert_called_once() - def test_handles_empty_comments(self): - context = _build_context() + def test_handles_empty_comments(self, asset_context): + context = asset_context reddit = Mock() md = Mock() @@ -193,8 +196,8 @@ def test_handles_empty_comments(self): assert _meta_val(result.metadata, "skipped") == 1 md.upsert_data.assert_not_called() - def test_handles_query_failure(self): - context = _build_context() + def test_handles_query_failure(self, asset_context): + context = asset_context reddit = Mock() md = Mock() @@ -210,8 +213,8 @@ def test_handles_query_failure(self): class TestRedditContentEmbeddings: """Tests for reddit_content_embeddings asset.""" - def test_embeds_posts_and_comments(self): - context = _build_context() + def test_embeds_posts_and_comments(self, asset_context): + context = asset_context ollama = Mock() md = Mock() @@ -240,8 +243,8 @@ def test_embeds_posts_and_comments(self): assert _meta_val(result.metadata, "failed") == 0 md.upsert_data.assert_called_once() - def test_skips_empty_text(self): - context = _build_context() + def test_skips_empty_text(self, asset_context): + context = asset_context ollama = Mock() md = Mock() @@ -267,8 +270,8 @@ def test_skips_empty_text(self): assert _meta_val(result.metadata, "skipped") == 2 ollama.get_embeddings.assert_not_called() - def test_falls_back_to_title_when_no_selftext(self): - context = _build_context() + def test_falls_back_to_title_when_no_selftext(self, asset_context): + context = asset_context ollama = Mock() md = Mock() @@ -290,8 +293,8 @@ def test_falls_back_to_title_when_no_selftext(self): assert _meta_val(result.metadata, "embedded") == 1 ollama.get_embeddings.assert_called_once_with(["Important Market Update"]) - def test_handles_embedding_failure(self): - context = _build_context() + def test_handles_embedding_failure(self, asset_context): + context = asset_context ollama = Mock() md = Mock() @@ -313,8 +316,8 @@ def test_handles_embedding_failure(self): assert _meta_val(result.metadata, "failed") == 1 assert _meta_val(result.metadata, "embedded") == 0 - def test_handles_query_failure_gracefully(self): - context = _build_context() + def test_handles_query_failure_gracefully(self, asset_context): + context = asset_context ollama = Mock() md = Mock() @@ -328,10 +331,10 @@ def test_handles_query_failure_gracefully(self): class TestRedditSentimentScored: """Tests for reddit_sentiment_scored asset.""" - def test_scores_posts_and_comments(self): + def test_scores_posts_and_comments(self, asset_context): from macro_agents.defs.domains.social_sentiment import reddit_sentiment_scored - context = _build_context() + context = asset_context md = Mock() md.execute_query.side_effect = [ @@ -358,10 +361,10 @@ def test_scores_posts_and_comments(self): assert _meta_val(result.metadata, "comments_scored") == 1 md.upsert_data.assert_called_once() - def test_skips_empty_text(self): + def test_skips_empty_text(self, asset_context): from macro_agents.defs.domains.social_sentiment import reddit_sentiment_scored - context = _build_context() + context = asset_context md = Mock() md.execute_query.side_effect = [ @@ -385,10 +388,10 @@ def test_skips_empty_text(self): assert _meta_val(result.metadata, "total_scored") == 0 md.upsert_data.assert_not_called() - def test_handles_query_failure(self): + def test_handles_query_failure(self, asset_context): from macro_agents.defs.domains.social_sentiment import reddit_sentiment_scored - context = _build_context() + context = asset_context md = Mock() md.execute_query.side_effect = Exception("DB error") @@ -414,10 +417,10 @@ def test_sentiment_labels_correct(self): assert results[1]["compound"] < -0.05 assert results[2]["label"] == "neutral" - def test_content_types_correct(self): + def test_content_types_correct(self, asset_context): from macro_agents.defs.domains.social_sentiment import reddit_sentiment_scored - context = _build_context() + context = asset_context md = Mock() md.execute_query.side_effect = [ @@ -483,10 +486,10 @@ def test_ignores_lowercase(self): class TestRedditTickerMentions: """Tests for reddit_ticker_mentions asset.""" - def test_extracts_from_posts_and_comments(self): + def test_extracts_from_posts_and_comments(self, asset_context): from macro_agents.defs.domains.social_tickers import reddit_ticker_mentions - context = _build_context() + context = asset_context md = Mock() md.execute_query.side_effect = [ @@ -511,10 +514,10 @@ def test_extracts_from_posts_and_comments(self): assert _meta_val(result.metadata, "unique_tickers") == 4 md.upsert_data.assert_called_once() - def test_no_tickers_found(self): + def test_no_tickers_found(self, asset_context): from macro_agents.defs.domains.social_tickers import reddit_ticker_mentions - context = _build_context() + context = asset_context md = Mock() md.execute_query.side_effect = [ @@ -539,10 +542,10 @@ def test_no_tickers_found(self): assert _meta_val(result.metadata, "unique_tickers") == 0 md.upsert_data.assert_not_called() - def test_handles_query_failure(self): + def test_handles_query_failure(self, asset_context): from macro_agents.defs.domains.social_tickers import reddit_ticker_mentions - context = _build_context() + context = asset_context md = Mock() md.execute_query.side_effect = Exception("DB error") diff --git a/makefile b/makefile index fcf00fb..a03a06a 100644 --- a/makefile +++ b/makefile @@ -8,10 +8,10 @@ lint: fix: cd macro_agents && uv sync --extra dev && uv run sqlfluff fix ../dbt_project/models --disable-progress-bar --processes 4 -test: dbt-manifest +test: cd macro_agents && uv sync --extra dev && uv run pytest tests/ -v -test-dagster: dbt-manifest +test-dagster: cd macro_agents && uv sync --extra dev && uv run pytest tests/ -v typecheck-dagster: @@ -27,15 +27,21 @@ typecheck-dagster: dbt-manifest: cd macro_agents && uv sync --extra dev && uv run dbt parse \ --project-dir ../dbt_project \ - --profiles-dir ../dbt_project + --profiles-dir ../dbt_project \ + --no-send-anonymous-usage-stats + +dbt-deps: + cd macro_agents && uv sync --extra dev && uv run dbt deps \ + --project-dir ../dbt_project \ + --profiles-dir ../dbt_project \ + --no-send-anonymous-usage-stats # ============================================================================= # Local Development # ============================================================================= # Dagster Setup (uses uv) -setup-dagster: - cd macro_agents && uv sync --extra dev +setup-dagster: dbt-deps @echo "Dagster environment synced with uv" @echo "Run 'make run-dagster' to start Dagster" From 2f1425f4eee6dc5e388e763fb27f301ade5e3089 Mon Sep 17 00:00:00 2001 From: Alex Noonan <48368867+C00ldudeNoonan@users.noreply.github.com> Date: Sat, 11 Jul 2026 15:31:09 -0400 Subject: [PATCH 2/2] fix: keep SQL lint credential-free in CI --- .github/workflows/dagster-ci.yml | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/.github/workflows/dagster-ci.yml b/.github/workflows/dagster-ci.yml index 0c2b878..0bcf513 100644 --- a/.github/workflows/dagster-ci.yml +++ b/.github/workflows/dagster-ci.yml @@ -90,7 +90,8 @@ jobs: - name: Lint dbt SQL with SQLFluff run: | cd macro_agents - uv run sqlfluff lint ../dbt_project/models --disable-progress-bar --processes 4 + # 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 != '' }}