datasette-init/tests/test_init.py

from datasette.app import Datasette
import pytest
import sqlite_utils
import httpx
 
 
def build_datasette(tmp_path_factory, init={}, db_init=None):
    db_directory = tmp_path_factory.mktemp("dbs")
    db_path = db_directory / "test.db"
    db = sqlite_utils.Database(db_path)
    if db_init:
        db_init(db)
    db.vacuum()
    ds = Datasette([db_path], metadata={"plugins": {"datasette-init": {"test": init}}})
    return ds
 
 
@pytest.mark.asyncio
async def test_tables(tmp_path_factory):
    ds = build_datasette(
        tmp_path_factory,
        {
            "tables": {
                "dogs": {
                    "columns": {
                        "id": "integer",
                        "name": "text",
                        "age": "integer",
                        "weight": "float",
                    },
                    "pk": "id",
                }
            }
        },
    )
    await ds.invoke_startup()
    async with httpx.AsyncClient(app=ds.app()) as client:
        response = await client.get("http://localhost/test/dogs.json")
        assert 200 == response.status_code
        data = response.json()
        assert ["id", "name", "age", "weight"] == data["columns"]
        assert ["id"] == data["primary_keys"]
 
 
@pytest.mark.asyncio
async def test_ignore_if_table_already_exists(tmp_path_factory):
    ds = build_datasette(
        tmp_path_factory,
        {"tables": {"dogs": {"columns": {"id": "integer"}, "pk": "id",}}},
        db_init=lambda db: db["dogs"].create({"id": int}),
    )
    await ds.invoke_startup()
 
 
@pytest.mark.asyncio
async def test_tables_compound_primary_key(tmp_path_factory):
    ds = build_datasette(
        tmp_path_factory,
        {
            "tables": {
                "dogs": {
                    "columns": {"id1": "integer", "id2": "integer", "name": "text",},
                    "pk": ["id1", "id2"],
                }
            }
        },
    )
    await ds.invoke_startup()
    async with httpx.AsyncClient(app=ds.app()) as client:
        response = await client.get("http://localhost/test/dogs.json")
        assert 200 == response.status_code
        data = response.json()
        assert ["id1", "id2", "name"] == data["columns"]
        assert ["id1", "id2"] == data["primary_keys"]
 
 
@pytest.mark.asyncio
async def test_views(tmp_path_factory):
    ds = build_datasette(tmp_path_factory, {"views": {"two": "select 1 + 1"}})
    await ds.invoke_startup()
    async with httpx.AsyncClient(app=ds.app()) as client:
        response = await client.get("http://localhost/test/two.json?_shape=array")
        assert 200 == response.status_code
        assert [{"1 + 1": 2}] == response.json()
 
 
@pytest.mark.asyncio
async def test_replace_view_if_already_exists(tmp_path_factory):
    ds = build_datasette(
        tmp_path_factory,
        {"views": {"two": "select 1 + 1"}},
        db_init=lambda db: db.create_view("two", "select 1 + 3"),
    )
    async with httpx.AsyncClient(app=ds.app()) as client:
        response = await client.get("http://localhost/test/two.json?_shape=array")
        assert [{"1 + 3": 4}] == response.json()
        await ds.invoke_startup()
        response2 = await client.get("http://localhost/test/two.json?_shape=array")
        assert [{"1 + 1": 2}] == response2.json()
 
 
@pytest.mark.asyncio
async def test_no_error_if_no_metadata():
    ds = Datasette([], memory=True)
    await ds.invoke_startup()
    async with httpx.AsyncClient(app=ds.app()) as client:
        assert 200 == (await client.get("http://localhost/")).status_code