from __future__ import annotations

import json

from pathlib import Path

from unittest.mock import patch

from click.testing import CliRunner

import pytest

from felderize.cli import _expand_query_paths, _read_text, _split_examples, cli

from felderize.models import Status, TranslationResult

# ---------------------------------------------------------------------------

# Shared fixtures

# ---------------------------------------------------------------------------

_SUCCESS = TranslationResult(

feldera_schema="CREATE TABLE t (x INT NOT NULL);",

feldera_query="CREATE VIEW v AS SELECT x FROM t;",

status=Status.SUCCESS,

)

_UNSUPPORTED = TranslationResult(

feldera_query="CREATE VIEW v AS SELECT x FROM t;",

unsupported=["PIVOT is not supported"],

status=Status.UNSUPPORTED,

)

_ERROR = TranslationResult(

warnings=["Syntax error at line 1"],

status=Status.ERROR,

)

def _schema(tmp_path: Path, content: str = "CREATE TABLE t (x INT NOT NULL);") -> Path:

p = tmp_path / "schema.sql"

p.write_text(content)

return p

def _query(tmp_path: Path, content: str = "CREATE VIEW v AS SELECT x FROM t;") -> Path:

p = tmp_path / "query.sql"

p.write_text(content)

return p

@pytest.fixture(autouse=True)

def _no_auto_download(monkeypatch):

"""Keep CLI tests hermetic: never reach out to GitHub to fetch a compiler.

With no --compiler/FELDERA_COMPILER configured, --validate would otherwise

auto-download the compiler; stub it to "not found" so tests stay offline.

"""

monkeypatch.setattr("felderize.cli.ensure_compiler", lambda *a, **k: None)

# ---------------------------------------------------------------------------

# _split_examples

# ---------------------------------------------------------------------------

class TestSplitExamples:

def test_empty_tuple(self):

dirs, files = _split_examples(())

assert dirs == [] and files == []

def test_directory_goes_to_dirs(self, tmp_path):

dirs, files = _split_examples((str(tmp_path),))

assert dirs == [tmp_path] and files == []

def test_file_goes_to_files(self, tmp_path):

f = tmp_path / "ex.md"

f.write_text("# example")

dirs, files = _split_examples((str(f),))

assert dirs == [] and files == [f]

def test_mixed_paths(self, tmp_path):

sub = tmp_path / "subdir"

sub.mkdir()

f = tmp_path / "ex.md"

f.write_text("# example")

dirs, files = _split_examples((str(sub), str(f)))

assert dirs == [sub] and files == [f]

# ---------------------------------------------------------------------------

# _expand_query_paths

# ---------------------------------------------------------------------------

class TestExpandQueryPaths:

def test_single_file_returned_as_is(self, tmp_path):

f = tmp_path / "q.sql"

f.write_text("SELECT 1")

assert _expand_query_paths((f,)) == [f]

def test_directory_expands_to_sorted_sql_files(self, tmp_path):

(tmp_path / "b.sql").write_text("SELECT 2")

(tmp_path / "a.sql").write_text("SELECT 1")

(tmp_path / "skip.txt").write_text("ignored")

result = _expand_query_paths((tmp_path,))

assert result == [tmp_path / "a.sql", tmp_path / "b.sql"]

def test_empty_directory_returns_empty_list(self, tmp_path):

assert _expand_query_paths((tmp_path,)) == []

def test_mixed_file_and_directory(self, tmp_path):

sub = tmp_path / "sub"

sub.mkdir()

(sub / "c.sql").write_text("SELECT 3")

lone = tmp_path / "lone.sql"

lone.write_text("SELECT 0")

result = _expand_query_paths((lone, sub))

assert result == [lone, sub / "c.sql"]

def test_non_sql_files_in_dir_excluded(self, tmp_path):

(tmp_path / "query.sql").write_text("SELECT 1")

(tmp_path / "notes.md").write_text("ignore me")

result = _expand_query_paths((tmp_path,))

assert all(p.suffix == ".sql" for p in result)

# ---------------------------------------------------------------------------

# download-compiler command

# ---------------------------------------------------------------------------

class TestDownloadCompilerCommand:

def test_success_prints_path_and_env_hint(self, tmp_path):

fake_jar = tmp_path / "sql2dbsp.jar"

runner = CliRunner()

with patch("felderize.cli.download_compiler", return_value=fake_jar):

result = runner.invoke(cli, ["download-compiler"])

assert result.exit_code == 0

