datasette/tests/test_table_html.py

from datasette.app import Datasette
from datasette.database import Database
from bs4 import BeautifulSoup as Soup
from .fixtures import make_app_client
import pathlib
import pytest
import urllib.parse
from .utils import inner_html


def table_data_from_soup(soup):
    import json
    import re

    table_script = [
        s for s in soup.find_all("script") if "_datasetteTableData" in (s.string or "")
    ][0]
    match = re.search(
        r"window\._datasetteTableData\s*=\s*({.*?});",
        table_script.string,
        re.DOTALL,
    )
    return json.loads(match.group(1))


@pytest.mark.asyncio
@pytest.mark.parametrize(
    "path,expected_definition_sql",
    [
        (
            "/fixtures/facet_cities",
            """
CREATE TABLE facet_cities (
    id integer primary key,
    name text
);
        """.strip(),
        ),
        (
            "/fixtures/compound_three_primary_keys",
            """
CREATE TABLE compound_three_primary_keys (
  pk1 varchar(30),
  pk2 varchar(30),
  pk3 varchar(30),
  content text,
  PRIMARY KEY (pk1, pk2, pk3)
);
CREATE INDEX idx_compound_three_primary_keys_content ON compound_three_primary_keys(content);
            """.strip(),
        ),
    ],
)
async def test_table_definition_sql(path, expected_definition_sql, ds_client):
    response = await ds_client.get(path)
    pre = Soup(response.text, "html.parser").select_one("pre.wrapped-sql")
    assert expected_definition_sql == pre.string


def test_table_cell_truncation():
    with make_app_client(settings={"truncate_cells_html": 5}) as client:
        response = client.get("/fixtures/facetable")
        assert response.status == 200
        table = Soup(response.body, "html.parser").find("table")
        assert table["class"] == ["rows-and-columns"]
        assert [
            "Missi…",
            "Dogpa…",
            "SOMA",
            "Tende…",
            "Berna…",
            "Hayes…",
            "Holly…",
            "Downt…",
            "Los F…",
            "Korea…",
            "Downt…",
            "Greek…",
            "Corkt…",
            "Mexic…",
            "Arcad…",
        ] == [
            td.string
            for td in table.find_all("td", {"class": "col-neighborhood-b352a7"})
        ]
        # URLs should be truncated too
        response2 = client.get("/fixtures/roadside_attractions")
        assert response2.status == 200
        table = Soup(response2.body, "html.parser").find("table")
        tds = table.find_all("td", {"class": "col-url"})
        assert [str(td) for td in tds] == [
            '<td class="col-url type-str"><a href="https://www.mysteryspot.com/">http…</a></td>',
            '<td class="col-url type-str"><a href="https://winchestermysteryhouse.com/">http…</a></td>',
            '<td class="col-url type-none">\xa0</td>',
            '<td class="col-url type-str"><a href="https://www.bigfootdiscoveryproject.com/">http…</a></td>',
        ]


@pytest.mark.asyncio
async def test_add_filter_redirects(ds_client):
    filter_args = urllib.parse.urlencode(
        {"_filter_column": "content", "_filter_op": "startswith", "_filter_value": "x"}
    )
    path_base = "/fixtures/simple_primary_key"
    path = path_base + "?" + filter_args
    response = await ds_client.get(path)
    assert response.status_code == 302
    assert response.headers["Location"].endswith("?content__startswith=x")

    # Adding a redirect to an existing query string:
    path = path_base + "?foo=bar&" + filter_args
    response = await ds_client.get(path)
    assert response.status_code == 302
    assert response.headers["Location"].endswith("?foo=bar&content__startswith=x")

    # Test that op with a __x suffix overrides the filter value
    path = (
        path_base
        + "?"
        + urllib.parse.urlencode(
            {
                "_filter_column": "content",
                "_filter_op": "isnull__5",
                "_filter_value": "x",
            }
        )
    )
    response = await ds_client.get(path)
    assert response.status_code == 302
    assert response.headers["Location"].endswith("?content__isnull=5")


@pytest.mark.asyncio
async def test_existing_filter_redirects(ds_client):
    filter_args = {
        "_filter_column_1": "name",
        "_filter_op_1": "contains",
        "_filter_value_1": "hello",
        "_filter_column_2": "age",
        "_filter_op_2": "gte",
        "_filter_value_2": "22",
        "_filter_column_3": "age",
        "_filter_op_3": "lt",
        "_filter_value_3": "30",
        "_filter_column_4": "name",
        "_filter_op_4": "contains",
        "_filter_value_4": "world",
    }
    path_base = "/fixtures/simple_primary_key"
    path = path_base + "?" + urllib.parse.urlencode(filter_args)
    response = await ds_client.get(path)
    assert response.status_code == 302
    assert_querystring_equal(
        "name__contains=hello&age__gte=22&age__lt=30&name__contains=world",
        response.headers["Location"].split("?")[1],
    )

    # Setting _filter_column_3 to empty string should remove *_3 entirely
    filter_args["_filter_column_3"] = ""
    path = path_base + "?" + urllib.parse.urlencode(filter_args)
    response = await ds_client.get(path)
    assert response.status_code == 302
    assert_querystring_equal(
        "name__contains=hello&age__gte=22&name__contains=world",
        response.headers["Location"].split("?")[1],
    )

    # ?_filter_op=exact should be removed if unaccompanied by _fiter_column
    response = await ds_client.get(path_base + "?_filter_op=exact")
    assert response.status_code == 302
    assert "?" not in response.headers["Location"]


@pytest.mark.asyncio
@pytest.mark.parametrize(
    "qs,expected_hidden",
    (
        # Things that should be reflected in hidden form fields:
        ("_facet=_neighborhood", {"_facet": "_neighborhood"}),
        ("_where=1+=+1&_col=_city_id", {"_where": "1 = 1", "_col": "_city_id"}),
        # Things that should NOT be reflected in hidden form fields:
        (
            "_facet=_neighborhood&_neighborhood__exact=Downtown",
            {"_facet": "_neighborhood"},
        ),
        ("_facet=_neighborhood&_city_id__gt=1", {"_facet": "_neighborhood"}),
    ),
)
async def test_reflected_hidden_form_fields(ds_client, qs, expected_hidden):
    # https://github.com/simonw/datasette/issues/1527
    response = await ds_client.get("/fixtures/facetable?{}".format(qs))
    # In this case we should NOT have a hidden _neighborhood__exact=Downtown field
    form = Soup(response.text, "html.parser").find("form")
    hidden_inputs = {
        input["name"]: input["value"] for input in form.select("input[type=hidden]")
    }
    assert hidden_inputs == expected_hidden


@pytest.mark.asyncio
async def test_empty_search_parameter_gets_removed(ds_client):
    path_base = "/fixtures/simple_primary_key"
    path = (
        path_base
        + "?"
        + urllib.parse.urlencode(
            {
                "_search": "",
                "_filter_column": "name",
                "_filter_op": "exact",
                "_filter_value": "chidi",
            }
        )
    )
    response = await ds_client.get(path)
    assert response.status_code == 302
    assert response.headers["Location"].endswith("?name__exact=chidi")


@pytest.mark.asyncio
async def test_searchable_view_persists_fts_table(ds_client):
    # The search form should persist ?_fts_table as a hidden field
    response = await ds_client.get(
        "/fixtures/searchable_view?_fts_table=searchable_fts&_fts_pk=pk"
    )
    inputs = Soup(response.text, "html.parser").find("form").find_all("input")
    hiddens = [i for i in inputs if i["type"] == "hidden"]
    assert [("_fts_table", "searchable_fts"), ("_fts_pk", "pk")] == [
        (hidden["name"], hidden["value"]) for hidden in hiddens
    ]


@pytest.mark.asyncio
async def test_sort_by_desc_redirects(ds_client):
    path_base = "/fixtures/sortable"
    path = (
        path_base
        + "?"
        + urllib.parse.urlencode({"_sort": "sortable", "_sort_by_desc": "1"})
    )
    response = await ds_client.get(path)
    assert response.status_code == 302
    assert response.headers["Location"].endswith("?_sort_desc=sortable")


