diff --git a/src/anthias_server/api/serializers/mixins.py b/src/anthias_server/api/serializers/mixins.py index 004767fe6..fa6d6d2ca 100644 --- a/src/anthias_server/api/serializers/mixins.py +++ b/src/anthias_server/api/serializers/mixins.py @@ -230,3 +230,12 @@ class RebootViewSerializerMixin(Serializer[Any]): class ShutdownViewSerializerMixin(Serializer[Any]): pass + + +class DisplayPowerViewSerializerMixin(Serializer[Any]): + # Read-only `message` so drf-spectacular generates an accurate + # schema for every status the endpoint returns (200/400/502/503). + # All four bodies share the same one-key shape; declaring it once + # keeps the OpenAPI document honest and clients can rely on the + # field being present regardless of outcome. + message = CharField(read_only=True) diff --git a/src/anthias_server/api/tests/test_v2_endpoints.py b/src/anthias_server/api/tests/test_v2_endpoints.py index 8f99a49e1..d00229af3 100644 --- a/src/anthias_server/api/tests/test_v2_endpoints.py +++ b/src/anthias_server/api/tests/test_v2_endpoints.py @@ -508,3 +508,104 @@ def test_integrations_non_balena_environment( 'balena_host_os_version': None, 'balena_device_name_at_init': None, } + + +# --- Display power (experimental, HDMI-CEC) ------------------------- + + +@pytest.mark.django_db +@mock.patch( + 'anthias_server.api.views.mixins.diagnostics.cec_available', + return_value=True, +) +@mock.patch( + 'anthias_server.api.views.mixins.diagnostics.set_display_power', + return_value=(True, 'Display turn-on command sent.'), +) +def test_display_power_on_success( + set_display_power_mock: Any, + _cec_available_mock: Any, + api_client: APIClient, +) -> None: + response = api_client.post( + reverse('api:display_power_v2', kwargs={'state': 'on'}) + ) + assert response.status_code == status.HTTP_200_OK + set_display_power_mock.assert_called_once_with(on=True) + assert 'sent' in response.data['message'] + + +@pytest.mark.django_db +@mock.patch( + 'anthias_server.api.views.mixins.diagnostics.cec_available', + return_value=True, +) +@mock.patch( + 'anthias_server.api.views.mixins.diagnostics.set_display_power', + return_value=(True, 'Display turn-off command sent.'), +) +def test_display_power_off_success( + set_display_power_mock: Any, + _cec_available_mock: Any, + api_client: APIClient, +) -> None: + response = api_client.post( + reverse('api:display_power_v2', kwargs={'state': 'off'}) + ) + assert response.status_code == status.HTTP_200_OK + set_display_power_mock.assert_called_once_with(on=False) + + +@pytest.mark.django_db +@mock.patch( + 'anthias_server.api.views.mixins.diagnostics.cec_available', + return_value=True, +) +@mock.patch( + 'anthias_server.api.views.mixins.diagnostics.set_display_power', + return_value=(False, 'Display turn-on failed: no adapter'), +) +def test_display_power_failure_returns_502( + _set_display_power_mock: Any, + _cec_available_mock: Any, + api_client: APIClient, +) -> None: + response = api_client.post( + reverse('api:display_power_v2', kwargs={'state': 'on'}) + ) + assert response.status_code == status.HTTP_502_BAD_GATEWAY + assert 'no adapter' in response.data['message'] + + +@pytest.mark.django_db +@mock.patch('anthias_server.api.views.mixins.diagnostics.set_display_power') +def test_display_power_invalid_state_returns_400( + set_display_power_mock: Any, api_client: APIClient +) -> None: + response = api_client.post( + reverse('api:display_power_v2', kwargs={'state': 'foo'}) + ) + assert response.status_code == status.HTTP_400_BAD_REQUEST + set_display_power_mock.assert_not_called() + + +@pytest.mark.django_db +@mock.patch( + 'anthias_server.api.views.mixins.diagnostics.cec_available', + return_value=False, +) +@mock.patch('anthias_server.api.views.mixins.diagnostics.set_display_power') +def test_display_power_returns_503_when_no_cec_adapter( + set_display_power_mock: Any, + _cec_available_mock: Any, + api_client: APIClient, +) -> None: + """The endpoint must fail fast (no 10 s subprocess) when neither + /dev/cec0 nor /dev/vchiq exists. 503 telegraphs 'this server lacks + the hardware to satisfy the request' more accurately than 502.""" + response = api_client.post( + reverse('api:display_power_v2', kwargs={'state': 'on'}) + ) + assert response.status_code == status.HTTP_503_SERVICE_UNAVAILABLE + set_display_power_mock.assert_not_called() + assert 'adapter' in response.data['message'] diff --git a/src/anthias_server/api/urls/v2.py b/src/anthias_server/api/urls/v2.py index 18010ed79..d3641ae39 100644 --- a/src/anthias_server/api/urls/v2.py +++ b/src/anthias_server/api/urls/v2.py @@ -8,6 +8,7 @@ AssetViewV2, BackupViewV2, DeviceSettingsViewV2, + DisplayPowerViewV2, FileAssetViewV2, InfoViewV2, IntegrationsViewV2, @@ -43,6 +44,11 @@ def get_url_patterns() -> list[URLPattern | URLResolver]: path('v2/recover', RecoverViewV2.as_view(), name='recover_v2'), path('v2/reboot', RebootViewV2.as_view(), name='reboot_v2'), path('v2/shutdown', ShutdownViewV2.as_view(), name='shutdown_v2'), + path( + 'v2/display/', + DisplayPowerViewV2.as_view(), + name='display_power_v2', + ), path('v2/file_asset', FileAssetViewV2.as_view(), name='file_asset_v2'), path( 'v2/assets//content', diff --git a/src/anthias_server/api/views/mixins.py b/src/anthias_server/api/views/mixins.py index 4e620ee64..a336d13b0 100644 --- a/src/anthias_server/api/views/mixins.py +++ b/src/anthias_server/api/views/mixins.py @@ -21,6 +21,7 @@ from anthias_server.api.helpers import save_active_assets_ordering from anthias_server.api.serializers.mixins import ( BackupViewSerializerMixin, + DisplayPowerViewSerializerMixin, PlaylistOrderSerializerMixin, RebootViewSerializerMixin, ShutdownViewSerializerMixin, @@ -179,6 +180,54 @@ def post(self, request: Request) -> Response: return Response(status=status.HTTP_200_OK) +class DisplayPowerViewMixin(APIView): + serializer_class = DisplayPowerViewSerializerMixin + + @extend_schema( + summary='Set display power state (experimental, HDMI-CEC)', + parameters=[ + OpenApiParameter( + name='state', + location=OpenApiParameter.PATH, + type=OpenApiTypes.STR, + enum=['on', 'off'], + description=( + 'Desired display power state. Only valid on ' + 'CEC-capable hardware.' + ), + ), + ], + # Every status returns the same `{message: ...}` shape. Mapping + # each code to the serializer keeps drf-spectacular's generated + # OpenAPI document accurate so clients know what to parse. + responses={ + 200: DisplayPowerViewSerializerMixin, + 400: DisplayPowerViewSerializerMixin, + 502: DisplayPowerViewSerializerMixin, + 503: DisplayPowerViewSerializerMixin, + }, + ) + @authorized + def post(self, request: Request, state: str) -> Response: + if state not in ('on', 'off'): + return Response( + {'message': 'Invalid display state.'}, + status=status.HTTP_400_BAD_REQUEST, + ) + # No /dev/cec0 or /dev/vchiq — fail fast with 503 rather than + # spawning a 10 s libcec subprocess that's guaranteed to error. + if not diagnostics.cec_available(): + return Response( + {'message': 'No HDMI-CEC adapter detected on this device.'}, + status=status.HTTP_503_SERVICE_UNAVAILABLE, + ) + ok, msg = diagnostics.set_display_power(on=(state == 'on')) + if ok: + return Response({'message': msg}, status=status.HTTP_200_OK) + # 502: upstream CEC adapter / TV refused or didn't respond. + return Response({'message': msg}, status=status.HTTP_502_BAD_GATEWAY) + + class FileAssetViewMixin(APIView): @extend_schema( summary='Upload file asset', diff --git a/src/anthias_server/api/views/v2.py b/src/anthias_server/api/views/v2.py index fc9ce7b62..fe05cdaaa 100644 --- a/src/anthias_server/api/views/v2.py +++ b/src/anthias_server/api/views/v2.py @@ -57,6 +57,7 @@ AssetsControlViewMixin, BackupViewMixin, DeleteAssetViewMixin, + DisplayPowerViewMixin, FileAssetViewMixin, InfoViewMixin, PlaylistOrderViewMixin, @@ -536,6 +537,10 @@ class ShutdownViewV2(ShutdownViewMixin): pass +class DisplayPowerViewV2(DisplayPowerViewMixin): + pass + + class FileAssetViewV2(FileAssetViewMixin): pass diff --git a/src/anthias_server/app/page_context.py b/src/anthias_server/app/page_context.py index 04b2eafce..63cd600e3 100644 --- a/src/anthias_server/app/page_context.py +++ b/src/anthias_server/app/page_context.py @@ -232,6 +232,10 @@ def device_settings() -> dict[str, Any]: # on that revision (matches the React audio-output dropdown). 'is_pi5': 'Raspberry Pi 5' in device_model, 'date_format_options': _DATE_FORMAT_OPTIONS, + # Render-time gate for the experimental CEC display-power + # buttons; cec_available() only stats device nodes, so it's + # cheap enough to call on every settings render. + 'cec_available': diagnostics.cec_available(), } diff --git a/src/anthias_server/app/static/sass/_styles.scss b/src/anthias_server/app/static/sass/_styles.scss index 48dcfd464..8b3739262 100644 --- a/src/anthias_server/app/static/sass/_styles.scss +++ b/src/anthias_server/app/static/sass/_styles.scss @@ -1532,6 +1532,19 @@ label { text-align: left; } display: flex; gap: var(--space-2); } + .settings-section__badge { + display: inline-block; + margin-left: var(--space-2); + padding: 0.125rem 0.5rem; + border-radius: var(--radius-md); + background: var(--color-warning-soft, #fef3c7); + color: var(--color-warning-strong, #92400e); + font-size: 0.6875rem; + font-weight: 600; + letter-spacing: 0.03em; + text-transform: uppercase; + vertical-align: middle; + } } .toggle-row { diff --git a/src/anthias_server/app/templates/settings.html b/src/anthias_server/app/templates/settings.html index 944a55653..e07a80599 100644 --- a/src/anthias_server/app/templates/settings.html +++ b/src/anthias_server/app/templates/settings.html @@ -247,5 +247,35 @@

