test_cli.py

from .fixtures import (
    make_app_client,
    TestClient as _TestClient,
    EXPECTED_PLUGINS,
)
from datasette.app import SETTINGS
from datasette.plugins import DEFAULT_PLUGINS, pm
from datasette.cli import cli, serve
from datasette.version import __version__
from datasette.utils import tilde_encode
from datasette.utils.sqlite import sqlite3
from click.testing import CliRunner
import io
import json
import pathlib
import pytest
import sys
import textwrap
from unittest import mock


def test_inspect_cli(app_client):
    runner = CliRunner()
    result = runner.invoke(cli, ["inspect", "fixtures.db"])
    data = json.loads(result.output)
    assert ["fixtures"] == list(data.keys())
    database = data["fixtures"]
    assert "fixtures.db" == database["file"]
    assert isinstance(database["hash"], str)
    assert 64 == len(database["hash"])
    for table_name, expected_count in {
        "Table With Space In Name": 0,
        "facetable": 15,
    }.items():
        assert expected_count == database["tables"][table_name]["count"]


def test_inspect_cli_writes_to_file(app_client):
    runner = CliRunner()
    result = runner.invoke(
        cli, ["inspect", "fixtures.db", "--inspect-file", "foo.json"]
    )
    assert 0 == result.exit_code, result.output
    with open("foo.json") as fp:
        data = json.load(fp)
    assert ["fixtures"] == list(data.keys())


def test_inspect_cli_returns_exact_counts_for_large_tables(tmp_path):
    db_path = tmp_path / "large.db"
    conn = sqlite3.connect(db_path)
    conn.execute("create table t(id integer primary key)")
    conn.executemany("insert into t(id) values (?)", [(i,) for i in range(10002)])
    conn.commit()
    conn.close()

    runner = CliRunner()
    result = runner.invoke(cli, ["inspect", str(db_path)])
    assert result.exit_code == 0, result.output
    data = json.loads(result.output)
    assert data["large"]["tables"]["t"]["count"] == 10002


def test_serve_with_inspect_file_prepopulates_table_counts_cache():
    inspect_data = {"fixtures": {"tables": {"hithere": {"count": 44}}}}
    with make_app_client(inspect_data=inspect_data, is_immutable=True) as client:
        assert inspect_data == client.ds.inspect_data
        db = client.ds.databases["fixtures"]
        assert {"hithere": 44} == db.cached_table_counts


@pytest.mark.parametrize(
    "spatialite_paths,should_suggest_load_extension",
    (
        ([], False),
        (["/tmp"], True),
    ),
)
def test_spatialite_error_if_attempt_to_open_spatialite(
    spatialite_paths, should_suggest_load_extension
):
    with mock.patch("datasette.utils.SPATIALITE_PATHS", spatialite_paths):
        runner = CliRunner()
        result = runner.invoke(
            cli, ["serve", str(pathlib.Path(__file__).parent / "spatialite.db")]
        )
        assert result.exit_code != 0
        assert "It looks like you're trying to load a SpatiaLite" in result.output
        suggestion = "--load-extension=spatialite"
        if should_suggest_load_extension:
            assert suggestion in result.output
        else:
            assert suggestion not in result.output


@mock.patch("datasette.utils.SPATIALITE_PATHS", ["/does/not/exist"])
def test_spatialite_error_if_cannot_find_load_extension_spatialite():
    runner = CliRunner()
    result = runner.invoke(
        cli,
        [
            "serve",
            str(pathlib.Path(__file__).parent / "spatialite.db"),
            "--load-extension",
            "spatialite",
        ],
    )
    assert result.exit_code != 0
    assert "Could not find SpatiaLite extension" in result.output