@pytest.mark.asyncio
async def test_sort_links(ds_client):
    response = await ds_client.get("/fixtures/sortable?_sort=sortable")
    assert response.status_code == 200
    ths = Soup(response.text, "html.parser").find_all("th")
    attrs_and_link_attrs = [
        {
            "attrs": th.attrs,
            "a_href": (th.find("a")["href"] if th.find("a") else None),
        }
        for th in ths
    ]
    assert attrs_and_link_attrs == [
        {
            "attrs": {
                "class": ["col-Link"],
                "scope": "col",
                "data-column": "Link",
                "data-column-type": "",
                "data-column-not-null": "0",
                "data-is-pk": "0",
                "data-is-link-column": "1",
            },
            "a_href": None,
        },
        {
            "attrs": {
                "class": ["col-pk1"],
                "scope": "col",
                "data-column": "pk1",
                "data-column-type": "varchar(30)",
                "data-column-not-null": "0",
                "data-is-pk": "1",
            },
            "a_href": None,
        },
        {
            "attrs": {
                "class": ["col-pk2"],
                "scope": "col",
                "data-column": "pk2",
                "data-column-type": "varchar(30)",
                "data-column-not-null": "0",
                "data-is-pk": "1",
            },
            "a_href": None,
        },
        {
            "attrs": {
                "class": ["col-content"],
                "scope": "col",
                "data-column": "content",
                "data-column-type": "text",
                "data-column-not-null": "0",
                "data-is-pk": "0",
            },
            "a_href": None,
        },
        {
            "attrs": {
                "class": ["col-sortable"],
                "scope": "col",
                "data-column": "sortable",
                "data-column-type": "integer",
                "data-column-not-null": "0",
                "data-is-pk": "0",
            },
            "a_href": "/fixtures/sortable?_sort_desc=sortable",
        },
        {
            "attrs": {
                "class": ["col-sortable_with_nulls"],
                "scope": "col",
                "data-column": "sortable_with_nulls",
                "data-column-type": "real",
                "data-column-not-null": "0",
                "data-is-pk": "0",
            },
            "a_href": "/fixtures/sortable?_sort=sortable_with_nulls",
        },
        {
            "attrs": {
                "class": ["col-sortable_with_nulls_2"],
                "scope": "col",
                "data-column": "sortable_with_nulls_2",
                "data-column-type": "real",
                "data-column-not-null": "0",
                "data-is-pk": "0",
            },
            "a_href": "/fixtures/sortable?_sort=sortable_with_nulls_2",
        },
        {
            "attrs": {
                "class": ["col-text"],
                "scope": "col",
                "data-column": "text",
                "data-column-type": "text",
                "data-column-not-null": "0",
                "data-is-pk": "0",
            },
            "a_href": "/fixtures/sortable?_sort=text",
        },
    ]


@pytest.mark.asyncio
async def test_facet_display(ds_client):
    response = await ds_client.get(
        "/fixtures/facetable?_facet=planet_int&_facet=_city_id&_facet=on_earth"
    )
    assert response.status_code == 200
    soup = Soup(response.text, "html.parser")
    divs = soup.find("div", {"class": "facet-results"}).find_all("div")
    actual = []
    for div in divs:
        actual.append(
            {
                "name": div.find("strong").text.split()[0],
                "items": [
                    {
                        "name": a.text,
                        "qs": a["href"].split("?")[-1],
                        "count": int(
                            a.parent.find(
                                "span", {"class": "facet-count"}
                            ).text.replace(",", "")
                        ),
                    }
                    for a in div.find("ul").find_all("a")
                ],
            }
        )
    assert actual == [
        {
            "name": "_city_id",
            "items": [
                {
                    "name": "San Francisco",
                    "qs": "_facet=planet_int&_facet=_city_id&_facet=on_earth&_city_id__exact=1",
                    "count": 6,
                },
                {
                    "name": "Los Angeles",
                    "qs": "_facet=planet_int&_facet=_city_id&_facet=on_earth&_city_id__exact=2",
                    "count": 4,
                },
                {
                    "name": "Detroit",
                    "qs": "_facet=planet_int&_facet=_city_id&_facet=on_earth&_city_id__exact=3",
                    "count": 4,
                },
                {
                    "name": "Memnonia",
                    "qs": "_facet=planet_int&_facet=_city_id&_facet=on_earth&_city_id__exact=4",
                    "count": 1,
                },
            ],
        },
        {
            "name": "planet_int",
            "items": [
                {
                    "name": "1",
                    "qs": "_facet=planet_int&_facet=_city_id&_facet=on_earth&planet_int=1",
                    "count": 14,
                },
                {
                    "name": "2",
                    "qs": "_facet=planet_int&_facet=_city_id&_facet=on_earth&planet_int=2",
                    "count": 1,
                },
            ],
        },
        {
            "name": "on_earth",
            "items": [
                {
                    "name": "1",
                    "qs": "_facet=planet_int&_facet=_city_id&_facet=on_earth&on_earth=1",
                    "count": 14,
                },
                {
                    "name": "0",
                    "qs": "_facet=planet_int&_facet=_city_id&_facet=on_earth&on_earth=0",
                    "count": 1,
                },
            ],
        },
    ]


@pytest.mark.asyncio
async def test_facets_persist_through_filter_form(ds_client):
    response = await ds_client.get(
        "/fixtures/facetable?_facet=planet_int&_facet=_city_id&_facet_array=tags"
    )
    assert response.status_code == 200
    inputs = Soup(response.text, "html.parser").find("form").find_all("input")
    hiddens = [i for i in inputs if i["type"] == "hidden"]
    assert [(hidden["name"], hidden["value"]) for hidden in hiddens] == [
        ("_facet", "planet_int"),
        ("_facet", "_city_id"),
        ("_facet_array", "tags"),
    ]


@pytest.mark.asyncio
async def test_next_does_not_persist_in_hidden_field(ds_client):
    response = await ds_client.get("/fixtures/searchable?_size=1&_next=1")
    assert response.status_code == 200
    inputs = Soup(response.text, "html.parser").find("form").find_all("input")
    hiddens = [i for i in inputs if i["type"] == "hidden"]
    assert [(hidden["name"], hidden["value"]) for hidden in hiddens] == [
        ("_size", "1"),
    ]


@pytest.mark.asyncio
async def test_table_html_simple_primary_key(ds_client):
    response = await ds_client.get("/fixtures/simple_primary_key?_size=3")
    assert response.status_code == 200
    table = Soup(response.text, "html.parser").find("table")
    assert table["class"] == ["rows-and-columns"]
    ths = table.find_all("th")
    assert "id\xa0▼" == ths[0].find("a").string.strip()
    for expected_col, th in zip(("content",), ths[1:]):
        a = th.find("a")
        assert expected_col == a.string
        assert a["href"].endswith(f"/simple_primary_key?_size=3&_sort={expected_col}")
        assert ["nofollow"] == a["rel"]
    assert [
        [
            '<td class="col-id type-pk"><a href="/fixtures/simple_primary_key/1">1</a></td>',
            '<td class="col-content type-str">hello</td>',
        ],
        [
            '<td class="col-id type-pk"><a href="/fixtures/simple_primary_key/2">2</a></td>',
            '<td class="col-content type-str">world</td>',
        ],
        [
            '<td class="col-id type-pk"><a href="/fixtures/simple_primary_key/3">3</a></td>',
            '<td class="col-content type-str">\xa0</td>',
        ],
    ] == [[str(td) for td in tr.select("td")] for tr in table.select("tbody tr")]


@pytest.mark.asyncio
async def test_table_csv_json_export_interface(ds_client):
    response = await ds_client.get("/fixtures/simple_primary_key?id__gt=2")
    assert response.status_code == 200
    # The links at the top of the page
    links = (
        Soup(response.text, "html.parser")
        .find("p", {"class": "export-links"})
        .find_all("a")
    )
    actual = [link["href"] for link in links]
    expected = [
        "/fixtures/simple_primary_key.json?id__gt=2",
        "/fixtures/simple_primary_key.testall?id__gt=2",
        "/fixtures/simple_primary_key.testnone?id__gt=2",
        "/fixtures/simple_primary_key.testresponse?id__gt=2",
        "/fixtures/simple_primary_key.csv?id__gt=2&_size=max",
        "#export",
    ]
    assert expected == actual
    # And the advanced export box at the bottom:
    div = Soup(response.text, "html.parser").find("div", {"class": "advanced-export"})
    json_links = [a["href"] for a in div.find("p").find_all("a")]
    assert [
        "/fixtures/simple_primary_key.json?id__gt=2",
        "/fixtures/simple_primary_key.json?id__gt=2&_shape=array",
        "/fixtures/simple_primary_key.json?id__gt=2&_shape=array&_nl=on",
        "/fixtures/simple_primary_key.json?id__gt=2&_shape=object",
    ] == json_links
    # And the CSV form
    form = div.find("form")
    assert form["action"].endswith("/simple_primary_key.csv")
    inputs = [str(input) for input in form.find_all("input")]
    assert [
        '<input name="_dl" type="checkbox"/>',
        '<input type="submit" value="Export CSV"/>',
        '<input name="id__gt" type="hidden" value="2"/>',
        '<input name="_size" type="hidden" value="max"/>',
    ] == inputs


