ticketfrei3/backend/tests/conftest.py

# 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
    )