import os import tempfile import unittest from app import create_app, db from app.models import Component, Ingredient, Recipe from config import TestingConfig class LegacyFifthBatchConfig(TestingConfig): _TMP_DIR = tempfile.mkdtemp(prefix="wesp-legacy-batch5-tests-") SQLALCHEMY_DATABASE_URI = f"sqlite:///{os.path.join(_TMP_DIR, 'recipes_test.db')}" SQLALCHEMY_BINDS = {"reports": f"sqlite:///{os.path.join(_TMP_DIR, 'reports_test.db')}"} AUTH_LOGIN = "legacy5-admin" AUTH_PASSWORD = "legacy5-secret" SYNC_CLIENT_VERSION = "1.3.0" class LegacyCompatFifthBatchTests(unittest.TestCase): def setUp(self) -> None: self._prev_sync_state_path = os.environ.get("WESP_SYNC_CLIENT_STATE_PATH") state_path = os.path.join(LegacyFifthBatchConfig._TMP_DIR, "sync_client_state.json") os.environ["WESP_SYNC_CLIENT_STATE_PATH"] = state_path if os.path.exists(state_path): os.unlink(state_path) self.app = create_app(LegacyFifthBatchConfig) self.client = self.app.test_client() self.ctx = self.app.app_context() self.ctx.push() db.create_all() login = self.client.post( "/api/auth/login", json={"login": "legacy5-admin", "password": "legacy5-secret"} ) self.assertEqual(login.status_code, 200) def tearDown(self) -> None: db.session.remove() db.drop_all() self.ctx.pop() if self._prev_sync_state_path is None: os.environ.pop("WESP_SYNC_CLIENT_STATE_PATH", None) else: os.environ["WESP_SYNC_CLIENT_STATE_PATH"] = self._prev_sync_state_path def test_recipes_wibor_and_updates_server_config(self) -> None: page_resp = self.client.get("/api/recipes_wibor") self.assertEqual(page_resp.status_code, 200) page_resp.close() check_resp = self.client.get("/api/updates/check") self.assertEqual(check_resp.status_code, 200) self.assertIn("update_available", check_resp.get_json()) self.assertEqual(check_resp.get_json()["current_version"], "1.3.0") status_resp = self.client.get("/api/updates/status") self.assertEqual(status_resp.status_code, 200) self.assertIn("current_version", status_resp.get_json()) self.assertEqual(status_resp.get_json()["current_version"], "1.3.0") install_resp = self.client.post("/api/updates/install") self.assertIn(install_resp.status_code, (404, 500)) self.assertIn("error", install_resp.get_json()) info_resp = self.client.get("/api/server/info") self.assertEqual(info_resp.status_code, 200) self.assertEqual(info_resp.get_json()["system"], "wesp") self.assertEqual(info_resp.get_json()["version"], "1.3.0") health_resp = self.client.get("/api/health") self.assertEqual(health_resp.status_code, 200) self.assertIn("ok", health_resp.get_json()) update_url_resp = self.client.get( "/api/config/update_server_url?url=https://example.test:5000" ) self.assertEqual(update_url_resp.status_code, 200) self.assertEqual(update_url_resp.get_json()["status"], "success") def test_loading_navigation_and_mixing_state(self) -> None: recipe = Recipe(id="r-legacy5", name="Recipe 5", heads_per_trip=10, mixing_time=5) db.session.add(recipe) db.session.flush() db.session.add_all( [ Ingredient(name="Comp 1", amount=10, recipe_id=recipe.id, order=1), Ingredient(name="Comp 2", amount=15, recipe_id=recipe.id, order=2), ] ) db.session.commit() set_recipe = self.client.post("/api/set_current_recipe", json={"recipe_id": recipe.id}) self.assertEqual(set_recipe.status_code, 200) current_recipe = self.client.get("/api/get_current_recipe") self.assertEqual(current_recipe.status_code, 200) self.assertEqual(current_recipe.get_json()["status"], "active") loading_state = self.client.get("/api/current_loading_state") self.assertEqual(loading_state.status_code, 200) self.assertEqual(loading_state.get_json()["status"], "active") display = self.client.get("/api/weight_display_data") self.assertEqual(display.status_code, 200) self.assertEqual(display.get_json()["status"], "active") nav_next = self.client.post("/api/navigate_component", json={"direction": "next"}) self.assertEqual(nav_next.status_code, 200) reset = self.client.post("/api/reset_component") self.assertEqual(reset.status_code, 200) send_cmd = self.client.post("/api/send_navigation_command", json={"command": "prev"}) self.assertEqual(send_cmd.status_code, 200) get_cmds = self.client.get("/api/get_navigation_commands") self.assertEqual(get_cmds.status_code, 200) self.assertEqual(get_cmds.get_json()["status"], "success") mix_mode = self.client.post("/api/set_mixing_mode", json={"is_mixing": True}) self.assertEqual(mix_mode.status_code, 200) mix_timer = self.client.post("/api/set_mixing_timer", json={"active": True}) self.assertEqual(mix_timer.status_code, 200) timer_state = self.client.get("/api/get_mixing_timer") self.assertEqual(timer_state.status_code, 200) self.assertTrue(timer_state.get_json()["active"]) def test_role_mac_and_soft_delete_restore_deleted(self) -> None: set_role = self.client.post( "/api/set_role", json={"role": "dispenser", "dispenser_name": "D-1"} ) self.assertEqual(set_role.status_code, 200) self.assertEqual(set_role.get_json()["status"], "ok") mac_info = self.client.get("/api/mac_info") self.assertEqual(mac_info.status_code, 200) self.assertIn("current_mac", mac_info.get_json()) component = Component(id="c-legacy5", name="Corn", type="grain") db.session.add(component) db.session.commit() soft_delete = self.client.delete("/api/components/c-legacy5/soft_delete") self.assertEqual(soft_delete.status_code, 200) self.assertTrue(soft_delete.get_json()["success"]) deleted_list = self.client.get("/api/components/deleted") self.assertEqual(deleted_list.status_code, 200) self.assertEqual(len(deleted_list.get_json()), 1) restore = self.client.post("/api/components/c-legacy5/restore") self.assertEqual(restore.status_code, 200) self.assertTrue(restore.get_json()["success"]) deleted_list_after = self.client.get("/api/components/deleted") self.assertEqual(deleted_list_after.status_code, 200) self.assertEqual(len(deleted_list_after.get_json()), 0) if __name__ == "__main__": unittest.main()