def test_base_url_export_links_are_not_double_prefixed(app_client_base_url_prefix):
    for path, expected_json, expected_csv in (
        (
            "/prefix/fixtures/simple_primary_key?id__gt=2",
            "/prefix/fixtures/simple_primary_key.json?id__gt=2",
            "/prefix/fixtures/simple_primary_key.csv?id__gt=2&_size=max",
        ),
        (
            "/prefix/fixtures/simple_primary_key/1",
            "/prefix/fixtures/simple_primary_key/1.json",
            None,
        ),
        (
            "/prefix/fixtures/-/query?sql=select+1",
            "/prefix/fixtures/-/query.json?sql=select+1",
            "/prefix/fixtures/-/query.csv?sql=select+1&_size=max",
        ),
    ):
        response = app_client_base_url_prefix.get(path)
        assert response.status_code == 200
        soup = Soup(response.text, "html.parser")
        export_links = soup.find("p", {"class": "export-links"}) or next(
            p for p in soup.find_all("p") if "This data as" in p.get_text()
        )
        hrefs = [a["href"] for a in export_links.find_all("a", href=True)]
        assert expected_json in hrefs
        if expected_csv:
            assert expected_csv in hrefs
        assert not [href for href in hrefs if href.startswith("/prefix/prefix/")]
        assert response.headers["link"] == (
            f"<http://localhost{expected_json}>; "
            'rel="alternate"; type="application/json+datasette"'
        )


def test_base_url_export_links_when_database_matches_prefix():
    with make_app_client(settings={"base_url": "/data/"}, filename="data.db") as client:
        response = client.get("/data/data/simple_primary_key?id__gt=2")
        assert response.status_code == 200
        export_links = Soup(response.text, "html.parser").find(
            "p", {"class": "export-links"}
        )
        hrefs = [a["href"] for a in export_links.find_all("a", href=True)]
        assert "/data/data/simple_primary_key.json?id__gt=2" in hrefs
        assert "/data/data/simple_primary_key.csv?id__gt=2&_size=max" in hrefs
        assert response.headers["link"] == (
            "<http://localhost/data/data/simple_primary_key.json?id__gt=2>; "
            'rel="alternate"; type="application/json+datasette"'
        )


@pytest.mark.asyncio
async def test_csv_json_export_links_include_labels_if_foreign_keys(ds_client):
    response = await ds_client.get("/fixtures/facetable")
    assert response.status_code == 200
    links = (
        Soup(response.text, "html.parser")
        .find("p", {"class": "export-links"})
        .find_all("a")
    )
    actual = [link["href"] for link in links]
    expected = [
        "/fixtures/facetable.json?_labels=on",
        "/fixtures/facetable.testall?_labels=on",
        "/fixtures/facetable.testnone?_labels=on",
        "/fixtures/facetable.testresponse?_labels=on",
        "/fixtures/facetable.csv?_labels=on&_size=max",
        "#export",
    ]
    assert expected == actual


@pytest.mark.asyncio
async def test_table_not_exists(ds_client):
    assert "Table not found" in (await ds_client.get("/fixtures/blah")).text


@pytest.mark.asyncio
async def test_table_html_no_primary_key(ds_client):
    response = await ds_client.get("/fixtures/no_primary_key")
    assert response.status_code == 200
    table = Soup(response.text, "html.parser").find("table")
    # We have disabled sorting for this table using metadata.json
    assert ["content", "a", "b", "c"] == [
        th.string.strip() for th in table.select("thead th")[2:]
    ]
    expected = [
        [
            '<td class="col-Link type-pk"><a href="/fixtures/no_primary_key/{}">{}</a></td>'.format(
                i, i
            ),
            f'<td class="col-rowid type-int">{i}</td>',
            f'<td class="col-content type-str">{i}</td>',
            f'<td class="col-a type-str">a{i}</td>',
            f'<td class="col-b type-str">b{i}</td>',
            f'<td class="col-c type-str">c{i}</td>',
        ]
        for i in range(1, 51)
    ]
    assert expected == [
        [str(td) for td in tr.select("td")] for tr in table.select("tbody tr")
    ]


@pytest.mark.asyncio
async def test_rowid_sortable_no_primary_key(ds_client):
    response = await ds_client.get("/fixtures/no_primary_key")
    assert response.status_code == 200
    table = Soup(response.text, "html.parser").find("table")
    assert table["class"] == ["rows-and-columns"]
    ths = table.find_all("th")
    assert "rowid\xa0▼" == ths[1].find("a").string.strip()


@pytest.mark.asyncio
async def test_table_html_compound_primary_key(ds_client):
    response = await ds_client.get("/fixtures/compound_primary_key")
    assert response.status_code == 200
    table = Soup(response.text, "html.parser").find("table")
    ths = table.find_all("th")
    assert "Link" == ths[0].string.strip()
    for expected_col, th in zip(("pk1", "pk2", "content"), ths[1:]):
        a = th.find("a")
        assert expected_col == a.string
        assert th["class"] == [f"col-{expected_col}"]
        assert a["href"].endswith(f"/compound_primary_key?_sort={expected_col}")
    expected = [
        [
            '<td class="col-Link type-pk"><a href="/fixtures/compound_primary_key/a,b">a,b</a></td>',
            '<td class="col-pk1 type-str">a</td>',
            '<td class="col-pk2 type-str">b</td>',
            '<td class="col-content type-str">c</td>',
        ],
        [
            '<td class="col-Link type-pk"><a href="/fixtures/compound_primary_key/a~2Fb,~2Ec-d">a/b,.c-d</a></td>',
            '<td class="col-pk1 type-str">a/b</td>',
            '<td class="col-pk2 type-str">.c-d</td>',
            '<td class="col-content type-str">c</td>',
        ],
        [
            '<td class="col-Link type-pk"><a href="/fixtures/compound_primary_key/d,e">d,e</a></td>',
            '<td class="col-pk1 type-str">d</td>',
            '<td class="col-pk2 type-str">e</td>',
            '<td class="col-content type-str">{"row": {"pk1": "d", "pk2": "e", "content": "RENDER_CELL_DEMO"}, "column": "content", "table": "compound_primary_key", "database": "fixtures", "pks": ["pk1", "pk2"], "config": {"depth": "database"}}</td>',
        ],
    ]
    assert [
        [str(td) for td in tr.select("td")] for tr in table.select("tbody tr")
    ] == expected
    rows = table.select("tbody tr")
    assert rows[0]["data-row"] == "a,b"
    assert "data-row-pk-path" not in rows[0].attrs
    assert "data-row-label" not in rows[0].attrs
    assert rows[1]["data-row"] == "a~2Fb,~2Ec-d"
    assert "data-row-pk-path" not in rows[1].attrs
    assert "data-row-label" not in rows[1].attrs


@pytest.mark.asyncio
async def test_table_html_foreign_key_links(ds_client):
    response = await ds_client.get("/fixtures/foreign_key_references")
    assert response.status_code == 200
    table = Soup(response.text, "html.parser").find("table")
    actual = [[str(td) for td in tr.select("td")] for tr in table.select("tbody tr")]
    assert actual == [
        [
            '<td class="col-pk type-pk"><a href="/fixtures/foreign_key_references/1">1</a></td>',
            '<td class="col-foreign_key_with_label type-int"><a href="/fixtures/simple_primary_key/1">hello</a>\xa0<em>1</em></td>',
            '<td class="col-foreign_key_with_blank_label type-int"><a href="/fixtures/simple_primary_key/3">-</a>\xa0<em>3</em></td>',
            '<td class="col-foreign_key_with_no_label type-str"><a href="/fixtures/primary_key_multiple_columns/1">1</a></td>',
            '<td class="col-foreign_key_compound_pk1 type-str">a</td>',
            '<td class="col-foreign_key_compound_pk2 type-str">b</td>',
        ],
        [
            '<td class="col-pk type-pk"><a href="/fixtures/foreign_key_references/2">2</a></td>',
            '<td class="col-foreign_key_with_label type-none">\xa0</td>',
            '<td class="col-foreign_key_with_blank_label type-none">\xa0</td>',
            '<td class="col-foreign_key_with_no_label type-none">\xa0</td>',
            '<td class="col-foreign_key_compound_pk1 type-none">\xa0</td>',
            '<td class="col-foreign_key_compound_pk2 type-none">\xa0</td>',
        ],
    ]


