test_utils.py

"""
Tests for various datasette helper functions.
"""

from datasette.app import Datasette
from datasette import utils
from datasette.utils.asgi import Request
from datasette.utils.sqlite import sqlite3
import json
import os
import pathlib
import pytest
import tempfile
from unittest.mock import patch


@pytest.mark.parametrize(
    "path,expected",
    [
        ("foo", ["foo"]),
        ("foo,bar", ["foo", "bar"]),
        ("123,433,112", ["123", "433", "112"]),
        ("123~2C433,112", ["123,433", "112"]),
        ("123~2F433~2F112", ["123/433/112"]),
    ],
)
def test_urlsafe_components(path, expected):
    assert expected == utils.urlsafe_components(path)


@pytest.mark.parametrize(
    "path,added_args,expected",
    [
        ("/foo", {"bar": 1}, "/foo?bar=1"),
        ("/foo?bar=1", {"baz": 2}, "/foo?bar=1&baz=2"),
        ("/foo?bar=1&bar=2", {"baz": 3}, "/foo?bar=1&bar=2&baz=3"),
        ("/foo?bar=1", {"bar": None}, "/foo"),
        # Test order is preserved
        (
            "/?_facet=prim_state&_facet=area_name",
            (("prim_state", "GA"),),
            "/?_facet=prim_state&_facet=area_name&prim_state=GA",
        ),
        (
            "/?_facet=state&_facet=city&state=MI",
            (("city", "Detroit"),),
            "/?_facet=state&_facet=city&state=MI&city=Detroit",
        ),
        (
            "/?_facet=state&_facet=city",
            (("_facet", "planet_int"),),
            "/?_facet=state&_facet=city&_facet=planet_int",
        ),
    ],
)
def test_path_with_added_args(path, added_args, expected):
    request = Request.fake(path)
    actual = utils.path_with_added_args(request, added_args)
    assert expected == actual


@pytest.mark.parametrize(
    "path,args,expected",
    [
        ("/foo?bar=1", {"bar"}, "/foo"),
        ("/foo?bar=1&baz=2", {"bar"}, "/foo?baz=2"),
        ("/foo?bar=1&bar=2&bar=3", {"bar": "2"}, "/foo?bar=1&bar=3"),
    ],
)
def test_path_with_removed_args(path, args, expected):
    request = Request.fake(path)
    actual = utils.path_with_removed_args(request, args)
    assert expected == actual
    # Run the test again but this time use the path= argument
    request = Request.fake("/")
    actual = utils.path_with_removed_args(request, args, path=path)
    assert expected == actual


@pytest.mark.parametrize(
    "path,args,expected",
    [
        ("/foo?bar=1", {"bar": 2}, "/foo?bar=2"),
        ("/foo?bar=1&baz=2", {"bar": None}, "/foo?baz=2"),
    ],
)
def test_path_with_replaced_args(path, args, expected):
    request = Request.fake(path)
    actual = utils.path_with_replaced_args(request, args)
    assert expected == actual


@pytest.mark.parametrize(
    "row,pks,expected_path",
    [
        ({"A": "foo", "B": "bar"}, ["A", "B"], "foo,bar"),
        ({"A": "f,o", "B": "bar"}, ["A", "B"], "f~2Co,bar"),
        ({"A": 123}, ["A"], "123"),
        (
            utils.CustomRow(
                ["searchable_id", "tag"],
                [
                    ("searchable_id", {"value": 1, "label": "1"}),
                    ("tag", {"value": "feline", "label": "feline"}),
                ],
            ),
            ["searchable_id", "tag"],
            "1,feline",
        ),
    ],
)
def test_path_from_row_pks(row, pks, expected_path):
    actual_path = utils.path_from_row_pks(row, pks, False)
    assert expected_path == actual_path


@pytest.mark.parametrize(
    "obj,expected",
    [
        (
            {
                "Description": "Soft drinks",
                "Picture": b"\x15\x1c\x02\xc7\xad\x05\xfe",
                "CategoryID": 1,
            },
            """
        {"CategoryID": 1, "Description": "Soft drinks", "Picture": {"$base64": true, "encoded": "FRwCx60F/g=="}}
    """.strip(),
        )
    ],
)
def test_custom_json_encoder(obj, expected):
    actual = json.dumps(obj, cls=utils.CustomJSONEncoder, sort_keys=True)
    assert expected == actual