assert str(fake_jar) in result.output

assert "FELDERA_COMPILER" in result.output

def test_exception_exits_nonzero(self):

runner = CliRunner()

with patch(

"felderize.cli.download_compiler", side_effect=RuntimeError("no network")

):

result = runner.invoke(cli, ["download-compiler"])

assert result.exit_code == 1

assert "no network" in result.output

def test_version_flag_forwarded(self, tmp_path):

fake_jar = tmp_path / "sql2dbsp.jar"

runner = CliRunner()

with patch("felderize.cli.download_compiler", return_value=fake_jar) as mock_dl:

runner.invoke(cli, ["download-compiler", "--version", "v0.291.0"])

mock_dl.assert_called_once_with(

output_dir=None, version="v0.291.0", force=False

)

# ---------------------------------------------------------------------------

# spark translate

# ---------------------------------------------------------------------------

class TestTranslateCommand:

_BASE = ["spark", "translate"]

def _invoke(self, runner: CliRunner, schema: Path, query: Path, *extra: str):

return runner.invoke(cli, self._BASE + [str(schema), str(query)] + list(extra))

def test_empty_schema_exits_with_error(self, tmp_path):

schema = _schema(tmp_path, "")

query = _query(tmp_path)

result = CliRunner().invoke(cli, self._BASE + [str(schema), str(query)])

assert result.exit_code == 1

assert "Error" in result.output

def test_empty_query_exits_with_error(self, tmp_path):

schema = _schema(tmp_path)

query = _query(tmp_path, "")

result = CliRunner().invoke(cli, self._BASE + [str(schema), str(query)])

assert result.exit_code == 1

def test_schema_without_create_table_exits(self, tmp_path):

schema = _schema(tmp_path, "SELECT 1;")

query = _query(tmp_path)

result = CliRunner().invoke(cli, self._BASE + [str(schema), str(query)])

assert result.exit_code == 1

def test_binary_schema_file_errors_cleanly(self, tmp_path):

"""A non-UTF-8 (binary) input is rejected with a message, not a traceback."""

schema = tmp_path / "schema.sql"

schema.write_bytes(b"\x88\x00\xff binary not text")

query = _query(tmp_path)

result = CliRunner().invoke(cli, self._BASE + [str(schema), str(query)])

assert result.exit_code == 1

assert "cannot read" in result.output and "not binary" in result.output

assert "Traceback" not in result.output

def test_success_text_output(self, tmp_path):

runner = CliRunner()

with patch("felderize.cli.translate_spark_to_feldera", return_value=_SUCCESS):

result = self._invoke(

runner, _schema(tmp_path), _query(tmp_path), "--validate"

)

assert result.exit_code == 0

assert "-- Schema --" in result.output

assert "-- Query --" in result.output

def test_validate_auto_resolves_compiler(self, tmp_path, monkeypatch):

"""With --validate and no --compiler, the auto-resolved JAR is passed

through to the translator via the config."""

fake_jar = tmp_path / "sql2dbsp-jar-with-dependencies-v0.310.0.jar"

monkeypatch.setattr("felderize.cli.ensure_compiler", lambda *a, **k: fake_jar)

runner = CliRunner()

with patch(

"felderize.cli.translate_spark_to_feldera", return_value=_SUCCESS

) as mock_fn:

self._invoke(runner, _schema(tmp_path), _query(tmp_path), "--validate")

config = mock_fn.call_args.args[2]

assert config.feldera_compiler == str(fake_jar)

def test_explicit_compiler_skips_auto_resolution(self, tmp_path, monkeypatch):

"""An explicit --compiler must win; auto-download is never consulted."""

called = []

monkeypatch.setattr(

"felderize.cli.ensure_compiler",

lambda *a, **k: called.append(True),

)

runner = CliRunner()

with patch("felderize.cli.translate_spark_to_feldera", return_value=_SUCCESS):

self._invoke(

runner,

_schema(tmp_path),

_query(tmp_path),

"--validate",

"--compiler",

"/opt/my-compiler.jar",

)

assert called == []

def test_env_compiler_skips_auto_resolution(self, tmp_path, monkeypatch):

"""A configured FELDERA_COMPILER must be used as-is; no auto-download."""

called = []

monkeypatch.setattr(

"felderize.cli.ensure_compiler",

lambda *a, **k: called.append(True),

)

monkeypatch.setenv("FELDERA_COMPILER", "/opt/env-compiler.jar")