@pytest.mark.asyncio
async def test_table_html_foreign_key_facets(ds_client):
    response = await ds_client.get(
        "/fixtures/foreign_key_references?_facet=foreign_key_with_blank_label"
    )
    assert response.status_code == 200
    assert (
        '<li><a href="http://localhost/fixtures/foreign_key_references?_facet=foreign_key_with_blank_label&amp;foreign_key_with_blank_label=3"'
        ' data-facet-value="3">-</a> <span class="facet-count">1</span></li>'
    ) in response.text


@pytest.mark.asyncio
async def test_table_html_disable_foreign_key_links_with_labels(ds_client):
    response = await ds_client.get(
        "/fixtures/foreign_key_references?_labels=off&_size=1"
    )
    assert response.status_code == 200
    table = Soup(response.text, "html.parser").find("table")
    actual = [[str(td) for td in tr.select("td")] for tr in table.select("tbody tr")]
    assert actual == [
        [
            '<td class="col-pk type-pk"><a href="/fixtures/foreign_key_references/1">1</a></td>',
            '<td class="col-foreign_key_with_label type-int">1</td>',
            '<td class="col-foreign_key_with_blank_label type-int">3</td>',
            '<td class="col-foreign_key_with_no_label type-str">1</td>',
            '<td class="col-foreign_key_compound_pk1 type-str">a</td>',
            '<td class="col-foreign_key_compound_pk2 type-str">b</td>',
        ]
    ]


@pytest.mark.asyncio
async def test_table_html_foreign_key_custom_label_column(ds_client):
    response = await ds_client.get("/fixtures/custom_foreign_key_label")
    assert response.status_code == 200
    table = Soup(response.text, "html.parser").find("table")
    expected = [
        [
            '<td class="col-pk type-pk"><a href="/fixtures/custom_foreign_key_label/1">1</a></td>',
            '<td class="col-foreign_key_with_custom_label type-str"><a href="/fixtures/primary_key_multiple_columns_explicit_label/1">world2</a>\xa0<em>1</em></td>',
        ]
    ]
    assert expected == [
        [str(td) for td in tr.select("td")] for tr in table.select("tbody tr")
    ]


@pytest.mark.asyncio
@pytest.mark.parametrize(
    "path,expected_column_options",
    [
        ("/fixtures/infinity", ["- column -", "rowid", "value"]),
        (
            "/fixtures/primary_key_multiple_columns",
            ["- column -", "id", "content", "content2"],
        ),
        ("/fixtures/compound_primary_key", ["- column -", "pk1", "pk2", "content"]),
    ],
)
async def test_table_html_filter_form_column_options(
    path, expected_column_options, ds_client
):
    response = await ds_client.get(path)
    assert response.status_code == 200
    form = Soup(response.text, "html.parser").find("form")
    column_options = [
        o.attrs.get("value") or o.string
        for o in form.select("select[name=_filter_column] option")
    ]
    assert expected_column_options == column_options


@pytest.mark.asyncio
async def test_table_html_filter_form_still_shows_nocol_columns(ds_client):
    # https://github.com/simonw/datasette/issues/1503
    response = await ds_client.get("/fixtures/sortable?_nocol=sortable")
    assert response.status_code == 200
    form = Soup(response.text, "html.parser").find("form")
    assert [
        o.string
        for o in form.select("select[name='_filter_column']")[0].select("option")
    ] == [
        "- column -",
        "pk1",
        "pk2",
        "content",
        "sortable_with_nulls",
        "sortable_with_nulls_2",
        "text",
        # Moved to the end because it is no longer returned by the query:
        "sortable",
    ]


@pytest.mark.asyncio
async def test_column_chooser_present(ds_client):
    response = await ds_client.get("/fixtures/facetable")
    assert response.status_code == 200
    soup = Soup(response.text, "html.parser")
    # Web component should be present
    chooser = soup.find("column-chooser")
    assert chooser is not None
    # Script block should contain column data as JSON

    scripts = soup.find_all("script")
    chooser_script = [s for s in scripts if "_columnChooserData" in (s.string or "")]
    assert len(chooser_script) == 1
    script_text = chooser_script[0].string
    # Extract the JSON data
    assert "allColumns" in script_text
    assert "selectedColumns" in script_text
    assert "primaryKeys" in script_text


@pytest.mark.asyncio
@pytest.mark.parametrize(
    "path", ["/fixtures/facetable", "/fixtures/123_starts_with_digits"]
)
async def test_mobile_column_actions_present(ds_client, path):
    response = await ds_client.get(path)
    assert response.status_code == 200
    soup = Soup(response.text, "html.parser")
    button = soup.select_one("button.column-actions-mobile.small-screen-only")
    assert button is not None
    assert button.text.strip() == "Column actions"
    assert button.find("svg") is not None
    assert any(
        "mobile-column-actions.js" in (script.get("src") or "")
        for script in soup.find_all("script")
    )
    # mobile-column-actions.js builds its dialog from <th data-column> elements,
    # so the thead must render even when the table has no rows.
    ths = soup.select("table.rows-and-columns thead th[data-column]")
    assert len(ths) >= 1


@pytest.mark.asyncio
async def test_row_delete_action_data_attributes():
    ds = Datasette(
        [],
        config={
            "databases": {
                "data": {
                    "tables": {
                        "items": {
                            "permissions": {
                                "update-row": {"id": "root"},
                                "delete-row": {"id": "root"},
                            },
                        },
                    },
                },
            },
        },
    )
    try:
        db = ds.add_database(
            Database(ds, memory_name="test_row_delete_actions"), name="data"
        )
        await db.execute_write_script("""
            create table items (id integer primary key, name text, score integer);
            insert into items (id, name, score) values (1, 'One', 5);
            """)
        response = await ds.client.get("/data/items", actor={"id": "root"})
        assert response.status_code == 200
        soup = Soup(response.text, "html.parser")
        assert table_data_from_soup(soup) == {
            "database": "data",
            "table": "items",
            "tableUrl": "/data/items",
        }
        assert soup.select_one('button[data-table-action="insert-row"]') is None

        row = soup.select_one("table.rows-and-columns tbody tr")
        assert row["data-row"] == "1"
        assert row["data-row-label"] == "One"
        assert {key for key in row.attrs if key.startswith("data-row")} == {
            "data-row",
            "data-row-label",
        }

        edit_button = row.select_one(
            'button.row-inline-action-edit[data-row-action="edit"]'
        )
        assert edit_button is not None
        assert edit_button["aria-label"] == "Edit row 1 One"
        assert edit_button["title"] == "Edit row"
        assert edit_button.find("svg") is not None

        button = row.select_one(
            'button.row-inline-action-delete[data-row-action="delete"]'
        )
        assert button is not None
        assert button["aria-label"] == "Delete row 1 One"
        assert button["title"] == "Delete row"
        assert button.find("svg") is not None

        response = await ds.client.get("/data/items?_col=score", actor={"id": "root"})
        assert response.status_code == 200
        soup = Soup(response.text, "html.parser")
        row = soup.select_one("table.rows-and-columns tbody tr")
        assert row["data-row"] == "1"
        assert "data-row-label" not in row.attrs

        edit_button = row.select_one(
            'button.row-inline-action-edit[data-row-action="edit"]'
        )
        assert edit_button is not None
        assert edit_button["aria-label"] == "Edit row 1"

        button = row.select_one(
            'button.row-inline-action-delete[data-row-action="delete"]'
        )
        assert button is not None
        assert button["aria-label"] == "Delete row 1"
    finally:
        ds.close()