@pytest.mark.parametrize(
    "bad_sql",
    [
        "update blah;",
        "-- sql comment to skip\nupdate blah;",
        "update blah set some_column='# Hello there\n\n* This is a list\n* of items\n--\n[And a link](https://github.com/simonw/datasette-render-markdown).'\nas demo_markdown",
        "PRAGMA case_sensitive_like = true",
        "SELECT * FROM pragma_not_on_allow_list('idx52')",
        "/* This comment is not valid. select 1",
        "/**/\nupdate foo set bar = 1\n/* test */ select 1",
    ],
)
def test_validate_sql_select_bad(bad_sql):
    with pytest.raises(utils.InvalidSql):
        utils.validate_sql_select(bad_sql)


@pytest.mark.parametrize(
    "good_sql",
    [
        "select count(*) from airports",
        "select foo from bar",
        "--sql comment to skip\nselect foo from bar",
        "select '# Hello there\n\n* This is a list\n* of items\n--\n[And a link](https://github.com/simonw/datasette-render-markdown).'\nas demo_markdown",
        "select 1 + 1",
        "explain select 1 + 1",
        "explain\nselect 1 + 1",
        "explain query plan select 1 + 1",
        "explain  query  plan\nselect 1 + 1",
        "SELECT\nblah FROM foo",
        "WITH RECURSIVE cnt(x) AS (SELECT 1 UNION ALL SELECT x+1 FROM cnt LIMIT 10) SELECT x FROM cnt;",
        "explain WITH RECURSIVE cnt(x) AS (SELECT 1 UNION ALL SELECT x+1 FROM cnt LIMIT 10) SELECT x FROM cnt;",
        "explain query plan WITH RECURSIVE cnt(x) AS (SELECT 1 UNION ALL SELECT x+1 FROM cnt LIMIT 10) SELECT x FROM cnt;",
        "SELECT * FROM pragma_index_info('idx52')",
        "select * from pragma_table_xinfo('table')",
        # Various types of comment
        "-- comment\nselect 1",
        "-- one line\n  -- two line\nselect 1",
        "  /* comment */\nselect 1",
        "  /* comment */select 1",
        "/* comment */\n -- another\n /* one more */ select 1",
        "/* This comment \n has multiple lines */\nselect 1",
    ],
)
def test_validate_sql_select_good(good_sql):
    utils.validate_sql_select(good_sql)


@pytest.mark.parametrize("open_quote,close_quote", [('"', '"'), ("[", "]")])
def test_detect_fts(open_quote, close_quote):
    sql = """
    CREATE TABLE "Dumb_Table" (
      "TreeID" INTEGER,
      "qSpecies" TEXT
    );
    CREATE TABLE "Street_Tree_List" (
      "TreeID" INTEGER,
      "qSpecies" TEXT,
      "qAddress" TEXT,
      "SiteOrder" INTEGER,
      "qSiteInfo" TEXT,
      "PlantType" TEXT,
      "qCaretaker" TEXT
    );
    CREATE VIEW Test_View AS SELECT * FROM Dumb_Table;
    CREATE VIRTUAL TABLE {open}Street_Tree_List_fts{close} USING FTS4 ("qAddress", "qCaretaker", "qSpecies", content={open}Street_Tree_List{close});
    CREATE VIRTUAL TABLE r USING rtree(a, b, c);
    """.format(
        open=open_quote, close=close_quote
    )
    conn = utils.sqlite3.connect(":memory:")
    conn.executescript(sql)
    assert None is utils.detect_fts(conn, "Dumb_Table")
    assert None is utils.detect_fts(conn, "Test_View")
    assert None is utils.detect_fts(conn, "r")
    assert "Street_Tree_List_fts" == utils.detect_fts(conn, "Street_Tree_List")


@pytest.mark.parametrize("table", ("regular", "has'single quote"))
def test_detect_fts_different_table_names(table):
    sql = """
    CREATE TABLE [{table}] (
      "TreeID" INTEGER,
      "qSpecies" TEXT
    );
    CREATE VIRTUAL TABLE [{table}_fts] USING FTS4 ("qSpecies", content="{table}");
    """.format(
        table=table
    )
    conn = utils.sqlite3.connect(":memory:")
    conn.executescript(sql)
    assert "{table}_fts".format(table=table) == utils.detect_fts(conn, table)


