feat: add DatasetValidator for testset structural and content validation

src/evalwire/cli.py

@@ -1,4 +1,4 @@
-"""evalwire CLI — ``evalwire upload``, ``evalwire run``, ``evalwire export``, ``evalwire compare``, and ``evalwire report`` commands."""
+"""evalwire CLI — ``evalwire upload``, ``evalwire run``, ``evalwire validate``, ``evalwire export``, ``evalwire compare``, and ``evalwire report`` commands."""
 
 import sys
 from typing import Literal, cast
@@ -58,6 +58,12 @@ def main() -> None:
     default=None,
     help="Pipe-split delimiter.",
 )
+@click.option(
+    "--strict",
+    is_flag=True,
+    default=False,
+    help="Abort upload if validation issues are found.",
+)
 @click.option(
     "--config",
     "config_path",
@@ -71,6 +77,7 @@ def upload_cmd(
     output_keys: str | None,
     tag_column: str | None,
     delimiter: str | None,
+    strict: bool,
     config_path: str | None,
 ) -> None:
     """Upload a CSV testset to Arize Phoenix as one or more named datasets."""
@@ -102,6 +109,26 @@ def upload_cmd(
                 "No CSV path provided. Use --csv or set csv_path in evalwire.toml."
             )
 
+        if strict:
+            from evalwire.validator import DatasetValidator
+
+            validator = DatasetValidator()
+            result = validator.validate(
+                csv_path=resolved_csv,
+                input_keys=resolved_input_keys,
+                output_keys=resolved_output_keys,
+                tag_column=resolved_tag_column,
+            )
+            if not result.is_valid:
+                for issue in result.issues:
+                    row_info = f"row {issue.row}: " if issue.row is not None else ""
+                    click.echo(f"  {row_info}{issue.message}", err=True)
+                click.echo(
+                    f"Validation failed: {len(result.issues)} issue(s) found. Upload aborted.",
+                    err=True,
+                )
+                sys.exit(2)
+
         from evalwire.uploader import DatasetUploader
 
         client = _make_client()
@@ -302,3 +329,60 @@ def report_cmd(experiment_id: str | None) -> None:
     except Exception as exc:
         click.echo(f"Error: {exc}", err=True)
         sys.exit(2)
+
+
+@main.command("validate")
+@click.option("--csv", "csv_path", default=None, help="Path to the CSV file.")
+@click.option(
+    "--input-keys",
+    default="user_query",
+    show_default=True,
+    help="Comma-separated input column names.",
+)
+@click.option(
+    "--output-keys",
+    default="expected_output",
+    show_default=True,
+    help="Comma-separated output column names.",
+)
+@click.option(
+    "--tag-column",
+    default="tags",
+    show_default=True,
+    help="Column used for dataset splitting.",
+)
+def validate_cmd(
+    csv_path: str | None,
+    input_keys: str,
+    output_keys: str,
+    tag_column: str,
+) -> None:
+    """Validate a CSV testset for structural and content correctness."""
+    if not csv_path:
+        raise click.UsageError("No CSV path provided. Use --csv.")
+    try:
+        from evalwire.validator import DatasetValidator
+
+        resolved_input_keys = [k.strip() for k in input_keys.split(",")]
+        resolved_output_keys = [k.strip() for k in output_keys.split(",")]
+
+        validator = DatasetValidator()
+        result = validator.validate(
+            csv_path=csv_path,
+            input_keys=resolved_input_keys,
+            output_keys=resolved_output_keys,
+            tag_column=tag_column,
+        )
+        if result.is_valid:
+            click.echo("Validation passed: testset is valid.")
+        else:
+            for issue in result.issues:
+                row_info = f"row {issue.row}: " if issue.row is not None else ""
+                click.echo(f"  {row_info}{issue.message}")
+            click.echo(f"Validation failed: {len(result.issues)} issue(s) found.")
+            sys.exit(1)
+    except click.UsageError:
+        raise
+    except Exception as exc:
+        click.echo(f"Error: {exc}", err=True)
+        sys.exit(2)

src/evalwire/validator.py

@@ -0,0 +1,109 @@
+"""Testset validation for evalwire CSV uploads."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from pathlib import Path
+
+import pandas as pd
+
+
+@dataclass
+class ValidationIssue:
+    """A single validation problem found in a testset."""
+
+    row: int | None
+    message: str
+
+
+@dataclass
+class ValidationResult:
+    """Aggregated result of a validation run."""
+
+    issues: list[ValidationIssue] = field(default_factory=list)
+
+    @property
+    def is_valid(self) -> bool:
+        return len(self.issues) == 0
+
+    def __repr__(self) -> str:
+        return f"ValidationResult(issues={len(self.issues)})"
+
+
+class DatasetValidator:
+    """Validate a CSV testset before uploading to Phoenix.
+
+    Checks performed:
+    - Required columns (input keys, output keys, tag column) are present.
+    - No row has an empty tag value.
+    - No row has an empty expected output value.
+    """
+
+    def validate(
+        self,
+        csv_path: Path | str,
+        input_keys: list[str],
+        output_keys: list[str],
+        tag_column: str = "tags",
+    ) -> ValidationResult:
+        """Validate *csv_path* against the given schema.
+
+        Parameters
+        ----------
+        csv_path:
+            Path to the CSV file to validate.
+        input_keys:
+            Expected input column names.
+        output_keys:
+            Expected output column names.
+        tag_column:
+            Name of the column used for dataset splitting.
+
+        Returns
+        -------
+        ValidationResult
+            Contains all discovered issues (structural and row-level).
+
+        Raises
+        ------
+        FileNotFoundError
+            If *csv_path* does not exist.
+        """
+        csv_path = Path(csv_path)
+        if not csv_path.exists():
+            raise FileNotFoundError(f"CSV file not found: {csv_path}")
+
+        df = pd.read_csv(csv_path)
+        issues: list[ValidationIssue] = []
+
+        required_columns = list(input_keys) + list(output_keys) + [tag_column]
+        for col in required_columns:
+            if col not in df.columns:
+                issues.append(
+                    ValidationIssue(row=None, message=f"missing column: {col}")
+                )
+
+        missing_cols = {col for col in required_columns if col not in df.columns}
+
+        if tag_column not in missing_cols:
+            for idx, value in enumerate(df[tag_column], start=1):
+                if pd.isna(value) or str(value).strip() == "":
+                    issues.append(
+                        ValidationIssue(
+                            row=idx, message=f"empty tag in column '{tag_column}'"
+                        )
+                    )
+
+        for output_key in output_keys:
+            if output_key in missing_cols:
+                continue
+            for idx, value in enumerate(df[output_key], start=1):
+                if pd.isna(value) or str(value).strip() == "":
+                    issues.append(
+                        ValidationIssue(
+                            row=idx,
+                            message=f"empty expected output in column '{output_key}'",
+                        )
+                    )
+
+        return ValidationResult(issues=issues)

tests/test_cli.py

@@ -342,3 +342,77 @@ def test_report_exits_zero(self):
     def test_report_missing_experiment_flag_exits_nonzero(self):
         result = _runner().invoke(main, ["report"])
         assert result.exit_code != 0
+
+
+class TestValidateCommand:
+    def test_valid_csv_exits_zero(self, sample_csv: Path):
+        result = _runner().invoke(main, ["validate", "--csv", str(sample_csv)])
+        assert result.exit_code == 0
+        assert "valid" in result.output.lower()
+
+    def test_missing_csv_exits_nonzero(self):
+        result = _runner().invoke(main, ["validate"])
+        assert result.exit_code != 0
+
+    def test_invalid_csv_exits_nonzero(self, tmp_path: Path):
+        bad = tmp_path / "bad.csv"
+        bad.write_text("wrong_col\nval\n")
+        result = _runner().invoke(main, ["validate", "--csv", str(bad)])
+        assert result.exit_code != 0
+        assert "issue" in result.output.lower() or "missing" in result.output.lower()
+
+    def test_invalid_csv_reports_all_issues(self, tmp_path: Path):
+        bad = tmp_path / "bad.csv"
+        bad.write_text("wrong,also_wrong\nval,val\n")
+        result = _runner().invoke(
+            main,
+            [
+                "validate",
+                "--csv",
+                str(bad),
+                "--input-keys",
+                "user_query",
+                "--output-keys",
+                "expected_output",
+            ],
+        )
+        assert result.exit_code != 0
+        assert "user_query" in result.output or "expected_output" in result.output
+
+    def test_custom_input_output_tag_columns(self, tmp_path: Path):
+        f = tmp_path / "custom.csv"
+        f.write_text("q,ans,grp\nhello,world,g1\n")
+        result = _runner().invoke(
+            main,
+            [
+                "validate",
+                "--csv",
+                str(f),
+                "--input-keys",
+                "q",
+                "--output-keys",
+                "ans",
+                "--tag-column",
+                "grp",
+            ],
+        )
+        assert result.exit_code == 0
+
+
+class TestUploadStrictFlag:
+    def test_upload_strict_passes_for_valid_csv(self, sample_csv: Path):
+        client = _mock_client()
+        with patch("evalwire.cli._make_client", return_value=client):
+            result = _runner().invoke(
+                main, ["upload", "--csv", str(sample_csv), "--strict"]
+            )
+        assert result.exit_code == 0
+
+    def test_upload_strict_fails_for_invalid_csv(self, tmp_path: Path):
+        bad = tmp_path / "bad.csv"
+        bad.write_text("wrong\nval\n")
+        client = _mock_client()
+        with patch("evalwire.cli._make_client", return_value=client):
+            result = _runner().invoke(main, ["upload", "--csv", str(bad), "--strict"])
+        assert result.exit_code != 0
+        assert client.datasets.create_dataset.call_count == 0