datasette-import-table/tests/test_import_table.py

from datasette.app import Datasette
import datasette_import_table
import pytest
import sqlite_utils
import json
import httpx
 
 
@pytest.fixture
def non_mocked_hosts():
    return ["localhost"]
 
 
@pytest.mark.asyncio
async def test_plugin_is_installed():
    ds = Datasette([], memory=True)
    response = await ds.client.get("/-/plugins.json")
    assert response.status_code == 200
    installed_plugins = {p["name"] for p in response.json()}
    assert "datasette-import-table" in installed_plugins
 
 
@pytest.mark.asyncio
async def test_import_table(tmpdir, httpx_mock):
    db_path = str(tmpdir / "test.db")
    httpx_mock.add_response(
        url="http://example/some/table.json?_shape=objects&_size=max",
        json={
            "table": "mytable",
            "rows": [{"foo": "bar"}],
            "primary_keys": [],
            "filtered_table_rows_count": 1,
            "next_url": None,
        },
        headers={"content-type": "application/json"},
    )
 
    ds = Datasette([db_path])
    cookies = {"ds_actor": ds.sign({"a": {"id": "root"}}, "actor")}
    response = await ds.client.get("/-/import-table", cookies=cookies)
    assert response.status_code == 200
    csrftoken = response.cookies["ds_csrftoken"]
    cookies["ds_csrftoken"] = csrftoken
    response = await ds.client.post(
        "/-/import-table",
        data={
            "url": "http://example/some/table",
            "csrftoken": csrftoken,
        },
        cookies=cookies,
    )
    assert response.status_code == 302
    assert response.headers["location"] == "/test/mytable?_import_expected_rows=1"
 
 
@pytest.mark.asyncio
async def test_import_table_multiple_databases(tmpdir):
    db_path1 = str(tmpdir / "test.db")
    db_path2 = str(tmpdir / "test2.db")
    ds = Datasette([db_path1, db_path2])
    cookies = {"ds_actor": ds.sign({"a": {"id": "root"}}, "actor")}
    response = await ds.client.get("/-/import-table", cookies=cookies)
    assert response.status_code == 200
    assert '<select name="database">' in response.text
    assert "<option>test</option>" in response.text
    assert "<option>test2</option>" in response.text
    response2 = await ds.client.get("/-/import-table?database=test2", cookies=cookies)
    assert response2.status_code == 200
    assert '<option selected="selected">test2</option>' in response2.text
 
 
@pytest.mark.asyncio
async def test_permissions(tmpdir):
    path = str(tmpdir / "test.db")
    ds = Datasette([path])
    response = await ds.client.get("/-/import-table")
    assert response.status_code == 403
    # Now try with a root actor
    response2 = await ds.client.get(
        "/-/import-table",
        cookies={"ds_actor": ds.sign({"a": {"id": "root"}}, "actor")},
    )
    assert response2.status_code != 403