@pytest.mark.parametrize(
    "url,expected",
    [
        ("http://www.google.com/", True),
        ("https://example.com/", True),
        ("www.google.com", False),
        ("http://www.google.com/ is a search engine", False),
    ],
)
def test_is_url(url, expected):
    assert expected == utils.is_url(url)


@pytest.mark.parametrize(
    "s,expected",
    [
        ("simple", "simple"),
        ("MixedCase", "MixedCase"),
        ("-no-leading-hyphens", "no-leading-hyphens-65bea6"),
        ("_no-leading-underscores", "no-leading-underscores-b921bc"),
        ("no spaces", "no-spaces-7088d7"),
        ("-", "336d5e"),
        ("no $ characters", "no--characters-59e024"),
    ],
)
def test_to_css_class(s, expected):
    assert expected == utils.to_css_class(s)


def test_temporary_docker_directory_uses_hard_link():
    with tempfile.TemporaryDirectory() as td:
        os.chdir(td)
        with open("hello", "w") as fp:
            fp.write("world")
        # Default usage of this should use symlink
        with utils.temporary_docker_directory(
            files=["hello"],
            name="t",
            metadata=None,
            extra_options=None,
            branch=None,
            template_dir=None,
            plugins_dir=None,
            static=[],
            install=[],
            spatialite=False,
            version_note=None,
            secret="secret",
        ) as temp_docker:
            hello = os.path.join(temp_docker, "hello")
            with open(hello) as fp:
                assert "world" == fp.read()
            # It should be a hard link
            assert 2 == os.stat(hello).st_nlink


@patch("os.link")
def test_temporary_docker_directory_uses_copy_if_hard_link_fails(mock_link):
    # Copy instead if os.link raises OSError (normally due to different device)
    mock_link.side_effect = OSError
    with tempfile.TemporaryDirectory() as td:
        os.chdir(td)
        with open("hello", "w") as fp:
            fp.write("world")
        # Default usage of this should use symlink
        with utils.temporary_docker_directory(
            files=["hello"],
            name="t",
            metadata=None,
            extra_options=None,
            branch=None,
            template_dir=None,
            plugins_dir=None,
            static=[],
            install=[],
            spatialite=False,
            version_note=None,
            secret=None,
        ) as temp_docker:
            hello = os.path.join(temp_docker, "hello")
            with open(hello) as fp:
                assert "world" == fp.read()
            # It should be a copy, not a hard link
            assert 1 == os.stat(hello).st_nlink


def test_temporary_docker_directory_quotes_args():
    with tempfile.TemporaryDirectory() as td:
        os.chdir(td)
        with open("hello", "w") as fp:
            fp.write("world")
        with utils.temporary_docker_directory(
            files=["hello"],
            name="t",
            metadata=None,
            extra_options="--$HOME",
            branch=None,
            template_dir=None,
            plugins_dir=None,
            static=[],
            install=[],
            spatialite=False,
            version_note="$PWD",
            secret="secret",
        ) as temp_docker:
            df = os.path.join(temp_docker, "Dockerfile")
            with open(df) as fp:
                df_contents = fp.read()
            assert "'$PWD'" in df_contents
            assert "'--$HOME'" in df_contents
            assert "ENV DATASETTE_SECRET 'secret'" in df_contents


def test_compound_keys_after_sql():
    assert "((a > :p0))" == utils.compound_keys_after_sql(["a"])
    assert """
((a > :p0)
  or
(a = :p0 and b > :p1))
    """.strip() == utils.compound_keys_after_sql(
        ["a", "b"]
    )
    assert """
((a > :p0)
  or
(a = :p0 and b > :p1)
  or
(a = :p0 and b = :p1 and c > :p2))
    """.strip() == utils.compound_keys_after_sql(
        ["a", "b", "c"]
    )


def test_table_columns():
    conn = sqlite3.connect(":memory:")
    conn.executescript(
        """
    create table places (id integer primary key, name text, bob integer)
    """
    )
    assert ["id", "name", "bob"] == utils.table_columns(conn, "places")


