"""
Tests for run_models tool.
"""
from typing import TYPE_CHECKING
import pytest
import pytest_asyncio
if TYPE_CHECKING:
from dbt_core_mcp.server import DbtCoreMcpServer
@pytest_asyncio.fixture
async def seeded_jaffle_shop_server(jaffle_shop_server: "DbtCoreMcpServer"):
"""Jaffle shop server with seeds already loaded."""
# Load seeds first since models depend on them
await jaffle_shop_server.toolImpl_seed_data()
return jaffle_shop_server
@pytest.mark.asyncio
async def test_run_models_all(seeded_jaffle_shop_server: "DbtCoreMcpServer") -> None:
"""Test running all models."""
result = await seeded_jaffle_shop_server.toolImpl_run_models(ctx=None)
assert result["status"] == "success"
assert "results" in result
assert "elapsed_time" in result
assert "command" in result
assert len(result["results"]) > 0
@pytest.mark.asyncio
async def test_run_models_select_specific(seeded_jaffle_shop_server: "DbtCoreMcpServer") -> None:
"""Test running a specific model."""
result = await seeded_jaffle_shop_server.toolImpl_run_models(ctx=None, select="customers")
assert result["status"] == "success"
assert "results" in result
# Should have run customers and possibly dependencies
assert len(result["results"]) >= 1
@pytest.mark.asyncio
async def test_run_models_invalid_selection_combination(jaffle_shop_server: "DbtCoreMcpServer") -> None:
"""Test that using both select_state_modified and select raises error."""
with pytest.raises(ValueError, match="Cannot use both select_state_modified\\* flags and select parameter"):
await jaffle_shop_server.toolImpl_run_models(ctx=None, select="customers", select_state_modified=True)
@pytest.mark.asyncio
async def test_run_models_modified_only_no_state_runs_all(jaffle_shop_server: "DbtCoreMcpServer") -> None:
"""Test select_state_modified without state returns success (cannot determine modifications)."""
# Clean any existing state
assert jaffle_shop_server.project_dir is not None
state_dir = jaffle_shop_server.project_dir / "target" / "state_last_run"
if state_dir.exists():
import shutil
shutil.rmtree(state_dir)
# With no state, select_state_modified should return success with message (not run anything)
result = await jaffle_shop_server.toolImpl_run_models(ctx=None, select_state_modified=True)
assert result["status"] == "success"
assert "No previous state" in result["message"]
assert result["results"] == []
@pytest.mark.asyncio
async def test_run_models_creates_state(seeded_jaffle_shop_server: "DbtCoreMcpServer") -> None:
"""Test that successful run creates state for next modified run."""
# Clean state first
assert seeded_jaffle_shop_server.project_dir is not None
state_dir = seeded_jaffle_shop_server.project_dir / "target" / "state_last_run"
if state_dir.exists():
import shutil
shutil.rmtree(state_dir)
# Run models
result = await seeded_jaffle_shop_server.toolImpl_run_models(ctx=None)
assert result["status"] == "success"
# State should be created
assert state_dir.exists()
assert (state_dir / "manifest.json").exists()
@pytest.mark.asyncio
async def test_run_models_full_refresh(seeded_jaffle_shop_server: "DbtCoreMcpServer") -> None:
"""Test run with full_refresh flag."""
result = await seeded_jaffle_shop_server.toolImpl_run_models(ctx=None, full_refresh=True)
assert result["status"] == "success"
assert "--full-refresh" in result["command"]