def test_plugins_cli(app_client):
    runner = CliRunner()
    result1 = runner.invoke(cli, ["plugins"])
    actual_plugins = sorted(
        [p for p in json.loads(result1.output) if p["name"] != "TrackEventPlugin"],
        key=lambda p: p["name"],
    )
    assert actual_plugins == EXPECTED_PLUGINS
    # Try with --all
    result2 = runner.invoke(cli, ["plugins", "--all"])
    names = [p["name"] for p in json.loads(result2.output)]
    # Should have all the EXPECTED_PLUGINS
    assert set(names).issuperset({p["name"] for p in EXPECTED_PLUGINS})
    # And the following too:
    assert set(names).issuperset(DEFAULT_PLUGINS)
    # --requirements should be empty because there are no installed non-plugins-dir plugins
    result3 = runner.invoke(cli, ["plugins", "--requirements"])
    assert result3.output == ""


def test_metadata_yaml():
    yaml_file = io.StringIO(textwrap.dedent("""
    title: Hello from YAML
    """))
    # Annoyingly we have to provide all default arguments here:
    ds = serve.callback(
        [],
        metadata=yaml_file,
        immutable=[],
        host="127.0.0.1",
        port=8001,
        uds=None,
        reload=False,
        cors=False,
        sqlite_extensions=[],
        inspect_file=None,
        template_dir=None,
        plugins_dir=None,
        static=[],
        memory=False,
        config=[],
        settings=[],
        secret=None,
        root=False,
        default_deny=False,
        token=None,
        actor=None,
        version_note=None,
        get=None,
        headers=False,
        help_settings=False,
        pdb=False,
        crossdb=False,
        nolock=False,
        open_browser=False,
        create=False,
        ssl_keyfile=None,
        ssl_certfile=None,
        return_instance=True,
        internal=None,
    )
    client = _TestClient(ds)
    response = client.get("/.json")
    assert {"title": "Hello from YAML"} == response.json["metadata"]


@mock.patch("datasette.cli.run_module")
def test_install(run_module):
    runner = CliRunner()
    runner.invoke(cli, ["install", "datasette-mock-plugin", "datasette-mock-plugin2"])
    run_module.assert_called_once_with("pip", run_name="__main__")
    assert sys.argv == [
        "pip",
        "install",
        "datasette-mock-plugin",
        "datasette-mock-plugin2",
    ]


@pytest.mark.parametrize("flag", ["-U", "--upgrade"])
@mock.patch("datasette.cli.run_module")
def test_install_upgrade(run_module, flag):
    runner = CliRunner()
    runner.invoke(cli, ["install", flag, "datasette"])
    run_module.assert_called_once_with("pip", run_name="__main__")
    assert sys.argv == ["pip", "install", "--upgrade", "datasette"]


@mock.patch("datasette.cli.run_module")
def test_install_requirements(run_module, tmpdir):
    path = tmpdir.join("requirements.txt")
    path.write("datasette-mock-plugin\ndatasette-plugin-2")
    runner = CliRunner()
    runner.invoke(cli, ["install", "-r", str(path)])
    run_module.assert_called_once_with("pip", run_name="__main__")
    assert sys.argv == ["pip", "install", "-r", str(path)]


def test_install_error_if_no_packages():
    runner = CliRunner()
    result = runner.invoke(cli, ["install"])
    assert result.exit_code == 2
    assert "Error: Please specify at least one package to install" in result.output


@mock.patch("datasette.cli.run_module")
def test_uninstall(run_module):
    runner = CliRunner()
    runner.invoke(cli, ["uninstall", "datasette-mock-plugin", "-y"])
    run_module.assert_called_once_with("pip", run_name="__main__")
    assert sys.argv == ["pip", "uninstall", "datasette-mock-plugin", "-y"]


def test_version():
    runner = CliRunner()
    result = runner.invoke(cli, ["--version"])
    assert result.output == f"cli, version {__version__}\n"