@pytest.mark.parametrize(
    "path,format,extra_qs,expected",
    [
        ("/foo?sql=select+1", "csv", {}, "/foo.csv?sql=select+1"),
        ("/foo?sql=select+1", "json", {}, "/foo.json?sql=select+1"),
        ("/foo/bar", "json", {}, "/foo/bar.json"),
        ("/foo/bar", "csv", {}, "/foo/bar.csv"),
        ("/foo/bar", "csv", {"_dl": 1}, "/foo/bar.csv?_dl=1"),
        (
            "/sf-trees/Street_Tree_List?_search=cherry&_size=1000",
            "csv",
            {"_dl": 1},
            "/sf-trees/Street_Tree_List.csv?_search=cherry&_size=1000&_dl=1",
        ),
    ],
)
def test_path_with_format(path, format, extra_qs, expected):
    request = Request.fake(path)
    actual = utils.path_with_format(request=request, format=format, extra_qs=extra_qs)
    assert expected == actual


@pytest.mark.parametrize(
    "bytes,expected",
    [
        (120, "120 bytes"),
        (1024, "1.0 KB"),
        (1024 * 1024, "1.0 MB"),
        (1024 * 1024 * 1024, "1.0 GB"),
        (1024 * 1024 * 1024 * 1.3, "1.3 GB"),
        (1024 * 1024 * 1024 * 1024, "1.0 TB"),
    ],
)
def test_format_bytes(bytes, expected):
    assert expected == utils.format_bytes(bytes)


@pytest.mark.parametrize(
    "query,expected",
    [
        ("dog", '"dog"'),
        ("cat,", '"cat,"'),
        ("cat dog", '"cat" "dog"'),
        # If a phrase is already double quoted, leave it so
        ('"cat dog"', '"cat dog"'),
        ('"cat dog" fish', '"cat dog" "fish"'),
        # Sensibly handle unbalanced double quotes
        ('cat"', '"cat"'),
        ('"cat dog" "fish', '"cat dog" "fish"'),
    ],
)
def test_escape_fts(query, expected):
    assert expected == utils.escape_fts(query)


@pytest.mark.parametrize(
    "input,expected",
    [
        ("dog", "dog"),
        ('dateutil_parse("1/2/2020")', r"dateutil_parse(\0000221/2/2020\000022)"),
        ("this\r\nand\r\nthat", r"this\00000Aand\00000Athat"),
    ],
)
def test_escape_css_string(input, expected):
    assert expected == utils.escape_css_string(input)


def test_check_connection_spatialite_raises():
    path = str(pathlib.Path(__file__).parent / "spatialite.db")
    conn = sqlite3.connect(path)
    with pytest.raises(utils.SpatialiteConnectionProblem):
        utils.check_connection(conn)


def test_check_connection_passes():
    conn = sqlite3.connect(":memory:")
    utils.check_connection(conn)


def test_call_with_supported_arguments():
    def foo(a, b):
        return f"{a}+{b}"

    assert "1+2" == utils.call_with_supported_arguments(foo, a=1, b=2)
    assert "1+2" == utils.call_with_supported_arguments(foo, a=1, b=2, c=3)

    with pytest.raises(TypeError):
        utils.call_with_supported_arguments(foo, a=1)


@pytest.mark.parametrize(
    "data,should_raise",
    [
        ([["foo", "bar"], ["foo", "baz"]], False),
        ([("foo", "bar"), ("foo", "baz")], False),
        ((["foo", "bar"], ["foo", "baz"]), False),
        ([["foo", "bar"], ["foo", "baz", "bax"]], True),
        ({"foo": ["bar", "baz"]}, False),
        ({"foo": ("bar", "baz")}, False),
        ({"foo": "bar"}, True),
    ],
)
def test_multi_params(data, should_raise):
    if should_raise:
        with pytest.raises(AssertionError):
            utils.MultiParams(data)
        return
    p1 = utils.MultiParams(data)
    assert "bar" == p1["foo"]
    assert ["bar", "baz"] == list(p1.getlist("foo"))