@pytest.mark.asyncio
async def test_table_insert_action_button_and_data():
    ds = Datasette(
        [],
        config={
            "databases": {
                "data": {
                    "tables": {
                        "items": {
                            "permissions": {
                                "insert-row": {"id": "root"},
                            },
                            "column_types": {"body": "textarea"},
                        },
                    },
                },
            },
        },
    )
    try:
        db = ds.add_database(
            Database(ds, memory_name="test_table_insert_action"), name="data"
        )
        await db.execute_write_script("""
            create table items (
                id integer primary key,
                name text not null,
                score integer default 5,
                price numeric,
                created text default (datetime('now')),
                body text,
                typeless
            );
            """)
        response = await ds.client.get("/data/items", actor={"id": "root"})
        assert response.status_code == 200
        soup = Soup(response.text, "html.parser")

        button = soup.select_one(
            'button.table-insert-row[data-table-action="insert-row"]'
        )
        assert button is not None
        assert button.text.strip() == "Insert row"
        assert button.find("svg") is not None
        assert button.find_parent("div", class_="table-row-toolbar") is not None

        insert_data = table_data_from_soup(soup)["insertRow"]
        assert insert_data["path"] == "/data/items/-/insert"
        assert insert_data["tableName"] == "items"
        assert insert_data["primaryKeys"] == ["id"]
        assert [column["name"] for column in insert_data["columns"]] == [
            "name",
            "score",
            "price",
            "created",
            "body",
            "typeless",
        ]
        name, score, price, created, body, typeless = insert_data["columns"]
        assert name["notnull"] == 1
        assert name["sqlite_type"] == "TEXT"
        assert name["value_kind"] == "string"
        assert not name["has_default"]
        assert score["default"] == "5"
        assert score["has_default"]
        assert score["sqlite_type"] == "INTEGER"
        assert score["value_kind"] == "number"
        assert price["sqlite_type"] == "NUMERIC"
        assert price["value_kind"] == "string"
        assert created["default"] == "datetime('now')"
        assert created["has_default"]
        assert created["sqlite_type"] == "TEXT"
        assert body["sqlite_type"] == "TEXT"
        assert body["value_kind"] == "string"
        assert body["column_type"] == {"type": "textarea", "config": None}
        assert typeless["sqlite_type"] == "BLOB"
        assert typeless["value_kind"] == "string"
    finally:
        ds.close()


@pytest.mark.asyncio
async def test_table_insert_action_includes_compound_primary_keys():
    ds = Datasette(
        [],
        config={
            "databases": {
                "data": {
                    "tables": {
                        "memberships": {
                            "permissions": {
                                "insert-row": {"id": "root"},
                            },
                        },
                    },
                },
            },
        },
    )
    try:
        db = ds.add_database(
            Database(ds, memory_name="test_table_insert_compound_pk"), name="data"
        )
        await db.execute_write_script("""
            create table memberships (
                account text,
                username text,
                role text,
                primary key (account, username)
            );
            """)
        response = await ds.client.get("/data/memberships", actor={"id": "root"})
        assert response.status_code == 200
        insert_data = table_data_from_soup(Soup(response.text, "html.parser"))[
            "insertRow"
        ]
        assert insert_data["tableName"] == "memberships"
        assert insert_data["primaryKeys"] == ["account", "username"]
        assert [column["name"] for column in insert_data["columns"]] == [
            "account",
            "username",
            "role",
        ]
        assert [column["is_pk"] for column in insert_data["columns"]] == [
            True,
            True,
            False,
        ]
    finally:
        ds.close()


@pytest.mark.asyncio
async def test_table_data_includes_foreign_key_autocomplete_urls():
    ds = Datasette([])
    try:
        db = ds.add_database(
            Database(ds, memory_name="test_table_foreign_key_autocomplete"), name="data"
        )
        await db.execute_write_script("""
            create table authors (
                id integer primary key,
                name text
            );
            create table tags (
                slug text unique,
                name text
            );
            create table articles (
                id integer primary key,
                author_id integer references authors(id),
                implicit_author_id integer references authors,
                tag_slug text references tags(slug),
                title text
            );
            insert into authors (id, name) values (1, 'Ada Lovelace');
            insert into tags (slug, name) values ('science', 'Science');
            insert into articles (
                id,
                author_id,
                implicit_author_id,
                tag_slug,
                title
            ) values (
                1,
                1,
                1,
                'science',
                'Notes'
            );
            """)
        response = await ds.client.get("/data/articles")
        assert response.status_code == 200
        soup = Soup(response.text, "html.parser")
        table_data = table_data_from_soup(soup)
        assert table_data["foreignKeys"] == {
            "author_id": "/data/authors/-/autocomplete",
            "implicit_author_id": "/data/authors/-/autocomplete",
        }
        assert any(
            "autocomplete.js" in (script.get("src") or "")
            for script in soup.find_all("script")
        )
    finally:
        ds.close()


@pytest.mark.asyncio
async def test_table_fragment_endpoint(ds_client):
    response = await ds_client.get("/fixtures/simple_primary_key/-/fragment?_row=1")
    assert response.status_code == 200
    assert response.headers["content-type"].startswith("text/html")
    soup = Soup(response.text, "html.parser")
    assert soup.find("html") is None
    rows = soup.select("[data-row]")
    assert len(rows) == 1
    assert rows[0]["data-row"] == "1"
    assert rows[0]["data-row-label"] == "hello"
    assert {key for key in rows[0].attrs if key.startswith("data-row")} == {
        "data-row",
        "data-row-label",
    }


@pytest.mark.asyncio
async def test_table_fragment_row_parameter_replaces_pk_filters(ds_client):
    response = await ds_client.get(
        "/fixtures/simple_primary_key/-/fragment?id=2&_row=1"
    )
    assert response.status_code == 200
    soup = Soup(response.text, "html.parser")
    rows = soup.select("[data-row]")
    assert len(rows) == 1
    assert rows[0]["data-row"] == "1"
    assert rows[0]["data-row-label"] == "hello"


@pytest.mark.asyncio
async def test_row_page_edit_delete_action_menu_buttons():
    ds = Datasette(
        [],
        config={
            "databases": {
                "data": {
                    "tables": {
                        "items": {
                            "permissions": {
                                "update-row": {"id": "root"},
                                "delete-row": {"id": "root"},
                            },
                        },
                    },
                },
            },
        },
    )
    try:
        db = ds.add_database(
            Database(ds, memory_name="test_row_page_edit_delete_actions"), name="data"
        )
        await db.execute_write_script("""
            create table items (id integer primary key, name text, score integer);
            insert into items (id, name, score) values (1, 'One', 5);
            """)
        response = await ds.client.get("/data/items/1", actor={"id": "root"})
        assert response.status_code == 200
        soup = Soup(response.text, "html.parser")
        assert table_data_from_soup(soup) == {
            "database": "data",
            "table": "items",
            "tableUrl": "/data/items",
        }
        script_srcs = [script.get("src") or "" for script in soup.find_all("script")]
        assert any("edit-tools.js" in src for src in script_srcs)
        assert not any("table.js" in src for src in script_srcs)

        row = soup.select_one("table.rows-and-columns tbody tr")
        assert row["data-row"] == "1"
        assert row["data-row-label"] == "One"

        edit_button = soup.select_one(
            'details.actions-menu-links button.action-menu-button[data-row-action="edit"]'
        )
        assert edit_button is not None
        assert edit_button["aria-label"] == "Edit row 1 One"
        assert edit_button["data-row"] == "1"
        assert edit_button["data-row-label"] == "One"
        assert edit_button["role"] == "menuitem"
        assert edit_button.find("span", class_="dropdown-description").text.strip() == (
            "Open a dialog to edit this row."
        )
        edit_button.find("span").extract()
        assert edit_button.text.strip() == "Edit row"

        delete_button = soup.select_one(
            'details.actions-menu-links button.action-menu-button[data-row-action="delete"]'
        )
        assert delete_button is not None
        assert delete_button["aria-label"] == "Delete row 1 One"
        assert delete_button["data-row"] == "1"
        assert delete_button["data-row-label"] == "One"
        assert delete_button["role"] == "menuitem"
        assert delete_button.find(
            "span", class_="dropdown-description"
        ).text.strip() == ("Open a confirmation dialog to delete this row.")
        delete_button.find("span").extract()
        assert delete_button.text.strip() == "Delete row"
    finally:
        ds.close()


@pytest.mark.asyncio
async def test_row_delete_redirect_to_table_sets_message():
    ds = Datasette(
        [],
        config={
            "databases": {
                "data": {
                    "tables": {
                        "items": {
                            "permissions": {
                                "delete-row": {"id": "root"},
                            },
                        },
                    },
                },
            },
        },
    )
    try:
        db = ds.add_database(
            Database(ds, memory_name="test_row_delete_redirect"), name="data"
        )
        await db.execute_write_script("""
            create table items (id integer primary key, name text);
            insert into items (id, name) values (1, 'One');
            """)
        response = await ds.client.post(
            "/data/items/1/-/delete?_redirect_to_table=1", actor={"id": "root"}
        )
        assert response.status_code == 200
        assert response.json() == {"ok": True, "redirect": "/data/items"}
        assert ds.unsign(response.cookies["ds_messages"], "messages") == [
            ["Deleted row 1 (One)", ds.INFO]
        ]
    finally:
        ds.close()