@pytest.mark.parametrize("invalid_port", ["-1", "0.5", "dog", "65536"])
def test_serve_invalid_ports(invalid_port):
    runner = CliRunner()
    result = runner.invoke(cli, ["--port", invalid_port])
    assert result.exit_code == 2
    assert "Invalid value for '-p'" in result.stderr


@pytest.mark.parametrize(
    "args",
    (
        ["--setting", "default_page_size", "5"],
        ["--setting", "settings.default_page_size", "5"],
        ["-s", "settings.default_page_size", "5"],
    ),
)
def test_setting(args):
    runner = CliRunner()
    result = runner.invoke(cli, ["--get", "/-/settings.json"] + args)
    assert result.exit_code == 0, result.output
    settings = json.loads(result.output)
    assert settings["default_page_size"] == 5


def test_setting_compatible_with_config(tmp_path):
    # https://github.com/simonw/datasette/issues/2389
    runner = CliRunner()
    config_path = tmp_path / "config.json"
    config_path.write_text(
        '{"settings": {"default_page_size": 5, "sql_time_limit_ms": 50}}', "utf-8"
    )
    result = runner.invoke(
        cli,
        [
            "--get",
            "/-/settings.json",
            "--config",
            str(config_path),
            "--setting",
            "default_page_size",
            "10",
        ],
    )
    assert result.exit_code == 0, result.output
    settings = json.loads(result.output)
    assert settings["default_page_size"] == 10
    assert settings["sql_time_limit_ms"] == 50


def test_plugin_s_overwrite():
    runner = CliRunner()
    plugins_dir = str(pathlib.Path(__file__).parent / "plugins")

    result = runner.invoke(
        cli,
        [
            "--plugins-dir",
            plugins_dir,
            "--get",
            "/_memory/-/query.json?sql=select+prepare_connection_args()",
        ],
    )
    assert result.exit_code == 0, result.output
    assert (
        json.loads(result.output).get("rows")[0].get("prepare_connection_args()")
        == 'database=_memory, datasette.plugin_config("name-of-plugin")=None'
    )

    result = runner.invoke(
        cli,
        [
            "--plugins-dir",
            plugins_dir,
            "--get",
            "/_memory/-/query.json?sql=select+prepare_connection_args()",
            "-s",
            "plugins.name-of-plugin",
            "OVERRIDE",
        ],
    )
    assert result.exit_code == 0, result.output
    assert (
        json.loads(result.output).get("rows")[0].get("prepare_connection_args()")
        == 'database=_memory, datasette.plugin_config("name-of-plugin")=OVERRIDE'
    )


def test_startup_error_from_plugin_is_click_exception(tmp_path):
    plugins_dir = tmp_path / "plugins"
    plugins_dir.mkdir()
    (plugins_dir / "startup_error.py").write_text(
        "from datasette import hookimpl\n"
        "from datasette.utils import StartupError\n"
        "\n"
        "@hookimpl\n"
        "def startup(datasette):\n"
        '    raise StartupError("boom")\n',
        "utf-8",
    )
    runner = CliRunner()
    result = runner.invoke(
        cli,
        [
            "--plugins-dir",
            str(plugins_dir),
            "--get",
            "/",
        ],
    )
    try:
        assert result.exit_code == 1
        assert "Error: boom" in result.output
    finally:
        # Cleanup: Unregister the plugin to avoid test isolation issues
        to_unregister = [
            p for p in pm.get_plugins() if p.__name__ == "startup_error.py"
        ]
        if to_unregister:
            pm.unregister(to_unregister[0])


def test_setting_type_validation():
    runner = CliRunner()
    result = runner.invoke(cli, ["--setting", "default_page_size", "dog"])
    assert result.exit_code == 2
    assert '"settings.default_page_size" should be an integer' in result.output


def test_setting_boolean_validation_invalid():
    """Test that invalid boolean values are rejected"""
    runner = CliRunner()
    result = runner.invoke(
        cli, ["--setting", "default_allow_sql", "invalid", "--get", "/-/settings.json"]
    )
    assert result.exit_code == 2
    assert (
        '"settings.default_allow_sql" should be on/off/true/false/1/0' in result.output
    )


