"""Tests for dbt output parsing using real and synthetic fixtures.
This validates that our output parsing logic handles various scenarios:
- Successful builds (real fixture)
- Failed builds (synthetic)
- Mixed scenarios (synthetic)
- Different resource types
- Edge cases (empty results, skipped)
"""
import json
from pathlib import Path
from typing import Any, Dict
import pytest
@pytest.fixture
def fixtures_dir() -> Path:
"""Path to test fixtures directory."""
return Path(__file__).parent / "fixtures"
@pytest.fixture
def real_run_results(fixtures_dir: Path) -> Dict[str, Any]:
"""Load real successful dbt run_results.json fixture."""
run_results_path = fixtures_dir / "target" / "run_results.json"
assert run_results_path.exists(), f"Fixture not found: {run_results_path}"
with open(run_results_path) as f:
return json.load(f)
@pytest.fixture
def failed_run_results() -> Dict[str, Any]:
"""Create synthetic failed run results for testing."""
return {
"metadata": {"dbt_schema_version": "https://schemas.getdbt.com/dbt/run-results/v6.json"},
"results": [
{
"status": "error",
"timing": [{"name": "compile", "started_at": "2026-01-13T13:11:56.602661Z", "completed_at": "2026-01-13T13:11:56.602661Z"}],
"thread_id": "Thread-1",
"execution_time": 0.5,
"adapter_response": {},
"message": "Compilation Error in model 'customers' (models/customers.sql)\n column \"invalid_column\" must appear in the GROUP BY clause",
"failures": ["compilation error"],
"unique_id": "model.jaffle_shop.customers",
"compiled": None,
"compiled_code": None,
"relation_name": None,
"batch_results": None,
}
],
"elapsed_time": 0.8,
"args": {"which": "build"},
}
@pytest.fixture
def mixed_run_results() -> Dict[str, Any]:
"""Create synthetic mixed results (success, fail, skip) for testing."""
return {
"metadata": {"dbt_schema_version": "https://schemas.getdbt.com/dbt/run-results/v6.json"},
"results": [
{
"status": "success",
"timing": [],
"thread_id": "Thread-1",
"execution_time": 0.2,
"adapter_response": {"_message": "CREATE TABLE", "code": "CREATE"},
"message": "CREATE TABLE (5 rows, 0 processed)",
"failures": None,
"unique_id": "model.jaffle_shop.stg_customers",
"compiled": None,
"compiled_code": None,
"relation_name": None,
"batch_results": None,
},
{
"status": "fail",
"timing": [],
"thread_id": "Thread-2",
"execution_time": 0.1,
"adapter_response": {},
"message": "FAIL 1 assert count(*) > 0",
"failures": ["assert count(*) > 0"],
"unique_id": "test.jaffle_shop.assert_positive_order_amount",
"compiled": None,
"compiled_code": None,
"relation_name": None,
"batch_results": None,
},
{
"status": "skipped",
"timing": [],
"thread_id": "Thread-3",
"execution_time": 0.0,
"adapter_response": {},
"message": "SKIPPED",
"failures": None,
"unique_id": "model.jaffle_shop.orders",
"compiled": None,
"compiled_code": None,
"relation_name": None,
"batch_results": None,
},
],
"elapsed_time": 1.5,
"args": {"which": "build"},
}
@pytest.fixture
def empty_run_results() -> Dict[str, Any]:
"""Create empty results for testing no-match scenarios."""
return {"metadata": {"dbt_schema_version": "https://schemas.getdbt.com/dbt/run-results/v6.json"}, "results": [], "elapsed_time": 0.1, "args": {"which": "build", "select": "nonexistent_model"}}
# Comprehensive Output Parsing Tests
def test_successful_results_parsing(real_run_results: Dict[str, Any]) -> None:
"""Test parsing successful dbt build output from real fixture."""
results_list = real_run_results.get("results", [])
# Assert basic structure
assert len(results_list) > 0, "Should have results in successful fixture"
assert "elapsed_time" in real_run_results, "Should have elapsed_time"
assert real_run_results["elapsed_time"] >= 0, "Elapsed time should be non-negative"
# Test counting logic with all successful results
passed_count = sum(1 for r in results_list if r.get("status") in ("success", "pass"))
failed_count = sum(1 for r in results_list if r.get("status") in ("error", "fail"))
skip_count = sum(1 for r in results_list if r.get("status") == "skipped")
assert passed_count == len(results_list), "All results should be successful in real fixture"
assert failed_count == 0, "Should have no failures in successful fixture"
assert skip_count == 0, "Should have no skips in successful fixture"
def test_failed_results_parsing(failed_run_results: Dict[str, Any]) -> None:
"""Test parsing failed dbt build output."""
results_list = failed_run_results.get("results", [])
# Test counting logic with failed results
passed_count = sum(1 for r in results_list if r.get("status") in ("success", "pass"))
failed_count = sum(1 for r in results_list if r.get("status") in ("error", "fail"))
skip_count = sum(1 for r in results_list if r.get("status") == "skipped")
total = len(results_list)
assert total > 0, "Should have results"
assert failed_count > 0, "Should have failures in failed fixture"
assert passed_count == 0, "Should have no successes in failed fixture"
assert passed_count + failed_count + skip_count == total, "Counts should add up"
# Test message construction for failures
parts = []
if passed_count > 0:
parts.append(f"✅ {passed_count} passed" if failed_count > 0 or skip_count > 0 else "✅ All passed")
if failed_count > 0:
parts.append(f"❌ {failed_count} failed")
if skip_count > 0:
parts.append(f"⏭️ {skip_count} skipped")
summary = f"Build: {total}/{total} resources completed ({', '.join(parts)})"
assert "❌" in summary, "Should have failure indicator"
assert "failed" in summary, "Should mention failures"
assert "✅" not in summary, "Should not have success indicator when all failed"
def test_mixed_results_parsing(mixed_run_results: Dict[str, Any]) -> None:
"""Test parsing mixed success/failure/skip results."""
results_list = mixed_run_results.get("results", [])
# Test counting logic with mixed results
passed_count = sum(1 for r in results_list if r.get("status") in ("success", "pass"))
failed_count = sum(1 for r in results_list if r.get("status") in ("error", "fail"))
skip_count = sum(1 for r in results_list if r.get("status") == "skipped")
total = len(results_list)
assert total == 3, "Mixed fixture should have 3 results"
assert passed_count == 1, "Should have 1 success"
assert failed_count == 1, "Should have 1 failure"
assert skip_count == 1, "Should have 1 skip"
assert passed_count + failed_count + skip_count == total, "Counts should add up"
# Test message construction for mixed results
parts = []
if passed_count > 0:
parts.append(f"✅ {passed_count} passed" if failed_count > 0 or skip_count > 0 else "✅ All passed")
if failed_count > 0:
parts.append(f"❌ {failed_count} failed")
if skip_count > 0:
parts.append(f"⏭️ {skip_count} skipped")
summary = f"Build: {total}/{total} resources completed ({', '.join(parts)})"
# Should show all three indicators
assert "✅ 1 passed" in summary, "Should show specific passed count"
assert "❌ 1 failed" in summary, "Should show failed count"
assert "⏭️ 1 skipped" in summary, "Should show skipped count"
assert "All passed" not in summary, "Should not show 'All passed' when mixed"
def test_empty_results_parsing(empty_run_results: Dict[str, Any]) -> None:
"""Test parsing empty results (no matches)."""
results_list = empty_run_results.get("results", [])
# Test counting logic with empty results
passed_count = sum(1 for r in results_list if r.get("status") in ("success", "pass"))
failed_count = sum(1 for r in results_list if r.get("status") in ("error", "fail"))
skip_count = sum(1 for r in results_list if r.get("status") == "skipped")
total = len(results_list)
assert total == 0, "Empty fixture should have no results"
assert passed_count == 0, "Should have no passes"
assert failed_count == 0, "Should have no failures"
assert skip_count == 0, "Should have no skips"
# Test message construction for empty results
parts = []
if passed_count > 0:
parts.append(f"✅ {passed_count} passed" if failed_count > 0 or skip_count > 0 else "✅ All passed")
if failed_count > 0:
parts.append(f"❌ {failed_count} failed")
if skip_count > 0:
parts.append(f"⏭️ {skip_count} skipped")
# Empty results should produce special message
if not parts:
summary = "Build: 0/0 resources completed (no resources matched selection)"
else:
summary = f"Build: {total}/{total} resources completed ({', '.join(parts)})"
assert "0/0" in summary, "Should show zero progress"
assert "no resources matched" in summary or len(parts) == 0, "Should handle empty case"
def test_resource_type_extraction(real_run_results: Dict[str, Any]) -> None:
"""Test extraction of resource types from unique_id."""
results_list = real_run_results.get("results", [])
# Extract resource types from unique_id (format: resource_type.package.name)
resource_types = set()
for result in results_list:
unique_id = result.get("unique_id", "")
if "." in unique_id:
resource_type = unique_id.split(".")[0]
resource_types.add(resource_type)
print(f"Found resource types: {resource_types}")
# Assert we found valid resource types
assert len(resource_types) > 0, "Should extract at least one resource type"
valid_types = {"model", "test", "seed", "snapshot", "analysis", "macro", "source", "unit_test"}
invalid_types = resource_types - valid_types
assert not invalid_types, f"Found invalid resource types: {invalid_types}"
def test_result_structure_validation(real_run_results: Dict[str, Any], failed_run_results: Dict[str, Any], mixed_run_results: Dict[str, Any]) -> None:
"""Test that all result structures have required fields."""
all_fixtures = [("real", real_run_results), ("failed", failed_run_results), ("mixed", mixed_run_results)]
for fixture_name, fixture_data in all_fixtures:
print(f"\nValidating {fixture_name} fixture structure:")
# Validate top-level structure
assert "results" in fixture_data, f"{fixture_name}: Should have 'results' key"
assert "elapsed_time" in fixture_data, f"{fixture_name}: Should have 'elapsed_time' key"
assert isinstance(fixture_data["results"], list), f"{fixture_name}: results should be list"
assert isinstance(fixture_data["elapsed_time"], (int, float)), f"{fixture_name}: elapsed_time should be numeric"
# Validate each result structure
results = fixture_data["results"]
for i, result in enumerate(results):
required_fields = ["status", "unique_id", "message", "timing", "execution_time"]
for field in required_fields:
assert field in result, f"{fixture_name} result {i}: Should have '{field}' field"
# Validate status values
status = result["status"]
valid_statuses = {"success", "pass", "error", "fail", "skipped"}
assert status in valid_statuses, f"{fixture_name} result {i}: Invalid status '{status}'"
# Validate unique_id format
unique_id = result["unique_id"]
assert isinstance(unique_id, str), f"{fixture_name} result {i}: unique_id should be string"
assert "." in unique_id, f"{fixture_name} result {i}: unique_id should have dots (resource_type.package.name)"
def test_fixture_structure_analysis(real_run_results: Dict[str, Any]) -> None:
"""Analyze the structure of real dbt output - kept for debugging purposes."""
print("\n=== Real dbt build output structure ===")
print(f"Top-level keys: {list(real_run_results.keys())}")
print(f"Total results: {len(real_run_results.get('results', []))}")
print(f"Elapsed time: {real_run_results.get('elapsed_time')} seconds")
results = real_run_results.get("results", [])
if results:
print(f"Sample result keys: {list(results[0].keys())}")
unique_id = results[0].get("unique_id", "")
if "." in unique_id:
resource_type = unique_id.split(".")[0]
print(f"Sample resource type: {resource_type}")