runner = CliRunner()

with patch(

"felderize.cli.translate_spark_to_feldera", return_value=_SUCCESS

) as mock_fn:

self._invoke(runner, _schema(tmp_path), _query(tmp_path), "--validate")

assert called == []

config = mock_fn.call_args.args[2]

assert config.feldera_compiler == "/opt/env-compiler.jar"

def test_success_json_output(self, tmp_path):

runner = CliRunner()

with patch("felderize.cli.translate_spark_to_feldera", return_value=_SUCCESS):

result = self._invoke(

runner,

_schema(tmp_path),

_query(tmp_path),

"--validate",

"--json-output",

)

assert result.exit_code == 0

data = json.loads(result.output)

assert data["status"] == "success"

assert "feldera_query" in data

assert "feldera_schema" in data

def test_output_writes_sql_file(self, tmp_path):

runner = CliRunner()

out = tmp_path / "out.sql"

with patch("felderize.cli.translate_spark_to_feldera", return_value=_SUCCESS):

result = self._invoke(

runner,

_schema(tmp_path),

_query(tmp_path),

"--validate",

"-o",

str(out),

)

assert result.exit_code == 0

text = out.read_text()

assert "CREATE TABLE t" in text and "CREATE VIEW v" in text

assert f"Wrote {out}" in result.output # status note (stderr)

# the human section-by-section print is skipped when writing a file

assert "Transformations" not in result.output

def test_output_file_includes_caveats_as_comments(self, tmp_path):

runner = CliRunner()

out = tmp_path / "out.sql"

with patch(

"felderize.cli.translate_spark_to_feldera", return_value=_UNSUPPORTED

):

result = self._invoke(

runner, _schema(tmp_path), _query(tmp_path), "-o", str(out)

)

assert result.exit_code == 0

text = out.read_text()

assert "-- Translated by felderize — status: unsupported" in text

assert "-- Unsupported" in text

assert "PIVOT is not supported" in text

# every header note line is a SQL comment (no leaked uncommented text)

for line in text.splitlines():

if line.strip() and not line.startswith("--"):

assert line.lstrip().upper().startswith("CREATE"), line

def test_output_with_json_writes_file_and_prints_json(self, tmp_path):

runner = CliRunner()

out = tmp_path / "out.sql"

with patch("felderize.cli.translate_spark_to_feldera", return_value=_SUCCESS):

result = self._invoke(

runner,

_schema(tmp_path),

_query(tmp_path),

"-o",

str(out),

"--json-output",

)

assert result.exit_code == 0

assert out.is_file() # the .sql file was written

assert '"status": "success"' in result.output # and JSON still printed

def test_no_docs_flag_forwarded(self, tmp_path):

runner = CliRunner()

with patch(

"felderize.cli.translate_spark_to_feldera", return_value=_SUCCESS

) as mock_fn:

self._invoke(

runner, _schema(tmp_path), _query(tmp_path), "--validate", "--no-docs"

)

assert mock_fn.call_args.kwargs["include_docs"] is False

def test_without_validate_prints_warning(self, tmp_path):

runner = CliRunner()

with patch("felderize.cli.translate_spark_to_feldera", return_value=_SUCCESS):

result = self._invoke(runner, _schema(tmp_path), _query(tmp_path))

assert "without validation" in result.output

def test_error_result_exits_zero_but_reports_failure(self, tmp_path):

runner = CliRunner()

with patch("felderize.cli.translate_spark_to_feldera", return_value=_ERROR):

result = self._invoke(

runner, _schema(tmp_path), _query(tmp_path), "--validate"

)

assert result.exit_code == 0

assert "Translation Failed" in result.output

def test_unsupported_result_shows_unsupported_section(self, tmp_path):

runner = CliRunner()

with patch(

"felderize.cli.translate_spark_to_feldera", return_value=_UNSUPPORTED

):

result = self._invoke(

runner, _schema(tmp_path), _query(tmp_path), "--validate"

)

assert result.exit_code == 0

assert "Unsupported" in result.output

assert "PIVOT" in result.output

# ---------------------------------------------------------------------------

# spark translate-file

# ---------------------------------------------------------------------------

class TestReadText:

def test_reads_utf8_file(self, tmp_path):

p = tmp_path / "x.sql"

p.write_text("CREATE TABLE t (x INT);")

assert _read_text(p) == "CREATE TABLE t (x INT);"

def test_binary_file_exits_cleanly(self, tmp_path):

