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():
    app = Datasette([], memory=True).app()
    async with httpx.AsyncClient(app=app) as client:
        response = await client.get("http://localhost/-/plugins.json")
        assert 200 == response.status_code
        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"},
    )
 
    datasette = Datasette([db_path])
    cookies = {"ds_actor": datasette.sign({"a": {"id": "root"}}, "actor")}
    async with httpx.AsyncClient(app=datasette.app()) as client:
        response = await client.get("http://localhost/-/import-table", cookies=cookies)
        assert 200 == response.status_code
        csrftoken = response.cookies["ds_csrftoken"]
        cookies["ds_csrftoken"] = csrftoken
        response = await client.post(
            "http://localhost/-/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")
    datasette = Datasette([db_path1, db_path2])
    cookies = {"ds_actor": datasette.sign({"a": {"id": "root"}}, "actor")}
    async with httpx.AsyncClient(app=datasette.app()) as client:
        response = await client.get("http://localhost/-/import-table", cookies=cookies)
        assert response.status_code == 200
        assert "<option>test</option>" in response.text
        assert "<option>test2</option>" in response.text
        response2 = await client.get(
            "http://localhost/-/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])
    app = ds.app()
    async with httpx.AsyncClient(app=app) as client:
        response = await client.get("http://localhost/-/import-table")
        assert 403 == response.status_code
    # Now try with a root actor
    async with httpx.AsyncClient(app=app) as client2:
        response2 = await client2.get(
            "http://localhost/-/import-table",
            cookies={"ds_actor": ds.sign({"a": {"id": "root"}}, "actor")},
        )
        assert 403 != response2.status_code