160 lines
6.7 KiB
Python
160 lines
6.7 KiB
Python
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()
|