@pytest.mark.parametrize("value", ("off", "false", "0"))
def test_setting_boolean_validation_false_values(value):
    """Test that 'off', 'false', '0' work for boolean settings"""
    runner = CliRunner()
    result = runner.invoke(
        cli,
        [
            "--setting",
            "default_allow_sql",
            value,
            "--get",
            "/_memory/-/query.json?sql=select+1",
        ],
    )
    # Should be forbidden (setting is false)
    assert result.exit_code == 1, result.output
    assert "Forbidden" in result.output


@pytest.mark.parametrize("value", ("on", "true", "1"))
def test_setting_boolean_validation_true_values(value):
    """Test that 'on', 'true', '1' work for boolean settings"""
    runner = CliRunner()
    result = runner.invoke(
        cli,
        [
            "--setting",
            "default_allow_sql",
            value,
            "--get",
            "/_memory/-/query.json?sql=select+1&_shape=objects",
        ],
    )
    # Should succeed (setting is true)
    assert result.exit_code == 0, result.output
    assert json.loads(result.output)["rows"][0] == {"1": 1}


@pytest.mark.parametrize("default_allow_sql", (True, False))
def test_setting_default_allow_sql(default_allow_sql):
    runner = CliRunner()
    result = runner.invoke(
        cli,
        [
            "--setting",
            "default_allow_sql",
            "on" if default_allow_sql else "off",
            "--get",
            "/_memory/-/query.json?sql=select+21&_shape=objects",
        ],
    )
    if default_allow_sql:
        assert result.exit_code == 0, result.output
        assert json.loads(result.output)["rows"][0] == {"21": 21}
    else:
        assert result.exit_code == 1, result.output
        # This isn't JSON at the moment, maybe it should be though
        assert "Forbidden" in result.output


def test_sql_errors_logged_to_stderr():
    runner = CliRunner()
    result = runner.invoke(cli, ["--get", "/_memory/-/query.json?sql=select+blah"])
    assert result.exit_code == 1
    assert "sql = 'select blah', params = {}: no such column: blah\n" in result.stderr


def test_serve_create(tmpdir):
    runner = CliRunner()
    db_path = tmpdir / "does_not_exist_yet.db"
    assert not db_path.exists()
    result = runner.invoke(
        cli, [str(db_path), "--create", "--get", "/-/databases.json"]
    )
    assert result.exit_code == 0, result.output
    databases = json.loads(result.output)
    assert {
        "name": "does_not_exist_yet",
        "is_mutable": True,
        "is_memory": False,
        "hash": None,
    }.items() <= databases[0].items()
    assert db_path.exists()


@pytest.mark.parametrize("argument", ("-c", "--config"))
@pytest.mark.parametrize("format_", ("json", "yaml"))
def test_serve_config(tmpdir, argument, format_):
    config_path = tmpdir / "datasette.{}".format(format_)
    config_path.write_text(
        (
            "settings:\n  default_page_size: 5\n"
            if format_ == "yaml"
            else '{"settings": {"default_page_size": 5}}'
        ),
        "utf-8",
    )
    runner = CliRunner()
    result = runner.invoke(
        cli,
        [
            argument,
            str(config_path),
            "--get",
            "/-/settings.json",
        ],
    )
    assert result.exit_code == 0, result.output
    assert json.loads(result.output)["default_page_size"] == 5


def test_serve_duplicate_database_names(tmpdir):
    "'datasette db.db nested/db.db' should attach two databases, /db and /db_2"
    runner = CliRunner()
    db_1_path = str(tmpdir / "db.db")
    nested = tmpdir / "nested"
    nested.mkdir()
    db_2_path = str(tmpdir / "nested" / "db.db")
    for path in (db_1_path, db_2_path):
        conn = sqlite3.connect(path)
        conn.execute("vacuum")
        conn.close()
    result = runner.invoke(cli, [db_1_path, db_2_path, "--get", "/-/databases.json"])
    assert result.exit_code == 0, result.output
    databases = json.loads(result.output)
    assert {db["name"] for db in databases} == {"db", "db_2"}