@pytest.mark.parametrize(
    "actor,allow,expected",
    [
        # Default is to allow:
        (None, None, True),
        # {} means deny-all:
        (None, {}, False),
        ({"id": "root"}, {}, False),
        # true means allow-all
        ({"id": "root"}, True, True),
        (None, True, True),
        # false means deny-all
        ({"id": "root"}, False, False),
        (None, False, False),
        # Special case for "unauthenticated": true
        (None, {"unauthenticated": True}, True),
        (None, {"unauthenticated": False}, False),
        # Match on just one property:
        (None, {"id": "root"}, False),
        ({"id": "root"}, None, True),
        ({"id": "simon", "staff": True}, {"staff": True}, True),
        ({"id": "simon", "staff": False}, {"staff": True}, False),
        # Special "*" value for any key:
        ({"id": "root"}, {"id": "*"}, True),
        ({}, {"id": "*"}, False),
        ({"name": "root"}, {"id": "*"}, False),
        # Supports single strings or list of values:
        ({"id": "root"}, {"id": "bob"}, False),
        ({"id": "root"}, {"id": ["bob"]}, False),
        ({"id": "root"}, {"id": "root"}, True),
        ({"id": "root"}, {"id": ["root"]}, True),
        # Any matching role will work:
        ({"id": "garry", "roles": ["staff", "dev"]}, {"roles": ["staff"]}, True),
        ({"id": "garry", "roles": ["staff", "dev"]}, {"roles": ["dev"]}, True),
        ({"id": "garry", "roles": ["staff", "dev"]}, {"roles": ["otter"]}, False),
        ({"id": "garry", "roles": ["staff", "dev"]}, {"roles": ["dev", "otter"]}, True),
        ({"id": "garry", "roles": []}, {"roles": ["staff"]}, False),
        ({"id": "garry"}, {"roles": ["staff"]}, False),
        # Any single matching key works:
        ({"id": "root"}, {"bot_id": "my-bot", "id": ["root"]}, True),
    ],
)
def test_actor_matches_allow(actor, allow, expected):
    assert expected == utils.actor_matches_allow(actor, allow)


@pytest.mark.parametrize(
    "config,expected",
    [
        ({"foo": "bar"}, {"foo": "bar"}),
        ({"$env": "FOO"}, "x"),
        ({"k": {"$env": "FOO"}}, {"k": "x"}),
        ([{"k": {"$env": "FOO"}}, {"z": {"$env": "FOO"}}], [{"k": "x"}, {"z": "x"}]),
        ({"k": [{"in_a_list": {"$env": "FOO"}}]}, {"k": [{"in_a_list": "x"}]}),
    ],
)
def test_resolve_env_secrets(config, expected):
    assert expected == utils.resolve_env_secrets(config, {"FOO": "x"})


@pytest.mark.parametrize(
    "actor,expected",
    [
        ({"id": "blah"}, "blah"),
        ({"id": "blah", "login": "l"}, "l"),
        ({"id": "blah", "login": "l"}, "l"),
        ({"id": "blah", "login": "l", "username": "u"}, "u"),
        ({"login": "l", "name": "n"}, "n"),
        (
            {"id": "blah", "login": "l", "username": "u", "name": "n", "display": "d"},
            "d",
        ),
        ({"weird": "shape"}, "{'weird': 'shape'}"),
    ],
)
def test_display_actor(actor, expected):
    assert expected == utils.display_actor(actor)


@pytest.mark.asyncio
@pytest.mark.parametrize(
    "dbs,expected_path",
    [
        (["one_table"], "/one/one"),
        (["two_tables"], "/two"),
        (["one_table", "two_tables"], "/"),
    ],
)
async def test_initial_path_for_datasette(tmp_path_factory, dbs, expected_path):
    db_dir = tmp_path_factory.mktemp("dbs")
    one_table = str(db_dir / "one.db")
    sqlite3.connect(one_table).execute("create table one (id integer primary key)")
    two_tables = str(db_dir / "two.db")
    sqlite3.connect(two_tables).execute("create table two (id integer primary key)")
    sqlite3.connect(two_tables).execute("create table three (id integer primary key)")
    datasette = Datasette(
        [{"one_table": one_table, "two_tables": two_tables}[db] for db in dbs]
    )
    path = await utils.initial_path_for_datasette(datasette)
    assert path == expected_path


@pytest.mark.parametrize(
    "content,expected",
    (
        ("title: Hello", {"title": "Hello"}),
        ('{"title": "Hello"}', {"title": "Hello"}),
        ("{{ this }} is {{ bad }}", None),
    ),
)
def test_parse_metadata(content, expected):
    if expected is None:
        with pytest.raises(utils.BadMetadataError):
            utils.parse_metadata(content)
    else:
        assert utils.parse_metadata(content) == expected


