dogsheep-beta/tests/test_plugin.py
from datasette.app import Datasette
from bs4 import BeautifulSoup as Soup
from dogsheep_beta.cli import index
from dogsheep_beta.utils import parse_metadata
import textwrap
import sqlite_utils
import pytest
import urllib
import httpx
@pytest.mark.asyncio
async def test_search(ds):
async with httpx.AsyncClient(app=ds.app()) as client:
response = await client.get("http://localhost/-/beta")
assert 200 == response.status_code
assert '<input type="search" name="q" value="" id="q">' in response.text
response = await client.get("http://localhost/-/beta?q=things")
assert 200 == response.status_code
for fragment in (
"<p>Got 3 results",
"<p>Email from blah@example.com, subject Hey there",
"<p>Email from blah@example.com, subject What's going on",
"<p>Commit to dogsheep/dogsheep-beta on 2020-08-01T00:05:02",
'<p>User searched for: "things"</p>',
):
assert fragment in response.text
# Test facets
soup = Soup(response.text, "html5lib")
facet_els = soup.select(".facet")
facets = [
{
"name": el.find("h2").text,
"values": [
{
"selected": "selected" in li.get("class", ""),
"count": int(li.select(".count")[0].text),
"url": li.find("a")["href"],
"label": li.select(".label")[0].text,
}
for li in el.findAll("li")
],
}
for el in facet_els
]
assert facets == [
{
"name": "type",
"values": [
{
"selected": False,
"count": 2,
"url": "?type=emails.db%2Femails&q=things",
"label": "emails.db/emails",
},
{
"selected": False,
"count": 1,
"url": "?type=github.db%2Fcommits&q=things",
"label": "github.db/commits",
},
],
},
{
"name": "category",
"values": [
{
"selected": False,
"count": 1,
"url": "?category=1&q=things",
"label": "created",
}
],
},
{
"name": "is_public",
"values": [
{
"selected": False,
"count": 2,
"url": "?is_public=0&q=things",
"label": "0",
},
{
"selected": False,
"count": 1,
"url": "?is_public=1&q=things",
"label": "1",
},
],
},
{
"name": "timestamp",
"values": [
{
"selected": False,
"count": 2,
"url": "?timestamp__date=2020-08-01&q=things",
"label": "2020-08-01",
},
{
"selected": False,
"count": 1,
"url": "?timestamp__date=2020-08-02&q=things",
"label": "2020-08-02",
},
],
},
]
all_results = [
"github.db/commits:5becbf70d64951e2910314ef5227d19b11c25b0c9586934941366da8997e57cb",
"emails.db/emails:2",
"github.db/commits:a5b39c5049b28997528bb0eca52730ab6febabeaba54cfcba0ab5d70e7207523",
"emails.db/emails:1",
]
@pytest.mark.asyncio
@pytest.mark.parametrize(
"q,expected",
(
("", all_results),
(" ", all_results),
(
"things NOT email",
[
"github.db/commits:a5b39c5049b28997528bb0eca52730ab6febabeaba54cfcba0ab5d70e7207523"
],
),
(
"#dogfest",
["emails.db/emails:1"],
),
(
"github-to-sqlite",
[],
),
),
)
async def test_advanced_search(ds, q, expected):
async with httpx.AsyncClient(app=ds.app()) as client:
response = await client.get(
"http://localhost/-/beta?" + urllib.parse.urlencode({"q": q})
)
assert response.status_code == 200
soup = Soup(response.text, "html5lib")
results = [el["data-table-key"] for el in soup.select("[data-table-key]")]
assert results == expected
if expected:
# Check that facets exist on the page
assert len(soup.select(".facet li")), "Could not see any facet results"
@pytest.mark.asyncio
@pytest.mark.parametrize(
"sort,expected",
(
(
"",
["emails.db/emails:1", "emails.db/emails:2"],
),
(
"newest",
["emails.db/emails:2", "emails.db/emails:1"],
),
(
"oldest",
["emails.db/emails:1", "emails.db/emails:2"],
),
),
)
async def test_search_order(ds, sort, expected):
async with httpx.AsyncClient(app=ds.app()) as client:
q = "email"
response = await client.get(
"http://localhost/-/beta?" + urllib.parse.urlencode({"q": q, "sort": sort})
)
assert response.status_code == 200
soup = Soup(response.text, "html5lib")
results = [el["data-table-key"] for el in soup.select("[data-table-key]")]
assert results == expected
# Check that sort links exist and are correct
sort_label = sort or "relevance"
assert "<strong>{}</strong>".format(sort_label) in response.text
if sort_label != "relevance":
assert (
'<input type="hidden" name="sort" value="{}">'.format(sort_label)
in response.text
)
for sort_order in ("relevance", "newest", "oldest"):
if sort_order != sort_label:
assert ">{}</a>".format(sort_order) in response.text
ALL_EXPECTED = [
"github.db/commits:5becbf70d64951e2910314ef5227d19b11c25b0c9586934941366da8997e57cb",
"emails.db/emails:2",
"github.db/commits:a5b39c5049b28997528bb0eca52730ab6febabeaba54cfcba0ab5d70e7207523",
"emails.db/emails:1",
]
@pytest.mark.asyncio
@pytest.mark.parametrize(
"sort,expected",
(
("", ALL_EXPECTED),
("newest", ALL_EXPECTED),
("oldest", list(reversed(ALL_EXPECTED))),
),
)
async def test_search_order_for_timeline(ds, sort, expected):
async with httpx.AsyncClient(app=ds.app()) as client:
url = "http://localhost/-/beta"
if sort:
url += "?sort=" + sort
response = await client.get(url)
assert response.status_code == 200
soup = Soup(response.text, "html5lib")
results = [el["data-table-key"] for el in soup.select("[data-table-key]")]
assert results == expected
# Check that sort links exist and are correct
sort_label = sort or "newest"
assert "<strong>{}</strong>".format(sort_label) in response.text
assert ">relevance</a>" not in response.text
assert (
">{}</a>".format("oldest" if sort_label == "newest" else "newest")
in response.text
)
@pytest.mark.asyncio
async def test_fixture(ds):
async with httpx.AsyncClient(app=ds.app()) as client:
response = await client.get("http://localhost/-/databases.json")
assert 200 == response.status_code
assert {d["name"] for d in response.json()} == {"beta", "emails", "github"}
@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 "dogsheep-beta" in installed_plugins
@pytest.fixture
def ds(tmp_path_factory, monkeypatch):
db_directory = tmp_path_factory.mktemp("dbs")
monkeypatch.chdir(db_directory)
github_path = db_directory / "github.db"
emails_path = db_directory / "emails.db"
beta_path = db_directory / "beta.db"
beta_config_path = db_directory / "dogsheep-beta.yml"
beta_config_path.write_text(
textwrap.dedent(
"""
emails.db:
emails:
display_sql: |-
select * from emails where id = :key
display: |-
<p>Email from {{ display.from_ }}, subject {{ display.subject }}
sql: |-
select
id as key,
subject as title,
date as timestamp,
0 as is_public,
body as search_1
from
emails
github.db:
commits:
display_sql: |-
select
commits.sha,
commits.message,
commits.committer_date,
commits.repo_name,
:q as their_query
from commits where sha = :key
display: |-
<p>Commit to {{ display.repo_name }} on {{ display.committer_date }}</p>
<p>{{ display.message }} - {{ display.sha }}</p>
<p>User searched for: "{{ display.their_query }}"</p>
sql: |-
select
sha as key,
'Commit to ' || commits.repo_name as title,
committer_date as timestamp,
1 as category,
1 as is_public,
message as search_1
from
commits
""",
),
"utf-8",
)
METADATA = textwrap.dedent(
"""
plugins:
dogsheep-beta:
database: beta
config_file: dogsheep-beta.yml
"""
)
github_db = sqlite_utils.Database(github_path)
github_db["commits"].insert_all(
[
{
"sha": "a5b39c5049b28997528bb0eca52730ab6febabeaba54cfcba0ab5d70e7207523",
"message": "Another commit to things",
"repo_name": "dogsheep/dogsheep-beta",
"committer_date": "2020-08-01T00:05:02",
},
{
"sha": "5becbf70d64951e2910314ef5227d19b11c25b0c9586934941366da8997e57cb",
"message": "Added some tests",
"repo_name": "dogsheep/dogsheep-beta",
"committer_date": "2020-08-02T12:35:48",
},
],
pk="sha",
)
emails_db = sqlite_utils.Database(emails_path)
emails_db["emails"].insert_all(
[
{
"id": 1,
"subject": "Hey there #dogfest",
"body": "An email about things",
"from_": "blah@example.com",
"date": "2020-08-01T00:05:02",
},
{
"id": 2,
"subject": "What's going on",
"body": "Another email about things",
"from_": "blah@example.com",
"date": "2020-08-02T00:05:02",
},
],
pk="id",
)
index.callback(beta_path, beta_config_path, None, [])
ds = Datasette(
[str(beta_path), str(github_path), str(emails_path)],
metadata=parse_metadata(METADATA),
)
return ds