diff --git a/.github/workflows/django-tests.yml b/.github/workflows/django-tests.yml new file mode 100644 index 00000000..83202785 --- /dev/null +++ b/.github/workflows/django-tests.yml @@ -0,0 +1,53 @@ +name: "Django tests" + +on: + pull_request: + paths: + - "docker/web/**" + - ".github/workflows/django-tests.yml" + push: + branches: + - "main" + - "devel" + - "beta" + paths: + - "docker/web/**" + - ".github/workflows/django-tests.yml" + +# Fork pull requests already get a read-only token and no repository secrets; this keeps +# same-repo runs read-only too. Don't switch to pull_request_target or add a workflow_run +# follow-up without reviewing what code it checks out: those run with secrets. +permissions: + contents: read + +jobs: + test: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + + - uses: actions/setup-python@v5 + with: + python-version: "3.12" # Matches the python:3.12 base image in docker/Dockerfile + cache: pip + cache-dependency-path: docker/web/requirements.txt + + - name: Install dependencies + run: pip install -r docker/web/requirements.txt unittest-xml-reporting + + - name: Run tests + run: docker/web/run_tests.sh -v 2 + env: + NSPM_TEST_JUNIT_DIR: ${{ github.workspace }}/test-results + + # annotate_only writes the job summary and failure annotations without creating a + # check run, which needs checks: write and so fails on pull requests from forks. + - name: Report test results + if: ${{ !cancelled() }} + uses: mikepenz/action-junit-report@v6 + with: + report_paths: test-results/*.xml + annotate_only: true + detailed_summary: true + include_passed: true + require_tests: true diff --git a/docker/web/nspanelmanager/nspanelmanager/settings.py b/docker/web/nspanelmanager/nspanelmanager/settings.py index 2044f3ec..cdf2c96b 100755 --- a/docker/web/nspanelmanager/nspanelmanager/settings.py +++ b/docker/web/nspanelmanager/nspanelmanager/settings.py @@ -28,9 +28,12 @@ # Quick-start development settings - unsuitable for production # See https://docs.djangoproject.com/en/4.1/howto/deployment/checklist/ +# Persistent data (secret key, database). Overridable so that tests can run outside the container. +DATA_DIR = environment("NSPM_DATA_DIR", default="/data") + # SECURITY WARNING: keep the secret key used in production secret! SECRET_KEY = "" -SECRET_KEY_PATH = "/data/secret.key" +SECRET_KEY_PATH = os.path.join(DATA_DIR, "secret.key") if os.path.exists(SECRET_KEY_PATH): with open(SECRET_KEY_PATH) as f: SECRET_KEY = f.read() @@ -107,7 +110,7 @@ # Database # https://docs.djangoproject.com/en/4.1/ref/settings/#databases -DATABASES = {"default": {"ENGINE": "django.db.backends.sqlite3", "NAME": "/data/nspanelmanager_db.sqlite3", "timeout": 20}} +DATABASES = {"default": {"ENGINE": "django.db.backends.sqlite3", "NAME": os.path.join(DATA_DIR, "nspanelmanager_db.sqlite3"), "timeout": 20}} LOGGING = { "version": 1, diff --git a/docker/web/nspanelmanager/nspanelmanager/test_settings.py b/docker/web/nspanelmanager/nspanelmanager/test_settings.py new file mode 100644 index 00000000..0ee438d0 --- /dev/null +++ b/docker/web/nspanelmanager/nspanelmanager/test_settings.py @@ -0,0 +1,30 @@ +""" +Settings used when running the Django test suite (see docker/web/run_tests.sh). + +Identical to the production settings except that persistent data (the secret key) +is written to a throw-away directory instead of /data. The Django test runner +automatically uses an in-memory SQLite database, so the real database is never touched. +""" + +import os +import tempfile + +os.environ.setdefault("NSPM_DATA_DIR", tempfile.mkdtemp(prefix="nspm-test-")) + +from .settings import * + +MEDIA_ROOT = os.path.join(DATA_DIR, "media") + +# Hashing speed is irrelevant for tests. +PASSWORD_HASHERS = ["django.contrib.auth.hashers.MD5PasswordHasher"] + +# Many views log full tracebacks when rejecting bad input, which the tests do on purpose. +# Set NSPM_TEST_LOG_LEVEL=INFO (or DEBUG) to see them when debugging a failing test. +LOGGING["root"]["level"] = os.environ.get("NSPM_TEST_LOG_LEVEL", "CRITICAL") +LOGGING["loggers"] = {"django.request": {"level": os.environ.get("NSPM_TEST_LOG_LEVEL", "CRITICAL")}} + +# CI sets NSPM_TEST_JUNIT_DIR to also write JUnit XML results (needs unittest-xml-reporting). +if os.environ.get("NSPM_TEST_JUNIT_DIR"): + TEST_RUNNER = "xmlrunner.extra.djangotestrunner.XMLTestRunner" + TEST_OUTPUT_DIR = os.environ["NSPM_TEST_JUNIT_DIR"] + TEST_OUTPUT_FILE_NAME = "django-tests.xml" diff --git a/docker/web/nspanelmanager/web/api.py b/docker/web/nspanelmanager/web/api.py index a78d1c93..7166eea2 100644 --- a/docker/web/nspanelmanager/web/api.py +++ b/docker/web/nspanelmanager/web/api.py @@ -231,13 +231,13 @@ def get_nspanel_config(request): base["button2_mqtt_topic"] = get_nspanel_setting_with_default(nspanel.id, "button2_mqtt_topic", "") base["button2_mqtt_payload"] = get_nspanel_setting_with_default(nspanel.id, "button2_mqtt_payload", "") - if nspanel.button1_detached_mode_light: - base["button1_detached_light"] = nspanel.button1_detached_mode_light.id + if nspanel.button1_detached_mode_entity: + base["button1_detached_light"] = nspanel.button1_detached_mode_entity.id else: - base["button1_detached_mode_light"] = -1 + base["button1_detached_light"] = -1 - if nspanel.button2_detached_mode_light: - base["button2_detached_light"] = nspanel.button2_detached_mode_light.id + if nspanel.button2_detached_mode_entity: + base["button2_detached_light"] = nspanel.button2_detached_mode_entity.id else: base["button2_detached_light"] = -1 base["rooms"] = [] diff --git a/docker/web/nspanelmanager/web/apps.py b/docker/web/nspanelmanager/web/apps.py index fa01240b..6816c0bb 100755 --- a/docker/web/nspanelmanager/web/apps.py +++ b/docker/web/nspanelmanager/web/apps.py @@ -74,28 +74,34 @@ def send_mqttmanager_reload_command(): def create_entity_pages_for_all_rooms(): - from .models import Room, RoomEntitiesPage + from .models import Room for room in Room.objects.all(): - # Check if an existing scenes page exists, if it does not, create one - existing_pages = RoomEntitiesPage.objects.filter(room=room, is_scenes_page=True) - if not existing_pages.exists(): - page = RoomEntitiesPage() - page.room = room - page.display_order = 0 - page.page_type = 4 - page.is_scenes_page = True - page.save() - - # Check if an existing entities page exists, if it doesn not, create one. - existing_pages = RoomEntitiesPage.objects.filter(room=room, is_scenes_page=False) - if not existing_pages.exists(): - page = RoomEntitiesPage() - page.room = room - page.display_order = 0 - page.page_type = 4 - page.is_scenes_page = False - page.save() + create_entity_pages_for_room(room) + + +def create_entity_pages_for_room(room): + from .models import RoomEntitiesPage + + # Check if an existing scenes page exists, if it does not, create one + existing_pages = RoomEntitiesPage.objects.filter(room=room, is_scenes_page=True) + if not existing_pages.exists(): + page = RoomEntitiesPage() + page.room = room + page.display_order = 0 + page.page_type = 4 + page.is_scenes_page = True + page.save() + + # Check if an existing entities page exists, if it doesn not, create one. + existing_pages = RoomEntitiesPage.objects.filter(room=room, is_scenes_page=False) + if not existing_pages.exists(): + page = RoomEntitiesPage() + page.room = room + page.display_order = 0 + page.page_type = 4 + page.is_scenes_page = False + page.save() def create_global_scenes_page(): diff --git a/docker/web/nspanelmanager/web/htmx.py b/docker/web/nspanelmanager/web/htmx.py index 73d194f6..7a4ceac5 100644 --- a/docker/web/nspanelmanager/web/htmx.py +++ b/docker/web/nspanelmanager/web/htmx.py @@ -58,63 +58,6 @@ def unblock_nspanel(request, nspanel_id): return response -@csrf_exempt -def nspanel_accept_register_request(request, nspanel_id): - try: - if request.method == "POST": - nspanel = NSPanel.objects.get(id=nspanel_id) - nspanel.denied = False - nspanel.accepted = True - nspanel.save() - send_mqttmanager_reload_command() - - response = HttpResponse("", status=200) - response["HX-Refresh"] = "true" - return response - else: - return JsonResponse({"status": "error"}, status=405) - except Exception as ex: - logging.exception(ex) - return JsonResponse({"status": "error"}, status=500) - - -@csrf_exempt -def nspanel_deny_register_request(request, nspanel_id): - try: - if request.method == "POST": - nspanel = NSPanel.objects.get(id=nspanel_id) - nspanel.accepted = False - nspanel.denied = True - nspanel.save() - - send_mqttmanager_reload_command() - - response = HttpResponse("", status=200) - response["HX-Refresh"] = "true" - return response - else: - return JsonResponse({"status": "error"}, status=405) - except Exception as ex: - logging.exception(ex) - return JsonResponse({"status": "error"}, status=500) - - -@csrf_exempt -def nspanel_delete(request, nspanel_id): - try: - if request.method == "DELETE": - nspanel = NSPanel.objects.get(id=nspanel_id) - nspanel.delete() - response = HttpResponse("", status=200) - response["HX-Redirect"] = "/" - return response - else: - return JsonResponse({"status": "error"}, status=405) - except Exception as ex: - logging.exception(ex) - return JsonResponse({"status": "error"}, status=500) - - def select_room_temperature_sensor_provider(request, room_id): if (get_setting_with_default("home_assistant_address") == "" or get_setting_with_default("home_assistant_token") == "") and get_setting_with_default("openhab_address") != "" and get_setting_with_default("openhab_token") != "": # OpenHAB connection configured but not Home Assistant. Skip selecting source: @@ -267,6 +210,7 @@ def relay_group_save(request): rg = RelayGroup() rg.friendly_name = request.POST["name"] rg.save() + send_mqttmanager_reload_command() response = HttpResponse() response["HX-Refresh"] = "true" @@ -278,6 +222,7 @@ def relay_group_delete(request, relay_group_id): if request.method == "DELETE": rg = RelayGroup.objects.get(id=relay_group_id) rg.delete() + send_mqttmanager_reload_command() response = HttpResponse() response["HX-Refresh"] = "true" @@ -538,32 +483,6 @@ def partial_remove_entity_from_page_slot(request, page_id, slot_id): ) -def partial_edit_entities_page(request, page_id): - data = { - "page_id": page_id, - } - return render(request, "partial/edit_entities_page.html", data) - - -def partial_save_edit_entities_page(request, page_id, page_type): - page = RoomEntitiesPage.objects.get(id=page_id) - page.page_type = page_type - page.save() - - room_id = 0 - if page.room: - room_id = page.room.id - send_mqttmanager_reload_command() - entities_pages = NSPanelRoomEntitiesPages() - return entities_pages.get( - request=request, - view="edit_room", - room_id=room_id, - is_scenes_pages=page.is_scenes_page, - is_global_scenes_page=(page.room is None), - ) - - def get_entity_in_page_slot(page_id, slot_id): page = RoomEntitiesPage.objects.get(id=page_id) entities = Entity.objects.filter(entities_page=page, room_view_position=slot_id) diff --git a/docker/web/nspanelmanager/web/rest.py b/docker/web/nspanelmanager/web/rest.py index ad756308..6f3a7f1c 100644 --- a/docker/web/nspanelmanager/web/rest.py +++ b/docker/web/nspanelmanager/web/rest.py @@ -7,6 +7,7 @@ import requests from django.core.files.storage import FileSystemStorage +from django.db import transaction from django.http import HttpResponse, JsonResponse from django.views.decorators.csrf import csrf_exempt @@ -14,11 +15,10 @@ import web.openhab_api from web.settings_helper import ( get_nspanel_setting_with_default, - get_setting_with_default, set_setting_value, ) -from .apps import send_mqttmanager_reload_command +from .apps import create_entity_pages_for_room, send_mqttmanager_reload_command from .models import Entity, LightState, NSPanel, RelayGroup, Room, RoomEntitiesPage, Scene, Settings ######################## @@ -88,42 +88,6 @@ def test_openhab(request): ## MQTTManager section ### ########################## -# Keys that are not allowed to be pulled though the REST API beacuse of security concerns. -banned_setting_keys = [ - "MQTT_SERVER", - "MQTT_PORT", - "MQTT_USERNAME", - "MQTT_PASSWORD", - "HOME_ASSISTANT_ADDRESS", - "HOME_ASSISTANT_TOKEN", - "OPENHAB_ADDRESS", - "OPENHAB_TOKEN", - "OPENHAB_TOKEN", -] - - -def mqttmanager_get_setting(request, setting_key): - if setting_key in banned_setting_keys: - return JsonResponse({"status": "error"}, status=403) # Return error forbidden - - try: - settings = {} - if request.method == "GET": - settings[setting_key] = get_setting_with_default(setting_key) - else: - return JsonResponse({"status": "error"}, status=405) - - return JsonResponse( - { - "status": "ok", - "settings": settings, - } - ) - except Exception as ex: - logging.exception(ex) - return JsonResponse({"status": "error"}, status=500) - return JsonResponse({"status": "error"}, status=500) - @csrf_exempt def test_mqttmanager(request): @@ -220,12 +184,9 @@ def settings_get(request): settings = {} for setting in Settings.objects.all(): settings[setting.name] = setting.value - settings["home_assistant_token_set"] = settings.get("home_assistant_token", "") != "" - del settings["home_assistant_token"] - settings["openhab_token_set"] = settings.get("openhab_token", "") != "" - del settings["openhab_token"] - settings["mqtt_password_set"] = settings.get("mqtt_password", "") != "" - del settings["mqtt_password"] + # Never return secrets, only whether they are set. Any of them may never have been saved. + for secret in ("home_assistant_token", "openhab_token", "mqtt_password"): + settings[f"{secret}_set"] = settings.pop(secret, "") != "" return JsonResponse({"status": "ok", "settings": settings}, status=200) @@ -273,20 +234,36 @@ def rooms_get(request): return JsonResponse({"status": "error"}, status=500) +# The entities page layouts the panel GUI has. +PAGE_SIZES = (4, 8, 12) + + +def get_page_size_error(page_type): + if page_type not in PAGE_SIZES: + return f"Unsupported page size {page_type!r}. Expected one of: {', '.join(map(str, PAGE_SIZES))}." + return None + + def put_room_entities_order(request, room_id): if request.method == "PUT": try: data = json.loads(request.body) - for entity in data["entities"]: - db_entity = Entity.objects.get(id=entity["id"]) - db_entity.room_view_position = entity["room_view_position"] - db_entity.entities_page_id = entity["entities_page_id"] - db_entity.save() - for scene in data["scenes"]: - db_scene = Scene.objects.get(id=scene["id"]) - db_scene.room_view_position = scene["room_view_position"] - db_scene.entities_page_id = scene["entities_page_id"] - db_scene.save() + for item in data["entities"] + data["scenes"]: + page = RoomEntitiesPage.objects.get(id=item["entities_page_id"]) + if not 0 <= item["room_view_position"] < page.page_type: + return JsonResponse({"status": "error", "message": f"room_view_position must be between 0 and {page.page_type - 1} on page {page.id}."}, status=400) + # All or nothing: a bad entry must not leave the entries before it already moved. + with transaction.atomic(): + for entity in data["entities"]: + db_entity = Entity.objects.get(id=entity["id"]) + db_entity.room_view_position = entity["room_view_position"] + db_entity.entities_page_id = entity["entities_page_id"] + db_entity.save() + for scene in data["scenes"]: + db_scene = Scene.objects.get(id=scene["id"]) + db_scene.room_view_position = scene["room_view_position"] + db_scene.entities_page_id = scene["entities_page_id"] + db_scene.save() send_mqttmanager_reload_command() return JsonResponse({"status": "ok"}, status=200) except Exception as ex: @@ -325,6 +302,9 @@ def room_entities_pages(request, room_id): for field in required_fields: if field not in data: return JsonResponse({"status": "error", "message": f"Missing required field: {field}"}, status=400) + page_size_error = get_page_size_error(data["type"]) + if page_size_error: + return JsonResponse({"status": "error", "message": page_size_error}, status=400) room = Room.objects.get(id=room_id) pages = RoomEntitiesPage.objects.filter(room=room).order_by("display_order") @@ -365,9 +345,18 @@ def room_entities_page(request, page_id): try: data = json.loads(request.body) db_page = RoomEntitiesPage.objects.get(id=page_id) - db_page.page_type = data.get("number_of_entities", db_page.page_type) + new_page_type = data.get("number_of_entities", db_page.page_type) + if new_page_type != db_page.page_type: + page_size_error = get_page_size_error(new_page_type) + if page_size_error: + return JsonResponse({"status": "error", "message": page_size_error}, status=400) + # Shrinking must not leave entities in slots the panel no longer shows. + if db_page.entity_set.filter(room_view_position__gte=new_page_type).exists() or db_page.scene_set.filter(room_view_position__gte=new_page_type).exists(): + return JsonResponse({"status": "error", "message": f"Move or remove the entities in slots {new_page_type + 1} and above before shrinking this page."}, status=400) + db_page.page_type = new_page_type db_page.display_order = data.get("display_order", db_page.display_order) db_page.save() + send_mqttmanager_reload_command() return JsonResponse({"status": "ok"}, status=200) except Exception as ex: logging.exception(ex) @@ -495,6 +484,9 @@ def global_entities_pages(request): for field in required_fields: if field not in data: return JsonResponse({"status": "error", "message": f"Missing required field: {field}"}, status=400) + page_size_error = get_page_size_error(data["type"]) + if page_size_error: + return JsonResponse({"status": "error", "message": page_size_error}, status=400) pages = RoomEntitiesPage.objects.filter(room=None).order_by("display_order") new_display_order = 0 # Default to zero of no pages exists @@ -523,7 +515,12 @@ def room_delete(request, room_id): if request.method == "DELETE": try: room = Room.objects.get(id=room_id) + # Panels are not deleted with their room. As in the UI's delete_room view, move them to another room. + new_room = Room.objects.exclude(id=room.id).first() + if new_room: + NSPanel.objects.filter(room=room).update(room=new_room) room.delete() + send_mqttmanager_reload_command() return JsonResponse({"status": "ok", "room_id": room_id}, status=200) except Exception as ex: logging.exception(ex) @@ -574,9 +571,15 @@ def room_create(request): if request.method == "POST": try: data = json.loads(request.body) - new_room = Room() - new_room.friendly_name = data["name"] - new_room.save() + max_length = Room._meta.get_field("friendly_name").max_length + if not data["name"] or len(data["name"]) > max_length: + return JsonResponse({"status": "error", "message": f"Room name must be 1 to {max_length} characters"}, status=400) + with transaction.atomic(): + new_room = Room() + new_room.friendly_name = data["name"] + new_room.save() + create_entity_pages_for_room(new_room) + send_mqttmanager_reload_command() return JsonResponse({"status": "ok", "room_id": new_room.id}, status=200) except Exception as ex: logging.exception(ex) @@ -628,7 +631,7 @@ def scenes(request): def scenes_get(request): try: scenes = [] - if request.GET.get("light_id"): + if request.GET.get("scene_id"): scenes_objects = Scene.objects.filter(id=request.GET.get("scene_id")) elif request.GET.get("room_id"): scenes_objects = Scene.objects.filter(room_id=request.GET.get("room_id")) @@ -732,6 +735,7 @@ def get_scene(request, scene_id): elif request.method == "DELETE": scene = Scene.objects.get(id=scene_id) scene.delete() + send_mqttmanager_reload_command() return JsonResponse({"status": "success"}) except Exception as ex: logging.exception(ex) @@ -740,6 +744,43 @@ def get_scene(request, scene_id): ### Generic Entity section ### +# The controllers MQTTManager's entity classes (docker/MQTTManager/include/{light,switch,button,thermostat}/) +# accept for each entity type. Anything else is logged as an error when it loads the entity. +ENTITY_CONTROLLERS = { + Entity.EntityType.LIGHT: ("home_assistant", "openhab"), + Entity.EntityType.SWITCH: ("home_assistant", "openhab"), + Entity.EntityType.BUTTON: ("home_assistant", "nspm"), + Entity.EntityType.THERMOSTAT: ("home_assistant", "openhab"), +} + +# The scene types MQTTManager's EntityManager loads. It ignores scenes of any other type. +SCENE_TYPES = ("home_assistant", "openhab", "nspm_scene") + + +def get_controller_error(entity_type, controller): + if controller not in ENTITY_CONTROLLERS[entity_type]: + return f"Unknown controller {controller!r}. Expected one of: {', '.join(ENTITY_CONTROLLERS[entity_type])}." + return None + + +def get_placement_error(obj, room, entities_page, room_view_position): + """Return why obj (an Entity or Scene) cannot go in that slot, or None if it can. + + Only a new or changed placement is checked, so entities saved before these checks + existed can still be edited without being moved. + """ + room_id = room.id if room else None + if obj.pk is not None and (obj.room_id, obj.entities_page_id, obj.room_view_position) == (room_id, entities_page.id, room_view_position): + return None + if entities_page.room_id != room_id: + return "The entities page belongs to a different room." + if not 0 <= room_view_position < entities_page.page_type: + return f"room_view_position must be between 0 and {entities_page.page_type - 1} on this page." + if type(obj).objects.filter(entities_page=entities_page, room_view_position=room_view_position).exclude(pk=obj.pk).exists(): + return "That position on the page is already taken." + return None + + def get_rest_entitiy_representation(entity_id): entity = Entity.objects.get(id=entity_id) return { @@ -787,7 +828,7 @@ def entities_lights(request): except Exception as ex: logging.exception(ex) return JsonResponse({"status": "error"}, status=500) - return JsonResponse({"status": "error", "error": "Unsupported method"}, status=403) + return JsonResponse({"status": "error", "error": "Unsupported method"}, status=405) def put_light_entity(request): @@ -814,6 +855,10 @@ def put_light_entity(request): if field not in data: return JsonResponse({"status": "error", "message": f"Missing required field: {field}"}, status=400) + controller_error = get_controller_error(Entity.EntityType.LIGHT, data["controller"]) + if controller_error: + return JsonResponse({"status": "error", "message": controller_error}, status=400) + entity_data = { "controller": data["controller"], "home_assistant_name": data.get("home_assistant_name", ""), @@ -833,10 +878,17 @@ def put_light_entity(request): new_light = Entity() new_light.entity_type = Entity.EntityType.LIGHT + room = Room.objects.get(id=int(data["room_id"])) + entities_page = RoomEntitiesPage.objects.get(id=int(data["entities_page_id"])) + room_view_position = int(data["room_view_position"]) + placement_error = get_placement_error(new_light, room, entities_page, room_view_position) + if placement_error: + return JsonResponse({"status": "error", "message": placement_error}, status=400) + new_light.friendly_name = data["friendly_name"] - new_light.room = Room.objects.get(id=int(data["room_id"])) - new_light.entities_page = RoomEntitiesPage.objects.get(id=int(data["entities_page_id"])) - new_light.room_view_position = int(data["room_view_position"]) + new_light.room = room + new_light.entities_page = entities_page + new_light.room_view_position = room_view_position new_light.entity_data = entity_data new_light.save() @@ -860,7 +912,7 @@ def entities_switches(request): except Exception as ex: logging.exception(ex) return JsonResponse({"status": "error"}, status=500) - return JsonResponse({"status": "error", "error": "Unsupported method"}, status=403) + return JsonResponse({"status": "error", "error": "Unsupported method"}, status=405) def put_switch_entity(request): @@ -880,6 +932,10 @@ def put_switch_entity(request): if field not in data: return JsonResponse({"status": "error", "message": f"Missing required field: {field}"}, status=400) + controller_error = get_controller_error(Entity.EntityType.SWITCH, data["controller"]) + if controller_error: + return JsonResponse({"status": "error", "message": controller_error}, status=400) + entity_data = { "openhab_item_switch": data.get("openhab_item_switch", ""), "home_assistant_name": data.get("home_assistant_name", ""), @@ -891,10 +947,17 @@ def put_switch_entity(request): new_switch = Entity() new_switch.entity_type = Entity.EntityType.SWITCH + room = Room.objects.get(id=int(data["room_id"])) + entities_page = RoomEntitiesPage.objects.get(id=int(data["entities_page_id"])) + room_view_position = int(data["room_view_position"]) + placement_error = get_placement_error(new_switch, room, entities_page, room_view_position) + if placement_error: + return JsonResponse({"status": "error", "message": placement_error}, status=400) + new_switch.friendly_name = data["friendly_name"] - new_switch.room = Room.objects.get(id=int(data["room_id"])) - new_switch.entities_page = RoomEntitiesPage.objects.get(id=int(data["entities_page_id"])) - new_switch.room_view_position = int(data["room_view_position"]) + new_switch.room = room + new_switch.entities_page = entities_page + new_switch.room_view_position = room_view_position new_switch.entity_data = entity_data new_switch.save() @@ -918,7 +981,7 @@ def entities_buttons(request): except Exception as ex: logging.exception(ex) return JsonResponse({"status": "error"}, status=500) - return JsonResponse({"status": "error", "error": "Unsupported method"}, status=403) + return JsonResponse({"status": "error", "error": "Unsupported method"}, status=405) def put_button_entity(request): @@ -939,6 +1002,10 @@ def put_button_entity(request): if field not in data: return JsonResponse({"status": "error", "message": f"Missing required field: {field}"}, status=400) + controller_error = get_controller_error(Entity.EntityType.BUTTON, data["controller"]) + if controller_error: + return JsonResponse({"status": "error", "message": controller_error}, status=400) + entity_data = { "mqtt_topic": data.get("mqtt_topic", ""), "mqtt_payload": data.get("mqtt_payload", ""), @@ -951,10 +1018,17 @@ def put_button_entity(request): new_button = Entity() new_button.entity_type = Entity.EntityType.BUTTON + room = Room.objects.get(id=int(data["room_id"])) + entities_page = RoomEntitiesPage.objects.get(id=int(data["entities_page_id"])) + room_view_position = int(data["room_view_position"]) + placement_error = get_placement_error(new_button, room, entities_page, room_view_position) + if placement_error: + return JsonResponse({"status": "error", "message": placement_error}, status=400) + new_button.friendly_name = data["friendly_name"] - new_button.room = Room.objects.get(id=int(data["room_id"])) - new_button.entities_page = RoomEntitiesPage.objects.get(id=int(data["entities_page_id"])) - new_button.room_view_position = int(data["room_view_position"]) + new_button.room = room + new_button.entities_page = entities_page + new_button.room_view_position = room_view_position new_button.entity_data = entity_data new_button.save() @@ -973,12 +1047,12 @@ def put_button_entity(request): def entities_thermostats(request): try: - # if request.method == "PUT": - return put_thermostat_entity(request) + if request.method == "PUT": + return put_thermostat_entity(request) except Exception as ex: logging.exception(ex) return JsonResponse({"status": "error"}, status=500) - return JsonResponse({"status": "error", "error": "Unsupported method"}, status=403) + return JsonResponse({"status": "error", "error": "Unsupported method"}, status=405) def put_thermostat_entity(request): @@ -1019,6 +1093,10 @@ def put_thermostat_entity(request): if field not in item: return JsonResponse({"status": "error", "message": f"Missing required field in {mode}. Missing field: {field}"}, status=400) + controller_error = get_controller_error(Entity.EntityType.THERMOSTAT, data["controller"]) + if controller_error: + return JsonResponse({"status": "error", "message": controller_error}, status=400) + entity_data = { "controller": data.get("controller", ""), "fan_modes": data.get("fan_modes", []), @@ -1043,10 +1121,17 @@ def put_thermostat_entity(request): new_thermostat = Entity() new_thermostat.entity_type = Entity.EntityType.THERMOSTAT + room = Room.objects.get(id=int(data["room_id"])) + entities_page = RoomEntitiesPage.objects.get(id=int(data["entities_page_id"])) + room_view_position = int(data["room_view_position"]) + placement_error = get_placement_error(new_thermostat, room, entities_page, room_view_position) + if placement_error: + return JsonResponse({"status": "error", "message": placement_error}, status=400) + new_thermostat.friendly_name = data["friendly_name"] - new_thermostat.room = Room.objects.get(id=int(data["room_id"])) - new_thermostat.entities_page = RoomEntitiesPage.objects.get(id=int(data["entities_page_id"])) - new_thermostat.room_view_position = int(data["room_view_position"]) + new_thermostat.room = room + new_thermostat.entities_page = entities_page + new_thermostat.room_view_position = room_view_position new_thermostat.entity_data = entity_data new_thermostat.save() @@ -1193,12 +1278,12 @@ def put_media_player_entity(request): def entities_scenes(request): try: - # if request.method == "PUT": - return put_scene_entity(request) + if request.method == "PUT": + return put_scene_entity(request) except Exception as ex: logging.exception(ex) return JsonResponse({"status": "error"}, status=500) - return JsonResponse({"status": "error", "error": "Unsupported method"}, status=403) + return JsonResponse({"status": "error", "error": "Unsupported method"}, status=405) def put_scene_entity(request): @@ -1219,15 +1304,27 @@ def put_scene_entity(request): if field not in data: return JsonResponse({"status": "error", "message": f"Missing required field: {field}"}, status=400) + if data["scene_type"] not in SCENE_TYPES: + return JsonResponse({"status": "error", "message": f"Unknown scene_type {data['scene_type']!r}. Expected one of: {', '.join(SCENE_TYPES)}."}, status=400) + if data["scene_type"] != "nspm_scene" and not data["backend_name"]: + return JsonResponse({"status": "error", "message": "backend_name is required for Home Assistant and OpenHAB scenes."}, status=400) + if "id" in data and data["id"]: new_scene = Scene.objects.get(id=int(data["id"])) else: new_scene = Scene() + room = Room.objects.get(id=int(data["room_id"])) if data["room_id"] else None + entities_page = RoomEntitiesPage.objects.get(id=int(data["entities_page_id"])) + room_view_position = int(data["room_view_position"]) + placement_error = get_placement_error(new_scene, room, entities_page, room_view_position) + if placement_error: + return JsonResponse({"status": "error", "message": placement_error}, status=400) + new_scene.friendly_name = data["friendly_name"] - new_scene.room = Room.objects.get(id=int(data["room_id"])) if data["room_id"] else None - new_scene.entities_page = RoomEntitiesPage.objects.get(id=int(data["entities_page_id"])) - new_scene.room_view_position = int(data["room_view_position"]) + new_scene.room = room + new_scene.entities_page = entities_page + new_scene.room_view_position = room_view_position new_scene.scene_type = data.get("scene_type", "") new_scene.backend_name = data.get("backend_name", "") diff --git a/docker/web/nspanelmanager/web/templates/base.html b/docker/web/nspanelmanager/web/templates/base.html index d4896ea0..7ce58cc7 100644 --- a/docker/web/nspanelmanager/web/templates/base.html +++ b/docker/web/nspanelmanager/web/templates/base.html @@ -68,9 +68,6 @@ {% endfor %}
- - {% if manager_address == "" %}{% include 'modals/initial_setup/welcome.html' %}{% endif %} - {% block content %} {% endblock %} diff --git a/docker/web/nspanelmanager/web/templates/index.html b/docker/web/nspanelmanager/web/templates/index.html index 5f3c2933..9951689f 100644 --- a/docker/web/nspanelmanager/web/templates/index.html +++ b/docker/web/nspanelmanager/web/templates/index.html @@ -4,7 +4,6 @@ {% block head %} {% load static %} -{% if manager_address == "" %}{% endif %} {% endblock %} {% block content %} diff --git a/docker/web/nspanelmanager/web/templates/modals/initial_setup/finished.html b/docker/web/nspanelmanager/web/templates/modals/initial_setup/finished.html deleted file mode 100644 index f611b63f..00000000 --- a/docker/web/nspanelmanager/web/templates/modals/initial_setup/finished.html +++ /dev/null @@ -1,61 +0,0 @@ -{% load static %} - - - - - - - - diff --git a/docker/web/nspanelmanager/web/templates/modals/initial_setup/home_assistant.html b/docker/web/nspanelmanager/web/templates/modals/initial_setup/home_assistant.html deleted file mode 100644 index 87f90def..00000000 --- a/docker/web/nspanelmanager/web/templates/modals/initial_setup/home_assistant.html +++ /dev/null @@ -1,74 +0,0 @@ -{% load static %} - - - - - - - - diff --git a/docker/web/nspanelmanager/web/templates/modals/initial_setup/manager_settings.html b/docker/web/nspanelmanager/web/templates/modals/initial_setup/manager_settings.html deleted file mode 100644 index 15aad28e..00000000 --- a/docker/web/nspanelmanager/web/templates/modals/initial_setup/manager_settings.html +++ /dev/null @@ -1,99 +0,0 @@ -{% load static %} - - - - - - - - - diff --git a/docker/web/nspanelmanager/web/templates/modals/initial_setup/mqtt.html b/docker/web/nspanelmanager/web/templates/modals/initial_setup/mqtt.html deleted file mode 100644 index 60e65169..00000000 --- a/docker/web/nspanelmanager/web/templates/modals/initial_setup/mqtt.html +++ /dev/null @@ -1,95 +0,0 @@ -{% load static %} - - - - - - - - diff --git a/docker/web/nspanelmanager/web/templates/modals/initial_setup/openhab.html b/docker/web/nspanelmanager/web/templates/modals/initial_setup/openhab.html deleted file mode 100644 index ca0f144d..00000000 --- a/docker/web/nspanelmanager/web/templates/modals/initial_setup/openhab.html +++ /dev/null @@ -1,70 +0,0 @@ -{% load static %} - - - - - - - - diff --git a/docker/web/nspanelmanager/web/templates/modals/initial_setup/welcome.html b/docker/web/nspanelmanager/web/templates/modals/initial_setup/welcome.html deleted file mode 100644 index 3f7313f3..00000000 --- a/docker/web/nspanelmanager/web/templates/modals/initial_setup/welcome.html +++ /dev/null @@ -1,59 +0,0 @@ -{% load static %} - - - - - - - - diff --git a/docker/web/nspanelmanager/web/tests.py b/docker/web/nspanelmanager/web/tests.py deleted file mode 100755 index 7ce503c2..00000000 --- a/docker/web/nspanelmanager/web/tests.py +++ /dev/null @@ -1,3 +0,0 @@ -from django.test import TestCase - -# Create your tests here. diff --git a/docker/web/nspanelmanager/web/tests/__init__.py b/docker/web/nspanelmanager/web/tests/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/docker/web/nspanelmanager/web/tests/harness.py b/docker/web/nspanelmanager/web/tests/harness.py new file mode 100644 index 00000000..105e68e0 --- /dev/null +++ b/docker/web/nspanelmanager/web/tests/harness.py @@ -0,0 +1,262 @@ +""" +Shared test harness for the NSPanelManager Django app. + +Every test should inherit from NSPMTestCase. It gives you: + + * An isolated in-memory database per test (via django.test.TestCase). + * self.mqttmanager -- a stand-in for the MQTTManager process. Views that change + config call send_mqttmanager_reload_command(); here that is recorded instead of + SIGUSR1-ing whatever nspm_mqttmanager happens to be running on the host. Use + self.assertManagerReloaded() / self.assertManagerNotReloaded() to verify the + manager would have been told to pick up the change. + * self.home_assistant -- a fake Home Assistant /api/states endpoint. Add states with + self.home_assistant.add("light.kitchen", friendly_name="Kitchen"). + self.openhab_items -- the item list returned by the (faked) OpenHAB API. + * assertLoadableByManager(obj) -- checks a saved Entity/Scene has the shape MQTTManager + parses when it reads the database, e.g. entity_data["controller"] is a known string. + * snapshot(obj) / assertOnlyChanged(obj, before, {...}) -- checks an edit changed the + named fields (entity_data keys as "entity_data.") and nothing else. + * self.client -- a Django test client that returns 500 responses instead of raising, + so tests can assert that bad input is rejected the same way a browser would see it. + * Builders for common fixtures (make_room, make_light, make_scene, ...) and put_json / + post_json helpers that send requests the way the React UI does. +""" + +import json +import sys +from typing import ClassVar +from unittest import mock + +from django.db import connection +from django.test import Client, TestCase + +import web.apps +from web.models import Entity, NSPanel, Room, RoomEntitiesPage, Scene +from web.settings_helper import get_setting_with_default, set_setting_value + +# MQTTManager process-control helpers in web.apps. Modules import these by name +# (`from .apps import x`), which binds a separate reference, so every web.* module holding +# one is patched -- found at runtime so that adding or removing an import needs no change here. +_MANAGER_HOOKS = ("send_mqttmanager_reload_command", "restart_mqtt_manager_process", "start_mqtt_manager") + + +def _modules_holding(name): + original = getattr(web.apps, name) + return [module_name for module_name, module in list(sys.modules.items()) if (module_name == "web" or module_name.startswith("web.")) and getattr(module, name, None) is original] + + +class FakeMQTTManager: + """Records the process-control calls the web app makes towards MQTTManager.""" + + def __init__(self): + self.reloads = 0 + self.restarts = 0 + + def _reload(self): + self.reloads += 1 + + def _restart(self): + self.restarts += 1 + + def reset(self): + self.reloads = 0 + self.restarts = 0 + + +class FakeHomeAssistant: + """Serves a canned /api/states response to web.home_assistant_api.""" + + def __init__(self): + self.states = [] + self.fail = False + + def add(self, entity_id, friendly_name=None, state="off", **attributes): + if friendly_name is not None: + attributes["friendly_name"] = friendly_name + self.states.append({"entity_id": entity_id, "state": state, "attributes": attributes}) + + def _get(self, url, *args, **kwargs): + response = mock.Mock() + if self.fail or not url.endswith("/api/states"): + response.status_code = 500 + response.json.return_value = {} + else: + response.status_code = 200 + response.json.return_value = self.states + return response + + +class NSPMTestCase(TestCase): + def setUp(self): + super().setUp() + self.client = Client(raise_request_exception=False) + + # Data migrations seed a "Dummy room" (and pages for it). Start every test from an + # empty configuration so tests only see what they create themselves. + Room.objects.all().delete() + RoomEntitiesPage.objects.all().delete() + + self.mqttmanager = FakeMQTTManager() + hooks = { + "send_mqttmanager_reload_command": self.mqttmanager._reload, + "restart_mqtt_manager_process": self.mqttmanager._restart, + "start_mqtt_manager": self.mqttmanager._restart, + } + for name in _MANAGER_HOOKS: + for module in _modules_holding(name): + self._patch(f"{module}.{name}", side_effect=hooks[name]) + # Safety net: even if a new code path finds another way to reach these, never + # let a test signal or spawn a real MQTTManager on the host. + self._patch("web.apps.psutil.process_iter", return_value=[]) + self._patch("web.apps.subprocess.Popen", side_effect=AssertionError("Tests must not start MQTTManager")) + + self.home_assistant = FakeHomeAssistant() + set_setting_value("home_assistant_address", "http://home-assistant.test") + set_setting_value("home_assistant_token", "test-token") + + # OpenHAB is faked one level higher, at its item-listing functions. + self.openhab_items = [] + self._patch("web.openhab_api.get_all_openhab_items", side_effect=lambda *args, **kwargs: {"items": self.openhab_items, "errors": []}) + self._patch("web.openhab_api.get_all_openhab_scenes", side_effect=lambda *args, **kwargs: {"items": self.openhab_items, "errors": []}) + + # Every module shares the one `requests` module, so route all HTTP through here: + # Home Assistant goes to the fake, anything else fails the test. + self._patch("requests.get", side_effect=self._fake_http_get) + + def _fake_http_get(self, url, *args, **kwargs): + if url.startswith(get_setting_with_default("home_assistant_address")): + return self.home_assistant._get(url, *args, **kwargs) + raise AssertionError(f"Unexpected network request from test: GET {url}") + + def _patch(self, target, **kwargs): + patcher = mock.patch(target, **kwargs) + patcher.start() + self.addCleanup(patcher.stop) + + # ---------------------------------------------------------------- assertions + + def assertManagerReloaded(self, times=None): + if times is None: + self.assertGreater(self.mqttmanager.reloads, 0, "Expected MQTTManager to be told to reload its config") + else: + self.assertEqual(self.mqttmanager.reloads, times, f"Expected MQTTManager to be told to reload {times} time(s)") + + def assertManagerNotReloaded(self): + self.assertEqual(self.mqttmanager.reloads, 0, "MQTTManager was told to reload although nothing should have changed") + + # MQTTManager reads entities straight from the SQLite database. These are the values its + # entity classes (docker/MQTTManager/include/{light,switch,button,thermostat}/) accept for + # entity_data["controller"]; anything else is logged as an error, and a non-string throws. + MANAGER_CONTROLLERS: ClassVar = { + Entity.EntityType.LIGHT: {"home_assistant", "openhab"}, + Entity.EntityType.SWITCH: {"home_assistant", "openhab"}, + Entity.EntityType.BUTTON: {"home_assistant", "nspm"}, + Entity.EntityType.THERMOSTAT: {"home_assistant", "openhab"}, + } + MANAGER_SCENE_TYPES: ClassVar = {"home_assistant", "openhab", "nspm_scene"} + + def assertLoadableByManager(self, obj): + """Check a saved Entity/Scene row has the shape MQTTManager expects to parse.""" + obj.refresh_from_db() + if isinstance(obj, Scene): + self.assertIn(obj.scene_type, self.MANAGER_SCENE_TYPES) + if obj.scene_type != "nspm_scene": + self.assertTrue(obj.backend_name, "Home Assistant/OpenHAB scenes need a backend_name to activate") + else: + # The ORM decodes the column for us, so check the stored text itself: that is what + # MQTTManager parses. A string that was JSON-encoded twice also fails here. + with connection.cursor() as cursor: + cursor.execute(f"SELECT entity_data FROM {Entity._meta.db_table} WHERE id = %s", [obj.id]) + (raw,) = cursor.fetchone() + try: + stored = json.loads(raw) + except (TypeError, ValueError): + self.fail(f"entity_data is not valid JSON: {raw!r}") + self.assertIsInstance(stored, dict, f"entity_data must be a JSON object, got {raw!r}") + controller = obj.entity_data.get("controller") + self.assertIsInstance(controller, str, f"entity_data['controller'] must be a string, got {controller!r}") + self.assertIn(controller, self.MANAGER_CONTROLLERS[obj.entity_type]) + self.assertIsNotNone(obj.entities_page_id) + self.assertLess(obj.room_view_position, obj.entities_page.page_type, "Entity is placed in a slot outside its page") + + def snapshot(self, obj): + """Every stored field of obj, with entity_data flattened to "entity_data.".""" + obj.refresh_from_db() + fields = {} + for field in obj._meta.concrete_fields: + value = getattr(obj, field.attname) + if field.attname == "entity_data": + fields.update({f"entity_data.{key}": item for key, item in value.items()}) + else: + fields[field.attname] = value + return fields + + def assertOnlyChanged(self, obj, before, expected): + """obj now differs from the snapshot before by exactly the fields in expected.""" + after = self.snapshot(obj) + changed = {key: after.get(key, "") for key in before.keys() | after.keys() if before.get(key, "") != after.get(key, "")} + self.assertEqual(changed, expected, "The edit changed fields other than the ones it was meant to") + + def assertRejected(self, response): + """The request was refused, however the view chose to signal it.""" + if response.status_code >= 400: + return + if response.get("Content-Type", "").startswith("application/json"): + self.assertEqual(response.json().get("status"), "error", f"Expected an error response, got: {response.content!r}") + return + self.fail(f"Expected request to be rejected, got HTTP {response.status_code}") + + # ---------------------------------------------------------------- builders + + def make_room(self, name="Room"): + """Create a room the way the UI does: with one entities page and one scenes page.""" + room = Room.objects.create(friendly_name=name) + RoomEntitiesPage.objects.create(room=room, page_type=4, is_scenes_page=True, display_order=0) + RoomEntitiesPage.objects.create(room=room, page_type=4, is_scenes_page=False, display_order=0) + return room + + def entities_page(self, room, index=0): + return RoomEntitiesPage.objects.filter(room=room, is_scenes_page=False).order_by("display_order")[index] + + def scenes_page(self, room, index=0): + return RoomEntitiesPage.objects.filter(room=room, is_scenes_page=True).order_by("display_order")[index] + + def global_scenes_page(self): + page, _ = RoomEntitiesPage.objects.get_or_create(room=None, is_scenes_page=True, defaults={"page_type": 4, "display_order": 0}) + return page + + def make_entity(self, room, entity_type=Entity.EntityType.LIGHT, name="Entity", page=None, slot=0, **entity_data): + entity_data.setdefault("controller", "home_assistant") + return Entity.objects.create( + room=room, + friendly_name=name, + entity_type=entity_type, + entities_page=page or self.entities_page(room), + room_view_position=slot, + entity_data=entity_data, + ) + + def make_light(self, room, name="Light", **kwargs): + return self.make_entity(room, Entity.EntityType.LIGHT, name, **kwargs) + + def make_scene(self, room, name="Scene", page=None, slot=0, scene_type="home_assistant", backend_name="scene.test"): + return Scene.objects.create( + room=room, + friendly_name=name, + scene_type=scene_type, + backend_name=backend_name, + entities_page=page or (self.scenes_page(room) if room else self.global_scenes_page()), + room_view_position=slot, + ) + + def make_panel(self, room, name="Panel", mac="AA:BB:CC:DD:EE:FF", **fields): + return NSPanel.objects.create(room=room, friendly_name=name, mac_address=mac, model=NSPanel.NSPanelModel.SONOFF, **fields) + + # ---------------------------------------------------------------- REST helpers + + def put_json(self, url, data): + """Send JSON the way the React UI does (fetch with a JSON body).""" + return self.client.put(url, json.dumps(data) if not isinstance(data, str) else data, content_type="application/json") + + def post_json(self, url, data): + return self.client.post(url, json.dumps(data) if not isinstance(data, str) else data, content_type="application/json") diff --git a/docker/web/nspanelmanager/web/tests/test_entities.py b/docker/web/nspanelmanager/web/tests/test_entities.py new file mode 100644 index 00000000..2cf59ee7 --- /dev/null +++ b/docker/web/nspanelmanager/web/tests/test_entities.py @@ -0,0 +1,351 @@ +""" +Creating and editing entities through the /rest/entities/* endpoints, using the same +payloads the React "add/edit entity" modal sends. +""" + +from django.urls import reverse + +from web.models import Entity, Scene + +from .harness import NSPMTestCase + + +def placement(page, slot): + return {"room_id": page.room_id, "entities_page_id": page.id, "room_view_position": slot} + + +class EntityRESTTestCase(NSPMTestCase): + """Base for the per-type entity endpoints. Subclasses define url and payload().""" + + url = None + model = Entity + + def setUp(self): + super().setUp() + self.room = self.make_room("Kitchen") + self.page = self.entities_page(self.room) + + def payload(self, slot=0, **overrides): + raise NotImplementedError + + def fields(self, payload): + """The dict holding the entity fields (lights wrap them in "values").""" + return payload + + def save(self, payload): + return self.put_json(reverse(self.url), payload) + + def assertCreatesNothing(self, response): + self.assertRejected(response) + self.assertFalse(self.model.objects.exists()) + self.assertManagerNotReloaded() + + +class EntityRESTCommonTests: + """Checks shared by every entity type. A mixin so it is not collected and run on its own.""" + + def test_every_required_field_is_enforced(self): + for field in list(self.fields(self.payload())): + with self.subTest(missing=field): + payload = self.payload() + del self.fields(payload)[field] + response = self.save(payload) + self.assertEqual(response.status_code, 400) + self.assertIn(field, response.json()["message"]) + self.assertFalse(self.model.objects.exists()) + self.assertManagerNotReloaded() + + def test_unknown_room_or_page_is_rejected(self): + for field in ("room_id", "entities_page_id"): + with self.subTest(field=field): + self.assertRejected(self.save(self.payload(**{field: 999}))) + self.assertFalse(self.model.objects.exists()) + self.assertManagerNotReloaded() + + def test_malformed_body_is_rejected(self): + self.assertCreatesNothing(self.put_json(reverse(self.url), "not json")) + + def test_edit_updates_in_place(self): + self.save(self.payload(slot=1)) + obj = self.model.objects.get() + before = self.snapshot(obj) + self.mqttmanager.reset() + + response = self.save(self.payload(slot=1, id=obj.id, friendly_name="Renamed")) + + self.assertEqual(response.status_code, 200) + self.assertEqual(self.model.objects.count(), 1) + self.assertOnlyChanged(obj, before, {"friendly_name": "Renamed"}) + self.assertLoadableByManager(obj) + self.assertManagerReloaded(times=1) + + def test_edit_unknown_id_is_rejected(self): + self.assertCreatesNothing(self.save(self.payload(id=999))) + + def test_other_methods_are_rejected(self): + # POST carries a valid payload, so only the method check can stop it being saved. + self.assertEqual(self.client.get(reverse(self.url)).status_code, 405) + self.assertCreatesNothing(self.post_json(reverse(self.url), self.payload())) + + def test_add_to_occupied_slot_is_rejected(self): + self.save(self.payload(slot=0)) + + self.assertRejected(self.save(self.payload(slot=0))) + self.assertEqual(self.model.objects.count(), 1) + + def test_add_to_slot_outside_page_is_rejected(self): + self.assertCreatesNothing(self.save(self.payload(slot=7))) + + +class LightTests(EntityRESTCommonTests, EntityRESTTestCase): + url = "rest_entities_lights" + + def payload(self, slot=0, **overrides): + values = { + **placement(self.page, slot), + "controller": "home_assistant", + "type": "light", + "friendly_name": "Ceiling", + "can_color_temperature": True, + "can_dim": True, + "can_rgb": False, + "controlled_by_nspanel_main_page": True, + "home_assistant_name": "light.kitchen_ceiling", + "is_ceiling_light": True, + "openhab_item_color_temp": "", + "openhab_item_dimmer": "", + "openhab_item_rgb": "", + } + return {"values": values | overrides} + + def fields(self, payload): + return payload["values"] + + def test_add_home_assistant_light(self): + response = self.save(self.payload(slot=2)) + + self.assertEqual(response.status_code, 200) + light = Entity.objects.get() + self.assertEqual(light.entity_type, Entity.EntityType.LIGHT) + self.assertEqual(light.friendly_name, "Ceiling") + self.assertEqual((light.room, light.entities_page, light.room_view_position), (self.room, self.page, 2)) + self.assertEqual(light.entity_data["home_assistant_name"], "light.kitchen_ceiling") + self.assertTrue(light.entity_data["can_dim"]) + self.assertTrue(light.entity_data["can_color_temperature"]) + self.assertFalse(light.entity_data["can_rgb"]) + self.assertTrue(light.entity_data["is_ceiling_light"]) + self.assertLoadableByManager(light) + self.assertManagerReloaded(times=1) + + def test_add_openhab_light(self): + payload = self.payload(controller="openhab", home_assistant_name="", can_dim=False, openhab_item_dimmer="Lamp_Switch") + + self.assertEqual(self.save(payload).status_code, 200) + + light = Entity.objects.get() + self.assertEqual(light.entity_data["openhab_control_mode"], "switch") + self.assertEqual(light.entity_data["openhab_item_dimmer"], "Lamp_Switch") + self.assertLoadableByManager(light) + + def test_boolean_flags_accept_strings(self): + self.save(self.payload(can_dim="false", can_rgb="true", is_ceiling_light="False")) + + data = Entity.objects.get().entity_data + self.assertEqual((data["can_dim"], data["can_rgb"], data["is_ceiling_light"]), (False, True, False)) + + def test_unknown_controller_is_rejected(self): + self.assertCreatesNothing(self.save(self.payload(controller="zigbee"))) + + def test_page_from_another_room_is_rejected(self): + other_page = self.entities_page(self.make_room("Lounge")) + + self.assertCreatesNothing(self.save(self.payload(entities_page_id=other_page.id))) + + def test_entity_saved_before_placement_checks_can_still_be_edited(self): + # Only a changed placement is validated, so rows already in a bad slot stay editable. + light = self.make_light(self.room, slot=7) + self.save(self.payload(slot=7, id=light.id)) + before = self.snapshot(light) + + response = self.save(self.payload(slot=7, id=light.id, friendly_name="Renamed")) + + self.assertEqual(response.status_code, 200) + self.assertOnlyChanged(light, before, {"friendly_name": "Renamed"}) + + +class SwitchTests(EntityRESTCommonTests, EntityRESTTestCase): + url = "rest_entities_switches" + + def payload(self, slot=0, **overrides): + return { + **placement(self.page, slot), + "controller": "home_assistant", + "type": "switch", + "friendly_name": "Fan", + "home_assistant_name": "switch.kitchen_fan", + "openhab_item_switch": "", + } | overrides + + def test_add_switch(self): + self.assertEqual(self.save(self.payload()).status_code, 200) + + switch = Entity.objects.get() + self.assertEqual(switch.entity_type, Entity.EntityType.SWITCH) + self.assertEqual(switch.entity_data["home_assistant_name"], "switch.kitchen_fan") + self.assertLoadableByManager(switch) + self.assertManagerReloaded(times=1) + + def test_add_openhab_switch(self): + self.save(self.payload(controller="openhab", home_assistant_name="", openhab_item_switch="Kitchen_Fan")) + + switch = Entity.objects.get() + self.assertEqual(switch.entity_data["openhab_item_switch"], "Kitchen_Fan") + self.assertLoadableByManager(switch) + + +class ButtonTests(EntityRESTCommonTests, EntityRESTTestCase): + url = "rest_entities_buttons" + + def payload(self, slot=0, **overrides): + return { + **placement(self.page, slot), + "controller": "nspm", + "type": "button", + "friendly_name": "Gate", + "home_assistant_name": "", + "mqtt_topic": "gate/cmd", + "mqtt_payload": "open", + } | overrides + + def test_add_mqtt_button(self): + self.assertEqual(self.save(self.payload()).status_code, 200) + + button = Entity.objects.get() + self.assertEqual(button.entity_type, Entity.EntityType.BUTTON) + self.assertEqual((button.entity_data["mqtt_topic"], button.entity_data["mqtt_payload"]), ("gate/cmd", "open")) + self.assertLoadableByManager(button) + self.assertManagerReloaded(times=1) + + def test_add_home_assistant_button(self): + self.save(self.payload(controller="home_assistant", home_assistant_name="input_button.doorbell", mqtt_topic="", mqtt_payload="")) + + button = Entity.objects.get() + self.assertEqual(button.entity_data["home_assistant_name"], "input_button.doorbell") + self.assertLoadableByManager(button) + + def test_openhab_controller_is_rejected(self): + # MQTTManager only has Home Assistant and MQTT ("nspm") buttons. + self.assertCreatesNothing(self.save(self.payload(controller="openhab"))) + + +class ThermostatTests(EntityRESTCommonTests, EntityRESTTestCase): + url = "rest_entities_thermostats" + + def payload(self, slot=0, **overrides): + return { + **placement(self.page, slot), + "controller": "home_assistant", + "type": "thermostat", + "friendly_name": "Heat pump", + "step_size": "0.5", + "use_current_temperature": "True", + "home_assistant_name": "climate.kitchen", + "openhab_fan_mode_item": "", + "openhab_hvac_mode_item": "", + "openhab_preset_mode_item": "", + "openhab_swing_mode_item": "", + "openhab_swingh_mode_item": "", + "openhab_temperature_item": "", + "fan_modes": [], + "hvac_modes": [{"value": "heat", "icon": "fire", "label": "Heat"}, {"value": "cool", "icon": "snow", "label": "Cool"}], + "preset_modes": [], + "swing_modes": [], + "swingh_modes": [], + } | overrides + + def test_add_thermostat(self): + self.assertEqual(self.save(self.payload()).status_code, 200) + + thermostat = Entity.objects.get() + self.assertEqual(thermostat.entity_type, Entity.EntityType.THERMOSTAT) + self.assertEqual(thermostat.entity_data["step_size"], 0.5) + self.assertEqual([m["value"] for m in thermostat.entity_data["hvac_modes"]], ["heat", "cool"]) + self.assertEqual(thermostat.entity_data["home_assistant_name"], "climate.kitchen") + self.assertIs(thermostat.entity_data["use_current_temperature"], True) + self.assertLoadableByManager(thermostat) + self.assertManagerReloaded(times=1) + + def test_mode_missing_a_field_is_rejected(self): + for field in ("value", "icon", "label"): + with self.subTest(missing=field): + mode = {"value": "heat", "icon": "fire", "label": "Heat"} + del mode[field] + self.assertEqual(self.save(self.payload(hvac_modes=[mode])).status_code, 400) + self.assertFalse(Entity.objects.exists()) + + def test_non_numeric_step_size_is_rejected(self): + self.assertCreatesNothing(self.save(self.payload(step_size="lots"))) + + +class SceneTests(EntityRESTCommonTests, EntityRESTTestCase): + url = "rest_entities_scenes" + model = Scene + + def setUp(self): + super().setUp() + self.page = self.scenes_page(self.room) + + def payload(self, slot=0, **overrides): + return { + **placement(self.page, slot), + "controller": "home_assistant", + "type": "scene", + "friendly_name": "Movie", + "scene_type": "home_assistant", + "backend_name": "scene.movie_time", + } | overrides + + def test_add_room_scene(self): + self.assertEqual(self.save(self.payload(slot=1)).status_code, 200) + + scene = Scene.objects.get() + self.assertEqual((scene.room, scene.entities_page, scene.room_view_position), (self.room, self.page, 1)) + self.assertEqual(scene.backend_name, "scene.movie_time") + self.assertLoadableByManager(scene) + self.assertManagerReloaded(times=1) + + def test_add_global_nspm_scene(self): + page = self.global_scenes_page() + + response = self.save(self.payload(room_id=None, entities_page_id=page.id, controller="nspm", scene_type="nspm_scene", backend_name="")) + + self.assertEqual(response.status_code, 200) + scene = Scene.objects.get() + self.assertIsNone(scene.room) + self.assertLoadableByManager(scene) + + def test_add_home_assistant_script_as_a_room_scene(self): + self.assertEqual(self.save(self.payload(friendly_name="Lights out", backend_name="script.lights_out")).status_code, 200) + + scene = Scene.objects.get() + self.assertEqual((scene.scene_type, scene.backend_name), ("home_assistant", "script.lights_out")) + self.assertEqual(scene.room, self.room) + self.assertLoadableByManager(scene) + self.assertManagerReloaded(times=1) + + def test_add_home_assistant_script_as_a_global_scene(self): + page = self.global_scenes_page() + + response = self.save(self.payload(room_id=None, entities_page_id=page.id, friendly_name="All off", backend_name="script.all_off")) + + self.assertEqual(response.status_code, 200) + scene = Scene.objects.get() + self.assertIsNone(scene.room) + self.assertEqual(scene.backend_name, "script.all_off") + self.assertLoadableByManager(scene) + + def test_home_assistant_scene_without_backend_name_is_rejected(self): + self.assertCreatesNothing(self.save(self.payload(backend_name=""))) + + def test_unknown_scene_type_is_rejected(self): + self.assertCreatesNothing(self.save(self.payload(scene_type="zigbee"))) diff --git a/docker/web/nspanelmanager/web/tests/test_files.py b/docker/web/nspanelmanager/web/tests/test_files.py new file mode 100644 index 00000000..0b3ef771 --- /dev/null +++ b/docker/web/nspanelmanager/web/tests/test_files.py @@ -0,0 +1,162 @@ +""" +Firmware, LittleFS and TFT images: uploading from the UI, and the checksum and download +endpoints panels use to decide whether to update and to fetch the image. +""" + +import hashlib +import shutil +import tempfile + +from django.core.files.uploadedfile import SimpleUploadedFile +from django.test import override_settings +from django.urls import reverse + +from web.settings_helper import set_nspanel_setting_value + +from .harness import NSPMTestCase + +FIRMWARE = bytes(range(256)) * 4 # 1 KiB with a recognisable pattern + + +def md5(data): + return hashlib.md5(data).hexdigest() + + +class FileTestCase(NSPMTestCase): + def setUp(self): + super().setUp() + media_root = tempfile.mkdtemp(prefix="nspm-media-") + self.addCleanup(shutil.rmtree, media_root, ignore_errors=True) + override = override_settings(MEDIA_ROOT=media_root) + override.enable() + self.addCleanup(override.disable) + + def upload(self, url_name, field, data, **form): + return self.client.post(reverse(url_name), {field: SimpleUploadedFile("upload.bin", data), **form}) + + +class FirmwareTests(FileTestCase): + def test_upload_then_checksum_and_download(self): + response = self.upload("save_new_firmware", "firmware", FIRMWARE, model="sonoff") + + self.assertRedirects(response, "/", fetch_redirect_response=False) + self.assertEqual(self.client.get(reverse("checksum_firmware")).content.decode(), md5(FIRMWARE)) + self.assertEqual(self.client.get(reverse("download_firmware")).content, FIRMWARE) + self.assertManagerReloaded(times=1) + + def test_models_are_stored_separately(self): + self.upload("save_new_firmware", "firmware", b"sonoff", model="sonoff") + self.upload("save_new_firmware", "firmware", b"custom", model="custom") + + self.assertEqual(self.client.get(reverse("download_firmware"), {"model": "custom"}).content, b"custom") + self.assertEqual(self.client.get(reverse("download_firmware"), {"model": "sonoff"}).content, b"sonoff") + self.assertEqual(self.client.get(reverse("checksum_firmware"), {"model": "custom"}).content.decode(), md5(b"custom")) + + def test_upload_replaces_previous_image(self): + self.upload("save_new_firmware", "firmware", b"old", model="sonoff") + self.upload("save_new_firmware", "firmware", b"new", model="sonoff") + + self.assertEqual(self.client.get(reverse("download_firmware")).content, b"new") + + def test_upload_without_model_is_ignored(self): + self.upload("save_new_firmware", "firmware", FIRMWARE) + + self.assertManagerNotReloaded() + self.assertRejected(self.client.get(reverse("download_firmware"))) + + def test_unknown_model_is_rejected(self): + for url_name in ("download_firmware", "checksum_firmware", "download_data_file", "checksum_data_file"): + with self.subTest(url_name=url_name): + self.assertEqual(self.client.get(reverse(url_name), {"model": "../../etc"}).status_code, 400) + + def test_range_requests_use_an_exclusive_end(self): + # CONTRACT with the panel firmware (NSPanelManager-firmware, UpdateManager.cpp): the + # "bytes=start-end" end is treated as EXCLUSIVE (Python slice), not inclusive as RFC 7233 + # says. The firmware deliberately sends ranges in the same convention. Changing either + # side alone corrupts OTA updates. + self.upload("save_new_firmware", "firmware", FIRMWARE, model="sonoff") + + response = self.client.get(reverse("download_firmware"), HTTP_RANGE="bytes=100-200") + + self.assertEqual(response.content, FIRMWARE[100:200]) + self.assertEqual(len(response.content), 100) + + def test_open_ended_range(self): + self.upload("save_new_firmware", "firmware", FIRMWARE, model="sonoff") + + self.assertEqual(self.client.get(reverse("download_firmware"), HTTP_RANGE="bytes=1000-").content, FIRMWARE[1000:]) + + def test_merged_flash_upload_does_not_reload_manager(self): + # MQTTManager does not track the merged image's checksum, so there is nothing to reload. + self.upload("save_new_merged_flash", "bin", FIRMWARE, model="sonoff") + + self.assertManagerNotReloaded() + + def test_checksum_of_missing_image_is_not_found(self): + panel = self.make_panel(self.make_room()) + for url in (reverse("checksum_firmware"), reverse("checksum_data_file"), reverse("checksum_tft_file", kwargs={"panel_id": panel.id})): + with self.subTest(url=url): + response = self.client.get(url) + self.assertEqual(response.status_code, 404) + self.assertEqual(response.content, b"") + + +class DataFileTests(FileTestCase): + def test_upload_then_checksum_and_download(self): + self.upload("save_new_data_file", "data_file", FIRMWARE, model="sonoff") + + self.assertEqual(self.client.get(reverse("checksum_data_file")).content.decode(), md5(FIRMWARE)) + self.assertEqual(self.client.get(reverse("download_data_file")).content, FIRMWARE) + self.assertEqual(self.client.get(reverse("download_data_file"), HTTP_RANGE="bytes=0-16").content, FIRMWARE[:16]) + self.assertManagerReloaded(times=1) + + +class TFTTests(FileTestCase): + def setUp(self): + super().setUp() + self.room = self.make_room() + self.panel = self.make_panel(self.room) + + def upload_tft(self, tft_file_type, data): + return self.upload("save_new_tft_file", "tft_file", data, tft_file_type=tft_file_type) + + def panel_tft(self, **headers): + return self.client.get(reverse("download_tft", kwargs={"panel_id": self.panel.id}), **headers).content + + def configure_panel(self, is_us_panel="False", orientation=None, tft="tft1"): + set_nspanel_setting_value(self.panel.id, "is_us_panel", is_us_panel) + set_nspanel_setting_value(self.panel.id, "selected_tft", tft) + if orientation: + set_nspanel_setting_value(self.panel.id, "us_panel_orientation", orientation) + + def test_each_panel_gets_the_image_for_its_type_and_selected_tft(self): + for tft_file_type in ("eu1", "eu3", "us2", "us4_horizontal_mirrored"): + self.upload_tft(tft_file_type, tft_file_type.encode()) + + cases = [ + ({}, b"eu1"), # defaults: EU panel, tft1 + ({"tft": "tft3"}, b"eu3"), + ({"is_us_panel": "True", "orientation": "vertical", "tft": "tft2"}, b"us2"), + # A horizontal US panel has the same screen orientation as an EU panel. + ({"is_us_panel": "True", "orientation": "horizontal", "tft": "tft3"}, b"eu3"), + ({"is_us_panel": "True", "orientation": "horizontal_mirrored", "tft": "tft4"}, b"us4_horizontal_mirrored"), + ] + for config, expected in cases: + with self.subTest(config=config): + self.configure_panel(**config) + self.assertEqual(self.panel_tft(), expected) + self.assertEqual(self.client.get(reverse("checksum_tft_file", kwargs={"panel_id": self.panel.id})).content.decode(), md5(expected)) + + def test_tft_range_download(self): + self.upload_tft("eu1", FIRMWARE) + + self.assertEqual(self.panel_tft(HTTP_RANGE="bytes=10-20"), FIRMWARE[10:20]) + + def test_upload_reloads_manager(self): + self.upload_tft("eu1", FIRMWARE) + + self.assertManagerReloaded(times=1) + + def test_unknown_tft_type_is_rejected(self): + self.assertEqual(self.upload_tft("eu9", FIRMWARE).status_code, 500) + self.assertManagerNotReloaded() diff --git a/docker/web/nspanelmanager/web/tests/test_layout.py b/docker/web/nspanelmanager/web/tests/test_layout.py new file mode 100644 index 00000000..91aa2e9d --- /dev/null +++ b/docker/web/nspanelmanager/web/tests/test_layout.py @@ -0,0 +1,286 @@ +""" +Moving and deleting entities and managing entities pages, through the REST endpoints the +React entities-page view calls. +""" + +from django.urls import reverse + +from web.models import Entity, RoomEntitiesPage, Scene + +from .harness import NSPMTestCase + + +def position(obj, page, slot): + return {"id": obj.id, "entities_page_id": page.id, "room_view_position": slot} + + +class ReadEntityTests(NSPMTestCase): + def test_get_entity(self): + room = self.make_room("Kitchen") + light = self.make_light(room, "Ceiling", slot=2, home_assistant_name="light.ceiling") + + response = self.client.get(reverse("rest_get_entity", kwargs={"entity_id": light.id})) + + self.assertEqual(response.status_code, 200) + body = response.json() + self.assertEqual((body["id"], body["friendly_name"], body["room_view_position"]), (light.id, "Ceiling", 2)) + self.assertEqual((body["controller"], body["home_assistant_name"]), ("home_assistant", "light.ceiling")) + + def test_get_room_pages_lists_entities_and_scenes(self): + room = self.make_room("Kitchen") + light = self.make_light(room) + scene = self.make_scene(room) + + response = self.client.get(reverse("rest_room_entities_pages", kwargs={"room_id": room.id})) + + self.assertEqual(response.status_code, 200) + pages = {page["type"]: page for page in response.json()["entities_pages"]} + self.assertEqual([e["id"] for e in pages["entity"]["entities"]], [light.id]) + self.assertEqual([s["id"] for s in pages["scene"]["scenes"]], [scene.id]) + + +class MoveEntityTests(NSPMTestCase): + def setUp(self): + super().setUp() + self.room = self.make_room("Kitchen") + self.page = self.entities_page(self.room) + self.second_page = RoomEntitiesPage.objects.create(room=self.room, page_type=8, is_scenes_page=False, display_order=1) + + def save_order(self, entities=(), scenes=()): + return self.put_json(reverse("rest_put_room_entities_order", kwargs={"room_id": self.room.id}), {"entities": list(entities), "scenes": list(scenes)}) + + def assertAt(self, obj, page, slot): + obj.refresh_from_db() + self.assertEqual((obj.entities_page_id, obj.room_view_position), (page.id, slot)) + + def test_move_entity(self): + light = self.make_light(self.room, slot=0) + + response = self.save_order([position(light, self.page, 3)]) + + self.assertEqual(response.status_code, 200) + self.assertAt(light, self.page, 3) + self.assertManagerReloaded(times=1) + + def test_swap_entities_across_pages(self): + ceiling = self.make_light(self.room, "Ceiling", slot=0) + lamp = self.make_light(self.room, "Lamp", page=self.second_page, slot=5) + + response = self.save_order([position(ceiling, self.second_page, 5), position(lamp, self.page, 0)]) + + self.assertEqual(response.status_code, 200) + self.assertAt(ceiling, self.second_page, 5) + self.assertAt(lamp, self.page, 0) + + def test_move_scenes(self): + scenes_page = self.scenes_page(self.room) + movie = self.make_scene(self.room, "Movie", slot=0) + dinner = self.make_scene(self.room, "Dinner", slot=1) + + response = self.save_order(scenes=[position(movie, scenes_page, 1), position(dinner, scenes_page, 0)]) + + self.assertEqual(response.status_code, 200) + self.assertAt(movie, scenes_page, 1) + self.assertAt(dinner, scenes_page, 0) + self.assertManagerReloaded(times=1) + + def test_bad_payloads_are_rejected(self): + light = self.make_light(self.room, slot=0) + url = reverse("rest_put_room_entities_order", kwargs={"room_id": self.room.id}) + for body in ({"entities": []}, {"scenes": []}, {"entities": [{"id": light.id}], "scenes": []}, "not json"): + with self.subTest(body=body): + self.assertRejected(self.put_json(url, body)) + self.assertAt(light, self.page, 0) + self.assertManagerNotReloaded() + + def test_unknown_entity_is_rejected(self): + self.assertRejected(self.save_order([{"id": 999, "entities_page_id": self.page.id, "room_view_position": 0}])) + self.assertManagerNotReloaded() + + def test_other_methods_are_rejected(self): + self.assertEqual(self.client.get(reverse("rest_put_room_entities_order", kwargs={"room_id": self.room.id})).status_code, 405) + + def test_failed_reorder_changes_nothing(self): + light = self.make_light(self.room, slot=0) + + response = self.save_order([position(light, self.page, 3), {"id": 999, "entities_page_id": self.page.id, "room_view_position": 0}]) + + self.assertRejected(response) + self.assertAt(light, self.page, 0) + self.assertManagerNotReloaded() + + def test_move_to_slot_outside_page_is_rejected(self): + light = self.make_light(self.room, slot=0) + + self.assertRejected(self.save_order([position(light, self.page, 11)])) + self.assertAt(light, self.page, 0) + + +class DeleteEntityTests(NSPMTestCase): + def setUp(self): + super().setUp() + self.room = self.make_room("Kitchen") + + def test_delete_entity(self): + doomed = self.make_light(self.room, "Ceiling", slot=1) + kept = self.make_light(self.room, "Lamp", slot=2) + + response = self.client.delete(reverse("rest_get_entity", kwargs={"entity_id": doomed.id})) + + self.assertEqual(response.status_code, 200) + self.assertEqual(list(Entity.objects.values_list("id", flat=True)), [kept.id]) + self.assertManagerReloaded(times=1) + + def test_delete_unknown_entity_is_rejected(self): + self.make_light(self.room) + + self.assertRejected(self.client.delete(reverse("rest_get_entity", kwargs={"entity_id": 999}))) + self.assertEqual(Entity.objects.count(), 1) + self.assertManagerNotReloaded() + + def test_delete_scene(self): + scene = self.make_scene(self.room) + + response = self.client.delete(reverse("rest_get_scene", kwargs={"scene_id": scene.id})) + + self.assertEqual(response.status_code, 200) + self.assertFalse(Scene.objects.exists()) + + def test_delete_scene_reloads_manager(self): + scene = self.make_scene(self.room) + + self.client.delete(reverse("rest_get_scene", kwargs={"scene_id": scene.id})) + + self.assertManagerReloaded(times=1) + + def test_delete_unknown_scene_is_rejected(self): + self.assertRejected(self.client.delete(reverse("rest_get_scene", kwargs={"scene_id": 999}))) + + +class EntitiesPageTests(NSPMTestCase): + def setUp(self): + super().setUp() + self.room = self.make_room("Kitchen") + + def pages(self, room, is_scenes_page=False): + return list(RoomEntitiesPage.objects.filter(room=room, is_scenes_page=is_scenes_page).order_by("display_order")) + + def create_page(self, page_type, is_scenes_page=False, room=None): + room = room or self.room + return self.put_json(reverse("rest_room_entities_pages", kwargs={"room_id": room.id}), {"type": page_type, "is_scenes_page": is_scenes_page}) + + def test_create_pages(self): + for page_type in (8, 12): + self.assertEqual(self.create_page(page_type).status_code, 200) + + pages = self.pages(self.room) + self.assertEqual([p.page_type for p in pages], [4, 8, 12]) + self.assertManagerReloaded(times=2) + + def test_create_global_scenes_page(self): + self.global_scenes_page() + + response = self.put_json(reverse("rest_global_entities_pages"), {"type": 4, "is_scenes_page": True}) + + self.assertEqual(response.status_code, 200) + self.assertEqual(len(self.pages(None, is_scenes_page=True)), 2) + + def test_create_page_missing_field_is_rejected(self): + before = RoomEntitiesPage.objects.count() + for body in ({"type": 4}, {"is_scenes_page": False}): + with self.subTest(body=body): + self.assertEqual(self.put_json(reverse("rest_room_entities_pages", kwargs={"room_id": self.room.id}), body).status_code, 400) + self.assertEqual(RoomEntitiesPage.objects.count(), before) + self.assertManagerNotReloaded() + + def test_create_page_in_unknown_room_is_rejected(self): + before = RoomEntitiesPage.objects.count() + + self.assertRejected(self.put_json(reverse("rest_room_entities_pages", kwargs={"room_id": 999}), {"type": 4, "is_scenes_page": False})) + + self.assertEqual(RoomEntitiesPage.objects.count(), before) + self.assertManagerNotReloaded() + + def test_create_page_with_unsupported_size_is_rejected(self): + before = RoomEntitiesPage.objects.count() + + self.assertRejected(self.create_page(6)) + self.assertEqual(RoomEntitiesPage.objects.count(), before) + + def test_delete_page_removes_its_entities(self): + second = RoomEntitiesPage.objects.create(room=self.room, page_type=8, is_scenes_page=False, display_order=2) + on_first = self.make_light(self.room, slot=0) + on_second = self.make_light(self.room, page=second, slot=0) + + response = self.client.delete(reverse("rest_room_entities_page", kwargs={"page_id": second.id})) + + self.assertEqual(response.status_code, 200) + self.assertFalse(RoomEntitiesPage.objects.filter(id=second.id).exists()) + self.assertEqual(list(Entity.objects.values_list("id", flat=True)), [on_first.id]) + self.assertNotEqual(on_second.id, on_first.id) + self.assertManagerReloaded(times=1) + + def test_delete_unknown_page_is_rejected(self): + self.assertRejected(self.client.delete(reverse("rest_room_entities_page", kwargs={"page_id": 999}))) + self.assertManagerNotReloaded() + + def test_reorder_pages(self): + first = self.entities_page(self.room) + second = RoomEntitiesPage.objects.create(room=self.room, page_type=8, is_scenes_page=False, display_order=2) + + response = self.put_json(reverse("rest_room_entities_pages_order"), {"order": [[second.id, 0], [first.id, 1]]}) + + self.assertEqual(response.status_code, 200) + self.assertEqual(self.pages(self.room), [second, first]) + self.assertManagerReloaded(times=1) + + def test_reorder_pages_bad_payload_is_rejected(self): + url = reverse("rest_room_entities_pages_order") + self.assertEqual(self.put_json(url, {}).status_code, 400) + self.assertRejected(self.put_json(url, {"order": [[999, 0]]})) + self.assertRejected(self.put_json(url, "not json")) + self.assertManagerNotReloaded() + + def test_resize_page(self): + page = self.entities_page(self.room) + + response = self.put_json(reverse("rest_room_entities_page", kwargs={"page_id": page.id}), {"id": page.id, "number_of_entities": 12}) + + self.assertEqual(response.status_code, 200) + page.refresh_from_db() + self.assertEqual(page.page_type, 12) + + def test_resize_page_reloads_manager(self): + page = self.entities_page(self.room) + + self.put_json(reverse("rest_room_entities_page", kwargs={"page_id": page.id}), {"number_of_entities": 12}) + + self.assertManagerReloaded(times=1) + + def test_shrinking_page_does_not_strand_entities(self): + page = RoomEntitiesPage.objects.create(room=self.room, page_type=12, is_scenes_page=False, display_order=2) + light = self.make_light(self.room, page=page, slot=9) + + response = self.put_json(reverse("rest_room_entities_page", kwargs={"page_id": page.id}), {"number_of_entities": 4}) + + self.assertRejected(response) + page.refresh_from_db() + self.assertEqual(page.page_type, 12) + self.assertLoadableByManager(light) + self.assertManagerNotReloaded() + + def test_shrinking_page_with_free_trailing_slots(self): + page = RoomEntitiesPage.objects.create(room=self.room, page_type=12, is_scenes_page=False, display_order=2) + light = self.make_light(self.room, page=page, slot=3) + + response = self.put_json(reverse("rest_room_entities_page", kwargs={"page_id": page.id}), {"number_of_entities": 4}) + + self.assertEqual(response.status_code, 200) + self.assertLoadableByManager(light) + + def test_resize_to_unsupported_size_is_rejected(self): + page = self.entities_page(self.room) + + self.assertRejected(self.put_json(reverse("rest_room_entities_page", kwargs={"page_id": page.id}), {"number_of_entities": 6})) + page.refresh_from_db() + self.assertEqual(page.page_type, 4) diff --git a/docker/web/nspanelmanager/web/tests/test_misc.py b/docker/web/nspanelmanager/web/tests/test_misc.py new file mode 100644 index 00000000..559752dd --- /dev/null +++ b/docker/web/nspanelmanager/web/tests/test_misc.py @@ -0,0 +1,174 @@ +""" +Smaller endpoints: messages, Home Assistant / OpenHAB entity lookups used by the entity +modal, hostname lookup and read-only listings. +""" + +import json +import socket +from unittest import mock + +from django.urls import reverse + +from web.models import Message + +from .harness import NSPMTestCase + + +class MessageTests(NSPMTestCase): + def test_unread_messages_are_flagged_and_can_be_marked_read(self): + message = Message.objects.create(title="Update available", message="v2.1 is out") + self.assertTrue(self.client.get(reverse("index")).context["has_unread_messages"]) + + response = self.client.get(reverse("htmx_mark_message_read", kwargs={"message_id": message.id})) + + self.assertContains(response, "Update available") + message.refresh_from_db() + self.assertTrue(message.read) + self.assertFalse(self.client.get(reverse("index")).context["has_unread_messages"]) + + def test_mark_unknown_message_is_rejected(self): + self.assertEqual(self.client.get(reverse("htmx_mark_message_read", kwargs={"message_id": 999})).status_code, 500) + + +class EntitySourceLookupTests(NSPMTestCase): + """The lists the React entity modal offers when picking a Home Assistant / OpenHAB item.""" + + def test_home_assistant_entities(self): + self.home_assistant.add("light.kitchen", friendly_name="Kitchen") + self.home_assistant.add("switch.fan", friendly_name="Fan") + + response = self.client.get(reverse("rest_get_home_assistant_entities")) + + self.assertEqual(response.status_code, 200) + self.assertEqual([(i["item_id"], i["label"]) for i in response.json()["items"]], [("light.kitchen", "Kitchen"), ("switch.fan", "Fan")]) + + def test_home_assistant_entities_filtered_by_domain(self): + self.home_assistant.add("light.kitchen") + self.home_assistant.add("switch.fan") + self.home_assistant.add("climate.lounge") + + response = self.client.get(reverse("rest_get_home_assistant_entities"), {"filter": json.dumps({"type": ["switch", "climate"]})}) + + self.assertEqual([i["item_id"] for i in response.json()["items"]], ["switch.fan", "climate.lounge"]) + + def test_home_assistant_scripts_are_offered_with_scenes(self): + # The React scene picker fetches every entity and keeps scene.* and script.* itself. + self.home_assistant.add("scene.movie_time", friendly_name="Movie time") + self.home_assistant.add("script.lights_out", friendly_name="Lights out") + self.home_assistant.add("light.kitchen") + + response = self.client.get(reverse("rest_get_home_assistant_entities")) + + items = {i["item_id"]: i["label"] for i in response.json()["items"]} + self.assertEqual(items["scene.movie_time"], "Movie time") + self.assertEqual(items["script.lights_out"], "Lights out") + + def test_entity_without_friendly_name_is_labelled_by_id(self): + self.home_assistant.add("light.unnamed") + + self.assertEqual(self.client.get(reverse("rest_get_home_assistant_entities")).json()["items"][0]["label"], "light.unnamed") + + def test_home_assistant_error_is_reported(self): + self.home_assistant.fail = True + + response = self.client.get(reverse("rest_get_home_assistant_entities")) + + self.assertEqual(response.json()["items"], []) + self.assertTrue(response.json()["errors"]) + + def test_home_assistant_not_configured_is_reported(self): + from web.settings_helper import set_setting_value + + set_setting_value("home_assistant_token", "") + + self.assertTrue(self.client.get(reverse("rest_get_home_assistant_entities")).json()["errors"]) + + def test_openhab_items_include_scenes(self): + self.openhab_items = [{"type": "openhab", "label": "Lamp", "item_id": "Lamp"}] + + response = self.client.get(reverse("rest_get_openhab_items")) + + # The fake returns the same list for items and for scenes (rules), so both are merged in. + self.assertEqual([i["item_id"] for i in response.json()["items"]], ["Lamp", "Lamp"]) + + def test_lookups_are_read_only(self): + self.assertEqual(self.client.post(reverse("rest_get_home_assistant_entities")).status_code, 405) + self.assertEqual(self.client.post(reverse("rest_get_openhab_items")).status_code, 405) + + +class HostnameLookupTests(NSPMTestCase): + def test_resolves_hostname(self): + with mock.patch("web.rest.socket.gethostbyname", return_value="192.168.1.20") as lookup: + response = self.post_json(reverse("get_ip_by_hostname"), {"hostname": "mqtt.local"}) + + lookup.assert_called_once_with("mqtt.local") + self.assertEqual(response.json(), {"ip": "192.168.1.20"}) + + def test_unknown_hostname_is_404(self): + with mock.patch("web.rest.socket.gethostbyname", side_effect=socket.gaierror): + self.assertEqual(self.post_json(reverse("get_ip_by_hostname"), {"hostname": "nope.invalid"}).status_code, 404) + + def test_missing_hostname_is_rejected(self): + with mock.patch("web.rest.socket.gethostbyname") as lookup: + self.assertEqual(self.post_json(reverse("get_ip_by_hostname"), {}).status_code, 500) + lookup.assert_not_called() + + +class ReadOnlyListingTests(NSPMTestCase): + def setUp(self): + super().setUp() + self.room = self.make_room("Kitchen") + + def test_room_entities(self): + light = self.make_light(self.room, "Ceiling") + self.make_light(self.make_room("Lounge"), "Lamp") + + response = self.client.get(reverse("rest_room_entities", kwargs={"room_id": self.room.id})) + + self.assertEqual([e["id"] for e in response.json()["entities"]], [light.id]) + + def test_global_entities_pages(self): + page = self.global_scenes_page() + scene = self.make_scene(None, "Goodnight") + + response = self.client.get(reverse("rest_global_entities_pages")) + + pages = response.json()["entities_pages"] + self.assertEqual([(p["id"], p["room_id"]) for p in pages], [(page.id, None)]) + self.assertEqual([s["id"] for s in pages[0]["scenes"]], [scene.id]) + + def test_single_entities_page(self): + light = self.make_light(self.room, slot=1) + + response = self.client.get(reverse("rest_room_entities_page", kwargs={"page_id": self.entities_page(self.room).id})) + + self.assertEqual([e["id"] for e in response.json()["entities"]], [light.id]) + + def test_get_scene(self): + scene = self.make_scene(self.room, "Dinner", backend_name="scene.dinner") + + result = self.client.get(reverse("rest_get_scene", kwargs={"scene_id": scene.id})).json()["result"] + + self.assertEqual((result["friendly_name"], result["backend_name"], result["room_id"]), ("Dinner", "scene.dinner", self.room.id)) + + def test_scenes_listing(self): + scene = self.make_scene(self.room, "Dinner") + self.make_scene(None, "Goodnight") + + response = self.client.get(reverse("rest_lights"), {"room_id": self.room.id}) + + self.assertEqual([s["scene_id"] for s in response.json()["scenes"]], [scene.id]) + + def test_legacy_scene_and_room_config(self): + scene = self.make_scene(self.room, "Dinner") + + self.assertEqual(self.client.get(reverse("get_scenes")).json()["scenes"][0]["scene_id"], scene.id) + self.assertEqual(self.client.get(reverse("get_nspanel_config", kwargs={"room_id": self.room.id})).json()["name"], "Kitchen") + + def test_scenes_listing_filters_by_scene_id(self): + scene = self.make_scene(self.room, "Dinner") + self.make_scene(self.room, "Breakfast") + + response = self.client.get(reverse("rest_lights"), {"scene_id": scene.id}) + + self.assertEqual([s["scene_id"] for s in response.json()["scenes"]], [scene.id]) diff --git a/docker/web/nspanelmanager/web/tests/test_pages.py b/docker/web/nspanelmanager/web/tests/test_pages.py new file mode 100644 index 00000000..41aaf033 --- /dev/null +++ b/docker/web/nspanelmanager/web/tests/test_pages.py @@ -0,0 +1,120 @@ +""" +Every page and htmx partial renders. These catch broken templates, template tags and +view context (e.g. a renamed model field referenced from a template), which otherwise +only show up when someone opens the page. +""" + +from django.urls import reverse + +from web.models import Message, RelayGroup +from web.settings_helper import set_setting_value + +from .harness import NSPMTestCase + + +class PageRenderTests(NSPMTestCase): + def setUp(self): + super().setUp() + self.room = self.make_room("Kitchen") + self.make_light(self.room, "Ceiling") + self.make_scene(self.room, "Dinner") + self.make_scene(None, "Goodnight", scene_type="nspm_scene", backend_name="") + self.panel = self.make_panel(self.room, "Kitchen panel", accepted=True) + self.denied_panel = self.make_panel(self.room, "Denied panel", mac="AA:BB:CC:DD:EE:01", denied=True) + self.relay_group = RelayGroup.objects.create(friendly_name="Hall lights") + Message.objects.create(title="Hello", message="World") + + def assertRenders(self, url): + response = self.client.get(url) + self.assertEqual(response.status_code, 200, f"GET {url} returned {response.status_code}") + return response + + def test_full_pages(self): + pages = [ + reverse("index"), + reverse("rooms"), + reverse("edit_room", kwargs={"room_id": self.room.id}), + reverse("edit_nspanel", kwargs={"panel_id": self.panel.id}), + reverse("settings"), + reverse("relay_groups"), + reverse("global_scenes"), + reverse("weather_and_time"), + reverse("denied_nspanels"), + ] + for url in pages: + with self.subTest(url=url): + self.assertRenders(url) + + def test_pages_render_with_no_configuration(self): + # A fresh install: no settings, no rooms, no panels. + self.room.delete() + for url in (reverse("index"), reverse("rooms"), reverse("settings"), reverse("weather_and_time"), reverse("global_scenes"), reverse("relay_groups")): + with self.subTest(url=url): + self.assertRenders(url) + + def test_index_shows_initial_setup_until_manager_address_is_configured(self): + self.assertContains(self.assertRenders(reverse("index")), "data-react-component='InitialSetup'") + + set_setting_value("manager_address", "192.168.1.10") + + self.assertNotContains(self.assertRenders(reverse("index")), "data-react-component='InitialSetup'") + + def test_htmx_partials(self): + partials = [ + reverse("htmx_partial_select_new_entity_type_react", kwargs={"entities_page_id": self.entities_page(self.room).id, "room_view_position": 1}), + reverse("htmx_partial_select_weather_location"), + reverse("htmx_partial_select_weather_outside_temperature_sensor"), + reverse("htmx_show_messages"), + reverse("htmx_modal_create_new_relay_group"), + reverse("htmx_modal_edit_relay_group", kwargs={"relay_group_id": self.relay_group.id}), + reverse("htmx_modal_add_relay_to_relay_group", kwargs={"relay_group_id": self.relay_group.id}), + ] + for url in partials: + with self.subTest(url=url): + self.assertRenders(url) + + def test_sensor_pickers_list_home_assistant_entities(self): + self.home_assistant.add("sensor.kitchen_temperature", friendly_name="Kitchen temperature") + + for url in ( + reverse("htmx_partial_select_room_temperature_sensor_from_list", kwargs={"entity_source": "home_assistant", "room_id": self.room.id}), + reverse("htmx_partial_select_weather_outside_temperature_sensor_from_list", kwargs={"entity_source": "home_assistant"}), + ): + with self.subTest(url=url): + self.assertContains(self.assertRenders(url), "Kitchen temperature") + + def test_sensor_provider_picker_skips_straight_to_the_only_configured_source(self): + room_list = reverse("htmx_partial_select_room_temperature_sensor_from_list", kwargs={"entity_source": "home_assistant", "room_id": self.room.id}) + weather_list = reverse("htmx_partial_select_weather_outside_temperature_sensor_from_list", kwargs={"entity_source": "home_assistant"}) + + self.assertRedirects(self.client.get(reverse("htmx_partial_select_room_temperature_sensor_provider", kwargs={"room_id": self.room.id})), room_list, fetch_redirect_response=False) + self.assertRedirects(self.client.get(reverse("htmx_partial_select_weather_outside_temperature_sensor_provider")), weather_list, fetch_redirect_response=False) + + set_setting_value("home_assistant_token", "") + set_setting_value("openhab_address", "http://openhab.test") + set_setting_value("openhab_token", "token") + self.assertRedirects( + self.client.get(reverse("htmx_partial_select_weather_outside_temperature_sensor_provider")), + reverse("htmx_partial_select_weather_outside_temperature_sensor_from_list", kwargs={"entity_source": "openhab"}), + fetch_redirect_response=False, + ) + + def test_sensor_provider_picker_offers_a_choice_when_both_are_configured(self): + set_setting_value("openhab_address", "http://openhab.test") + set_setting_value("openhab_token", "token") + + self.assertRenders(reverse("htmx_partial_select_room_temperature_sensor_provider", kwargs={"room_id": self.room.id})) + self.assertRenders(reverse("htmx_partial_select_weather_outside_temperature_sensor_provider")) + + def test_sensor_provider_picker_with_nothing_configured_is_an_error(self): + set_setting_value("home_assistant_token", "") + + self.assertRejected(self.client.get(reverse("htmx_partial_select_weather_outside_temperature_sensor_provider"))) + + def test_unknown_objects_are_rejected(self): + for url in ( + reverse("edit_nspanel", kwargs={"panel_id": 999}), + reverse("htmx_modal_edit_relay_group", kwargs={"relay_group_id": 999}), + ): + with self.subTest(url=url): + self.assertRejected(self.client.get(url)) diff --git a/docker/web/nspanelmanager/web/tests/test_panels.py b/docker/web/nspanelmanager/web/tests/test_panels.py new file mode 100644 index 00000000..c661d5e3 --- /dev/null +++ b/docker/web/nspanelmanager/web/tests/test_panels.py @@ -0,0 +1,276 @@ +""" +Panel lifecycle (accept, deny, delete, unblock), the per-panel settings form, and the +legacy panel-facing api/ endpoints. +""" + +from django.urls import reverse + +from web.models import Entity, NSPanel +from web.settings_helper import get_nspanel_setting_with_default, set_nspanel_setting_value + +from .harness import NSPMTestCase + + +class PanelLifecycleTests(NSPMTestCase): + def setUp(self): + super().setUp() + self.room = self.make_room("Kitchen") + self.panel = self.make_panel(self.room, "New panel") + + def test_accept_register_request(self): + lounge = self.make_room("Lounge") + self.panel.denied = True + self.panel.save() + + response = self.post_json(reverse("nspanel_accept", kwargs={"nspanel_id": self.panel.id}), {"room_id": lounge.id}) + + self.assertEqual(response.json(), {"status": "ok", "id": self.panel.id}) + self.panel.refresh_from_db() + self.assertEqual((self.panel.accepted, self.panel.denied, self.panel.room), (True, False, lounge)) + self.assertManagerReloaded(times=1) + + def test_accept_without_room_is_rejected(self): + self.assertRejected(self.post_json(reverse("nspanel_accept", kwargs={"nspanel_id": self.panel.id}), {})) + self.panel.refresh_from_db() + self.assertFalse(self.panel.accepted) + self.assertManagerNotReloaded() + + def test_deny_register_request(self): + response = self.client.post(reverse("nspanel_deny", kwargs={"nspanel_id": self.panel.id})) + + self.assertEqual(response.json(), {"status": "ok", "id": self.panel.id}) + self.panel.refresh_from_db() + self.assertEqual((self.panel.accepted, self.panel.denied), (False, True)) + self.assertManagerReloaded(times=1) + + def test_denied_panels_page_lists_denied_panel(self): + self.make_panel(self.room, "Blocked panel", mac="AA:BB:CC:DD:EE:01", denied=True) + + response = self.client.get(reverse("denied_nspanels")) + + self.assertContains(response, "Blocked panel") + self.assertNotContains(response, "New panel") + + def test_accept_and_deny_require_post(self): + for name in ("nspanel_accept", "nspanel_deny"): + with self.subTest(name=name): + self.assertEqual(self.client.get(reverse(name, kwargs={"nspanel_id": self.panel.id})).status_code, 405) + self.panel.refresh_from_db() + self.assertEqual((self.panel.accepted, self.panel.denied), (False, False)) + self.assertManagerNotReloaded() + + def test_accept_unknown_panel_is_rejected(self): + self.assertRejected(self.post_json(reverse("nspanel_accept", kwargs={"nspanel_id": 999}), {"room_id": self.room.id})) + self.assertManagerNotReloaded() + + def test_delete_panel(self): + set_nspanel_setting_value(self.panel.id, "screen_dim_level", "50") + + response = self.client.delete(reverse("delete_nspanel", kwargs={"nspanel_id": self.panel.id})) + + self.assertEqual(response.json(), {"status": "ok"}) + self.assertFalse(NSPanel.objects.exists()) + self.assertEqual(get_nspanel_setting_with_default(self.panel.id, "screen_dim_level", "gone"), "gone") + self.assertManagerReloaded(times=1) + + def test_delete_requires_delete_method(self): + self.assertEqual(self.client.post(reverse("delete_nspanel", kwargs={"nspanel_id": self.panel.id})).status_code, 405) + self.assertTrue(NSPanel.objects.exists()) + self.assertManagerNotReloaded() + + def test_delete_unknown_panel_is_rejected(self): + self.assertRejected(self.client.delete(reverse("delete_nspanel", kwargs={"nspanel_id": 999}))) + self.assertManagerNotReloaded() + + def test_unblock_removes_the_panel_so_it_can_register_again(self): + for url_name, method in (("htmx_unblock_nspanel", "delete"), ("unblock_nspanel", "get")): + with self.subTest(url_name=url_name): + panel = self.make_panel(self.room, "Blocked", mac=f"AA:BB:CC:DD:EE:{len(url_name):02d}", denied=True) + getattr(self.client, method)(reverse(url_name, kwargs={"nspanel_id": panel.id})) + self.assertFalse(NSPanel.objects.filter(id=panel.id).exists()) + self.assertManagerReloaded(times=2) + + def test_deleting_a_room_keeps_its_panels(self): + other = self.make_room("Lounge") + + self.client.get(reverse("delete_room", kwargs={"room_id": self.room.id})) + + self.panel.refresh_from_db() + self.assertEqual(self.panel.room, other) + + +# Every field the panel settings form submits, with "use global default" choices. +PANEL_FORM = { + "name": "Kitchen panel", + "button1_mode": "0", + "button1_detached_mode_light": "", + "button1_custom_mqtt_topic": "", + "button1_custom_mqtt_payload": "", + "button2_mode": "0", + "button2_detached_mode_light": "", + "button2_custom_mqtt_topic": "", + "button2_custom_mqtt_payload": "", + "screen_dim_level": "", + "screensaver_dim_level": "", + "screensaver_activation_timeout": "", + "screensaver_mode": "global", + "show_screensaver_inside_temperature": "global", + "show_screensaver_outside_temperature": "global", + "temperature_calibration": "", + "relay1_default_mode": "False", + "relay2_default_mode": "False", + "default_page": "0", + "lock_to_default_room": "False", + "reverse_relays": "False", + "button1_relay_lower_temperature": "", + "button1_relay_upper_temperature": "", + "button2_relay_lower_temperature": "", + "button2_relay_upper_temperature": "", + "panel_type": "eu", + "selected_tft": "tft1", + "relay1_is_light": "False", + "relay2_is_light": "False", +} + + +class PanelSettingsFormTests(NSPMTestCase): + def setUp(self): + super().setUp() + self.room = self.make_room("Kitchen") + self.panel = self.make_panel(self.room, "Panel", accepted=True) + + def save(self, **overrides): + return self.client.post(reverse("save_panel_settings", kwargs={"panel_id": self.panel.id}), PANEL_FORM | {"room_id": self.room.id} | overrides) + + def panel_setting(self, name, default=None): + return get_nspanel_setting_with_default(self.panel.id, name, default) + + def test_save_basic_settings(self): + lounge = self.make_room("Lounge") + + response = self.save(room_id=lounge.id, name="Renamed", default_page="2", temperature_calibration="-1.5") + + self.assertRedirects(response, reverse("edit_nspanel", kwargs={"panel_id": self.panel.id}), fetch_redirect_response=False) + self.panel.refresh_from_db() + self.assertEqual((self.panel.friendly_name, self.panel.room), ("Renamed", lounge)) + self.assertEqual(self.panel_setting("default_page"), "2") + self.assertEqual(self.panel_setting("temperature_calibration"), "-1.5") + self.assertManagerReloaded(times=1) + + def test_blank_overrides_fall_back_to_global_settings(self): + # Start with per-panel overrides, then submit the form with those fields blank + # ("use global default"): saving must delete the overrides, not keep them. + set_nspanel_setting_value(self.panel.id, "screen_dim_level", "40") + set_nspanel_setting_value(self.panel.id, "screensaver_mode", "no_background") + + self.save() + + self.assertIsNone(self.panel_setting("screen_dim_level")) + self.assertIsNone(self.panel_setting("screensaver_mode")) + self.assertEqual(self.panel_setting("temperature_calibration"), "0.0") + + def test_per_panel_overrides_are_stored(self): + self.save(screen_dim_level="40", screensaver_mode="no_background", show_screensaver_inside_temperature="False") + + self.assertEqual(self.panel_setting("screen_dim_level"), "40") + self.assertEqual(self.panel_setting("screensaver_mode"), "no_background") + self.assertEqual(self.panel_setting("show_screensaver_inside_temperature"), "False") + + def test_panel_types(self): + cases = { + "eu": ("False", None), + "us_vertical": ("True", "vertical"), + "us_horizontal": ("True", "horizontal"), + "us_horizontal_mirrored": ("True", "horizontal_mirrored"), + } + for panel_type, expected in cases.items(): + with self.subTest(panel_type=panel_type): + self.save(panel_type=panel_type) + is_us = self.panel_setting("is_us_panel") + self.assertEqual(is_us, expected[0]) + if expected[1]: + self.assertEqual(self.panel_setting("us_panel_orientation"), expected[1]) + + def test_button_detached_mode_links_an_entity(self): + light = self.make_light(self.room, "Ceiling") + + self.save(button1_mode="1", button1_detached_mode_light=str(light.id)) + + self.panel.refresh_from_db() + self.assertEqual(self.panel.button1_mode, 1) + self.assertEqual(self.panel.button1_detached_mode_entity, light) + + self.save(button1_mode="0") + + self.panel.refresh_from_db() + self.assertIsNone(self.panel.button1_detached_mode_entity) + + def test_deleting_the_detached_entity_unlinks_it(self): + light = self.make_light(self.room, "Ceiling") + self.save(button2_mode="1", button2_detached_mode_light=str(light.id)) + + Entity.objects.all().delete() + + self.panel.refresh_from_db() + self.assertIsNone(self.panel.button2_detached_mode_entity) + + def test_button_custom_mqtt_mode(self): + self.save(button2_mode="2", button2_custom_mqtt_topic="garage/door", button2_custom_mqtt_payload="toggle") + + self.assertEqual((self.panel_setting("button2_mqtt_topic"), self.panel_setting("button2_mqtt_payload")), ("garage/door", "toggle")) + + self.save(button2_mode="0") + + self.assertIsNone(self.panel_setting("button2_mqtt_topic")) + + def test_detached_mode_with_unknown_entity_is_rejected(self): + self.assertRejected(self.save(button1_mode="1", button1_detached_mode_light="999")) + self.panel.refresh_from_db() + self.assertEqual(self.panel.button1_mode, 0) + self.assertManagerNotReloaded() + + def test_non_numeric_temperature_calibration_is_rejected(self): + self.assertRejected(self.save(temperature_calibration="warm")) + self.assertManagerNotReloaded() + + def test_unknown_room_is_rejected(self): + self.assertRejected(self.save(room_id=999)) + self.panel.refresh_from_db() + self.assertEqual(self.panel.room, self.room) + + def test_edit_page_shows_saved_values(self): + self.save(name="Hallway", screen_dim_level="40") + + response = self.client.get(reverse("edit_nspanel", kwargs={"panel_id": self.panel.id})) + + self.assertContains(response, "Hallway") + self.assertEqual(response.context["settings"]["screen_dim_level"], "40") + + +class LegacyPanelAPITests(NSPMTestCase): + """api/* endpoints from before panel config moved to protobuf over MQTT.""" + + def setUp(self): + super().setUp() + self.room = self.make_room("Kitchen") + self.panel = self.make_panel(self.room, "Kitchen panel", mac="C0:49:EF:00:00:01") + + def test_get_nspanel_config(self): + response = self.client.get(reverse("get_nspanel_config"), {"mac": self.panel.mac_address}) + + self.assertEqual(response.status_code, 200) + self.assertEqual(response.json()["name"], "Kitchen panel") + + def test_get_nspanel_config_for_unknown_panel_is_rejected(self): + self.assertEqual(self.client.get(reverse("get_nspanel_config"), {"mac": "00:00:00:00:00:00"}).status_code, 500) + + def test_online_status_for_unknown_panel_is_rejected(self): + response = self.client.post(reverse("set_panel_online_status", kwargs={"panel_mac": "00:00:00:00:00:00"}), '{"state": "online"}', content_type="application/json") + + self.assertEqual(response.status_code, 500) + + def test_restart_mqtt_manager(self): + response = self.client.post(reverse("restart_mqtt_manager")) + + self.assertEqual(response.json(), {"result": "OK"}) + self.assertEqual(self.mqttmanager.restarts, 1) diff --git a/docker/web/nspanelmanager/web/tests/test_relay_groups.py b/docker/web/nspanelmanager/web/tests/test_relay_groups.py new file mode 100644 index 00000000..eb74882a --- /dev/null +++ b/docker/web/nspanelmanager/web/tests/test_relay_groups.py @@ -0,0 +1,166 @@ +""" +Relay groups: tie panel relays together so switching one switches them all. +Covered through both the form views and the htmx endpoints, plus the REST listing +MQTTManager reads. +""" + +from django.urls import reverse + +from web.models import NSPanel, RelayGroup, RelayGroupBinding + +from .harness import NSPMTestCase + + +class RelayGroupTestCase(NSPMTestCase): + def setUp(self): + super().setUp() + self.room = self.make_room("Hall") + self.panel = self.make_panel(self.room, "Hall panel", accepted=True) + self.other_panel = self.make_panel(self.room, "Stairs panel", mac="AA:BB:CC:DD:EE:01", accepted=True) + self.group = RelayGroup.objects.create(friendly_name="Hall lights") + + def bind(self, panel, relay_num): + return RelayGroupBinding.objects.create(relay_group=self.group, nspanel=panel, relay_num=relay_num) + + +class RelayGroupViewTests(RelayGroupTestCase): + def test_create_group(self): + response = self.client.post(reverse("create_or_update_relay_group"), {"relay_group_id": "", "relay_group_name": "Porch"}) + + self.assertRedirects(response, reverse("relay_groups"), fetch_redirect_response=False) + self.assertTrue(RelayGroup.objects.filter(friendly_name="Porch").exists()) + self.assertManagerReloaded(times=1) + + def test_rename_group(self): + self.client.post(reverse("create_or_update_relay_group"), {"relay_group_id": self.group.id, "relay_group_name": "Landing"}) + + self.group.refresh_from_db() + self.assertEqual(self.group.friendly_name, "Landing") + self.assertEqual(RelayGroup.objects.count(), 1) + + def test_add_relay_to_group(self): + response = self.client.post(reverse("add_nspanel_relay_to_group"), {"relay_group_id": self.group.id, "nspanel_id": self.panel.id, "relay_num": "2"}) + + self.assertEqual(response.status_code, 302) + binding = RelayGroupBinding.objects.get() + self.assertEqual((binding.relay_group, binding.nspanel, binding.relay_num), (self.group, self.panel, 2)) + self.assertManagerReloaded(times=1) + + def test_adding_the_same_relay_twice_is_a_no_op(self): + self.bind(self.panel, 1) + + self.client.post(reverse("add_nspanel_relay_to_group"), {"relay_group_id": self.group.id, "nspanel_id": self.panel.id, "relay_num": "1"}) + + self.assertEqual(RelayGroupBinding.objects.count(), 1) + self.assertManagerNotReloaded() + + def test_add_relay_with_unknown_group_or_panel_is_rejected(self): + for form in ({"relay_group_id": 999, "nspanel_id": self.panel.id, "relay_num": "1"}, {"relay_group_id": self.group.id, "nspanel_id": 999, "relay_num": "1"}): + with self.subTest(form=form): + self.assertRejected(self.client.post(reverse("add_nspanel_relay_to_group"), form)) + self.assertFalse(RelayGroupBinding.objects.exists()) + self.assertManagerNotReloaded() + + def test_remove_relay_from_group(self): + binding = self.bind(self.panel, 1) + kept = self.bind(self.other_panel, 1) + + self.client.get(reverse("delete_relay_group_binding", kwargs={"relay_binding_id": binding.id})) + + self.assertEqual(list(RelayGroupBinding.objects.all()), [kept]) + self.assertManagerReloaded(times=1) + + def test_delete_group_removes_its_bindings(self): + self.bind(self.panel, 1) + + self.client.get(reverse("delete_relay_group", kwargs={"relay_group_id": self.group.id})) + + self.assertFalse(RelayGroup.objects.exists()) + self.assertFalse(RelayGroupBinding.objects.exists()) + self.assertManagerReloaded(times=1) + + def test_delete_unknown_group_is_rejected(self): + self.assertRejected(self.client.get(reverse("delete_relay_group", kwargs={"relay_group_id": 999}))) + self.assertManagerNotReloaded() + + def test_deleting_a_panel_removes_its_bindings(self): + self.bind(self.panel, 1) + + NSPanel.objects.filter(id=self.panel.id).delete() + + self.assertFalse(RelayGroupBinding.objects.exists()) + self.assertTrue(RelayGroup.objects.exists()) + + def test_relay_groups_page_lists_groups_and_panels(self): + self.bind(self.panel, 2) + + response = self.client.get(reverse("relay_groups")) + + self.assertContains(response, "Hall lights") + self.assertContains(response, "Hall panel") + + +class RelayGroupHTMXTests(RelayGroupTestCase): + def test_create_and_rename(self): + response = self.client.post(reverse("htmx_save_relay_group"), {"name": "Porch"}) + + self.assertEqual(response["HX-Refresh"], "true") + porch = RelayGroup.objects.get(friendly_name="Porch") + + self.client.post(reverse("htmx_save_relay_group"), {"relay_group_id": porch.id, "name": "Deck"}) + + porch.refresh_from_db() + self.assertEqual(porch.friendly_name, "Deck") + + def test_add_and_remove_relay(self): + response = self.client.post(reverse("htmx_add_relay_to_relay_group", kwargs={"relay_group_id": self.group.id}), {"nspanel_id": self.panel.id, "relay_selection": "1"}) + + self.assertEqual(response["HX-Refresh"], "true") + binding = RelayGroupBinding.objects.get() + self.assertManagerReloaded(times=1) + + self.client.post(reverse("htmx_add_relay_to_relay_group", kwargs={"relay_group_id": self.group.id}), {"nspanel_id": self.panel.id, "relay_selection": "1"}) + self.assertEqual(RelayGroupBinding.objects.count(), 1) + + self.client.delete(reverse("htmx_remove_relay_from_relay_group", kwargs={"relay_binding_id": binding.id})) + + self.assertFalse(RelayGroupBinding.objects.exists()) + self.assertManagerReloaded(times=2) + + def test_add_relay_modal_only_offers_accepted_panels(self): + self.make_panel(self.room, "Pending panel", mac="AA:BB:CC:DD:EE:02") + + response = self.client.get(reverse("htmx_modal_add_relay_to_relay_group", kwargs={"relay_group_id": self.group.id})) + + self.assertContains(response, "Hall panel") + self.assertNotContains(response, "Pending panel") + + def test_delete_group(self): + self.bind(self.panel, 1) + + self.client.delete(reverse("htmx_delete_relay_group", kwargs={"relay_group_id": self.group.id})) + + self.assertFalse(RelayGroup.objects.exists()) + self.assertFalse(RelayGroupBinding.objects.exists()) + + def test_save_and_delete_group_reload_manager(self): + self.client.post(reverse("htmx_save_relay_group"), {"name": "Porch"}) + self.client.delete(reverse("htmx_delete_relay_group", kwargs={"relay_group_id": self.group.id})) + + self.assertManagerReloaded(times=2) + + +class RelayGroupRESTTests(RelayGroupTestCase): + def test_list_relay_groups(self): + self.bind(self.panel, 1) + self.bind(self.other_panel, 2) + + response = self.client.get(reverse("rest_relay_groups")) + + self.assertEqual(response.status_code, 200) + group = response.json()["relay_groups"][0] + self.assertEqual((group["relay_group_id"], group["name"]), (self.group.id, "Hall lights")) + self.assertCountEqual(group["relays"], [{"nspanel_id": self.panel.id, "relay_num": 1}, {"nspanel_id": self.other_panel.id, "relay_num": 2}]) + + def test_relay_groups_endpoint_is_read_only(self): + self.assertEqual(self.client.post(reverse("rest_relay_groups")).status_code, 405) diff --git a/docker/web/nspanelmanager/web/tests/test_rooms.py b/docker/web/nspanelmanager/web/tests/test_rooms.py new file mode 100644 index 00000000..a3883885 --- /dev/null +++ b/docker/web/nspanelmanager/web/tests/test_rooms.py @@ -0,0 +1,199 @@ +import json +from unittest import expectedFailure + +from django.urls import reverse + +from web.models import Entity, Room, RoomEntitiesPage, Scene + +from .harness import NSPMTestCase + + +class RoomUITests(NSPMTestCase): + """Rooms created/edited/deleted through the web UI views.""" + + def test_create_room(self): + response = self.client.post(reverse("save_new_room"), {"friendly_name": "Kitchen"}) + + room = Room.objects.get(friendly_name="Kitchen") + self.assertRedirects(response, reverse("edit_room", kwargs={"room_id": room.id}), fetch_redirect_response=False) + # A new room always gets one entities page and one scenes page. + self.assertEqual(RoomEntitiesPage.objects.filter(room=room, is_scenes_page=False).count(), 1) + self.assertEqual(RoomEntitiesPage.objects.filter(room=room, is_scenes_page=True).count(), 1) + self.assertManagerReloaded(times=1) + + def test_create_room_without_name_is_rejected(self): + response = self.client.post(reverse("save_new_room"), {}) + + self.assertRejected(response) + self.assertFalse(Room.objects.exists()) + self.assertManagerNotReloaded() + + def test_rename_room(self): + room = self.make_room("Kitchen") + + response = self.client.post( + reverse("update_room_form", kwargs={"room_id": room.id}), + {"friendly_name": "Galley", "room_temp_provider": "home_assistant", "room_temp_sensor": "sensor.galley_temp"}, + ) + + self.assertEqual(response.status_code, 302) + room.refresh_from_db() + self.assertEqual(room.friendly_name, "Galley") + self.assertEqual(room.room_temp_sensor, "sensor.galley_temp") + self.assertManagerReloaded(times=1) + + def test_delete_room_cascades_and_rehomes_panels(self): + kitchen = self.make_room("Kitchen") + lounge = self.make_room("Lounge") + panel = self.make_panel(kitchen) + self.make_light(kitchen) + self.make_scene(kitchen) + + self.client.get(reverse("delete_room", kwargs={"room_id": kitchen.id})) + + self.assertFalse(Room.objects.filter(id=kitchen.id).exists()) + self.assertFalse(Entity.objects.exists()) + self.assertFalse(Scene.objects.exists()) + self.assertFalse(RoomEntitiesPage.objects.filter(room_id=kitchen.id).exists()) + # The panel is not deleted with its room; it moves to another room. + panel.refresh_from_db() + self.assertEqual(panel.room, lounge) + self.assertManagerReloaded(times=1) + + def test_last_room_cannot_be_deleted(self): + room = self.make_room("Only room") + + self.client.get(reverse("delete_room", kwargs={"room_id": room.id})) + + self.assertTrue(Room.objects.filter(id=room.id).exists()) + self.assertManagerNotReloaded() + + def test_reorder_rooms(self): + a, b, c = self.make_room("A"), self.make_room("B"), self.make_room("C") + + response = self.client.post(reverse("htmx_partial_reorder_rooms"), {"htmx_form_save_rooms_order_field": json.dumps({"rooms": [c.id, a.id, b.id]})}) + + self.assertEqual(response.status_code, 200) + self.assertEqual(list(Room.objects.order_by("displayOrder").values_list("friendly_name", flat=True)), ["C", "A", "B"]) + self.assertManagerReloaded(times=1) + + def test_reorder_rooms_with_missing_or_empty_payload_is_rejected(self): + self.make_room("A") + for payload in ({}, {"htmx_form_save_rooms_order_field": json.dumps({})}, {"htmx_form_save_rooms_order_field": json.dumps({"rooms": []})}): + with self.subTest(payload=payload): + self.assertRejected(self.client.post(reverse("htmx_partial_reorder_rooms"), payload)) + self.assertManagerNotReloaded() + + +class RoomRESTTests(NSPMTestCase): + """The /rest/rooms JSON API used by external tools.""" + + def test_create_room(self): + response = self.post_json(reverse("rest_rooms_create"), {"name": "Office"}) + + self.assertEqual(response.status_code, 200) + room_id = response.json()["room_id"] + self.assertEqual(Room.objects.get(id=room_id).friendly_name, "Office") + + def test_create_room_rejects_bad_payloads(self): + for body in ({}, {"friendly_name": "wrong key"}, "not json"): + with self.subTest(body=body): + response = self.post_json(reverse("rest_rooms_create"), body) + self.assertEqual(response.status_code, 500) + self.assertEqual(response.json()["status"], "error") + self.assertFalse(Room.objects.exists()) + + def test_rooms_endpoint_rejects_unsupported_methods(self): + self.assertEqual(self.client.delete(reverse("rest_rooms_create")).status_code, 405) + room = self.make_room() + self.assertEqual(self.client.get(reverse("rest_room_delete", kwargs={"room_id": room.id})).status_code, 405) + + def test_delete_room(self): + room = self.make_room("Office") + self.make_light(room) + + response = self.client.delete(reverse("rest_room_delete", kwargs={"room_id": room.id})) + + self.assertEqual(response.status_code, 200) + self.assertEqual(response.json(), {"status": "ok", "room_id": room.id}) + self.assertFalse(Room.objects.exists()) + self.assertFalse(Entity.objects.exists()) + + def test_delete_unknown_room_is_rejected(self): + room = self.make_room("Office") + + response = self.client.delete(reverse("rest_room_delete", kwargs={"room_id": room.id + 100})) + + self.assertRejected(response) + self.assertTrue(Room.objects.filter(id=room.id).exists()) + + def test_list_rooms(self): + office = self.make_room("Office") + lounge = self.make_room("Lounge") + + response = self.client.get(reverse("rest_rooms_create")) + + self.assertEqual(response.status_code, 200) + self.assertEqual([(r["id"], r["friendly_name"]) for r in response.json()["rooms"]], [(office.id, "Office"), (lounge.id, "Lounge")]) + self.assertEqual(set(response.json()["rooms"][0]), {"id", "friendly_name", "display_order", "room_temp_provider", "room_temp_sensor"}) + + def test_get_single_room(self): + self.make_room("Office") + lounge = self.make_room("Lounge") + + response = self.client.get(reverse("rest_rooms_create"), {"id": lounge.id}) + + self.assertEqual([(r["id"], r["friendly_name"]) for r in response.json()["rooms"]], [(lounge.id, "Lounge")]) + + def test_update_room(self): + room = self.make_room("Office") + + response = self.put_json(reverse("rest_rooms_create"), {"id": room.id, "friendly_name": "Study", "room_temp_provider": "home_assistant", "room_temp_sensor": "sensor.study"}) + + self.assertEqual(response.status_code, 200) + room.refresh_from_db() + self.assertEqual((room.friendly_name, room.room_temp_provider, room.room_temp_sensor), ("Study", "home_assistant", "sensor.study")) + self.assertManagerReloaded(times=1) + + @expectedFailure + def test_update_room_display_order(self): + # KNOWN BUG: room_put checks for "displayOrder" but reads data["display_order"], so + # sending display_order (the key GET /rest/rooms returns) is silently ignored and + # sending displayOrder raises KeyError and returns 500. + room = self.make_room("Office") + + response = self.put_json(reverse("rest_rooms_create"), {"id": room.id, "display_order": 7}) + + self.assertEqual(response.status_code, 200) + room.refresh_from_db() + self.assertEqual(room.displayOrder, 7) + + def test_delete_room_reloads_manager(self): + room = self.make_room("Office") + self.make_room("Lounge") + + self.client.delete(reverse("rest_room_delete", kwargs={"room_id": room.id})) + + self.assertManagerReloaded(times=1) + + def test_delete_room_rehomes_panels(self): + office = self.make_room("Office") + lounge = self.make_room("Lounge") + panel = self.make_panel(office) + + self.client.delete(reverse("rest_room_delete", kwargs={"room_id": office.id})) + + panel.refresh_from_db() + self.assertEqual(panel.room, lounge) + + def test_created_room_has_entity_pages(self): + response = self.post_json(reverse("rest_rooms_create"), {"name": "Office"}) + + room_id = response.json()["room_id"] + self.assertEqual(RoomEntitiesPage.objects.filter(room_id=room_id).count(), 2) + + def test_create_room_rejects_overlong_name(self): + response = self.post_json(reverse("rest_rooms_create"), {"name": "x" * 100}) + + self.assertRejected(response) + self.assertFalse(Room.objects.exists()) diff --git a/docker/web/nspanelmanager/web/tests/test_settings.py b/docker/web/nspanelmanager/web/tests/test_settings.py new file mode 100644 index 00000000..7e9efebc --- /dev/null +++ b/docker/web/nspanelmanager/web/tests/test_settings.py @@ -0,0 +1,287 @@ +""" +Global settings: the settings page form, weather & time, theme, the first-run setup +wizard, the settings REST endpoints and web.settings_helper. +""" + +from django.urls import reverse + +from web.models import NSPanelSettings, Settings +from web.settings_helper import ( + delete_nspanel_setting, + does_setting_exist, + get_nspanel_setting_with_default, + get_setting_with_default, + set_nspanel_setting_value, + set_setting_value, +) + +from .harness import NSPMTestCase + +# Every field the settings page form submits. +SETTINGS_FORM = { + "mqtt_server": "mqtt.local", + "mqtt_port": "1883", + "mqtt_username": "nspm", + "mqtt_password": "", + "home_assistant_address": "http://ha.local:8123", + "home_assistant_token": "", + "openhab_address": "", + "openhab_token": "", + "raise_to_100_light_level": "95", + "default_nspanel_type": "eu", + "color_temp_min": "2000", + "color_temp_max": "6000", + "reverse_color_temp": "False", + "min_button_push_time": "50", + "button_long_press_time": "300", + "special_mode_trigger_time": "300", + "special_mode_release_time": "5000", + "screensaver_activation_timeout": "30000", + "screen_dim_level": "100", + "screensaver_dim_level": "0", + "screensaver_mode": "with_background", + "show_screensaver_inside_temperature": "True", + "show_screensaver_outside_temperature": "True", + "turn_on_behavior": "color_temp", + "max_live_log_messages": "10", + "max_log_buffer_size": "10", + "mqttmanager_log_level": "info", + "manager_address": "192.168.1.10", + "manager_port": "8000", + "optimistic_mode": "optimistic", + "all_rooms_status_backoff_time": "250", + "room_status_backoff_time": "250", + "light_turn_on_brightness": "50", +} + + +def setting(name): + return get_setting_with_default(name) + + +class SettingsPageTests(NSPMTestCase): + def save(self, **overrides): + return self.client.post(reverse("save_settings"), SETTINGS_FORM | overrides) + + def test_save_settings(self): + response = self.save() + + self.assertRedirects(response, reverse("settings"), fetch_redirect_response=False) + for name in ("mqtt_server", "mqtt_port", "color_temp_min", "screensaver_mode", "manager_address", "light_turn_on_brightness"): + with self.subTest(name=name): + self.assertEqual(setting(name), SETTINGS_FORM[name]) + self.assertEqual(setting("optimistic_mode"), "True") + self.assertManagerReloaded(times=1) + + def test_trailing_slash_is_stripped_from_addresses(self): + self.save(home_assistant_address="http://ha.local:8123/", openhab_address="http://openhab.local/") + + self.assertEqual(setting("home_assistant_address"), "http://ha.local:8123") + self.assertEqual(setting("openhab_address"), "http://openhab.local") + + def test_blank_secret_fields_keep_the_stored_secret(self): + # The form never echoes secrets back, so a blank field means "unchanged". + set_setting_value("mqtt_password", "hunter2") + set_setting_value("home_assistant_token", "ha-token") + set_setting_value("openhab_token", "oh-token") + + self.save() + + self.assertEqual((setting("mqtt_password"), setting("home_assistant_token"), setting("openhab_token")), ("hunter2", "ha-token", "oh-token")) + + def test_new_secrets_replace_the_stored_ones(self): + set_setting_value("mqtt_password", "hunter2") + + self.save(mqtt_password="correct horse", home_assistant_token="new-token") + + self.assertEqual((setting("mqtt_password"), setting("home_assistant_token")), ("correct horse", "new-token")) + + def test_clear_checkboxes_remove_secrets(self): + set_setting_value("mqtt_password", "hunter2") + set_setting_value("openhab_token", "oh-token") + + self.save(clear_mqtt_password="on", clear_home_assistant_token="on", clear_openhab_token="on", mqtt_password="ignored") + + self.assertEqual((setting("mqtt_password"), setting("home_assistant_token"), setting("openhab_token")), ("", "", "")) + + def test_optional_fields_fall_back_to_defaults(self): + form = {k: v for k, v in SETTINGS_FORM.items() if k not in ("show_screensaver_inside_temperature", "turn_on_behavior", "optimistic_mode")} + set_setting_value("optimistic_mode", "False") + + self.client.post(reverse("save_settings"), form) + + self.assertEqual(setting("show_screensaver_inside_temperature"), "True") + self.assertEqual(setting("turn_on_behavior"), "color_temp") + self.assertEqual(setting("optimistic_mode"), "False") # left alone when not submitted + + def test_missing_required_field_is_rejected(self): + form = {k: v for k, v in SETTINGS_FORM.items() if k != "manager_port"} + + self.assertRejected(self.client.post(reverse("save_settings"), form)) + self.assertManagerNotReloaded() + + def test_settings_page_shows_saved_values_but_not_secrets(self): + self.save(mqtt_password="hunter2", home_assistant_token="ha-secret-token") + + response = self.client.get(reverse("settings")) + + self.assertContains(response, "mqtt.local") + self.assertNotContains(response, "hunter2") + self.assertNotContains(response, "ha-secret-token") + + +WEATHER_FORM = { + "location_latitude": "-36.85", + "location_longitude": "174.76", + "wind_speed_format": "kmh", + "precipitation_format": "mm", + "outside_temp_provider": "home_assistant", + "outside_temp_sensor": "sensor.outside", + "weather_update_interval": "10", + "date_format": "%d/%m/%Y", + "clock_us_style": "False", + "use_fahrenheit": "True", +} + + +class WeatherAndThemeTests(NSPMTestCase): + def test_save_weather_and_time(self): + response = self.client.post(reverse("weather_and_time"), WEATHER_FORM) + + self.assertRedirects(response, reverse("weather_and_time"), fetch_redirect_response=False) + self.assertEqual(setting("location_latitude"), "-36.85") + self.assertEqual(setting("weather_wind_speed_format"), "kmh") + self.assertEqual(setting("outside_temp_sensor_entity_id"), "sensor.outside") + self.assertEqual(setting("use_fahrenheit"), "True") + self.assertManagerReloaded(times=1) + + def test_weather_form_missing_field_is_rejected(self): + form = {k: v for k, v in WEATHER_FORM.items() if k != "use_fahrenheit"} + + self.assertRejected(self.client.post(reverse("weather_and_time"), form)) + self.assertManagerNotReloaded() + + def test_interface_theme(self): + response = self.client.post(reverse("htmx_interface_theme"), {"theme-dropdown": "dracula"}) + + self.assertEqual(response.status_code, 200) + self.assertEqual(setting("theme"), "dracula") + self.assertContains(self.client.get(reverse("index")), "dracula") + + +# Everything the React first-run wizard submits. +WIZARD_SETTINGS = { + "manager_address": "192.168.1.10", + "manager_port": "8000", + "mqtt_server": "mqtt.local", + "mqtt_port": "1883", + "mqtt_username": "u", + "mqtt_password": "p", + "home_assistant_address": "http://ha.local", + "home_assistant_token": "t", + "openhab_address": "", + "openhab_token": "", +} + + +class InitialSetupTests(NSPMTestCase): + """The first-run wizard (React InitialSetup) saves everything in one POST /rest/settings.""" + + def test_wizard_saves_all_settings(self): + response = self.post_json(reverse("rest_settings"), {"settings": WIZARD_SETTINGS}) + + self.assertEqual(response.status_code, 200) + for name, value in WIZARD_SETTINGS.items(): + self.assertEqual(setting(name), value) + self.assertManagerReloaded(times=1) + + def test_only_submitted_settings_change(self): + set_setting_value("mqtt_password", "keep-me") + + self.post_json(reverse("rest_settings"), {"settings": {"mqtt_server": "mqtt.local"}}) + + self.assertEqual(setting("mqtt_server"), "mqtt.local") + self.assertEqual(setting("mqtt_password"), "keep-me") + + def test_bad_payloads_are_rejected(self): + before = list(Settings.objects.values_list("name", "value")) + for body in ({}, {"settings": ["mqtt_server"]}, {"settings": "mqtt_server"}): + with self.subTest(body=body): + self.assertEqual(self.post_json(reverse("rest_settings"), body).status_code, 400) + self.assertEqual(list(Settings.objects.values_list("name", "value")), before) + self.assertManagerNotReloaded() + + +class SettingsRESTTests(NSPMTestCase): + def configure_secrets(self): + set_setting_value("mqtt_password", "hunter2") + set_setting_value("home_assistant_token", "ha-secret") + set_setting_value("openhab_token", "") + + def test_settings_endpoint_masks_secrets(self): + self.configure_secrets() + set_setting_value("mqtt_server", "mqtt.local") + + response = self.client.get(reverse("rest_settings")) + + self.assertEqual(response.status_code, 200) + settings = response.json()["settings"] + self.assertEqual(settings["mqtt_server"], "mqtt.local") + for secret in ("mqtt_password", "home_assistant_token", "openhab_token"): + self.assertNotIn(secret, settings) + self.assertEqual((settings["mqtt_password_set"], settings["home_assistant_token_set"], settings["openhab_token_set"]), (True, True, False)) + self.assertNotIn("hunter2", response.content.decode()) + + def test_settings_endpoint_rejects_unsupported_methods(self): + self.assertEqual(self.client.put(reverse("rest_settings")).status_code, 405) + + def test_settings_endpoint_works_before_all_secrets_are_set(self): + set_setting_value("mqtt_password", "hunter2") + + response = self.client.get(reverse("rest_settings")) + + self.assertEqual(response.status_code, 200) + settings = response.json()["settings"] + self.assertEqual((settings["mqtt_password_set"], settings["home_assistant_token_set"], settings["openhab_token_set"]), (True, True, False)) + self.assertNotIn("mqtt_password", settings) + self.assertNotIn("home_assistant_token", settings) + + +class SettingsHelperTests(NSPMTestCase): + def test_global_settings(self): + self.assertEqual(get_setting_with_default("not_set"), "") + self.assertFalse(does_setting_exist("not_set")) + + set_setting_value("colour", "blue") + set_setting_value("colour", "green") # update, not duplicate + + self.assertEqual(get_setting_with_default("colour"), "green") + self.assertTrue(does_setting_exist("colour")) + self.assertEqual(Settings.objects.filter(name="colour").count(), 1) + + def test_values_are_returned_as_strings(self): + set_setting_value("level", 42) + + self.assertEqual(get_setting_with_default("level"), "42") + + def test_panel_settings(self): + room = self.make_room() + panel = self.make_panel(room) + other = self.make_panel(room, mac="AA:BB:CC:DD:EE:01") + + self.assertEqual(get_nspanel_setting_with_default(panel.id, "dim", "default"), "default") + set_nspanel_setting_value(panel.id, "dim", "50") + set_nspanel_setting_value(panel.id, "dim", "60") + + self.assertEqual(get_nspanel_setting_with_default(panel.id, "dim", "default"), "60") + self.assertEqual(get_nspanel_setting_with_default(other.id, "dim", "default"), "default") + self.assertEqual(NSPanelSettings.objects.filter(nspanel=panel, name="dim").count(), 1) + + delete_nspanel_setting(panel.id, "dim") + self.assertEqual(get_nspanel_setting_with_default(panel.id, "dim", "default"), "default") + + def test_panel_settings_for_unknown_panel(self): + self.assertEqual(get_nspanel_setting_with_default(999, "dim", "default"), "default") + set_nspanel_setting_value(999, "dim", "50") # silently ignored + delete_nspanel_setting(999, "dim") + self.assertFalse(NSPanelSettings.objects.exists()) diff --git a/docker/web/nspanelmanager/web/urls.py b/docker/web/nspanelmanager/web/urls.py index 4d38d8f3..306874a7 100644 --- a/docker/web/nspanelmanager/web/urls.py +++ b/docker/web/nspanelmanager/web/urls.py @@ -56,7 +56,6 @@ path("rest/openhab/items", rest.get_openhab_items, name="rest_get_openhab_items"), path("rest/openhab/test", rest.test_openhab, name="rest_test_openhab"), # MQTTManager endpoints - path("rest/mqttmanager/settings/", rest.mqttmanager_get_setting, name="rest_mqttmanager_get_setting"), # Test if we can react manager on given IP and port path("rest/mqttmanager/test", rest.test_mqttmanager, name="rest_test_mqttmanager"), # Relay groups @@ -112,16 +111,11 @@ path("htmx/partial/edit_light_entity_in_room/", htmx.partial_entity_edit_light_entity, name="htmx_partial_edit_light_entity"), path("htmx/partial/edit_scene_entity/", htmx.partial_entity_edit_scene_entity, name="htmx_partial_edit_scene_entity"), path("htmx/partial/remove_entity_from_page_slot//", htmx.partial_remove_entity_from_page_slot, name="htmx_partial_remove_entity_from_page_slot"), - path("htmx/partial/edit_entities_page/", htmx.partial_edit_entities_page, name="htmx_partial_edit_entities_page"), - path("htmx/partial/edit_entities_page//save/", htmx.partial_save_edit_entities_page, name="htmx_partial_save_edit_entities_page"), path("htmx/partial/move_entity", htmx.partial_move_entity, name="htmx_partial_move_entity"), path("htmx/partial/reorder_entity_pages", htmx.partial_move_entities_pages, name="htmx_partial_move_entities_pages"), path("htmx/partial/delete_entities_page/", htmx.partial_delete_entities_page, name="htmx_partial_delete_entities_page"), path("htmx/partial/create_entities_page_in_room////", htmx.create_entities_page_in_room, name="htmx_create_entities_page_in_room"), path("htmx/partial/reorder_rooms", htmx.partial_reorder_rooms, name="htmx_partial_reorder_rooms"), - path("htmx/nspanels//accept_register_request", htmx.nspanel_accept_register_request, name="htmx_nspanel_accept_register_request"), - path("htmx/nspanels//deny_register_request", htmx.nspanel_deny_register_request, name="htmx_nspanel_deny_register_request"), - path("htmx/nspanels//delete", htmx.nspanel_delete, name="htmx_nspanel_delete"), path("htmx/partial/select_room_temperature_sensor_from_list//", htmx.select_room_temperature_sensor_from_list, name="htmx_partial_select_room_temperature_sensor_from_list"), path("htmx/partial/select_room_temperature_sensor_provider/", htmx.select_room_temperature_sensor_provider, name="htmx_partial_select_room_temperature_sensor_provider"), path("htmx/partial/select_weather_location", htmx.select_weather_location, name="htmx_partial_select_weather_location"), diff --git a/docker/web/nspanelmanager/web/views.py b/docker/web/nspanelmanager/web/views.py index 8ede62f8..88dc89ec 100755 --- a/docker/web/nspanelmanager/web/views.py +++ b/docker/web/nspanelmanager/web/views.py @@ -697,6 +697,13 @@ def download_tft(request, panel_id): return HttpResponse(fs.open(tft_file).read(), content_type="application/octet-stream") +def checksum_response(md5sum): + # A panel compares the body against its stored checksum, so never send "None" as if it were one. + if md5sum is None: + return HttpResponse("", status=404) + return HttpResponse(md5sum) + + def checksum_firmware(request): if "model" in request.GET: model = request.GET["model"] @@ -706,7 +713,7 @@ def checksum_firmware(request): # As model was not specified, default to sonoff as that was the original model model = "sonoff" - return HttpResponse(get_file_md5sum(f"firmware/{model}/firmware.bin")) + return checksum_response(get_file_md5sum(f"firmware/{model}/firmware.bin")) def checksum_data_file(request): @@ -718,7 +725,7 @@ def checksum_data_file(request): # As model was not specified, default to sonoff as that was the original model model = "sonoff" - return HttpResponse(get_file_md5sum(f"firmware/{model}/data_file.bin")) + return checksum_response(get_file_md5sum(f"firmware/{model}/data_file.bin")) def checksum_tft_file(request, panel_id): @@ -739,7 +746,7 @@ def checksum_tft_file(request, panel_id): else: print(f"ERROR! Could not determine TFT file for NSPanel with ID {panel_id}") - return HttpResponse(get_file_md5sum(tft_file)) + return checksum_response(get_file_md5sum(tft_file)) def get_manual(request): diff --git a/docker/web/run_tests.sh b/docker/web/run_tests.sh new file mode 100755 index 00000000..b922c44f --- /dev/null +++ b/docker/web/run_tests.sh @@ -0,0 +1,6 @@ +#!/bin/sh +# Run the Django test suite. Extra arguments are passed to "manage.py test", e.g. +# ./run_tests.sh web.tests.test_entities +# ./run_tests.sh -v 2 +cd "$(dirname "$0")/nspanelmanager" || exit 1 +exec python manage.py test --settings=nspanelmanager.test_settings "$@"