p = tmp_path / "x.bin"

p.write_bytes(b"\x88\x00\xff not text")

with pytest.raises(SystemExit) as exc:

_read_text(p)

assert exc.value.code == 1

def test_missing_file_exits_cleanly(self, tmp_path):

with pytest.raises(SystemExit) as exc:

_read_text(tmp_path / "does_not_exist.sql")

assert exc.value.code == 1

class TestTranslateFileCommand:

_BASE = ["spark", "translate-file"]

def test_binary_file_errors_cleanly(self, tmp_path):

combined = tmp_path / "combined.sql"

combined.write_bytes(b"\x88\xff binary")

result = CliRunner().invoke(cli, self._BASE + [str(combined)])

assert result.exit_code == 1

assert "cannot read" in result.output

assert "Traceback" not in result.output

def test_empty_file_exits_with_error(self, tmp_path):

combined = tmp_path / "combined.sql"

combined.write_text("")

result = CliRunner().invoke(cli, self._BASE + [str(combined)])

assert result.exit_code == 1

def test_success_with_combined_sql(self, tmp_path):

combined = tmp_path / "combined.sql"

combined.write_text(

"CREATE TABLE t (x INT NOT NULL);\nCREATE VIEW v AS SELECT x FROM t;\n"

)

runner = CliRunner()

with patch("felderize.cli.translate_spark_to_feldera", return_value=_SUCCESS):

result = runner.invoke(cli, self._BASE + [str(combined), "--validate"])

assert result.exit_code == 0

assert "-- Query --" in result.output

def test_json_output_flag(self, tmp_path):

combined = tmp_path / "combined.sql"

combined.write_text(

"CREATE TABLE t (x INT NOT NULL);\nCREATE VIEW v AS SELECT x FROM t;\n"

)

runner = CliRunner()

with patch("felderize.cli.translate_spark_to_feldera", return_value=_SUCCESS):

result = runner.invoke(

cli, self._BASE + [str(combined), "--validate", "--json-output"]

)

assert result.exit_code == 0

data = json.loads(result.output)

assert data["status"] == "success"

# ---------------------------------------------------------------------------

# spark translate-batch

# ---------------------------------------------------------------------------

class TestTranslateBatchCommand:

_BASE = ["spark", "translate-batch"]

def _invoke(self, runner: CliRunner, schema: Path, *extra: str):

return runner.invoke(cli, self._BASE + [str(schema)] + list(extra))

def test_invalid_schema_exits_before_any_translation(self, tmp_path):

schema = _schema(tmp_path, "")

query = _query(tmp_path)

runner = CliRunner()

with patch("felderize.cli.translate_spark_to_feldera") as mock_fn:

result = self._invoke(runner, schema, str(query))

assert result.exit_code == 1

mock_fn.assert_not_called()

def test_directory_with_no_sql_files_exits(self, tmp_path):

schema = _schema(tmp_path)

query_dir = tmp_path / "queries"

query_dir.mkdir()

result = CliRunner().invoke(cli, self._BASE + [str(schema), str(query_dir)])

assert result.exit_code == 1

assert "no .sql" in result.output.lower()

def test_translates_multiple_explicit_files(self, tmp_path):

schema = _schema(tmp_path)

q1 = tmp_path / "q1.sql"

q1.write_text("CREATE VIEW v1 AS SELECT x FROM t;")

q2 = tmp_path / "q2.sql"

q2.write_text("CREATE VIEW v2 AS SELECT x FROM t;")

runner = CliRunner()

with patch(

"felderize.cli.translate_spark_to_feldera", return_value=_SUCCESS

) as mock_fn:

result = self._invoke(runner, schema, str(q1), str(q2))

assert result.exit_code == 0

assert mock_fn.call_count == 2

def test_binary_query_is_skipped_not_crashed(self, tmp_path):

"""A binary query file is reported and skipped; good queries still run."""

schema = _schema(tmp_path)

good = tmp_path / "good.sql"

good.write_text("CREATE VIEW v AS SELECT x FROM t;")

bad = tmp_path / "bad.sql"

bad.write_bytes(b"\x88\xff not text")

runner = CliRunner()

with patch(

"felderize.cli.translate_spark_to_feldera", return_value=_SUCCESS

) as mock_fn:

result = self._invoke(runner, schema, str(good), str(bad))

assert "cannot read" in result.output

assert "Traceback" not in result.output

assert mock_fn.call_count == 1 # only the good query was translated

