mirror of
https://github.com/mealie-recipes/mealie.git
synced 2025-07-24 07:39:41 +02:00
feat: mealplan-webhooks (#1403)
* fix type errors on event bus * webhooks fields required for new implementation * db migration * wip: webhook query + tests and stub function * ignore type checker error * type and method cleanup * datetime and time utc validator * update testing code for utc scheduled time * fix file cmp function call * update version_number * add support for translating "time" objects when restoring backup * bump recipe-scrapers * use specific import syntax * generate frontend types * utilize names exports * use utc times * add task to scheduler * implement new scheduler functionality * stub for type annotation * implement meal-plan data getter * add experimental banner
This commit is contained in:
parent
b1256f4ad2
commit
5a053cdcd6
22 changed files with 428 additions and 93 deletions
|
@ -1,71 +1,74 @@
|
|||
from datetime import datetime, timezone
|
||||
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from tests.utils import assert_derserialize, jsonify
|
||||
from tests.utils.fixture_schemas import TestUser
|
||||
|
||||
|
||||
class Routes:
|
||||
base = "/api/groups/webhooks"
|
||||
|
||||
@staticmethod
|
||||
def item(item_id: int) -> str:
|
||||
return f"{Routes.base}/{item_id}"
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def webhook_data():
|
||||
return {"enabled": True, "name": "Test-Name", "url": "https://my-fake-url.com", "time": "00:00"}
|
||||
return {
|
||||
"enabled": True,
|
||||
"name": "Test-Name",
|
||||
"url": "https://my-fake-url.com",
|
||||
"time": "00:00",
|
||||
"scheduledTime": datetime.now(),
|
||||
}
|
||||
|
||||
|
||||
def test_create_webhook(api_client: TestClient, unique_user: TestUser, webhook_data):
|
||||
response = api_client.post(Routes.base, json=webhook_data, headers=unique_user.token)
|
||||
|
||||
response = api_client.post(Routes.base, json=jsonify(webhook_data), headers=unique_user.token)
|
||||
assert response.status_code == 201
|
||||
|
||||
|
||||
def test_read_webhook(api_client: TestClient, unique_user: TestUser, webhook_data):
|
||||
response = api_client.post(Routes.base, json=webhook_data, headers=unique_user.token)
|
||||
id = response.json()["id"]
|
||||
response = api_client.post(Routes.base, json=jsonify(webhook_data), headers=unique_user.token)
|
||||
item_id = response.json()["id"]
|
||||
|
||||
response = api_client.get(Routes.item(id), headers=unique_user.token)
|
||||
response = api_client.get(Routes.item(item_id), headers=unique_user.token)
|
||||
webhook = assert_derserialize(response, 200)
|
||||
|
||||
webhook = response.json()
|
||||
|
||||
assert webhook["id"]
|
||||
assert webhook["id"] == item_id
|
||||
assert webhook["name"] == webhook_data["name"]
|
||||
assert webhook["url"] == webhook_data["url"]
|
||||
assert webhook["time"] == webhook_data["time"]
|
||||
assert webhook["scheduledTime"] == str(webhook_data["scheduledTime"].astimezone(timezone.utc).time())
|
||||
assert webhook["enabled"] == webhook_data["enabled"]
|
||||
|
||||
|
||||
def test_update_webhook(api_client: TestClient, webhook_data, unique_user: TestUser):
|
||||
response = api_client.post(Routes.base, json=webhook_data, headers=unique_user.token)
|
||||
id = response.json()["id"]
|
||||
response = api_client.post(Routes.base, json=jsonify(webhook_data), headers=unique_user.token)
|
||||
item_dict = assert_derserialize(response, 201)
|
||||
item_id = item_dict["id"]
|
||||
|
||||
webhook_data["name"] = "My New Name"
|
||||
webhook_data["url"] = "https://my-new-fake-url.com"
|
||||
webhook_data["time"] = "01:00"
|
||||
webhook_data["enabled"] = False
|
||||
|
||||
response = api_client.put(Routes.item(id), json=webhook_data, headers=unique_user.token)
|
||||
response = api_client.put(Routes.item(item_id), json=jsonify(webhook_data), headers=unique_user.token)
|
||||
updated_webhook = assert_derserialize(response, 200)
|
||||
|
||||
assert response.status_code == 200
|
||||
|
||||
updated_webhook = response.json()
|
||||
assert updated_webhook["name"] == webhook_data["name"]
|
||||
assert updated_webhook["url"] == webhook_data["url"]
|
||||
assert updated_webhook["time"] == webhook_data["time"]
|
||||
assert updated_webhook["enabled"] == webhook_data["enabled"]
|
||||
|
||||
assert response.status_code == 200
|
||||
|
||||
|
||||
def test_delete_webhook(api_client: TestClient, webhook_data, unique_user: TestUser):
|
||||
response = api_client.post(Routes.base, json=webhook_data, headers=unique_user.token)
|
||||
id = response.json()["id"]
|
||||
|
||||
response = api_client.delete(Routes.item(id), headers=unique_user.token)
|
||||
response = api_client.post(Routes.base, json=jsonify(webhook_data), headers=unique_user.token)
|
||||
item_dict = assert_derserialize(response, 201)
|
||||
item_id = item_dict["id"]
|
||||
|
||||
response = api_client.delete(Routes.item(item_id), headers=unique_user.token)
|
||||
assert response.status_code == 200
|
||||
|
||||
response = api_client.get(Routes.item(id), headers=unique_user.token)
|
||||
response = api_client.get(Routes.item(item_id), headers=unique_user.token)
|
||||
assert response.status_code == 404
|
||||
|
|
Loading…
Add table
Add a link
Reference in a new issue