@pytest.mark.asyncio
async def test_row_update_sets_message():
    ds = Datasette(
        [],
        config={
            "databases": {
                "data": {
                    "tables": {
                        "items": {
                            "permissions": {
                                "update-row": {"id": "root"},
                            },
                        },
                    },
                },
            },
        },
    )
    try:
        db = ds.add_database(
            Database(ds, memory_name="test_row_update_message"), name="data"
        )
        await db.execute_write_script("""
            create table items (id integer primary key, name text);
            insert into items (id, name) values (1, 'One');
            """)
        long_name = "Two " + ("long label " * 12)
        truncated_name = long_name[:79] + "\u2026"
        response = await ds.client.post(
            "/data/items/1/-/update?_message=1",
            actor={"id": "root"},
            json={"update": {"name": long_name}, "return": True},
        )
        assert response.status_code == 200
        assert response.json()["row"]["name"] == long_name
        assert ds.unsign(response.cookies["ds_messages"], "messages") == [
            ["Updated row 1 ({})".format(truncated_name), ds.INFO]
        ]
    finally:
        ds.close()


def test_table_data_uses_base_url(app_client_base_url_prefix):
    response = app_client_base_url_prefix.get("/prefix/fixtures/simple_primary_key")
    assert response.status_code == 200
    import json
    import re

    soup = Soup(response.text, "html.parser")
    table_script = [
        s for s in soup.find_all("script") if "_datasetteTableData" in (s.string or "")
    ][0]
    match = re.search(
        r"window\._datasetteTableData\s*=\s*({.*?});",
        table_script.string,
        re.DOTALL,
    )
    assert json.loads(match.group(1)) == {
        "database": "fixtures",
        "table": "simple_primary_key",
        "tableUrl": "/prefix/fixtures/simple_primary_key",
    }


def test_table_fragment_custom_table_include():
    with make_app_client(
        template_dir=str(pathlib.Path(__file__).parent / "test_templates")
    ) as client:
        response = client.get("/fixtures/complex_foreign_keys/-/fragment?f1=1&f2=2")
        assert response.status == 200
        assert (
            '<div class="custom-table-row">'
            '1 - 2 - <a href="/fixtures/simple_primary_key/1">hello</a> <em>1</em>'
            "</div>"
        ) == str(Soup(response.text, "html.parser").select_one("div.custom-table-row"))


@pytest.mark.asyncio
async def test_table_fragment_uses_render_cell_hook():
    from datasette import hookimpl
    from markupsafe import Markup

    class TestRenderCellPlugin:
        __name__ = "TestRenderCellPlugin"

        @hookimpl
        def render_cell(self, value, column, table, database):
            if database == "data" and table == "items" and column == "name":
                return Markup("<strong>{}</strong>".format(value))
            return None

    ds = Datasette(memory=True)
    await ds.invoke_startup()
    db = ds.add_memory_database("data")
    await db.execute_write("create table items (id integer primary key, name text)")
    await db.execute_write("insert into items values (1, 'Alice')")
    ds.pm.register(TestRenderCellPlugin(), name="TestRenderCellPlugin")
    try:
        response = await ds.client.get("/data/items/-/fragment?id=1")
        assert response.status_code == 200
        assert "<strong>Alice</strong>" in response.text
    finally:
        ds.pm.unregister(name="TestRenderCellPlugin")
        ds.close()


@pytest.mark.asyncio
async def test_zero_row_table_renders_thead(ds_client):
    response = await ds_client.get("/fixtures/123_starts_with_digits")
    assert response.status_code == 200
    soup = Soup(response.text, "html.parser")
    table = soup.select_one("table.rows-and-columns")
    assert table is not None
    column_names = [
        th.get("data-column") for th in table.select("thead th[data-column]")
    ]
    assert "content" in column_names
    assert table.select_one("tbody tr") is None
    assert soup.select_one("p.zero-results") is not None


@pytest.mark.asyncio
async def test_column_chooser_data_reflects_col_filtering(ds_client):
    response = await ds_client.get("/fixtures/facetable?_col=state&_col=created")
    assert response.status_code == 200
    import json
    import re

    soup = Soup(response.text, "html.parser")
    chooser = soup.find("column-chooser")
    assert chooser is not None
    scripts = soup.find_all("script")
    chooser_script = [s for s in scripts if "_columnChooserData" in (s.string or "")]
    script_text = chooser_script[0].string
    # Parse the JSON object from the script
    match = re.search(
        r"window\._columnChooserData\s*=\s*({.*?});", script_text, re.DOTALL
    )
    data = json.loads(match.group(1))
    # All non-PK columns should still be listed in allColumns
    assert "state" in data["allColumns"]
    assert "created" in data["allColumns"]
    assert "planet_int" in data["allColumns"]
    # Only state and created should be in selectedColumns (plus pk)
    non_pk_selected = [
        c for c in data["selectedColumns"] if c not in data["primaryKeys"]
    ]
    assert "state" in non_pk_selected
    assert "created" in non_pk_selected
    assert "planet_int" not in non_pk_selected


@pytest.mark.asyncio
async def test_column_chooser_shown_for_views(ds_client):
    response = await ds_client.get("/fixtures/simple_view")
    assert response.status_code == 200
    soup = Soup(response.text, "html.parser")
    chooser = soup.find("column-chooser")
    assert chooser is not None
    scripts = soup.find_all("script")
    chooser_script = [s for s in scripts if "_columnChooserData" in (s.string or "")]
    assert len(chooser_script) == 1


@pytest.mark.asyncio
async def test_compound_primary_key_with_foreign_key_references(ds_client):
    # e.g. a many-to-many table with a compound primary key on the two columns
    response = await ds_client.get("/fixtures/searchable_tags")
    assert response.status_code == 200
    table = Soup(response.text, "html.parser").find("table")
    expected = [
        [
            '<td class="col-Link type-pk"><a href="/fixtures/searchable_tags/1,feline">1,feline</a></td>',
            '<td class="col-searchable_id type-int"><a href="/fixtures/searchable/1">1</a>\xa0<em>1</em></td>',
            '<td class="col-tag type-str"><a href="/fixtures/tags/feline">feline</a></td>',
        ],
        [
            '<td class="col-Link type-pk"><a href="/fixtures/searchable_tags/2,canine">2,canine</a></td>',
            '<td class="col-searchable_id type-int"><a href="/fixtures/searchable/2">2</a>\xa0<em>2</em></td>',
            '<td class="col-tag type-str"><a href="/fixtures/tags/canine">canine</a></td>',
        ],
    ]
    assert expected == [
        [str(td) for td in tr.select("td")] for tr in table.select("tbody tr")
    ]


@pytest.mark.asyncio
async def test_view_html(ds_client):
    response = await ds_client.get("/fixtures/simple_view?_size=3")
    assert response.status_code == 200
    table = Soup(response.text, "html.parser").find("table")
    ths = table.select("thead th")
    assert 2 == len(ths)
    assert ths[0].find("a") is not None
    assert ths[0].find("a")["href"].endswith("/simple_view?_size=3&_sort=content")
    assert ths[0].find("a").string.strip() == "content"
    assert ths[1].find("a") is None
    assert ths[1].string.strip() == "upper_content"
    expected = [
        [
            '<td class="col-content type-str">hello</td>',
            '<td class="col-upper_content type-str">HELLO</td>',
        ],
        [
            '<td class="col-content type-str">world</td>',
            '<td class="col-upper_content type-str">WORLD</td>',
        ],
        [
            '<td class="col-content type-str">\xa0</td>',
            '<td class="col-upper_content type-str">\xa0</td>',
        ],
    ]
    assert expected == [
        [str(td) for td in tr.select("td")] for tr in table.select("tbody tr")
    ]


@pytest.mark.asyncio
async def test_table_metadata(ds_client):
    response = await ds_client.get("/fixtures/simple_primary_key")
    assert response.status_code == 200
    soup = Soup(response.text, "html.parser")
    # Page title should be custom and should be HTML escaped
    assert "This &lt;em&gt;HTML&lt;/em&gt; is escaped" == inner_html(soup.find("h1"))
    # Description should be custom and NOT escaped (we used description_html)
    assert "Simple <em>primary</em> key" == inner_html(
        soup.find("div", {"class": "metadata-description"})
    )


@pytest.mark.asyncio
@pytest.mark.parametrize(
    "path,has_object,has_stream,has_expand",
    [
        ("/fixtures/no_primary_key", False, True, False),
        ("/fixtures/complex_foreign_keys", True, False, True),
    ],
)
async def test_advanced_export_box(ds_client, path, has_object, has_stream, has_expand):
    response = await ds_client.get(path)
    assert response.status_code == 200
    soup = Soup(response.text, "html.parser")
    # JSON shape options
    expected_json_shapes = ["default", "array", "newline-delimited"]
    if has_object:
        expected_json_shapes.append("object")
    div = soup.find("div", {"class": "advanced-export"})
    assert expected_json_shapes == [a.text for a in div.find("p").find_all("a")]
    # "stream all rows" option
    if has_stream:
        assert "stream all rows" in str(div)
    # "expand labels" option
    if has_expand:
        assert "expand labels" in str(div)


