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

[tests] Ported telegram tests to async

Browse source
This commit is contained in:
missytake 2023-04-01 15:23:33 +02:00
parent f0e15f1d37
commit 16c4ebc84b
5 changed files with 102 additions and 81 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

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

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

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

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

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

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

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

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

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