+ + {% if cec_available %} + {# ===== Display power (experimental, HDMI-CEC) ===== #} +
+
+
+

Display power Experimental

+

+ Toggle the connected display over HDMI-CEC. Works only on + CEC-capable hardware (most x86 PCs are not). The outcome + will appear as a toast. +

+
+
+
+ {% csrf_token %} + +
+
+ {% csrf_token %} + +
+
+
+
+ {% endif %} {% endblock %} diff --git a/src/anthias_server/app/urls.py b/src/anthias_server/app/urls.py index 8fc0f1e66..850ec0f5b 100644 --- a/src/anthias_server/app/urls.py +++ b/src/anthias_server/app/urls.py @@ -29,6 +29,11 @@ views.settings_shutdown, name='settings_shutdown', ), + path( + 'settings/display//', + views.settings_display_power, + name='settings_display_power', + ), path( 'settings/migrate-to-screenly/', views.migrate_to_screenly, diff --git a/src/anthias_server/app/views.py b/src/anthias_server/app/views.py index 35d97a57b..114cdee38 100644 --- a/src/anthias_server/app/views.py +++ b/src/anthias_server/app/views.py @@ -990,6 +990,26 @@ def settings_shutdown(request: HttpRequest) -> HttpResponse: return redirect(reverse('anthias_app:settings')) +@authorized +@require_http_methods(['POST']) +def settings_display_power(request: HttpRequest, state: str) -> HttpResponse: + if state not in ('on', 'off'): + messages.error(request, 'Invalid display state.') + return redirect(reverse('anthias_app:settings')) + # Guard the (otherwise hidden) endpoint so a stale form or a + # direct curl on non-CEC hardware doesn't burn the request thread + # on a 10 s libcec subprocess that's guaranteed to fail. + if not diagnostics.cec_available(): + messages.error( + request, + 'No HDMI-CEC adapter detected on this device.', + ) + return redirect(reverse('anthias_app:settings')) + ok, msg = diagnostics.set_display_power(on=(state == 'on')) + (messages.success if ok else messages.error)(request, msg) + return redirect(reverse('anthias_app:settings')) + + @authorized @require_http_methods(['GET']) def system_info(request: HttpRequest) -> HttpResponse: diff --git a/src/anthias_server/lib/diagnostics.py b/src/anthias_server/lib/diagnostics.py index 1acb681ec..2dba9de87 100755 --- a/src/anthias_server/lib/diagnostics.py +++ b/src/anthias_server/lib/diagnostics.py @@ -23,6 +23,30 @@ sys.stdout.write('Unknown') """ +# Issued from the settings page / REST endpoint, *not* from a celery +# worker, so a hung libcec call would block the request thread until +# the subprocess timeout fires. Same subprocess+timeout shape as +# `_CEC_QUERY_SCRIPT` for the same reason: libcec C calls don't +# honour Python signals. +_CEC_SET_SCRIPT = """ +import sys +try: + import cec + cec.init() + tv = cec.Device(cec.CECDEVICE_TV) +except Exception as exc: + sys.stdout.write('ERROR: ' + (str(exc) or 'CEC stack unavailable')) + sys.exit(0) +try: + if {on}: + tv.power_on() + else: + tv.standby() + sys.stdout.write('OK') +except Exception as exc: + sys.stdout.write('ERROR: ' + (str(exc) or 'CEC command failed')) +""" + def get_display_power() -> str | bool: """ @@ -51,6 +75,80 @@ def get_display_power() -> str | bool: return output or 'CEC error' +def set_display_power(on: bool) -> tuple[bool, str]: + """Send a CEC power_on / standby to the connected TV. + + Returns ``(ok, message)`` for direct surfacing to the operator as + a toast. Stays synchronous on purpose — the issue brief asks for + an immediate feedback loop so failed CEC commands aren't silent. + """ + script = _CEC_SET_SCRIPT.format(on='True' if on else 'False') + verb = 'on' if on else 'off' + try: + result = subprocess.run( + [sys.executable, '-c', script], + capture_output=True, + timeout=10, + ) + except subprocess.TimeoutExpired: + return ( + False, + f'Display turn-{verb} timed out — CEC adapter unresponsive.', + ) + + output = result.stdout.decode('utf-8', errors='replace').strip() + if output == 'OK': + return True, f'Display turn-{verb} command sent.' + if output.startswith('ERROR: '): + return False, ( + f'Display turn-{verb} failed: ' + f'{_trim_cec_detail(output[len("ERROR: ") :])}' + ) + + # Subprocess didn't emit one of the two contract sentinels. The + # likely causes are an interpreter crash (returncode != 0) or + # libcec writing its diagnostic to stderr instead of stdout — both + # would surface as "unexpected CEC response." without further + # detail, which is useless to an operator. Fall back to stderr (or + # the raw stdout if non-empty) so the toast / API response carries + # something actionable. + stderr = result.stderr.decode('utf-8', errors='replace').strip() + detail = ( + stderr or output + ) or f'subprocess exited with status {result.returncode}' + return False, f'Display turn-{verb} failed: {_trim_cec_detail(detail)}' + + +def _trim_cec_detail(detail: str) -> str: + """Sanitize an arbitrarily-sized libcec / Python error blob into a + one-line, length-capped toast / JSON message. + + libcec (and the in-subprocess Python) can emit multi-line tracebacks + or kilobyte-scale diagnostics on either stdout or stderr. The last + non-empty line is almost always the actual exception/error message, + so we keep that and drop the rest, then cap to 240 chars so the toast + stack doesn't overflow and JSON responses stay small. + """ + lines = [line for line in detail.splitlines() if line.strip()] + one_line = lines[-1].strip() if lines else detail.strip() + if len(one_line) > 240: + one_line = one_line[:237] + '...' + return one_line + + +def cec_available() -> bool: + """Cheap render-time gate for whether to show CEC controls. + + Probes only for the device nodes libcec consumes — `/dev/cec0` + on mainline kernels (Pi 5, x86 USB adapters when exposed) and + `/dev/vchiq` on Pi 1-4 (currently the only one passed into the + server container by `docker-compose.yml.tmpl`). A positive result + means the adapter *could* work, not that it will: the actual + success/failure is surfaced by ``set_display_power``'s toast. + """ + return os.path.exists('/dev/cec0') or os.path.exists('/dev/vchiq') + + def get_uptime() -> float: with open('/proc/uptime', 'r') as f: uptime_seconds = float(f.readline().split()[0]) diff --git a/tests/test_app.py b/tests/test_app.py index c63167412..dbffb48da 100644 --- a/tests/test_app.py +++ b/tests/test_app.py @@ -873,3 +873,166 @@ def test_skip_buttons_publish_correct_command( sub.close() except Exception: pass + + +# --------------------------------------------------------------------------- +# 8. Display power (experimental, HDMI-CEC) — issue #2575 +# --------------------------------------------------------------------------- +# +# The section is gated on cec_available(), which stats /dev/cec0 and +# /dev/vchiq. Neither exists in the test container by default, so the +# section is hidden on every other settings test. To exercise the +# visible state we stub /dev/vchiq with a plain file before navigating +# and remove it on teardown. + +# Relative to wherever pytest is invoked. Matches the `--output +# test-artifacts` convention pytest-playwright already uses in +# pyproject.toml's addopts, so CI's upload-artifact step picks up +# these PNGs alongside any failure traces. +_SCREENSHOT_DIR = 'test-artifacts/cec' + + +def _ensure_screenshot_dir() -> str: + os.makedirs(_SCREENSHOT_DIR, exist_ok=True) + return _SCREENSHOT_DIR + + +@pytest.fixture +def cec_stub_device() -> Any: + """Create a stub `/dev/vchiq` so `diagnostics.cec_available()` + returns True. /dev is tmpfs+writable in the test container; we + create a plain file (not a real device node) — the gate only + `os.path.exists`s the path. + """ + path = '/dev/vchiq' + created = False + if not os.path.exists(path): + try: + open(path, 'w').close() + created = True + except OSError: + pytest.skip('cannot stub /dev/vchiq in this environment') + try: + yield path + finally: + if created: + try: + os.remove(path) + except FileNotFoundError: + pass + + +@pytest.mark.integration +@pytest.mark.django_db(transaction=True) +def test_display_power_section_hidden_without_cec_adapter( + reset_assets: None, page: Page +) -> None: + """No /dev/cec0 or /dev/vchiq in the container by default — the + experimental section must NOT render. Guards against accidentally + surfacing CEC controls on x86 / non-CEC hardware.""" + if os.path.exists('/dev/vchiq') or os.path.exists('/dev/cec0'): + pytest.skip('CEC device present; cannot test the hidden case') + page.goto(SETTINGS_URL) + expect( + page.get_by_role('heading', name='Settings', exact=True) + ).to_be_visible() + expect(page.get_by_role('heading', name='Display power')).to_have_count(0) + + +@pytest.mark.integration +@pytest.mark.django_db(transaction=True) +def test_display_power_section_visible_with_cec_adapter( + reset_assets: None, page: Page, cec_stub_device: str +) -> None: + """With a CEC device node present, both buttons render under an + Experimental badge inside the System controls neighbourhood.""" + page.goto(SETTINGS_URL) + expect(page.get_by_role('heading', name='Display power')).to_be_visible() + expect(page.get_by_role('button', name='Turn display on')).to_be_visible() + expect(page.get_by_role('button', name='Turn display off')).to_be_visible() + # Experimental badge sits next to the heading. + expect(page.locator('.settings-section__badge')).to_have_text( + 'Experimental' + ) + + # Screenshot 1: full settings page with the new section + _ensure_screenshot_dir() + page.screenshot( + path=f'{_SCREENSHOT_DIR}/01-settings-page-with-cec.png', + full_page=True, + ) + + # Screenshot 2: just the Display power card (tight crop) + section = page.locator('section', has_text='Display power').last + section.scroll_into_view_if_needed() + box = section.bounding_box() + assert box, 'display-power section has no bounding box' + page.screenshot( + path=f'{_SCREENSHOT_DIR}/02-display-power-card.png', + clip={ + 'x': max(box['x'] - 8, 0), + 'y': max(box['y'] - 8, 0), + 'width': box['width'] + 16, + 'height': box['height'] + 16, + }, + ) + + +@pytest.mark.integration +@pytest.mark.django_db(transaction=True) +def test_display_power_button_click_surfaces_error_toast( + reset_assets: None, page: Page, cec_stub_device: str +) -> None: + """Issue #2575's feedback-loop requirement: a failing CEC command + must surface to the operator as a visible toast, not silently + succeed or no-op. The container has no real CEC adapter, so the + inner subprocess fails — exactly the path we want to exercise.""" + page.goto(SETTINGS_URL) + expect(page.get_by_role('heading', name='Display power')).to_be_visible() + + page.get_by_role('button', name='Turn display on').click() + + # After the form post + redirect, the toast pipeline reads + # django-messages and pushes an app-toast--error item. Match by + # the CSS class the toast template sets per-kind. + error_toast = page.locator('.app-toast--error').first + expect(error_toast).to_be_visible() + # The message should namespace the failure as a display action. + expect(error_toast).to_contain_text('Display turn-on') + + # Screenshot 3: error toast in context + _ensure_screenshot_dir() + page.screenshot( + path=f'{_SCREENSHOT_DIR}/03-error-toast.png', + full_page=False, + ) + + +@pytest.mark.integration +@pytest.mark.django_db(transaction=True) +def test_display_power_success_toast_appearance( + reset_assets: None, page: Page, cec_stub_device: str +) -> None: + """Real success path requires a working HDMI-CEC TV, which the + test container cannot supply. To exercise the *visual* success + treatment (green check, dismissible) we drive Alpine's toast store + directly — same call path the server-rendered django-messages + drain uses (vendor.ts:50). This isn't asserting on the redirect + flow; it's a UX-coverage screenshot for the success kind.""" + page.goto(SETTINGS_URL) + expect(page.get_by_role('heading', name='Display power')).to_be_visible() + + page.evaluate( + """() => window.Alpine.store('toasts').push( + 'success', 'Display turn-on command sent.', 60000 + )""" + ) + success_toast = page.locator('.app-toast--success').first + expect(success_toast).to_be_visible() + expect(success_toast).to_contain_text('Display turn-on command sent.') + + _ensure_screenshot_dir() + page.screenshot( + path=f'{_SCREENSHOT_DIR}/04-success-toast.png', + full_page=False, + ) diff --git a/tests/test_diagnostics.py b/tests/test_diagnostics.py index 5b82c22b5..1335c37e0 100644 --- a/tests/test_diagnostics.py +++ b/tests/test_diagnostics.py @@ -148,6 +148,159 @@ def test_get_display_power_empty_output_returns_cec_error() -> None: assert diagnostics.get_display_power() == 'CEC error' +def test_set_display_power_on_success() -> None: + completed = mock.MagicMock(spec=subprocess.CompletedProcess) + completed.stdout = b'OK' + with mock.patch.object(subprocess, 'run', return_value=completed): + ok, msg = diagnostics.set_display_power(on=True) + assert ok is True + assert 'on' in msg + + +def test_set_display_power_off_success() -> None: + completed = mock.MagicMock(spec=subprocess.CompletedProcess) + completed.stdout = b'OK' + with mock.patch.object(subprocess, 'run', return_value=completed): + ok, msg = diagnostics.set_display_power(on=False) + assert ok is True + assert 'off' in msg + + +def test_set_display_power_cec_error_passes_through_reason() -> None: + completed = mock.MagicMock(spec=subprocess.CompletedProcess) + completed.stdout = b'ERROR: no adapter' + with mock.patch.object(subprocess, 'run', return_value=completed): + ok, msg = diagnostics.set_display_power(on=True) + assert ok is False + assert 'no adapter' in msg + + +def test_set_display_power_timeout_returns_failure_message() -> None: + with mock.patch.object( + subprocess, + 'run', + side_effect=subprocess.TimeoutExpired(cmd='python', timeout=10), + ): + ok, msg = diagnostics.set_display_power(on=True) + assert ok is False + assert 'timed out' in msg.lower() + + +def test_set_display_power_unexpected_stdout_falls_through_to_stdout() -> None: + """No 'OK' / 'ERROR:' sentinel — the helper still has to return + something actionable. With non-empty stdout and a clean exit, that + becomes the raw line itself (capped).""" + completed = mock.MagicMock(spec=subprocess.CompletedProcess) + completed.stdout = b'something weird' + completed.stderr = b'' + completed.returncode = 0 + with mock.patch.object(subprocess, 'run', return_value=completed): + ok, msg = diagnostics.set_display_power(on=True) + assert ok is False + assert 'something weird' in msg + + +def test_set_display_power_subprocess_crash_surfaces_stderr() -> None: + """When stdout is empty and stderr has content (interpreter crash, + libcec writing to stderr), the last line of stderr is what reaches + the toast — gives the operator a real reason instead of a generic + 'unexpected response.'""" + completed = mock.MagicMock(spec=subprocess.CompletedProcess) + completed.stdout = b'' + completed.stderr = ( + b'Traceback (most recent call last):\n' + b' File "", line 4, in \n' + b'RuntimeError: cec init failed: no adapter\n' + ) + completed.returncode = 1 + with mock.patch.object(subprocess, 'run', return_value=completed): + ok, msg = diagnostics.set_display_power(on=True) + assert ok is False + assert 'RuntimeError: cec init failed: no adapter' in msg + + +def test_set_display_power_subprocess_crash_with_empty_streams_reports_status() -> ( + None +): + """Last-resort fallback: subprocess exits non-zero with no stderr + and no stdout. Still has to report something — surface the returncode.""" + completed = mock.MagicMock(spec=subprocess.CompletedProcess) + completed.stdout = b'' + completed.stderr = b'' + completed.returncode = 137 + with mock.patch.object(subprocess, 'run', return_value=completed): + ok, msg = diagnostics.set_display_power(on=True) + assert ok is False + assert '137' in msg + + +def test_set_display_power_caps_long_error_message() -> None: + """libcec can spew kilobytes of diagnostic output; the toast / API + body must not carry an unbounded blob.""" + completed = mock.MagicMock(spec=subprocess.CompletedProcess) + completed.stdout = b'' + completed.stderr = ('X' * 4000).encode() + completed.returncode = 1 + with mock.patch.object(subprocess, 'run', return_value=completed): + ok, msg = diagnostics.set_display_power(on=True) + assert ok is False + # Cap is 240; message has prefix "Display turn-on failed: " so total + # is under ~280 chars and ends with the ellipsis sentinel. + assert len(msg) < 300 + assert msg.endswith('...') + + +def test_set_display_power_caps_long_error_sentinel_reason() -> None: + """The ERROR: sentinel branch must apply the same length cap + + last-line trim as the unexpected-stdout fallback; a hostile or + chatty libcec build could otherwise smuggle a multi-line / huge + string into the toast via the contract path.""" + long_reason = 'X' * 4000 + completed = mock.MagicMock(spec=subprocess.CompletedProcess) + completed.stdout = f'ERROR: {long_reason}'.encode() + completed.stderr = b'' + completed.returncode = 0 + with mock.patch.object(subprocess, 'run', return_value=completed): + ok, msg = diagnostics.set_display_power(on=True) + assert ok is False + assert len(msg) < 300 + assert msg.endswith('...') + + +def test_set_display_power_error_sentinel_strips_multiline() -> None: + """Multi-line reason on the ERROR: branch — we keep only the last + non-empty line so the toast stays one row tall.""" + completed = mock.MagicMock(spec=subprocess.CompletedProcess) + completed.stdout = b'ERROR: first line\nmiddle line\nactual failure reason' + completed.stderr = b'' + completed.returncode = 0 + with mock.patch.object(subprocess, 'run', return_value=completed): + ok, msg = diagnostics.set_display_power(on=True) + assert ok is False + assert 'actual failure reason' in msg + assert 'first line' not in msg + assert 'middle line' not in msg + + +def test_cec_available_true_when_cec0_present() -> None: + with mock.patch.object( + os.path, 'exists', side_effect=lambda p: p == '/dev/cec0' + ): + assert diagnostics.cec_available() is True + + +def test_cec_available_true_when_vchiq_present() -> None: + with mock.patch.object( + os.path, 'exists', side_effect=lambda p: p == '/dev/vchiq' + ): + assert diagnostics.cec_available() is True + + +def test_cec_available_false_when_neither_present() -> None: + with mock.patch.object(os.path, 'exists', return_value=False): + assert diagnostics.cec_available() is False + + def test_get_display_power_subprocess_timeout() -> None: with mock.patch.object( subprocess, diff --git a/tests/test_template_views.py b/tests/test_template_views.py index 42b5715b1..a402f9b1c 100644 --- a/tests/test_template_views.py +++ b/tests/test_template_views.py @@ -480,6 +480,106 @@ def test_settings_shutdown(shutdown_mock: Any, client: Client) -> None: assert shutdown_mock.called +@pytest.mark.django_db +@mock.patch( + 'anthias_server.app.views.diagnostics.cec_available', return_value=True +) +@mock.patch( + 'anthias_server.app.views.diagnostics.set_display_power', + return_value=(True, 'Display turn-on command sent.'), +) +def test_settings_display_on( + set_display_power_mock: Any, + _cec_available_mock: Any, + client: Client, +) -> None: + response = client.post( + reverse('anthias_app:settings_display_power', kwargs={'state': 'on'}) + ) + assert response.status_code in (200, 302) + set_display_power_mock.assert_called_once_with(on=True) + + +@pytest.mark.django_db +@mock.patch( + 'anthias_server.app.views.diagnostics.cec_available', return_value=True +) +@mock.patch( + 'anthias_server.app.views.diagnostics.set_display_power', + return_value=(True, 'Display turn-off command sent.'), +) +def test_settings_display_off( + set_display_power_mock: Any, + _cec_available_mock: Any, + client: Client, +) -> None: + response = client.post( + reverse('anthias_app:settings_display_power', kwargs={'state': 'off'}) + ) + assert response.status_code in (200, 302) + set_display_power_mock.assert_called_once_with(on=False) + + +@pytest.mark.django_db +@mock.patch('anthias_server.app.views.diagnostics.set_display_power') +def test_settings_display_invalid_state( + set_display_power_mock: Any, client: Client +) -> None: + response = client.post( + reverse('anthias_app:settings_display_power', kwargs={'state': 'foo'}) + ) + assert response.status_code in (200, 302) + set_display_power_mock.assert_not_called() + + +@pytest.mark.django_db +@mock.patch( + 'anthias_server.app.views.diagnostics.cec_available', return_value=False +) +@mock.patch('anthias_server.app.views.diagnostics.set_display_power') +def test_settings_display_blocked_without_cec( + set_display_power_mock: Any, + _cec_available_mock: Any, + client: Client, +) -> None: + """A stale form (or direct curl) against a non-CEC device must + short-circuit before the 10 s libcec subprocess ever runs.""" + from django.contrib.messages import get_messages + + response = client.post( + reverse('anthias_app:settings_display_power', kwargs={'state': 'on'}) + ) + assert response.status_code in (200, 302) + set_display_power_mock.assert_not_called() + messages_out = [m.message for m in get_messages(response.wsgi_request)] + assert any('CEC' in m or 'adapter' in m for m in messages_out) + + +@pytest.mark.django_db +@mock.patch( + 'anthias_server.app.views.diagnostics.cec_available', return_value=True +) +@mock.patch( + 'anthias_server.app.views.diagnostics.set_display_power', + return_value=(False, 'Display turn-on failed: no adapter'), +) +def test_settings_display_surfaces_error_message( + _set_display_power_mock: Any, + _cec_available_mock: Any, + client: Client, +) -> None: + """Failed CEC commands must reach the operator via a flash message + (the feedback loop called out in issue #2575).""" + from django.contrib.messages import get_messages + + response = client.post( + reverse('anthias_app:settings_display_power', kwargs={'state': 'on'}) + ) + assert response.status_code in (200, 302) + messages_out = [m.message for m in get_messages(response.wsgi_request)] + assert any('no adapter' in m for m in messages_out) + + @pytest.mark.django_db def test_assets_update_via_post(client: Client, asset: Asset) -> None: new_name = 'Renamed asset'