def test_expands_directory_of_sql_files(self, tmp_path):

schema = _schema(tmp_path)

query_dir = tmp_path / "queries"

query_dir.mkdir()

(query_dir / "a.sql").write_text("CREATE VIEW a AS SELECT x FROM t;")

(query_dir / "b.sql").write_text("CREATE VIEW b AS SELECT x FROM t;")

runner = CliRunner()

with patch(

"felderize.cli.translate_spark_to_feldera", return_value=_SUCCESS

) as mock_fn:

result = self._invoke(runner, schema, str(query_dir))

assert result.exit_code == 0

assert mock_fn.call_count == 2

def test_output_dir_creates_one_file_per_query(self, tmp_path):

schema = _schema(tmp_path)

q1 = tmp_path / "q1.sql"

q1.write_text("CREATE VIEW v AS SELECT x FROM t;")

out = tmp_path / "out"

runner = CliRunner()

with patch("felderize.cli.translate_spark_to_feldera", return_value=_SUCCESS):

result = self._invoke(runner, schema, str(q1), "--output-dir", str(out))

assert result.exit_code == 0

assert (out / "q1_feldera.sql").is_file()

content = (out / "q1_feldera.sql").read_text()

assert "Schema" in content

assert "Query" in content

def test_output_dir_created_if_missing(self, tmp_path):

schema = _schema(tmp_path)

q1 = tmp_path / "q1.sql"

q1.write_text("CREATE VIEW v AS SELECT x FROM t;")

out = tmp_path / "nested" / "out"

runner = CliRunner()

with patch("felderize.cli.translate_spark_to_feldera", return_value=_SUCCESS):

result = self._invoke(runner, schema, str(q1), "--output-dir", str(out))

assert result.exit_code == 0

assert out.is_dir()

def test_bad_query_skipped_loop_continues(self, tmp_path):

schema = _schema(tmp_path)

bad = tmp_path / "bad.sql"

bad.write_text("") # empty → validate_query fails

good = tmp_path / "good.sql"

good.write_text("CREATE VIEW v AS SELECT x FROM t;")

runner = CliRunner()

with patch(

"felderize.cli.translate_spark_to_feldera", return_value=_SUCCESS

) as mock_fn:

result = self._invoke(runner, schema, str(bad), str(good))

assert result.exit_code == 0

mock_fn.assert_called_once() # only the good query reaches translation

def test_summary_line_shows_counts(self, tmp_path):

schema = _schema(tmp_path)

q1 = tmp_path / "q1.sql"

q1.write_text("CREATE VIEW v AS SELECT x FROM t;")

runner = CliRunner()

with patch("felderize.cli.translate_spark_to_feldera", return_value=_SUCCESS):

result = self._invoke(runner, schema, str(q1))

assert "1/1" in result.output

# ---------------------------------------------------------------------------

# spark example

# ---------------------------------------------------------------------------

class TestExampleCommand:

_BASE = ["spark", "example"]

def test_no_name_lists_examples(self):

result = CliRunner().invoke(cli, self._BASE)

assert result.exit_code == 0

assert "simple" in result.output

def test_no_name_shows_schema_query_and_combined_tags(self):

result = CliRunner().invoke(cli, self._BASE)

assert "[schema+query]" in result.output or "[combined]" in result.output

def test_unknown_name_exits_nonzero(self):

result = CliRunner().invoke(cli, self._BASE + ["__no_such_example__"])

assert result.exit_code == 1

def test_schema_query_pair_example(self):

runner = CliRunner()

with patch("felderize.cli.translate_spark_to_feldera", return_value=_SUCCESS):

result = runner.invoke(cli, self._BASE + ["simple", "--validate"])

assert result.exit_code == 0

def test_combined_file_example(self):

runner = CliRunner()

with patch("felderize.cli.translate_spark_to_feldera", return_value=_SUCCESS):

result = runner.invoke(cli, self._BASE + ["arithmetic", "--validate"])

assert result.exit_code == 0

def test_json_output_for_example(self):

runner = CliRunner()

with patch("felderize.cli.translate_spark_to_feldera", return_value=_SUCCESS):

result = runner.invoke(

cli, self._BASE + ["simple", "--validate", "--json-output"]

)

assert result.exit_code == 0

# The example command echoes the Spark SQL to stderr before the JSON, so

# result.output contains the preamble followed by the JSON object.

json_start = result.output.index("{")

data = json.loads(result.output[json_start:])

assert data["status"] == "success"