fix(validation): classify warnings independently of message text

.github/workflows/validation-ci.yml

@@ -60,7 +60,7 @@ jobs:
         run: uv run validation/test_validate.py
 
       - name: Run validation pytest tests
-        run: uv run --with pytest --with pyyaml --with jsonschema -m pytest validation/tests/
+        run: uv run --with pytest --with pyyaml --with jsonschema --with sqlglot -m pytest validation/tests/
 
       - name: Validate canonical example
         run: uv run validation/validate.py examples/tpcds_semantic_model.yaml

validation/tests/test_validate.py

@@ -292,6 +292,107 @@ def test_skips_malformed_flat_unique_keys() -> None:
     assert errors == []
 
 
+@pytest.fixture
+def run_validator(tmp_path, monkeypatch, capsys):
+    def run(document):
+        model_path = tmp_path / "model.json"
+        model_path.write_text(json.dumps(document))
+        monkeypatch.setattr(_VALIDATE.sys, "argv", [str(_VALIDATE_PATH), str(model_path)])
+        with pytest.raises(SystemExit) as caught:
+            _VALIDATE.main()
+        return caught.value.code, capsys.readouterr().out
+
+    return run
+
+
+@pytest.mark.parametrize("target", [
+    "missing_customers",
+    "Warning: missing_customers",
+    "[SQL] Warning: missing_customers",
+    "[Reference] Warning: missing_customers",
+])
+def test_unknown_dataset_is_an_error_regardless_of_its_name(run_validator, target):
+    document = _document([_ORDERS], [_relationship(to_columns=["id"], to=target)])
+
+    exit_code, output = run_validator(document)
+
+    assert exit_code == 1
+    assert "Validation FAILED with 1 error(s)" in output
+    assert f"references unknown dataset '{target}'" in output
+    assert "Validation PASSED" not in output
+
+
+def test_duplicate_dataset_with_warning_in_name_is_an_error(run_validator):
+    dataset = {"name": "Warning: orders", "source": "db.s.orders"}
+
+    exit_code, output = run_validator(_document([dataset, dataset], []))
+
+    assert exit_code == 1
+    assert "Validation FAILED with 1 error(s)" in output
+    assert "Duplicate dataset name 'Warning: orders'" in output
+
+
+def test_schema_error_containing_warning_text_is_an_error(run_validator):
+    document = _document([_ORDERS], [])
+    document["Warning: unexpected"] = True
+
+    exit_code, output = run_validator(document)
+
+    assert exit_code == 1
+    assert "Validation FAILED with 1 error(s)" in output
+    assert "[Schema]" in output
+    assert "Warning: unexpected" in output
+
+
+@pytest.mark.skipif(not _VALIDATE.SQLGLOT_AVAILABLE, reason="sqlglot is not installed")
+def test_sql_error_in_metric_with_warning_in_name_is_an_error(run_validator):
+    document = _document([_ORDERS], [])
+    document["metrics"] = [{
+        "name": "Warning: broken_metric",
+        "expression": {"dialects": [{"dialect": "ANSI_SQL", "expression": "SUM("}]},
+    }]
+
+    exit_code, output = run_validator(document)
+
+    assert exit_code == 1
+    assert "Validation FAILED with 1 error(s)" in output
+    assert "[SQL] Metric 'Warning: broken_metric'" in output
+
+
+def test_key_coverage_warning_remains_nonfatal(run_validator):
+    document = _document([_ORDERS, _CUSTOMERS], [_relationship(to_columns=["region"])])
+
+    exit_code, output = run_validator(document)
+
+    assert exit_code == 0
+    assert "[Reference] Warning:" in output
+    assert "Validation PASSED" in output
+
+
+def test_missing_sqlglot_warning_remains_nonfatal(run_validator, monkeypatch):
+    monkeypatch.setattr(_VALIDATE, "SQLGLOT_AVAILABLE", False)
+
+    exit_code, output = run_validator(_document([_ORDERS], []))
+
+    assert exit_code == 0
+    assert "[SQL] Warning: sqlglot not installed" in output
+    assert "Validation PASSED" in output
+
+
+def test_genuine_warning_does_not_hide_reference_error(run_validator):
+    document = _document([_ORDERS, _CUSTOMERS], [
+        _relationship(to_columns=["region"]),
+        {**_relationship(to_columns=["id"], to="Warning: missing"), "name": "broken"},
+    ])
+
+    exit_code, output = run_validator(document)
+
+    assert exit_code == 1
+    assert "[Reference] Warning:" in output
+    assert "references unknown dataset 'Warning: missing'" in output
+    assert "Validation FAILED with 1 error(s)" in output
+
+
 def _arity_relationship(from_columns: list[str], to_columns: list[str]) -> dict:
     return {
         "name": "orders_to_customers",

validation/validate.py

@@ -81,6 +81,10 @@
 SKIP_SQL_VALIDATION = {"MDX", "TABLEAU", "MAQL", "SIGMA", "THOUGHTSPOT", "DAX"}
 
 
+class ValidationWarning(str):
+    """An explicitly nonfatal diagnostic, compatible with existing string callers."""
+
+
 class UniqueKeyLoader(yaml.SafeLoader):
     """Safe YAML loader that rejects duplicate explicit mapping keys."""
 
@@ -234,7 +238,9 @@ def validate_references(data: dict) -> list[str]:
             declared_keys = [k for k in candidate_keys if isinstance(k, list) and k]
             to_column_set = set(to_columns)
             if declared_keys and not any(set(key) <= to_column_set for key in declared_keys):
-                errors.append(f"[Reference] Warning: Relationship '{rel_name}' in model '{model_name}': to_columns {to_columns} does not cover the primary key or a unique key of dataset '{to_ds}'")
+                errors.append(ValidationWarning(
+                    f"[Reference] Warning: Relationship '{rel_name}' in model '{model_name}': to_columns {to_columns} does not cover the primary key or a unique key of dataset '{to_ds}'"
+                ))
 
     return errors
 
@@ -304,7 +310,9 @@ def validate_sql(data: dict) -> list[str]:
         return []
 
     if not SQLGLOT_AVAILABLE:
-        return ["[SQL] Warning: sqlglot not installed, skipping SQL validation. Install with: pip install sqlglot"]
+        return [ValidationWarning(
+            "[SQL] Warning: sqlglot not installed, skipping SQL validation. Install with: pip install sqlglot"
+        )]
 
     model = data
     errors = []
@@ -391,9 +399,9 @@ def main():
 
     # Report results
     if errors:
-        # Separate warnings from errors
-        warnings = [e for e in errors if "Warning:" in e]
-        actual_errors = [e for e in errors if "Warning:" not in e]
+        # Severity must not depend on user-controlled text in a diagnostic.
+        warnings = [e for e in errors if isinstance(e, ValidationWarning)]
+        actual_errors = [e for e in errors if not isinstance(e, ValidationWarning)]
 
         for warning in warnings:
             print(f"  {warning}")