@pytest.mark.parametrize(
    "filename", ["test-database (1).sqlite", "database (1).sqlite"]
)
def test_weird_database_names(tmpdir, filename):
    # https://github.com/simonw/datasette/issues/1181
    runner = CliRunner()
    db_path = str(tmpdir / filename)
    conn = sqlite3.connect(db_path)
    conn.execute("vacuum")
    conn.close()
    result1 = runner.invoke(cli, [db_path, "--get", "/"])
    assert result1.exit_code == 0, result1.output
    filename_no_stem = filename.rsplit(".", 1)[0]
    expected_link = '<a href="/{}">{}</a>'.format(
        tilde_encode(filename_no_stem), filename_no_stem
    )
    assert expected_link in result1.output
    # Now try hitting that database page
    result2 = runner.invoke(
        cli, [db_path, "--get", "/{}".format(tilde_encode(filename_no_stem))]
    )
    assert result2.exit_code == 0, result2.output


def test_help_settings():
    runner = CliRunner()
    result = runner.invoke(cli, ["--help-settings"])
    for setting in SETTINGS:
        assert setting.name in result.output


def test_internal_db(tmpdir):
    runner = CliRunner()
    internal_path = tmpdir / "internal.db"
    assert not internal_path.exists()
    result = runner.invoke(
        cli, ["--memory", "--internal", str(internal_path), "--get", "/"]
    )
    assert result.exit_code == 0
    assert internal_path.exists()


def test_duplicate_database_files_error(tmpdir):
    """Test that passing the same database file multiple times raises an error"""
    runner = CliRunner()
    db_path = str(tmpdir / "test.db")
    conn = sqlite3.connect(db_path)
    conn.execute("vacuum")
    conn.close()

    # Test with exact duplicate
    result = runner.invoke(cli, ["serve", db_path, db_path, "--get", "/"])
    assert result.exit_code == 1
    assert "Duplicate database file" in result.output
    assert "both refer to" in result.output

    # Test with different paths to same file (relative vs absolute)
    result2 = runner.invoke(
        cli, ["serve", db_path, str(pathlib.Path(db_path).resolve()), "--get", "/"]
    )
    assert result2.exit_code == 1
    assert "Duplicate database file" in result2.output

    # Test that a file in the config_dir can't also be passed explicitly
    config_dir = tmpdir / "config"
    config_dir.mkdir()
    config_db_path = str(config_dir / "data.db")
    conn = sqlite3.connect(config_db_path)
    conn.execute("vacuum")
    conn.close()

    result3 = runner.invoke(
        cli, ["serve", config_db_path, str(config_dir), "--get", "/"]
    )
    assert result3.exit_code == 1
    assert "Duplicate database file" in result3.output
    assert "both refer to" in result3.output

    # Test that mixing a file NOT in the directory with a directory works fine
    other_db_path = str(tmpdir / "other.db")
    conn = sqlite3.connect(other_db_path)
    conn.execute("vacuum")
    conn.close()

    result4 = runner.invoke(
        cli, ["serve", other_db_path, str(config_dir), "--get", "/-/databases.json"]
    )
    assert result4.exit_code == 0
    databases = json.loads(result4.output)
    assert {db["name"] for db in databases} == {"other", "data"}

    # Test that multiple directories raise an error
    config_dir2 = tmpdir / "config2"
    config_dir2.mkdir()

    result5 = runner.invoke(
        cli, ["serve", str(config_dir), str(config_dir2), "--get", "/"]
    )
    assert result5.exit_code == 1
    assert "Cannot pass multiple directories" in result5.output