ticketfrei/ticketfrei3
8
1
Fork 0
Code Issues 11 Pull requests 2 Projects 1 Releases Wiki Activity

Port tests to async #25

Merged
vmann merged 8 commits from async-pytest-23 into development 2023-04-01 13:57:15 +00:00
Conversation 0 Commits 8 Files changed 16 +332 -253
16 changed files with 332 additions and 253 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

78
backend/tests/conftest.py
View file

@ -7,24 +7,29 @@
from urllib.parse import urlparse
from fastapi import FastAPI, status
from fastapi.testclient import TestClient
from pytest import fixture
from httpx import AsyncClient
import pytest
from kibicara import email
from kibicara.model import Mapping
from kibicara.webapi import router
@fixture(scope="module")
@pytest.fixture(scope="module")
def anyio_backend():
return "asyncio"
@pytest.fixture(scope="module")
def client():
Mapping.drop_all()
Mapping.create_all()
app = FastAPI()
app.include_router(router, prefix="/api")
return TestClient(app)
return AsyncClient(app=app, base_url="http://test")
@fixture(scope="module")
@pytest.fixture(scope="module")
def monkeymodule():
from _pytest.monkeypatch import MonkeyPatch
@ -33,7 +38,7 @@ def monkeymodule():
mpatch.undo()
@fixture(scope="module")
@pytest.fixture(scope="module")
def receive_email(monkeymodule):
mailbox = []
@ -47,47 +52,54 @@ def receive_email(monkeymodule):
return mock_receive_email
@fixture(scope="module")
def register_token(client, receive_email):
response = client.post(
@pytest.fixture(scope="module")
@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]
@fixture(scope="module")
def register_confirmed(client, register_token):
response = client.post("/api/admin/confirm/{0}".format(register_token))
@pytest.fixture(scope="module")
@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
@fixture(scope="module")
def access_token(client, register_confirmed):
response = client.post(
@pytest.fixture(scope="module")
@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"]
@fixture(scope="module")
@pytest.fixture(scope="module")
def auth_header(access_token):
return {"Authorization": "Bearer {0}".format(access_token)}
@fixture(scope="function")
def hood_id(client, auth_header):
response = client.post("/api/hoods/", json={"name": "hood"}, headers=auth_header)
@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
client.delete("/api/hoods/{0}".format(hood_id), headers=auth_header)
await client.delete("/api/hoods/{0}".format(hood_id), headers=auth_header)
@fixture(scope="function")
def trigger_id(client, hood_id, auth_header):
response = client.post(
@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,
@ -95,14 +107,15 @@ def trigger_id(client, hood_id, auth_header):
assert response.status_code == status.HTTP_201_CREATED
trigger_id = int(response.headers["Location"])
yield trigger_id
client.delete(
await client.delete(
"/api/hoods/{0}/triggers/{1}".format(hood_id, trigger_id), headers=auth_header
)
@fixture(scope="function")
def badword_id(client, hood_id, auth_header):
response = client.post(
@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,
@ -110,19 +123,20 @@ def badword_id(client, hood_id, auth_header):
assert response.status_code == status.HTTP_201_CREATED
badword_id = int(response.headers["Location"])
yield badword_id
client.delete(
await client.delete(
"/api/hoods/{0}/badwords/{1}".format(hood_id, badword_id), headers=auth_header
)
@fixture(scope="function")
def test_id(client, hood_id, auth_header):
response = client.post(
@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
client.delete(
await client.delete(
"/api/hoods/{0}/test/{1}".format(hood_id, test_id), headers=auth_header
)

11
backend/tests/tests_core/test_api_admin.py
View file

@ -3,13 +3,16 @@
# SPDX-License-Identifier: 0BSD
from fastapi import status
import pytest
def test_hoods_unauthorized(client):
response = client.get("/api/admin/hoods/")
@pytest.mark.anyio
async def test_hoods_unauthorized(client):
response = await client.get("/api/admin/hoods/")
assert response.status_code == status.HTTP_401_UNAUTHORIZED
def test_hoods_success(client, auth_header):
response = client.get("/api/admin/hoods/", headers=auth_header)
@pytest.mark.anyio
async def test_hoods_success(client, auth_header):
response = await client.get("/api/admin/hoods/", headers=auth_header)
assert response.status_code == status.HTTP_200_OK

89
backend/tests/tests_core/test_api_hoods.py
View file

@ -3,80 +3,107 @@
# Copyright (C) 2020 by Martin Rey <martin.rey@mailbox.org>
#
# SPDX-License-Identifier: 0BSD
import pytest
from fastapi import status
def test_hood_read_all(client):
response = client.get("/api/hoods/")
@pytest.mark.anyio
async def test_hood_read_all(client):
response = await client.get("/api/hoods/")
assert response.status_code == status.HTTP_200_OK
def test_hood_create_unauthorized(client, hood_id):
response = client.post("/api/hoods/")
@pytest.mark.anyio
async def test_hood_create_unauthorized(client, hood_id):
response = await client.post("/api/hoods/")
assert response.status_code == status.HTTP_401_UNAUTHORIZED
def test_hood_read(client, hood_id):
response = client.get("/api/hoods/{0}".format(hood_id))
@pytest.mark.anyio
async def test_hood_read(client, hood_id):
response = await client.get("/api/hoods/{0}".format(hood_id))
assert response.status_code == status.HTTP_200_OK
def test_hood_update_unauthorized(client, hood_id):
response = client.put("/api/hoods/{0}".format(hood_id))
@pytest.mark.anyio
async def test_hood_update_unauthorized(client, hood_id):
response = await client.put("/api/hoods/{0}".format(hood_id))
assert response.status_code == status.HTTP_401_UNAUTHORIZED
def test_hood_delete_unauthorized(client, hood_id):
response = client.delete("/api/hoods/{0}".format(hood_id))
@pytest.mark.anyio
async def test_hood_delete_unauthorized(client, hood_id):
response = await client.delete("/api/hoods/{0}".format(hood_id))
assert response.status_code == status.HTTP_401_UNAUTHORIZED
def test_trigger_read_all_unauthorized(client, hood_id):
response = client.get("/api/hoods/{0}/triggers/".format(hood_id))
@pytest.mark.anyio
async def test_trigger_read_all_unauthorized(client, hood_id):
response = await client.get("/api/hoods/{0}/triggers/".format(hood_id))
assert response.status_code == status.HTTP_401_UNAUTHORIZED
def test_trigger_create_unauthorized(client, hood_id):
response = client.post("/api/hoods/{0}/triggers/".format(hood_id))
@pytest.mark.anyio
async def test_trigger_create_unauthorized(client, hood_id):
response = await client.post("/api/hoods/{0}/triggers/".format(hood_id))
assert response.status_code == status.HTTP_401_UNAUTHORIZED
def test_trigger_read_unauthorized(client, hood_id, trigger_id):
response = client.get("/api/hoods/{0}/triggers/{1}".format(hood_id, trigger_id))
@pytest.mark.anyio
async def test_trigger_read_unauthorized(client, hood_id, trigger_id):
response = await client.get(
"/api/hoods/{0}/triggers/{1}".format(hood_id, trigger_id)
)
assert response.status_code == status.HTTP_401_UNAUTHORIZED
def test_trigger_update_unauthorized(client, hood_id, trigger_id):
response = client.put("/api/hoods/{0}/triggers/{1}".format(hood_id, trigger_id))
@pytest.mark.anyio
async def test_trigger_update_unauthorized(client, hood_id, trigger_id):
response = await client.put(
"/api/hoods/{0}/triggers/{1}".format(hood_id, trigger_id)
)
assert response.status_code == status.HTTP_401_UNAUTHORIZED
def test_trigger_delete_unauthorized(client, hood_id, trigger_id):
response = client.delete("/api/hoods/{0}/triggers/{1}".format(hood_id, trigger_id))
@pytest.mark.anyio
async def test_trigger_delete_unauthorized(client, hood_id, trigger_id):
response = await client.delete(
"/api/hoods/{0}/triggers/{1}".format(hood_id, trigger_id)
)
assert response.status_code == status.HTTP_401_UNAUTHORIZED
def test_badword_read_all_unauthorized(client, hood_id):
response = client.get("/api/hoods/{0}/badwords/".format(hood_id))
@pytest.mark.anyio
async def test_badword_read_all_unauthorized(client, hood_id):
response = await client.get("/api/hoods/{0}/badwords/".format(hood_id))
assert response.status_code == status.HTTP_401_UNAUTHORIZED
def test_badword_create_unauthorized(client, hood_id):
response = client.post("/api/hoods/{0}/badwords/".format(hood_id))
@pytest.mark.anyio
async def test_badword_create_unauthorized(client, hood_id):
response = await client.post("/api/hoods/{0}/badwords/".format(hood_id))
assert response.status_code == status.HTTP_401_UNAUTHORIZED
def test_badword_read_unauthorized(client, hood_id, badword_id):
response = client.get("/api/hoods/{0}/badwords/{1}".format(hood_id, badword_id))
@pytest.mark.anyio
async def test_badword_read_unauthorized(client, hood_id, badword_id):
response = await client.get(
"/api/hoods/{0}/badwords/{1}".format(hood_id, badword_id)
)
assert response.status_code == status.HTTP_401_UNAUTHORIZED
def test_badword_update_unauthorized(client, hood_id, badword_id):
response = client.put("/api/hoods/{0}/badwords/{1}".format(hood_id, badword_id))
@pytest.mark.anyio
async def test_badword_update_unauthorized(client, hood_id, badword_id):
response = await client.put(
"/api/hoods/{0}/badwords/{1}".format(hood_id, badword_id)
)
assert response.status_code == status.HTTP_401_UNAUTHORIZED
def test_badword_delete_unauthorized(client, hood_id, badword_id):
response = client.delete("/api/hoods/{0}/badwords/{1}".format(hood_id, badword_id))
@pytest.mark.anyio
async def test_badword_delete_unauthorized(client, hood_id, badword_id):
response = await client.delete(
"/api/hoods/{0}/badwords/{1}".format(hood_id, badword_id)
)
assert response.status_code == status.HTTP_401_UNAUTHORIZED

11
backend/tests/tests_email/conftest.py
View file

@ -6,12 +6,13 @@
from fastapi import status
from pytest import fixture
import pytest
@fixture(scope="function")
def email_row(client, hood_id, auth_header):
response = client.post(
@pytest.fixture(scope="function")
@pytest.mark.anyio
async def email_row(client, hood_id, auth_header):
response = await client.post(
"/api/hoods/{0}/email/".format(hood_id),
json={"name": "kibicara-test"},
headers=auth_header,
@ -19,6 +20,6 @@ def email_row(client, hood_id, auth_header):
assert response.status_code == status.HTTP_201_CREATED
email_id = int(response.headers["Location"])
yield response.json()
client.delete(
await client.delete(
"/api/hoods/{0}/email/{1}".format(hood_id, email_id), headers=auth_header
)

25
backend/tests/tests_email/test_api_email_happy_path.py
View file

@ -8,13 +8,14 @@ from re import findall
from urllib.parse import urlparse
from fastapi import status
from pytest import skip
import pytest
from kibicara.webapi.admin import to_token
def test_email_subscribe_unsubscribe(client, hood_id, receive_email):
response = client.post(
@pytest.mark.anyio
async def test_email_subscribe_unsubscribe(client, hood_id, receive_email):
response = await client.post(
"/api/hoods/{0}/email/subscribe/".format(hood_id),
json={"email": "test@localhost"},
)
@ -27,37 +28,41 @@ def test_email_subscribe_unsubscribe(client, hood_id, receive_email):
body,
)[0]
start = len("token=")
response = client.post(
response = await client.post(
"/api/hoods/{0}/email/subscribe/confirm/{1}".format(
hood_id, urlparse(confirm_url).query[start:]
)
)
assert response.status_code == status.HTTP_201_CREATED
response = client.post(
response = await client.post(
"/api/hoods/{0}/email/subscribe/confirm/{1}".format(
hood_id, urlparse(confirm_url).query[start:]
)
)
assert response.status_code == status.HTTP_409_CONFLICT
token = to_token(email=mail["to"], hood=hood_id)
response = client.delete(
response = await client.delete(
"/api/hoods/{0}/email/unsubscribe/{1}".format(hood_id, token)
)
assert response.status_code == status.HTTP_204_NO_CONTENT
def test_email_message(client, hood_id, trigger_id, email_row):
@pytest.mark.anyio
async def test_email_message(client, hood_id, trigger_id, email_row):
body = {
"text": "test",
"author": "test@localhost",
"secret": email_row["secret"],
}
response = client.post("/api/hoods/{0}/email/messages/".format(hood_id), json=body)
response = await client.post(
"/api/hoods/{0}/email/messages/".format(hood_id), json=body
)
assert response.status_code == status.HTTP_201_CREATED
def test_email_send_mda(trigger_id, email_row):
skip("Only works if kibicara is listening on port 8000, and only sometimes")
@pytest.mark.anyio
async def test_email_send_mda(trigger_id, email_row):
pytest.skip("Only works if kibicara is listening on port 8000, and only sometimes")
mail = """From test@example.com Tue Jun 16 15:33:19 2020
Return-path: <test@example.com>
Envelope-to: hood@localhost

18
backend/tests/tests_email/test_api_email_unauthorized.py
View file

@ -5,21 +5,27 @@
# SPDX-License-Identifier: 0BSD
from fastapi import status
import pytest
def test_email_create_unauthorized(client, hood_id):
response = client.post("/api/hoods/{0}/email/".format(hood_id))
@pytest.mark.anyio
async def test_email_create_unauthorized(client, hood_id):
response = await client.post("/api/hoods/{0}/email/".format(hood_id))
assert response.status_code == status.HTTP_401_UNAUTHORIZED
def test_email_delete_unauthorized(client, hood_id, email_row):
response = client.delete(
@pytest.mark.anyio
async def test_email_delete_unauthorized(client, hood_id, email_row):
response = await client.delete(
"/api/hoods/{0}/email/{1}".format(hood_id, email_row["id"])
)
assert response.status_code == status.HTTP_401_UNAUTHORIZED
def test_email_message_unauthorized(client, hood_id, email_row):
@pytest.mark.anyio
async def test_email_message_unauthorized(client, hood_id, email_row):
body = {"text": "test", "author": "author", "secret": "wrong"}
response = client.post("/api/hoods/{0}/email/messages/".format(hood_id), json=body)
response = await client.post(
"/api/hoods/{0}/email/messages/".format(hood_id), json=body
)
assert response.status_code == status.HTTP_401_UNAUTHORIZED

33
backend/tests/tests_email/test_api_email_wrong.py
View file

@ -5,16 +5,19 @@
from fastapi import status
from nacl.exceptions import CryptoError
import pytest
def test_email_subscribe_empty(client, hood_id):
response = client.post("/api/hoods/{0}/email/subscribe/".format(hood_id))
@pytest.mark.anyio
async def test_email_subscribe_empty(client, hood_id):
response = await client.post("/api/hoods/{0}/email/subscribe/".format(hood_id))
assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
def test_email_subscribe_confirm_wrong_token(client, hood_id):
@pytest.mark.anyio
async def test_email_subscribe_confirm_wrong_token(client, hood_id):
try:
response = client.post(
response = await client.post(
"/api/hoods/{0}/email/subscribe/confirm/".format(hood_id)
+ "asdfasdfasdfasdfasdfasdfasdfasdf"
)
@ -23,26 +26,31 @@ def test_email_subscribe_confirm_wrong_token(client, hood_id):
pass
def test_email_subscribe_confirm_wrong_hood(client):
response = client.delete(
@pytest.mark.anyio
async def test_email_subscribe_confirm_wrong_hood(client):
response = await client.delete(
"/api/hoods/99999/email/unsubscribe/asdfasdfasdfasdfasdfasdfasdfasdf"
)
assert response.json()["detail"] == "Not Found"
def test_email_message_wrong(client, hood_id, email_row):
@pytest.mark.anyio
async def test_email_message_wrong(client, hood_id, email_row):
body = {
"text": "",
"author": "test@localhost",
"secret": email_row["secret"],
}
response = client.post("/api/hoods/{0}/email/messages/".format(hood_id), json=body)
response = await client.post(
"/api/hoods/{0}/email/messages/".format(hood_id), json=body
)
assert response.status_code == status.HTTP_451_UNAVAILABLE_FOR_LEGAL_REASONS
def test_email_unsubscribe_wrong_token(client, hood_id):
@pytest.mark.anyio
async def test_email_unsubscribe_wrong_token(client, hood_id):
try:
client.delete(
await client.delete(
"/api/hoods/{0}/email/unsubscribe/asdfasdfasdfasdfasdfasdfasdfasdf".format(
hood_id
)
@ -51,8 +59,9 @@ def test_email_unsubscribe_wrong_token(client, hood_id):
pass
def test_email_unsubscribe_wrong_hood(client):
response = client.delete(
@pytest.mark.anyio
async def test_email_unsubscribe_wrong_hood(client):
response = await client.delete(
"/api/hoods/99999/email/unsubscribe/asdfasdfasdfasdfasdfasdfasdfasdf"
)
assert response.json()["detail"] == "Not Found"

38
backend/tests/tests_mastodon/conftest.py
View file

@ -3,32 +3,30 @@
#
# SPDX-License-Identifier: 0BSD
from pytest import fixture
import pytest
from kibicara.model import Hood
from kibicara.platforms.mastodon.model import MastodonAccount, MastodonInstance
@fixture(scope="function")
def mastodon_instance(event_loop):
return event_loop.run_until_complete(
MastodonInstance.objects.create(
name="inst4nce",
client_id="cl13nt_id",
client_secret="cl13nt_s3cr3t",
)
@pytest.fixture(scope="function")
@pytest.mark.anyio
async def mastodon_instance():
return await MastodonInstance.objects.create(
name="inst4nce",
client_id="cl13nt_id",
client_secret="cl13nt_s3cr3t",
)
@fixture(scope="function")
def mastodon_account(event_loop, hood_id, mastodon_instance):
hood = event_loop.run_until_complete(Hood.objects.get(id=hood_id))
return event_loop.run_until_complete(
MastodonAccount.objects.create(
hood=hood,
instance=mastodon_instance,
access_token="t0k3n",
enabled=True,
username="us3r",
)
@pytest.fixture(scope="function")
@pytest.mark.anyio
async def mastodon_account(hood_id, mastodon_instance):
hood = await Hood.objects.get(id=hood_id)
return await MastodonAccount.objects.create(
hood=hood,
instance=mastodon_instance,
access_token="t0k3n",
enabled=True,
username="us3r",
)

38
backend/tests/tests_mastodon/test_api_mastodon_create_bot.py
View file

@ -4,14 +4,14 @@
# SPDX-License-Identifier: 0BSD
from fastapi import status
from pytest import fixture, mark
import pytest
from mastodon.Mastodon import Mastodon
from kibicara.platforms import mastodon
from kibicara.platforms.mastodon.model import MastodonAccount, MastodonInstance
@fixture(scope="function")
@pytest.fixture(scope="function")
def disable_spawner(monkeypatch):
class DoNothing:
def start(self, bot):
@ -20,7 +20,7 @@ def disable_spawner(monkeypatch):
monkeypatch.setattr(mastodon.webapi, "spawner", DoNothing())
@mark.parametrize(
@pytest.mark.parametrize(
"body",
[
{
@ -30,8 +30,8 @@ def disable_spawner(monkeypatch):
}
],
)
def test_mastodon_create_bot(
event_loop,
@pytest.mark.anyio
async def test_mastodon_create_bot(
client,
disable_spawner,
hood_id,
@ -44,7 +44,7 @@ def test_mastodon_create_bot(
monkeypatch.setattr(Mastodon, "log_in", log_in_mock)
response = client.post(
response = await client.post(
"/api/hoods/{0}/mastodon/".format(hood_id),
json=body,
headers=auth_header,
@ -52,11 +52,9 @@ def test_mastodon_create_bot(
print(response.json())
assert response.status_code == status.HTTP_201_CREATED
bot_id = response.json()["id"]
mastodon_obj = event_loop.run_until_complete(MastodonAccount.objects.get(id=bot_id))
mastodon_obj = await MastodonAccount.objects.get(id=bot_id)
assert response.json()["access_token"] == mastodon_obj.access_token
mastodon_instance = event_loop.run_until_complete(
MastodonInstance.objects.get(id=mastodon_obj.instance.id)
)
mastodon_instance = await MastodonInstance.objects.get(id=mastodon_obj.instance.id)
assert (
response.json()["instance"]["name"]
== body["instance_url"]
@ -70,15 +68,15 @@ def test_mastodon_create_bot(
assert mastodon_obj.enabled
@mark.parametrize(
@pytest.mark.parametrize(
"body",
[
{"instance_url": "botsin.space", "email": "notanemail", "password": "asdf1234"},
{"instance_url": "wrong", "email": "asdf@example.org", "password": "asdf1234"},
],
)
def test_mastodon_invalid_input(
event_loop,
@pytest.mark.anyio
async def test_mastodon_invalid_input(
client,
disable_spawner,
hood_id,
@ -86,7 +84,7 @@ def test_mastodon_invalid_input(
monkeypatch,
body,
):
response = client.post(
response = await client.post(
"/api/hoods/{0}/mastodon/".format(hood_id),
json=body,
headers=auth_header,
@ -94,13 +92,15 @@ def test_mastodon_invalid_input(
assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
def test_mastodon_create_mastodon_invalid_id(client, auth_header):
response = client.post("/api/hoods/1337/mastodon/", headers=auth_header)
@pytest.mark.anyio
async def test_mastodon_create_mastodon_invalid_id(client, auth_header):
response = await client.post("/api/hoods/1337/mastodon/", headers=auth_header)
assert response.status_code == status.HTTP_404_NOT_FOUND
response = client.post("/api/hoods/wrong/mastodon/", headers=auth_header)
response = await client.post("/api/hoods/wrong/mastodon/", headers=auth_header)
assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
def test_mastodon_create_unauthorized(client, hood_id):
response = client.post("/api/hoods/{hood_id}/mastodon/")
@pytest.mark.anyio
async def test_mastodon_create_unauthorized(client, hood_id):
response = await client.post("/api/hoods/{hood_id}/mastodon/")
assert response.status_code == status.HTTP_401_UNAUTHORIZED

29
backend/tests/tests_mastodon/test_api_mastodon_delete_bot.py
View file

@ -5,42 +5,43 @@
from fastapi import status
from ormantic.exceptions import NoMatch
from pytest import raises
import pytest
from kibicara.platforms.mastodon.model import MastodonAccount
def test_mastodon_delete_bot(client, event_loop, mastodon_account, auth_header):
response = client.delete(
@pytest.mark.anyio
async def test_mastodon_delete_bot(client, mastodon_account, auth_header):
response = await client.delete(
"/api/hoods/{0}/mastodon/{1}".format(
mastodon_account.hood.id, mastodon_account.id
),
headers=auth_header,
)
assert response.status_code == status.HTTP_204_NO_CONTENT
with raises(NoMatch):
event_loop.run_until_complete(
MastodonAccount.objects.get(id=mastodon_account.id)
)
with pytest.raises(NoMatch):
await MastodonAccount.objects.get(id=mastodon_account.id)
def test_mastodon_delete_bot_invalid_id(client, auth_header, hood_id):
response = client.delete("/api/hoods/1337/mastodon/123", headers=auth_header)
@pytest.mark.anyio
async def test_mastodon_delete_bot_invalid_id(client, auth_header, hood_id):
response = await client.delete("/api/hoods/1337/mastodon/123", headers=auth_header)
assert response.status_code == status.HTTP_404_NOT_FOUND
response = client.delete("/api/hoods/wrong/mastodon/123", headers=auth_header)
response = await client.delete("/api/hoods/wrong/mastodon/123", headers=auth_header)
assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
response = client.delete(
response = await client.delete(
"/api/hoods/{0}/mastodon/7331".format(hood_id), headers=auth_header
)
assert response.status_code == status.HTTP_404_NOT_FOUND
response = client.delete(
response = await client.delete(
"/api/hoods/{0}/mastodon/wrong".format(hood_id), headers=auth_header
)
assert response.status_code == status.HTTP_404_NOT_FOUND
def test_mastodon_delete_bot_unauthorized(client, mastodon_account):
response = client.delete(
@pytest.mark.anyio
async def test_mastodon_delete_bot_unauthorized(client, mastodon_account):
response = await client.delete(
"/api/hoods/{0}/mastodon/{1}".format(
mastodon_account.hood.id, mastodon_account.id
)

42
backend/tests/tests_mastodon/test_api_mastodon_get_bots.py
View file

@ -4,25 +4,26 @@
# SPDX-License-Identifier: 0BSD
from fastapi import status
import pytest
from kibicara.platforms.mastodon.model import MastodonAccount
def test_mastodon_get_bots(
client, auth_header, event_loop, hood_id, mastodon_account, mastodon_instance
@pytest.mark.anyio
async def test_mastodon_get_bots(
client, auth_header, hood_id, mastodon_account, mastodon_instance
):
mastodon2 = event_loop.run_until_complete(
MastodonAccount.objects.create(
hood=mastodon_account.hood,
instance=mastodon_instance,
access_token="4cc3ss",
enabled=True,
username="us4r",
)
mastodon2 = await MastodonAccount.objects.create(
hood=mastodon_account.hood,
instance=mastodon_instance,
access_token="4cc3ss",
enabled=True,
username="us4r",
)
response = client.get(
"/api/hoods/{0}/mastodon".format(mastodon_account.hood.id), headers=auth_header
response = await client.get(
"/api/hoods/{0}/mastodon/".format(mastodon_account.hood.id), headers=auth_header
)
print(response.headers)
assert response.status_code == status.HTTP_200_OK
assert response.json()[0]["id"] == mastodon_account.id
assert response.json()[0]["access_token"] == mastodon_account.access_token
@ -30,20 +31,23 @@ def test_mastodon_get_bots(
assert response.json()[1]["access_token"] == mastodon2.access_token
def test_mastodon_get_bots_invalid_id(client, auth_header, hood_id):
response = client.get("/api/hoods/1337/mastodon", headers=auth_header)
@pytest.mark.anyio
async def test_mastodon_get_bots_invalid_id(client, auth_header, hood_id):
response = await client.get("/api/hoods/1337/mastodon/", headers=auth_header)
assert response.status_code == status.HTTP_404_NOT_FOUND
response = client.get("/api/hoods/wrong/mastodon", headers=auth_header)
response = await client.get("/api/hoods/wrong/mastodon/", headers=auth_header)
assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
def test_mastodon_get_bots_unauthorized(client, hood_id):
response = client.get("/api/hoods/{0}/mastodon".format(hood_id))
@pytest.mark.anyio
async def test_mastodon_get_bots_unauthorized(client, hood_id):
response = await client.get("/api/hoods/{0}/mastodon/".format(hood_id))
assert response.status_code == status.HTTP_401_UNAUTHORIZED
def test_mastodon_public(client, mastodon_account, mastodon_instance, event_loop):
response = client.get(
@pytest.mark.anyio
async def test_mastodon_public(client, mastodon_account, mastodon_instance):
response = await client.get(
"/api/hoods/{0}/mastodon/public".format(mastodon_account.hood.id)
)
assert response.json()[0]["username"] == mastodon_account.username

19
backend/tests/tests_telegram/conftest.py
View file

@ -3,19 +3,18 @@
#
# SPDX-License-Identifier: 0BSD
from pytest import fixture
import pytest
from kibicara.model import Hood
from kibicara.platforms.telegram.model import Telegram
@fixture(scope="function")
def telegram(event_loop, hood_id, bot):
hood = event_loop.run_until_complete(Hood.objects.get(id=hood_id))
return event_loop.run_until_complete(
Telegram.objects.create(
hood=hood,
api_token=bot["api_token"],
welcome_message=bot["welcome_message"],
)
@pytest.fixture(scope="function")
@pytest.mark.anyio
async def telegram(hood_id, bot):
hood = await Hood.objects.get(id=hood_id)
return await Telegram.objects.create(
hood=hood,
api_token=bot["api_token"],
welcome_message=bot["welcome_message"],
)

34
backend/tests/tests_telegram/test_api_telegram_create_bot.py
View file

@ -4,13 +4,13 @@
# SPDX-License-Identifier: 0BSD
from fastapi import status
from pytest import fixture, mark
import pytest
from kibicara.platforms import telegram
from kibicara.platforms.telegram.model import Telegram
@fixture(scope="function")
@pytest.fixture(scope="function")
def disable_spawner(monkeypatch):
class DoNothing:
def start(self, bot):
@ -19,9 +19,9 @@ def disable_spawner(monkeypatch):
monkeypatch.setattr(telegram.webapi, "spawner", DoNothing())
@mark.parametrize("body", [{"api_token": "string", "welcome_message": "string"}])
def test_telegram_create_bot(
event_loop,
@pytest.mark.parametrize("body", [{"api_token": "string", "welcome_message": "string"}])
@pytest.mark.anyio
async def test_telegram_create_bot(
client,
disable_spawner,
hood_id,
@ -34,14 +34,14 @@ def test_telegram_create_bot(
monkeypatch.setattr(telegram.webapi, "check_token", check_token_mock)
response = client.post(
response = await client.post(
"/api/hoods/{0}/telegram/".format(hood_id),
json=body,
headers=auth_header,
)
assert response.status_code == status.HTTP_201_CREATED
bot_id = response.json()["id"]
telegram_obj = event_loop.run_until_complete(Telegram.objects.get(id=bot_id))
telegram_obj = await Telegram.objects.get(id=bot_id)
assert response.json()["api_token"] == body["api_token"] == telegram_obj.api_token
assert (
response.json()["welcome_message"]
@ -52,9 +52,9 @@ def test_telegram_create_bot(
assert telegram_obj.enabled
@mark.parametrize("body", [{"api_token": "string", "welcome_message": "string"}])
def test_telegram_invalid_api_token(
event_loop,
@pytest.mark.parametrize("body", [{"api_token": "string", "welcome_message": "string"}])
@pytest.mark.anyio
async def test_telegram_invalid_api_token(
client,
disable_spawner,
hood_id,
@ -62,7 +62,7 @@ def test_telegram_invalid_api_token(
monkeypatch,
body,
):
response = client.post(
response = await client.post(
"/api/hoods/{0}/telegram/".format(hood_id),
json=body,
headers=auth_header,
@ -70,13 +70,15 @@ def test_telegram_invalid_api_token(
assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
def test_telegram_create_telegram_invalid_id(client, auth_header):
response = client.post("/api/hoods/1337/telegram/", headers=auth_header)
@pytest.mark.anyio
async def test_telegram_create_telegram_invalid_id(client, auth_header):
response = await client.post("/api/hoods/1337/telegram/", headers=auth_header)
assert response.status_code == status.HTTP_404_NOT_FOUND
response = client.post("/api/hoods/wrong/telegram/", headers=auth_header)
response = await client.post("/api/hoods/wrong/telegram/", headers=auth_header)
assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
def test_telegram_create_unauthorized(client, hood_id):
response = client.post("/api/hoods/{hood_id}/telegram/")
@pytest.mark.anyio
async def test_telegram_create_unauthorized(client, hood_id):
response = await client.post("/api/hoods/{hood_id}/telegram/")
assert response.status_code == status.HTTP_401_UNAUTHORIZED

47
backend/tests/tests_telegram/test_api_telegram_delete_bot.py
View file

@ -5,48 +5,51 @@
from fastapi import status
from ormantic.exceptions import NoMatch
from pytest import mark, raises
import pytest
from kibicara.platforms.telegram.model import Telegram, TelegramUser
@mark.parametrize("bot", [{"api_token": "apitoken123", "welcome_message": "msg"}])
def test_telegram_delete_bot(client, event_loop, bot, telegram, auth_header):
event_loop.run_until_complete(
TelegramUser.objects.create(user_id=1234, bot=telegram.id)
)
event_loop.run_until_complete(
TelegramUser.objects.create(user_id=5678, bot=telegram.id)
)
response = client.delete(
@pytest.mark.parametrize(
"bot", [{"api_token": "apitoken123", "welcome_message": "msg"}]
)
@pytest.mark.anyio
async def test_telegram_delete_bot(client, bot, telegram, auth_header):
await TelegramUser.objects.create(user_id=1234, bot=telegram.id)
await TelegramUser.objects.create(user_id=5678, bot=telegram.id)
response = await client.delete(
"/api/hoods/{0}/telegram/{1}".format(telegram.hood.id, telegram.id),
headers=auth_header,
)
assert response.status_code == status.HTTP_204_NO_CONTENT
with raises(NoMatch):
event_loop.run_until_complete(Telegram.objects.get(id=telegram.id))
with raises(NoMatch):
event_loop.run_until_complete(TelegramUser.objects.get(id=telegram.id))
with pytest.raises(NoMatch):
await Telegram.objects.get(id=telegram.id)
with pytest.raises(NoMatch):
await TelegramUser.objects.get(id=telegram.id)
def test_telegram_delete_bot_invalid_id(client, auth_header, hood_id):
response = client.delete("/api/hoods/1337/telegram/123", headers=auth_header)
@pytest.mark.anyio
async def test_telegram_delete_bot_invalid_id(client, auth_header, hood_id):
response = await client.delete("/api/hoods/1337/telegram/123", headers=auth_header)
assert response.status_code == status.HTTP_404_NOT_FOUND
response = client.delete("/api/hoods/wrong/telegram/123", headers=auth_header)
response = await client.delete("/api/hoods/wrong/telegram/123", headers=auth_header)
assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
response = client.delete(
response = await client.delete(
"/api/hoods/{0}/telegram/7331".format(hood_id), headers=auth_header
)
assert response.status_code == status.HTTP_404_NOT_FOUND
response = client.delete(
response = await client.delete(
"/api/hoods/{0}/telegram/wrong".format(hood_id), headers=auth_header
)
assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
@mark.parametrize("bot", [{"api_token": "apitoken123", "welcome_message": "msg"}])
def test_telegram_delete_bot_unauthorized(client, bot, telegram):
response = client.delete(
@pytest.mark.parametrize(
"bot", [{"api_token": "apitoken123", "welcome_message": "msg"}]
)
@pytest.mark.anyio
async def test_telegram_delete_bot_unauthorized(client, bot, telegram):
response = await client.delete(
"/api/hoods/{0}/telegram/{1}".format(telegram.hood.id, telegram.id)
)
assert response.status_code == status.HTTP_401_UNAUTHORIZED

31
backend/tests/tests_telegram/test_api_telegram_get_bot.py
View file

@ -4,12 +4,15 @@
# SPDX-License-Identifier: 0BSD
from fastapi import status
from pytest import mark
import pytest
@mark.parametrize("bot", [{"api_token": "apitoken123", "welcome_message": "msg"}])
def test_telegram_get_bot(client, auth_header, event_loop, bot, telegram):
response = client.get(
@pytest.mark.parametrize(
"bot", [{"api_token": "apitoken123", "welcome_message": "msg"}]
)
@pytest.mark.anyio
async def test_telegram_get_bot(client, auth_header, bot, telegram):
response = await client.get(
"/api/hoods/{0}/telegram/{1}".format(telegram.hood.id, telegram.id),
headers=auth_header,
)
@ -19,24 +22,28 @@ def test_telegram_get_bot(client, auth_header, event_loop, bot, telegram):
assert response.json()["welcome_message"] == telegram.welcome_message
def test_telegram_get_bot_invalid_id(client, auth_header, hood_id):
response = client.get("/api/hoods/1337/telegram/123", headers=auth_header)
@pytest.mark.anyio
async def test_telegram_get_bot_invalid_id(client, auth_header, hood_id):
response = await client.get("/api/hoods/1337/telegram/123", headers=auth_header)
assert response.status_code == status.HTTP_404_NOT_FOUND
response = client.get("/api/hoods/wrong/telegram/123", headers=auth_header)
response = await client.get("/api/hoods/wrong/telegram/123", headers=auth_header)
assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
response = client.get(
response = await client.get(
"/api/hoods/{0}/telegram/7331".format(hood_id), headers=auth_header
)
assert response.status_code == status.HTTP_404_NOT_FOUND
response = client.get(
response = await client.get(
"/api/hoods/{0}/telegram/wrong".format(hood_id), headers=auth_header
)
assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
@mark.parametrize("bot", [{"api_token": "apitoken456", "welcome_message": "msg"}])
def test_telegram_get_bot_unauthorized(client, bot, telegram):
response = client.get(
@pytest.mark.parametrize(
"bot", [{"api_token": "apitoken456", "welcome_message": "msg"}]
)
@pytest.mark.anyio
async def test_telegram_get_bot_unauthorized(client, bot, telegram):
response = await client.get(
"/api/hoods/{0}/telegram/{1}".format(telegram.hood.id, telegram.id)
)
assert response.status_code == status.HTTP_401_UNAUTHORIZED

42
backend/tests/tests_telegram/test_api_telegram_get_bots.py
View file

@ -4,29 +4,27 @@
# SPDX-License-Identifier: 0BSD
from fastapi import status
import pytest
from kibicara.model import Hood
from kibicara.platforms.telegram.model import Telegram
def test_telegram_get_bots(client, auth_header, event_loop, hood_id):
hood = event_loop.run_until_complete(Hood.objects.get(id=hood_id))
telegram0 = event_loop.run_until_complete(
Telegram.objects.create(
hood=hood,
api_token="api_token123",
welcome_message="welcome_message123",
)
@pytest.mark.anyio
async def test_telegram_get_bots(client, auth_header, hood_id):
hood = await Hood.objects.get(id=hood_id)
telegram0 = await Telegram.objects.create(
hood=hood,
api_token="api_token123",
welcome_message="welcome_message123",
)
telegram1 = event_loop.run_until_complete(
Telegram.objects.create(
hood=hood,
api_token="api_token456",
welcome_message="welcome_message123",
)
telegram1 = await Telegram.objects.create(
hood=hood,
api_token="api_token456",
welcome_message="welcome_message123",
)
response = client.get(
"/api/hoods/{0}/telegram".format(telegram0.hood.id), headers=auth_header
response = await client.get(
"/api/hoods/{0}/telegram/".format(telegram0.hood.id), headers=auth_header
)
assert response.status_code == status.HTTP_200_OK
assert response.json()[0]["id"] == telegram0.id
@ -35,13 +33,15 @@ def test_telegram_get_bots(client, auth_header, event_loop, hood_id):
assert response.json()[1]["api_token"] == telegram1.api_token
def test_telegram_get_bots_invalid_id(client, auth_header, hood_id):
response = client.get("/api/hoods/1337/telegram", headers=auth_header)
@pytest.mark.anyio
async def test_telegram_get_bots_invalid_id(client, auth_header, hood_id):
response = await client.get("/api/hoods/1337/telegram/", headers=auth_header)
assert response.status_code == status.HTTP_404_NOT_FOUND
response = client.get("/api/hoods/wrong/telegram", headers=auth_header)
response = await client.get("/api/hoods/wrong/telegram/", headers=auth_header)
assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
def test_telegram_get_bots_unauthorized(client, hood_id):
response = client.get("/api/hoods/{0}/telegram".format(hood_id))
@pytest.mark.anyio
async def test_telegram_get_bots_unauthorized(client, hood_id):
response = await client.get("/api/hoods/{0}/telegram/".format(hood_id))
assert response.status_code == status.HTTP_401_UNAUTHORIZED