import io
import json
import pathlib
import sys
import textwrap
from unittest import mock

import pytest
from click.testing import CliRunner

from datasette.app import SETTINGS
from datasette.cli import cli, serve
from datasette.plugins import DEFAULT_PLUGINS, pm
from datasette.utils import tilde_encode
from datasette.utils.sqlite import sqlite3
from datasette.version import __version__

from .fixtures import (
    EXPECTED_PLUGINS,
    make_app_client,
)
from .fixtures import (
    TestClient as _TestClient,
)


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_counts_all_rows(tmp_path):
    db_path = tmp_path / "big.db"
    conn = sqlite3.connect(db_path)
    with conn:
        conn.execute("create table t (id integer primary key)")
        conn.executemany("insert into t (id) values (?)", ((i,) for i in range(10002)))
    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["big"]["tables"]["t"]["count"] == 10002


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


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
    error = json.loads(result.output)
    assert error["ok"] is False
    assert error["status"] == 403


@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
        error = json.loads(result.output)
        assert error["ok"] is False
        assert error["status"] == 403


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)["databases"]
    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 / f"datasette.{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)["databases"]
    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 = (
        f'<a href="/{tilde_encode(filename_no_stem)}">{filename_no_stem}</a>'
    )
    assert expected_link in result1.output
    # Now try hitting that database page
    result2 = runner.invoke(
        cli, [db_path, "--get", f"/{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)["databases"]
    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