157 lines
4.6 KiB
Python
157 lines
4.6 KiB
Python
# Copyright (C) 2020, 2023 by Thomas Lindner <tom@dl6tom.de>
|
|
# Copyright (C) 2020 by Christian Hagenest <c.hagenest@pm.me>
|
|
# Copyright (C) 2020 by Martin Rey <martin.rey@mailbox.org>
|
|
#
|
|
# SPDX-License-Identifier: 0BSD
|
|
|
|
from urllib.parse import urlparse
|
|
|
|
from fastapi import FastAPI, status
|
|
from httpx import AsyncClient
|
|
import pytest
|
|
from tortoise import Tortoise
|
|
|
|
from kibicara import email
|
|
from kibicara.webapi import router
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def anyio_backend():
|
|
return "asyncio"
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
@pytest.mark.anyio
|
|
async def client():
|
|
await Tortoise.init(
|
|
db_url="sqlite://:memory:",
|
|
modules={
|
|
"models": [
|
|
"kibicara.model",
|
|
"kibicara.platforms.email.model",
|
|
"kibicara.platforms.mastodon.model",
|
|
"kibicara.platforms.telegram.model",
|
|
"kibicara.platforms.test.model",
|
|
"kibicara.platforms.twitter.model",
|
|
]
|
|
},
|
|
)
|
|
await Tortoise.generate_schemas()
|
|
app = FastAPI()
|
|
app.include_router(router, prefix="/api")
|
|
yield AsyncClient(app=app, base_url="http://test")
|
|
await Tortoise.close_connections()
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def monkeymodule():
|
|
from _pytest.monkeypatch import MonkeyPatch
|
|
|
|
mpatch = MonkeyPatch()
|
|
yield mpatch
|
|
mpatch.undo()
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def receive_email(monkeymodule):
|
|
mailbox = []
|
|
|
|
def mock_send_email(to, subject, sender="kibicara", body=""):
|
|
mailbox.append(dict(to=to, subject=subject, sender=sender, body=body))
|
|
|
|
def mock_receive_email():
|
|
return mailbox.pop()
|
|
|
|
monkeymodule.setattr(email, "send_email", mock_send_email)
|
|
return mock_receive_email
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
@pytest.mark.anyio
|
|
async def register_token(client, receive_email):
|
|
response = await client.post(
|
|
"/api/admin/register/", json={"email": "user", "password": "password"}
|
|
)
|
|
assert response.status_code == status.HTTP_202_ACCEPTED
|
|
return urlparse(receive_email()["body"]).query.split("=", 1)[1]
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
@pytest.mark.anyio
|
|
async def register_confirmed(client, register_token):
|
|
response = await client.post("/api/admin/confirm/{0}".format(register_token))
|
|
assert response.status_code == status.HTTP_200_OK
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
@pytest.mark.anyio
|
|
async def access_token(client, register_confirmed):
|
|
response = await client.post(
|
|
"/api/admin/login/", data={"username": "user", "password": "password"}
|
|
)
|
|
assert response.status_code == status.HTTP_200_OK
|
|
return response.json()["access_token"]
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def auth_header(access_token):
|
|
return {"Authorization": "Bearer {0}".format(access_token)}
|
|
|
|
|
|
@pytest.fixture(scope="function")
|
|
@pytest.mark.anyio
|
|
async def hood_id(client, auth_header):
|
|
response = await client.post(
|
|
"/api/hoods/", json={"name": "hood"}, headers=auth_header
|
|
)
|
|
assert response.status_code == status.HTTP_201_CREATED
|
|
hood_id = int(response.headers["Location"])
|
|
yield hood_id
|
|
await client.delete("/api/hoods/{0}".format(hood_id), headers=auth_header)
|
|
|
|
|
|
@pytest.fixture(scope="function")
|
|
@pytest.mark.anyio
|
|
async def trigger_id(client, hood_id, auth_header):
|
|
response = await client.post(
|
|
"/api/hoods/{0}/triggers/".format(hood_id),
|
|
json={"pattern": "test"},
|
|
headers=auth_header,
|
|
)
|
|
assert response.status_code == status.HTTP_201_CREATED
|
|
trigger_id = int(response.headers["Location"])
|
|
yield trigger_id
|
|
await client.delete(
|
|
"/api/hoods/{0}/triggers/{1}".format(hood_id, trigger_id), headers=auth_header
|
|
)
|
|
|
|
|
|
@pytest.fixture(scope="function")
|
|
@pytest.mark.anyio
|
|
async def badword_id(client, hood_id, auth_header):
|
|
response = await client.post(
|
|
"/api/hoods/{0}/badwords/".format(hood_id),
|
|
json={"pattern": ""},
|
|
headers=auth_header,
|
|
)
|
|
assert response.status_code == status.HTTP_201_CREATED
|
|
badword_id = int(response.headers["Location"])
|
|
yield badword_id
|
|
await client.delete(
|
|
"/api/hoods/{0}/badwords/{1}".format(hood_id, badword_id), headers=auth_header
|
|
)
|
|
|
|
|
|
@pytest.fixture(scope="function")
|
|
@pytest.mark.anyio
|
|
async def test_id(client, hood_id, auth_header):
|
|
response = await client.post(
|
|
"/api/hoods/{0}/test/".format(hood_id), json={}, headers=auth_header
|
|
)
|
|
assert response.status_code == status.HTTP_201_CREATED
|
|
test_id = int(response.headers["Location"])
|
|
yield test_id
|
|
await client.delete(
|
|
"/api/hoods/{0}/test/{1}".format(hood_id, test_id), headers=auth_header
|
|
)
|