@pytest.mark.asyncio
@pytest.mark.parametrize(
    "sql,expected",
    (
        ("select 1", []),
        ("select 1 + :one", ["one"]),
        ("select 1 + :one + :two", ["one", "two"]),
        ("select 'bob' || '0:00' || :cat", ["cat"]),
        ("select this is invalid :one, :two, :three", ["one", "two", "three"]),
    ),
)
async def test_derive_named_parameters(sql, expected):
    ds = Datasette([], memory=True)
    db = ds.get_database("_memory")
    params = await utils.derive_named_parameters(db, sql)
    assert params == expected


@pytest.mark.parametrize(
    "original,expected",
    (
        ("abc", "abc"),
        ("/foo/bar", "~2Ffoo~2Fbar"),
        ("/-/bar", "~2F-~2Fbar"),
        ("-/db-/table.csv", "-~2Fdb-~2Ftable~2Ecsv"),
        (r"%~-/", "~25~7E-~2F"),
        ("~25~7E~2D~2F", "~7E25~7E7E~7E2D~7E2F"),
        ("with space", "with+space"),
    ),
)
def test_tilde_encoding(original, expected):
    actual = utils.tilde_encode(original)
    assert actual == expected
    # And test round-trip
    assert original == utils.tilde_decode(actual)


@pytest.mark.parametrize(
    "url,length,expected",
    (
        ("https://example.com/", 5, "http…"),
        ("https://example.com/foo/bar", 15, "https://exampl…"),
        ("https://example.com/foo/bar/baz.jpg", 30, "https://example.com/foo/ba….jpg"),
        # Extensions longer than 4 characters are not treated specially:
        ("https://example.com/foo/bar/baz.jpeg2", 30, "https://example.com/foo/bar/b…"),
        (
            "https://example.com/foo/bar/baz.jpeg2",
            None,
            "https://example.com/foo/bar/baz.jpeg2",
        ),
    ),
)
def test_truncate_url(url, length, expected):
    actual = utils.truncate_url(url, length)
    assert actual == expected


@pytest.mark.parametrize(
    "pairs,expected",
    (
        # Simple nested objects
        ([("a", "b")], {"a": "b"}),
        ([("a.b", "c")], {"a": {"b": "c"}}),
        # JSON literals
        ([("a.b", "true")], {"a": {"b": True}}),
        ([("a.b", "false")], {"a": {"b": False}}),
        ([("a.b", "null")], {"a": {"b": None}}),
        ([("a.b", "1")], {"a": {"b": 1}}),
        ([("a.b", "1.1")], {"a": {"b": 1.1}}),
        # Nested JSON literals
        ([("a.b", '{"foo": "bar"}')], {"a": {"b": {"foo": "bar"}}}),
        ([("a.b", "[1, 2, 3]")], {"a": {"b": [1, 2, 3]}}),
        # JSON strings are preserved
        ([("a.b", '"true"')], {"a": {"b": "true"}}),
        ([("a.b", '"[1, 2, 3]"')], {"a": {"b": "[1, 2, 3]"}}),
        # Later keys over-ride the previous
        (
            [
                ("a", "b"),
                ("a.b", "c"),
            ],
            {"a": {"b": "c"}},
        ),
        (
            [
                ("settings.trace_debug", "true"),
                ("plugins.datasette-ripgrep.path", "/etc"),
                ("settings.trace_debug", "false"),
            ],
            {
                "settings": {
                    "trace_debug": False,
                },
                "plugins": {
                    "datasette-ripgrep": {
                        "path": "/etc",
                    }
                },
            },
        ),
    ),
)
def test_pairs_to_nested_config(pairs, expected):
    actual = utils.pairs_to_nested_config(pairs)
    assert actual == expected


@pytest.mark.asyncio
async def test_calculate_etag(tmp_path):
    path = tmp_path / "test.txt"
    path.write_text("hello")
    etag = '"5d41402abc4b2a76b9719d911017c592"'
    assert etag == await utils.calculate_etag(path)
    assert utils._etag_cache[path] == etag
    utils._etag_cache[path] = "hash"
    assert "hash" == await utils.calculate_etag(path)
    utils._etag_cache.clear()