@pytest.mark.asyncio
async def test_extra_where_clauses(ds_client):
    response = await ds_client.get(
        "/fixtures/facetable?_where=_neighborhood='Dogpatch'&_where=_city_id=1"
    )
    soup = Soup(response.text, "html.parser")
    div = soup.select(".extra-wheres")[0]
    assert "2 extra where clauses" == div.find("h3").text
    hrefs = [a["href"] for a in div.find_all("a")]
    assert [
        "/fixtures/facetable?_where=_city_id%3D1",
        "/fixtures/facetable?_where=_neighborhood%3D%27Dogpatch%27",
    ] == hrefs
    # These should also be persisted as hidden fields
    inputs = soup.find("form").find_all("input")
    hiddens = [i for i in inputs if i["type"] == "hidden"]
    assert [("_where", "_neighborhood='Dogpatch'"), ("_where", "_city_id=1")] == [
        (hidden["name"], hidden["value"]) for hidden in hiddens
    ]


@pytest.mark.asyncio
@pytest.mark.parametrize(
    "path,expected_hidden",
    [
        ("/fixtures/facetable?_size=10", [("_size", "10")]),
        (
            "/fixtures/facetable?_size=10&_ignore=1&_ignore=2",
            [
                ("_size", "10"),
                ("_ignore", "1"),
                ("_ignore", "2"),
            ],
        ),
    ],
)
async def test_other_hidden_form_fields(ds_client, path, expected_hidden):
    response = await ds_client.get(path)
    soup = Soup(response.text, "html.parser")
    inputs = soup.find("form").find_all("input")
    hiddens = [i for i in inputs if i["type"] == "hidden"]
    assert [(hidden["name"], hidden["value"]) for hidden in hiddens] == expected_hidden


@pytest.mark.asyncio
@pytest.mark.parametrize(
    "path,expected_hidden",
    [
        ("/fixtures/searchable?_search=terry", []),
        ("/fixtures/searchable?_sort=text2", []),
        ("/fixtures/searchable?_sort_desc=text2", []),
        ("/fixtures/searchable?_sort=text2&_where=1", [("_where", "1")]),
    ],
)
async def test_search_and_sort_fields_not_duplicated(ds_client, path, expected_hidden):
    # https://github.com/simonw/datasette/issues/1214
    response = await ds_client.get(path)
    soup = Soup(response.text, "html.parser")
    inputs = soup.find("form").find_all("input")
    hiddens = [i for i in inputs if i["type"] == "hidden"]
    assert [(hidden["name"], hidden["value"]) for hidden in hiddens] == expected_hidden


@pytest.mark.asyncio
async def test_binary_data_display_in_table(ds_client):
    response = await ds_client.get("/fixtures/binary_data")
    assert response.status_code == 200
    table = Soup(response.text, "html.parser").find("table")
    expected_tds = [
        [
            '<td class="col-Link type-pk"><a href="/fixtures/binary_data/1">1</a></td>',
            '<td class="col-rowid type-int">1</td>',
            '<td class="col-data type-bytes"><a class="blob-download" href="/fixtures/binary_data/1.blob?_blob_column=data">&lt;Binary:\xa07\xa0bytes&gt;</a></td>',
        ],
        [
            '<td class="col-Link type-pk"><a href="/fixtures/binary_data/2">2</a></td>',
            '<td class="col-rowid type-int">2</td>',
            '<td class="col-data type-bytes"><a class="blob-download" href="/fixtures/binary_data/2.blob?_blob_column=data">&lt;Binary:\xa07\xa0bytes&gt;</a></td>',
        ],
        [
            '<td class="col-Link type-pk"><a href="/fixtures/binary_data/3">3</a></td>',
            '<td class="col-rowid type-int">3</td>',
            '<td class="col-data type-none">\xa0</td>',
        ],
    ]
    assert expected_tds == [
        [str(td) for td in tr.select("td")] for tr in table.select("tbody tr")
    ]


def test_custom_table_include():
    with make_app_client(
        template_dir=str(pathlib.Path(__file__).parent / "test_templates")
    ) as client:
        response = client.get("/fixtures/complex_foreign_keys")
        assert response.status == 200
        assert (
            '<div class="custom-table-row">'
            '1 - 2 - <a href="/fixtures/simple_primary_key/1">hello</a> <em>1</em>'
            "</div>"
        ) == str(Soup(response.text, "html.parser").select_one("div.custom-table-row"))


@pytest.mark.asyncio
@pytest.mark.parametrize("json", (True, False))
@pytest.mark.parametrize(
    "params,error",
    (
        ("?_sort=bad", "Cannot sort table by bad"),
        ("?_sort_desc=bad", "Cannot sort table by bad"),
        (
            "?_sort=state&_sort_desc=state",
            "Cannot use _sort and _sort_desc at the same time",
        ),
    ),
)
async def test_sort_errors(ds_client, json, params, error):
    path = "/fixtures/facetable{}{}".format(
        ".json" if json else "",
        params,
    )
    response = await ds_client.get(path)
    assert response.status_code == 400
    if json:
        assert response.json() == {
            "ok": False,
            "error": error,
            "status": 400,
            "title": None,
        }
    else:
        assert error in response.text


@pytest.mark.asyncio
async def test_metadata_sort(ds_client):
    response = await ds_client.get("/fixtures/facet_cities")
    assert response.status_code == 200
    table = Soup(response.text, "html.parser").find("table")
    assert table["class"] == ["rows-and-columns"]
    ths = table.find_all("th")
    assert ["id", "name\xa0▼"] == [th.find("a").string.strip() for th in ths]
    rows = [[str(td) for td in tr.select("td")] for tr in table.select("tbody tr")]
    expected = [
        [
            '<td class="col-id type-pk"><a href="/fixtures/facet_cities/3">3</a></td>',
            '<td class="col-name type-str">Detroit</td>',
        ],
        [
            '<td class="col-id type-pk"><a href="/fixtures/facet_cities/2">2</a></td>',
            '<td class="col-name type-str">Los Angeles</td>',
        ],
        [
            '<td class="col-id type-pk"><a href="/fixtures/facet_cities/4">4</a></td>',
            '<td class="col-name type-str">Memnonia</td>',
        ],
        [
            '<td class="col-id type-pk"><a href="/fixtures/facet_cities/1">1</a></td>',
            '<td class="col-name type-str">San Francisco</td>',
        ],
    ]
    assert expected == rows
    # Make sure you can reverse that sort order
    response = await ds_client.get("/fixtures/facet_cities?_sort_desc=name")
    assert response.status_code == 200
    table = Soup(response.text, "html.parser").find("table")
    rows = [[str(td) for td in tr.select("td")] for tr in table.select("tbody tr")]
    assert list(reversed(expected)) == rows


@pytest.mark.asyncio
async def test_metadata_sort_desc(ds_client):
    response = await ds_client.get("/fixtures/attraction_characteristic")
    assert response.status_code == 200
    table = Soup(response.text, "html.parser").find("table")
    assert table["class"] == ["rows-and-columns"]
    ths = table.find_all("th")
    assert ["pk\xa0▲", "name"] == [th.find("a").string.strip() for th in ths]
    rows = [[str(td) for td in tr.select("td")] for tr in table.select("tbody tr")]
    expected = [
        [
            '<td class="col-pk type-pk"><a href="/fixtures/attraction_characteristic/2">2</a></td>',
            '<td class="col-name type-str">Paranormal</td>',
        ],
        [
            '<td class="col-pk type-pk"><a href="/fixtures/attraction_characteristic/1">1</a></td>',
            '<td class="col-name type-str">Museum</td>',
        ],
    ]
    assert expected == rows
    # Make sure you can reverse that sort order
    response = await ds_client.get("/fixtures/attraction_characteristic?_sort=pk")
    assert response.status_code == 200
    table = Soup(response.text, "html.parser").find("table")
    rows = [[str(td) for td in tr.select("td")] for tr in table.select("tbody tr")]
    assert list(reversed(expected)) == rows


