import unittest import datetime import tests.test_data as test_data import importlib def reload_test_data(): global test_data test_data = importlib.reload(test_data) from db import connect, create import recipes.db as recipes_db def unique(lst: list, key: callable): seen = set() for item in lst: k = key(item) if k not in seen: seen.add(k) yield item import main class TestRecipe(unittest.IsolatedAsyncioTestCase): async def asyncSetUp(self): self.conn = await connect(':memory:') await create(self.conn) await test_data.create_persons(self.conn) reload_test_data() return await super().asyncSetUp() async def asyncTearDown(self) -> None: await self.conn.close() return await super().asyncTearDown() async def testCreateAndFind(self) -> None: recipe = test_data.Recipes.broccoli_soup ingredient = recipe.ingredients[0] product = ingredient.product person = test_data.Persons.jacob await products_db.insert_product(self.conn, product, {}) create_response = await main.create_recipe(recipe, self.conn, person) self.assertIsNotNone(create_response) self.assertIsInstance(create_response, recipes_db.Recipe, msg=create_response.body if hasattr(create_response, 'body') else create_response) recipe_by_id = await main.get_recipe(recipe.id, self.conn) self.assertIsNotNone(recipe_by_id) self.assertIsInstance(recipe_by_id, recipes_db.Recipe, msg=recipe_by_id.body if hasattr(recipe_by_id, 'body') else recipe_by_id) self.assertIsNotNone(recipe_by_id) self.assertEqual(recipe_by_id.id, recipe.id) self.assertEqual(recipe_by_id.name, recipe.name) self.assertEqual(recipe_by_id.link, recipe.link) self.assertEqual(recipe_by_id.image_urls, recipe.image_urls) self.assertEqual(len(recipe_by_id.ingredients), 1) self.assertEqual(recipe_by_id.ingredients[0].id, ingredient.id) self.assertEqual(recipe_by_id.ingredients[0].line, ingredient.line) self.assertEqual(recipe_by_id.ingredients[0].name, ingredient.name) self.assertEqual(recipe_by_id.ingredients[0].unit, ingredient.unit) self.assertEqual(recipe_by_id.ingredients[0].quantity, ingredient.quantity) self.assertEqual(recipe_by_id.ingredients[0].preparation, ingredient.preparation) self.assertEqual(recipe_by_id.ingredients[0].product.id, product.id) self.assertEqual(recipe_by_id.ingredients[0].product.name, product.name) self.assertEqual(recipe_by_id.ingredients[0].product.link, product.link) self.assertEqual(recipe_by_id.ingredients[0].product.img_large, product.img_large) self.assertEqual(recipe_by_id.ingredients[0].product.img_small, product.img_small) self.assertEqual(recipe_by_id.created_by.id, person.id) self.assertEqual(recipe_by_id.created_by.name, person.name) all_recipes = await main.get_recipes(None, self.conn) self.assertIsNotNone(all_recipes) self.assertIsInstance(all_recipes, list, msg=all_recipes.body if hasattr(all_recipes, 'body') else all_recipes) self.assertEqual(len(all_recipes), 1) self.assertEqual(all_recipes[0].id, recipe.id) await main.delete_recipe(recipe.id, self.conn, test_data.Persons.jacob) all_recipes = await main.get_recipes(None, self.conn) self.assertIsNotNone(all_recipes) self.assertIsInstance(all_recipes, list, msg=all_recipes.body if hasattr(all_recipes, 'body') else all_recipes) self.assertEqual(len(all_recipes), 0) import products.db as products_db import meals class TestMeals(unittest.IsolatedAsyncioTestCase): async def asyncSetUp(self): self.conn = await connect(':memory:') await create(self.conn) await test_data.create_persons(self.conn) reload_test_data() return await super().asyncSetUp() async def asyncTearDown(self) -> None: await self.conn.close() return await super().asyncTearDown() async def testMultiplePariticpants(self) -> None: meal = test_data.Meals.broccoli_soup_for_jacob recipe = meal.recipes[0].recipe recipe_ingredient = recipe.ingredients[0] recipe_product = recipe_ingredient.product extra_ingredient = meal.extra_ingredients[0] extra_product = extra_ingredient.product person = test_data.Persons.jacob await products_db.insert_product(self.conn, recipe_product, {}) await products_db.insert_product(self.conn, extra_product, {}) created_recipe = await main.create_recipe(recipe, self.conn, person) recipe.id = created_recipe.id meal.chefs = [test_data.Persons.ryan, test_data.Persons.chris, test_data.Persons.ryan] error_response = await main.create_meal(meal, self.conn) self.assertIsNotNone(error_response) self.assertEqual(error_response.status_code, 400) self.assertEqual(error_response.body, b'{"message":"Duplicate chef: Ryan"}') meal.chefs = [test_data.Persons.ryan, test_data.Persons.chris] meal.consumers = [test_data.Persons.ryan, test_data.Persons.chris, test_data.Persons.ryan] error_response = await main.create_meal(meal, self.conn) self.assertIsNotNone(error_response) self.assertEqual(error_response.status_code, 400) self.assertEqual(error_response.body, b'{"message":"Duplicate consumer: Ryan"}') meal.consumers = [test_data.Persons.ryan, test_data.Persons.chris] meal.cleanup = [test_data.Persons.ryan, test_data.Persons.chris, test_data.Persons.ryan] error_response = await main.create_meal(meal, self.conn) self.assertIsNotNone(error_response) self.assertEqual(error_response.status_code, 400) self.assertEqual(error_response.body, b'{"message":"Duplicate cleanup person: Ryan"}') meal.cleanup = [test_data.Persons.ryan, test_data.Persons.chris] response = await main.create_meal(meal, self.conn) self.assertIsNotNone(response) self.assertIsInstance(response, meals.Meal, msg=response.body if hasattr(response, 'body') else response) async def testCreateAndFind(self) -> None: meal = test_data.Meals.broccoli_soup_for_jacob meal_recipe = meal.recipes[0] recipe = meal_recipe.recipe recipe_ingredient = recipe.ingredients[0] recipe_product = recipe_ingredient.product extra_ingredient = meal.extra_ingredients[0] extra_product = extra_ingredient.product person = test_data.Persons.jacob await products_db.insert_product(self.conn, recipe_product, {}) await products_db.insert_product(self.conn, extra_product, {}) created_recipe = await main.create_recipe(recipe, self.conn, person) recipe.id = created_recipe.id create_response = await main.create_meal(meal, self.conn) self.assertIsNotNone(create_response) self.assertIsInstance(create_response, meals.Meal, msg=create_response.body if hasattr(create_response, 'body') else create_response) meal_by_id = await main.get_meal(meal.id, self.conn) self.assertIsNotNone(meal_by_id) self.assertIsInstance(meal_by_id, meals.Meal, msg=meal_by_id.body if hasattr(meal_by_id, 'body') else meal_by_id) self.assertIsNotNone(meal_by_id) self.assertEqual(meal_by_id.id, meal.id) self.assertEqual(meal_by_id.suggested_date, meal.suggested_date) self.assertEqual(len(meal_by_id.chefs), 1) self.assertEqual(meal_by_id.chefs[0].id, meal.chefs[0].id) self.assertEqual(meal_by_id.chefs[0].name, meal.chefs[0].name) self.assertEqual(len(meal_by_id.cleanup), 1) self.assertEqual(meal_by_id.cleanup[0].id, meal.cleanup[0].id) self.assertEqual(meal_by_id.cleanup[0].name, meal.cleanup[0].name) self.assertEqual(len(meal_by_id.consumers), 2) expected = [p.id for p in meal.consumers] actual = [p.id for p in meal_by_id.consumers] self.assertEqual(sorted(expected), sorted(actual)) self.assertEqual(len(meal_by_id.recipes), 1) self.assertEqual(meal_by_id.recipes[0].servings, meal_recipe.servings) self.assertIsNotNone(meal_by_id.recipes[0].recipe) self.assertEqual(meal_by_id.recipes[0].recipe.id, recipe.id) self.assertEqual(meal_by_id.recipes[0].recipe.name, recipe.name) self.assertEqual(meal_by_id.recipes[0].recipe.link, recipe.link) self.assertEqual(meal_by_id.recipes[0].recipe.image_urls, recipe.image_urls) self.assertEqual(len(meal_by_id.recipes[0].recipe.ingredients), 1) self.assertEqual(meal_by_id.recipes[0].recipe.ingredients[0].id, recipe_ingredient.id) self.assertEqual(meal_by_id.recipes[0].recipe.ingredients[0].line, recipe_ingredient.line) self.assertEqual(meal_by_id.extra_ingredients[0].id, extra_ingredient.id) self.assertEqual(meal_by_id.extra_ingredients[0].line, extra_ingredient.line) meals_by_date_range = await main.get_upcoming_meals(meal.suggested_date, meal.suggested_date, self.conn) self.assertIsNotNone(meals_by_date_range) self.assertIsInstance(meals_by_date_range, list, msg=meals_by_date_range.body if hasattr(meals_by_date_range, 'body') else meals_by_date_range) self.assertEqual(len(meals_by_date_range), 1) self.assertEqual(meals_by_date_range[0].id, meal.id) self.assertEqual(meals_by_date_range[0].suggested_date, meal.suggested_date) self.assertEqual(len(meals_by_date_range[0].chefs), 1) self.assertEqual(meals_by_date_range[0].chefs[0].id, meal.chefs[0].id) self.assertEqual(meals_by_date_range[0].chefs[0].name, meal.chefs[0].name) self.assertEqual(len(meals_by_date_range[0].cleanup), 1) self.assertEqual(meals_by_date_range[0].cleanup[0].id, meal.cleanup[0].id) self.assertEqual(meals_by_date_range[0].cleanup[0].name, meal.cleanup[0].name) self.assertEqual(len(meals_by_date_range[0].consumers), 2) self.assertEqual(meals_by_date_range[0].recipes[0].recipe.id, recipe.id) self.assertEqual(meals_by_date_range[0].recipes[0].recipe.name, recipe.name) self.assertEqual(meals_by_date_range[0].recipes[0].servings, meal_recipe.servings) async def testMarkConsumed(self) -> None: meal = test_data.Meals.broccoli_soup_for_jacob recipe = meal.recipes[0].recipe recipe_ingredient = recipe.ingredients[0] recipe_product = recipe_ingredient.product extra_ingredient = meal.extra_ingredients[0] extra_product = extra_ingredient.product person = test_data.Persons.jacob await products_db.insert_product(self.conn, recipe_product, {}) await products_db.insert_product(self.conn, extra_product, {}) created_recipe = await main.create_recipe(recipe, self.conn, person) recipe.id = created_recipe.id create_response = await main.create_meal(meal, self.conn) self.assertIsNotNone(create_response) self.assertIsInstance(create_response, meals.Meal, msg=create_response.body if hasattr(create_response, 'body') else create_response) meal_by_id = await main.get_meal(meal.id, self.conn) self.assertIsNotNone(meal_by_id) self.assertIsInstance(meal_by_id, meals.Meal, msg=meal_by_id.body if hasattr(meal_by_id, 'body') else meal_by_id) self.assertIsNone(meal_by_id.consumed_date) updated_meal = await main.mark_consumed(meal_id=meal.id, consumed_date=datetime.datetime.now().astimezone(), conn=self.conn) self.assertIsNotNone(updated_meal) self.assertIsInstance(updated_meal, meals.Meal, msg=updated_meal.body if hasattr(updated_meal, 'body') else updated_meal) self.assertIsNotNone(updated_meal.consumed_date) self.assertLessEqual(datetime.datetime.now().astimezone() - updated_meal.consumed_date, datetime.timedelta(seconds=1)) upcoming_meals = await main.get_upcoming_meals(meal.suggested_date, meal.suggested_date, self.conn) self.assertIsNotNone(upcoming_meals) self.assertIsInstance(upcoming_meals, list, msg=upcoming_meals.body if hasattr(upcoming_meals, 'body') else upcoming_meals) self.assertEqual(len(upcoming_meals), 0) meal_by_id = await main.get_meal(meal.id, self.conn) self.assertIsNotNone(meal_by_id) self.assertIsInstance(meal_by_id, meals.Meal, msg=meal_by_id.body if hasattr(meal_by_id, 'body') else meal_by_id) self.assertIsNotNone(meal_by_id.consumed_date) self.assertLessEqual(datetime.datetime.now().astimezone() - meal_by_id.consumed_date, datetime.timedelta(seconds=1)) async def testUpdate(self) -> None: meal = test_data.Meals.broccoli_soup_for_jacob person = test_data.Persons.jacob recipes = [r.recipe for r in meal.recipes] + [test_data.Recipes.how_to_steam_green_beans] ingredients = meal.extra_ingredients + [ingredient for recipe in recipes for ingredient in recipe.ingredients] + [test_data.Ingredients.butter] products = [ingredient.product for ingredient in ingredients] for product in unique(products, lambda p: p.name): await products_db.insert_product(self.conn, product, {}) for mr in recipes: created_recipe = await main.create_recipe(mr, self.conn, person) mr.id = created_recipe.id create_response = await main.create_meal(meal, self.conn) self.assertIsInstance(create_response, meals.Meal, msg=create_response.body if hasattr(create_response, 'body') else create_response) saved_meal = await main.get_meal(meal.id, self.conn) self.assertIsNotNone(saved_meal) self.assertIsInstance(saved_meal, meals.Meal, msg=saved_meal.body if hasattr(saved_meal, 'body') else saved_meal) saved_meal.suggested_date = datetime.datetime.fromisoformat('2021-01-01T12:00:00') saved_meal.recipes = [meals.MealRecipe(recipe_id=-1, meal_id=-1, recipe=test_data.Recipes.how_to_steam_green_beans, servings=4)] saved_meal.extra_ingredients.append(test_data.Ingredients.butter) saved_meal.chefs.append(test_data.Persons.ryan) saved_meal.cleanup = [test_data.Persons.chris] saved_meal.consumers = [test_data.Persons.ellie, test_data.Persons.jacob, test_data.Persons.ryan] updated_meal = await main.update_meal(saved_meal.id, saved_meal, self.conn) self.assertIsNotNone(updated_meal) self.assertIsInstance(updated_meal, meals.Meal, msg=updated_meal.body if hasattr(updated_meal, 'body') else updated_meal) self.assertEqual(updated_meal.id, saved_meal.id) self.assertEqual(updated_meal.suggested_date, saved_meal.suggested_date) self.assertPersons(updated_meal.chefs, saved_meal.chefs) self.assertPersons(updated_meal.cleanup, saved_meal.cleanup) self.assertPersons(updated_meal.consumers, saved_meal.consumers) self.assertEqual(len(updated_meal.recipes), 1) self.assertEqual(updated_meal.recipes[0].servings, 4) self.assertEqual(updated_meal.recipes[0].recipe.id, test_data.Recipes.how_to_steam_green_beans.id) self.assertEqual(updated_meal.recipes[0].recipe.name, test_data.Recipes.how_to_steam_green_beans.name) self.assertEqual(updated_meal.recipes[0].recipe.link, test_data.Recipes.how_to_steam_green_beans.link) self.assertEqual(updated_meal.recipes[0].recipe.image_urls, test_data.Recipes.how_to_steam_green_beans.image_urls) expected = [i.name for i in test_data.Recipes.how_to_steam_green_beans.ingredients] actual = [i.name for i in updated_meal.recipes[0].recipe.ingredients] self.assertEqual(sorted(expected), sorted(actual)) expected = [i.name for i in saved_meal.extra_ingredients] actual = [i.name for i in updated_meal.extra_ingredients] self.assertEqual(sorted(expected), sorted(actual)) def assertPersons(self, expected, actual): expected_names = [p.name for p in expected] actual_names = [p.name for p in actual] self.assertEqual(sorted(expected_names), sorted(actual_names)) if __name__ == '__main__': unittest.main()