@pytest.mark.parametrize(
    "max_returned_rows,path,expected_num_facets,expected_ellipses,expected_ellipses_url",
    (
        (
            5,
            # Default should show 2 facets
            "/fixtures/facetable?_facet=_neighborhood",
            2,
            True,
            "/fixtures/facetable?_facet=_neighborhood&_facet_size=max",
        ),
        # _facet_size above max_returned_rows should show max_returned_rows (5)
        (
            5,
            "/fixtures/facetable?_facet=_neighborhood&_facet_size=50",
            5,
            True,
            "/fixtures/facetable?_facet=_neighborhood&_facet_size=max",
        ),
        # If max_returned_rows is high enough, should return all
        (
            20,
            "/fixtures/facetable?_facet=_neighborhood&_facet_size=max",
            14,
            False,
            None,
        ),
        # If num facets > max_returned_rows, show ... without a link
        # _facet_size above max_returned_rows should show max_returned_rows (5)
        (
            5,
            "/fixtures/facetable?_facet=_neighborhood&_facet_size=max",
            5,
            True,
            None,
        ),
    ),
)
def test_facet_more_links(
    max_returned_rows,
    path,
    expected_num_facets,
    expected_ellipses,
    expected_ellipses_url,
):
    with make_app_client(
        settings={"max_returned_rows": max_returned_rows, "default_facet_size": 2}
    ) as client:
        response = client.get(path)
        soup = Soup(response.body, "html.parser")
        lis = soup.select("#facet-neighborhood-b352a7 ul li:not(.facet-truncated)")
        facet_truncated = soup.select_one(".facet-truncated")
        assert len(lis) == expected_num_facets
        if not expected_ellipses:
            assert facet_truncated is None
        else:
            if expected_ellipses_url:
                assert facet_truncated.find("a")["href"] == expected_ellipses_url
            else:
                assert facet_truncated.find("a") is None


def test_unavailable_table_does_not_break_sort_relationships():
    # https://github.com/simonw/datasette/issues/1305
    with make_app_client(
        config={
            "databases": {
                "fixtures": {"tables": {"foreign_key_references": {"allow": False}}}
            }
        }
    ) as client:
        response = client.get("/?_sort=relationships")
        assert response.status == 200


@pytest.mark.asyncio
async def test_column_metadata(ds_client):
    response = await ds_client.get("/fixtures/roadside_attractions")
    soup = Soup(response.text, "html.parser")
    dl = soup.find("dl")
    assert [(dt.text, dt.next_sibling.text) for dt in dl.find_all("dt")] == [
        ("address", "The street address for the attraction"),
        ("name", "The name of the attraction"),
    ]
    assert (
        soup.select("th[data-column=name]")[0]["data-column-description"]
        == "The name of the attraction"
    )
    assert (
        soup.select("th[data-column=address]")[0]["data-column-description"]
        == "The street address for the attraction"
    )


def test_facet_total():
    # https://github.com/simonw/datasette/issues/1423
    # https://github.com/simonw/datasette/issues/1556
    with make_app_client(settings={"max_returned_rows": 100}) as client:
        path = "/fixtures/sortable?_facet=content&_facet=pk1"
        response = client.get(path)
        assert response.status == 200
    fragments = (
        '<span class="facet-info-total">&gt;30</span>',
        '<span class="facet-info-total">8</span>',
    )
    for fragment in fragments:
        assert fragment in response.text


@pytest.mark.asyncio
async def test_sort_rowid_with_next(ds_client):
    # https://github.com/simonw/datasette/issues/1470
    response = await ds_client.get("/fixtures/binary_data?_size=1&_next=1&_sort=rowid")
    assert response.status_code == 200


def assert_querystring_equal(expected, actual):
    assert sorted(expected.split("&")) == sorted(actual.split("&"))


@pytest.mark.asyncio
@pytest.mark.parametrize(
    "path,expected",
    (
        (
            "/fixtures/facetable",
            "fixtures: facetable: 15 rows",
        ),
        (
            "/fixtures/facetable?on_earth__exact=1",
            "fixtures: facetable: 14 rows where on_earth = 1",
        ),
    ),
)
async def test_table_page_title(ds_client, path, expected):
    response = await ds_client.get(path)
    title = Soup(response.text, "html.parser").find("title").text
    assert title == expected


@pytest.mark.asyncio
async def test_table_post_method_not_allowed(ds_client):
    response = await ds_client.post("/fixtures/facetable")
    assert response.status_code == 405
    assert "Method not allowed" in response.text


@pytest.mark.parametrize("allow_facet", (True, False))
def test_allow_facet_off(allow_facet):
    with make_app_client(settings={"allow_facet": allow_facet}) as client:
        response = client.get("/fixtures/facetable")
        expected = "DATASETTE_ALLOW_FACET = {};".format(
            "true" if allow_facet else "false"
        )
        assert expected in response.text
        if allow_facet:
            assert "Suggested facets" in response.text
        else:
            assert "Suggested facets" not in response.text


@pytest.mark.asyncio
@pytest.mark.parametrize(
    "size,title,length_bytes",
    (
        (2000, ' title="2.0 KB"', "2,000"),
        (20000, ' title="19.5 KB"', "20,000"),
        (20, "", "20"),
    ),
)
async def test_format_of_binary_links(size, title, length_bytes):
    ds = Datasette()
    db_name = "binary-links-{}".format(size)
    db = ds.add_memory_database(db_name)
    sql = "select zeroblob({}) as blob".format(size)
    await db.execute_write("create table blobs as {}".format(sql))
    response = await ds.client.get("/{}/blobs".format(db_name))
    assert response.status_code == 200
    expected = "{}>&lt;Binary:&nbsp;{}&nbsp;bytes&gt;</a>".format(title, length_bytes)
    assert expected in response.text
    # And test with arbitrary SQL query too
    sql_response = await ds.client.get(
        "{}/-/query".format(db_name), params={"sql": sql}
    )
    assert sql_response.status_code == 200
    assert expected in sql_response.text


@pytest.mark.asyncio
@pytest.mark.parametrize(
    "config",
    (
        # Blocked at table level
        {
            "databases": {
                "foreign_key_labels": {
                    "tables": {
                        # Table a is only visible to root
                        "a": {"allow": {"id": "root"}},
                    }
                }
            }
        },
        # Blocked at database level
        {
            "databases": {
                "foreign_key_labels": {
                    # Only root can view this database
                    "allow": {"id": "root"},
                    "tables": {
                        # But table b is visible to everyone
                        "b": {"allow": True},
                    },
                }
            }
        },
        # Blocked at the instance level
        {
            "allow": {"id": "root"},
            "databases": {
                "foreign_key_labels": {
                    "tables": {
                        # Table b is visible to everyone
                        "b": {"allow": True},
                    }
                }
            },
        },
    ),
)
async def test_foreign_key_labels_obey_permissions(config):
    ds = Datasette(config=config)
    db = ds.add_memory_database("foreign_key_labels")
    await db.execute_write(
        "create table if not exists a(id integer primary key, name text)"
    )
    await db.execute_write("insert or replace into a (id, name) values (1, 'hello')")
    await db.execute_write(
        "create table if not exists b(id integer primary key, name text, a_id integer references a(id))"
    )
    await db.execute_write(
        "insert or replace into b (id, name, a_id) values (1, 'world', 1)"
    )
    # Anonymous user can see table b but not table a
    await ds.client.get("/foreign_key_labels.json")
    anon_a = await ds.client.get("/foreign_key_labels/a.json?_labels=on")
    assert anon_a.status_code == 403
    anon_b = await ds.client.get("/foreign_key_labels/b.json?_labels=on")
    assert anon_b.status_code == 200
    # root user can see both
    cookies = {"ds_actor": ds.sign({"a": {"id": "root"}}, "actor")}
    root_a = await ds.client.get(
        "/foreign_key_labels/a.json?_labels=on", cookies=cookies
    )
    assert root_a.status_code == 200
    root_b = await ds.client.get(
        "/foreign_key_labels/b.json?_labels=on", cookies=cookies
    )
    assert root_b.status_code == 200
    # Labels should have been expanded for root
    assert root_b.json() == {
        "ok": True,
        "next": None,
        "rows": [{"id": 1, "name": "world", "a_id": {"value": 1, "label": "hello"}}],
        "truncated": False,
    }
    # But not for anon
    assert anon_b.json() == {
        "ok": True,
        "next": None,
        "rows": [{"id": 1, "name": "world", "a_id": 1}],
        "truncated": False,
    }


def test_foreign_keys_special_character_in_database_name(app_client_with_dot):
    # https://github.com/simonw/datasette/pull/2476
    response = app_client_with_dot.get("/fixtures~2Edot/complex_foreign_keys")
    assert '<a href="/fixtures~2Edot/simple_primary_key/2">world</a>' in response.text