mirror of
https://github.com/ChuckBuilds/LEDMatrix.git
synced 2026-10-04 22:35:08 +00:00
fix: web UI and src.common bugs (wifi wrong-password, plugin icon, starlark toggle, API caching, scroll/logo/font helpers) (#646)
- wifi: keep the "wrong_password:" prefix through the restore/AP fallback so the UI's incorrect-password prompt fires again. - /plugins/installed returns the manifest's icon (string only). - /starlark/apps/<id>/toggle coerces `enabled` and delegates to _toggle_starlark_app (disk before memory, no KeyError, "false" is false). - /api/v3/ JSON GETs are sent Cache-Control: no-store; non-JSON keeps 5s. - ScrollHelper.set_scrolling_image converts non-RGB input (alpha onto black); create/set_scrolling_image reset last_update_time like reset_scroll. - LogoHelper backs off a failed download per path for MISSING_LOGO_RECHECK_SECONDS; cleared on invalidate/clear_cache. - refresh_placeholder_timestamp saves atomically. - FontManager.clear_cache / _clear_plugin_font_cache bump cache_generation. - Odds manager: per-game logs to DEBUG; JSON decode error caught before RequestException (same cooldown). - element_style mangled continuations; startup validator skips null plugin blocks and reuses the controller's discovery. - src/common/README lists frame_timing, json_body, render_gate. Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,46 @@
|
||||
"""GET /api/v3/plugins/installed carries the manifest's `icon`.
|
||||
|
||||
The tab nav sets ``iconEl.className = plugin.icon || 'fas fa-puzzle-piece'``
|
||||
(app-shell.js, app-early.js), but the route never included `icon`, so every
|
||||
plugin tab showed the default puzzle piece.
|
||||
"""
|
||||
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import pytest
|
||||
|
||||
from test._api_v3_test_helpers import ( # noqa: F401 - fixtures
|
||||
api_v3_client, api_v3_module,
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def installed(api_v3_module, api_v3_client, tmp_path):
|
||||
def _get(manifest_extra):
|
||||
api = api_v3_module.api_v3
|
||||
info = {'id': 'demo', 'name': 'Demo', 'version': '1.0.0', 'loaded': False}
|
||||
info.update(manifest_extra)
|
||||
api.plugin_manager.plugins_dir = str(tmp_path) # no manifest on disk
|
||||
api.plugin_manager.get_all_plugin_info = MagicMock(return_value=[info])
|
||||
api.plugin_manager.get_plugin = MagicMock(return_value=None)
|
||||
api.plugin_store_manager.get_registry_info = MagicMock(return_value=None)
|
||||
api.config_manager.load_config = MagicMock(return_value={})
|
||||
response = api_v3_client.get('/api/v3/plugins/installed')
|
||||
assert response.status_code == 200
|
||||
plugins = [p for p in response.get_json()['data']['plugins'] if p['id'] == 'demo']
|
||||
assert len(plugins) == 1
|
||||
return plugins[0]
|
||||
return _get
|
||||
|
||||
|
||||
def test_the_manifest_icon_is_passed_through(installed):
|
||||
assert installed({'icon': 'fas fa-cloud-sun'})['icon'] == 'fas fa-cloud-sun'
|
||||
|
||||
|
||||
def test_no_icon_comes_back_as_null(installed):
|
||||
# The JS falls back to the puzzle piece on a falsy value.
|
||||
assert installed({})['icon'] is None
|
||||
|
||||
|
||||
def test_a_non_string_icon_is_not_passed_through(installed):
|
||||
assert installed({'icon': {'class': 'fas fa-star'}})['icon'] is None
|
||||
@@ -13,6 +13,7 @@ its requests through a session so it can identify itself to ESPN, so patching
|
||||
the module-level requests.get would no longer intercept anything.
|
||||
"""
|
||||
|
||||
import logging
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
@@ -161,6 +162,34 @@ class TestGetOdds:
|
||||
assert result == {'stale': True}
|
||||
assert cache_manager.get_with_auto_strategy.call_count == 2
|
||||
|
||||
def test_a_bad_body_is_reported_as_a_parse_error(
|
||||
self, manager, cache_manager, mock_get, caplog):
|
||||
# requests' JSONDecodeError is a RequestException, so the JSON branch
|
||||
# sat unreachable behind it and a bad body read as a failed fetch.
|
||||
response = _make_response(None)
|
||||
response.json.side_effect = requests.exceptions.JSONDecodeError('x', 'doc', 0)
|
||||
mock_get.return_value = response
|
||||
cache_manager.get_with_auto_strategy.side_effect = [None, {'stale': True}]
|
||||
|
||||
with caplog.at_level(logging.ERROR, logger=manager.logger.name):
|
||||
result = manager.get_odds('football', 'nfl', '401')
|
||||
|
||||
assert result == {'stale': True}
|
||||
messages = [r.getMessage() for r in caplog.records]
|
||||
assert any('decoding JSON' in m for m in messages), messages
|
||||
assert not any('Error fetching odds' in m for m in messages), messages
|
||||
# The hold-off is unchanged: a bad body still backs off like a failed
|
||||
# fetch did.
|
||||
assert manager._skip_network_until > 0
|
||||
|
||||
def test_routine_fetches_log_nothing_at_info(
|
||||
self, manager, cache_manager, mock_get, caplog):
|
||||
with caplog.at_level(logging.INFO, logger=manager.logger.name):
|
||||
manager.get_odds('football', 'nfl', '401') # miss + fetch
|
||||
cache_manager.get_with_auto_strategy.return_value = {'spread': 1}
|
||||
manager.get_odds('football', 'nfl', '401') # hit
|
||||
assert [r for r in caplog.records if r.levelno == logging.INFO] == []
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# _extract_espn_data
|
||||
|
||||
@@ -0,0 +1,21 @@
|
||||
"""src/common/README.md says it lists every module; keep it true.
|
||||
|
||||
frame_timing, json_body and render_gate landed without rows, so the page a
|
||||
plugin author reads to see what core offers was missing three modules.
|
||||
"""
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
COMMON = Path(__file__).resolve().parent.parent / "src" / "common"
|
||||
|
||||
|
||||
def _modules():
|
||||
return sorted(p.stem for p in COMMON.glob("*.py") if p.stem != "__init__")
|
||||
|
||||
|
||||
def test_every_module_has_a_table_row_and_a_section():
|
||||
readme = (COMMON / "README.md").read_text(encoding="utf-8")
|
||||
missing_rows = [m for m in _modules() if f"| [`{m}`](#{m}) |" not in readme]
|
||||
missing_sections = [m for m in _modules() if f"\n### {m}\n" not in readme]
|
||||
assert not missing_rows, f"no summary-table row for: {missing_rows}"
|
||||
assert not missing_sections, f"no '### <module>' section for: {missing_sections}"
|
||||
@@ -25,8 +25,9 @@ class RecordingValidator:
|
||||
self.warnings = ['config warning']
|
||||
return True, [], list(self.warnings)
|
||||
|
||||
def _validate_plugins(self):
|
||||
def _validate_plugins(self, discovered_plugins=None):
|
||||
RecordingValidator.calls.append(('plugins',))
|
||||
RecordingValidator.discovered = discovered_plugins
|
||||
self.warnings = ['plugin warning']
|
||||
|
||||
|
||||
@@ -75,3 +76,19 @@ def test_each_check_runs_once(mock_config_manager, mock_display_manager,
|
||||
assert any('plugin warning' in r.getMessage() for r in caplog.records)
|
||||
finally:
|
||||
controller.cleanup()
|
||||
|
||||
|
||||
def test_plugins_are_discovered_once(mock_config_manager, mock_display_manager,
|
||||
mock_cache_manager, test_config_with_plugins,
|
||||
emulator_mode):
|
||||
"""The plugin checks reuse the controller's discovery rather than rescanning
|
||||
the plugins directory (and logging every plugin) a second time."""
|
||||
RecordingValidator.calls = []
|
||||
RecordingValidator.discovered = None
|
||||
controller = _build_controller(mock_config_manager, mock_display_manager,
|
||||
mock_cache_manager, test_config_with_plugins)
|
||||
try:
|
||||
assert controller.plugin_manager.discover_plugins.call_count == 1
|
||||
assert RecordingValidator.discovered == []
|
||||
finally:
|
||||
controller.cleanup()
|
||||
|
||||
@@ -137,6 +137,24 @@ class TestCacheLifecycle:
|
||||
assert fm.cache_generation == gen_before + 1
|
||||
assert not fm.font_cache
|
||||
|
||||
def test_clear_cache_bumps_generation(self, fm):
|
||||
# Layout contexts and font-usage results are keyed off
|
||||
# cache_generation; clear_cache used to drop the fonts without
|
||||
# telling them.
|
||||
gen_before = fm.cache_generation
|
||||
fm.clear_cache()
|
||||
assert fm.cache_generation == gen_before + 1
|
||||
|
||||
def test_clearing_a_plugins_cached_fonts_bumps_generation(self, fm):
|
||||
fm.font_cache["demo::tiny_8"] = object()
|
||||
gen_before = fm.cache_generation
|
||||
fm._clear_plugin_font_cache("demo")
|
||||
assert "demo::tiny_8" not in fm.font_cache
|
||||
assert fm.cache_generation == gen_before + 1
|
||||
# Nothing to drop, nothing to rebuild.
|
||||
fm._clear_plugin_font_cache("demo")
|
||||
assert fm.cache_generation == gen_before + 1
|
||||
|
||||
|
||||
class TestPluginFonts:
|
||||
"""plugin:// sources resolve against the plugin's own directory, which
|
||||
|
||||
@@ -356,6 +356,18 @@ class TestRefreshPlaceholderTimestamp:
|
||||
def test_missing_file_is_not_an_error(self, tmp_path):
|
||||
assert refresh_placeholder_timestamp(tmp_path / "nope.png") is False
|
||||
|
||||
def test_the_restamp_is_atomic(self, tmp_path):
|
||||
# It used to save over the file in place, so a renderer opening the
|
||||
# logo mid-write read a truncated PNG. A failed write must now leave
|
||||
# the previous placeholder intact and no temp file behind.
|
||||
assert LogoDownloader().create_placeholder_logo("COLL", str(tmp_path))
|
||||
path = tmp_path / "COLL.png"
|
||||
before = path.read_bytes()
|
||||
with patch("src.logo_downloader.os.replace", side_effect=OSError("disk full")):
|
||||
assert refresh_placeholder_timestamp(path) is False
|
||||
assert path.read_bytes() == before
|
||||
assert sorted(p.name for p in tmp_path.iterdir()) == ["COLL.png"]
|
||||
|
||||
|
||||
class TestFailurePaths:
|
||||
def test_a_team_without_logos_is_a_failed_download(self, tmp_path):
|
||||
|
||||
@@ -0,0 +1,89 @@
|
||||
"""A failed logo download is not retried on every call.
|
||||
|
||||
A logo that is simply absent (not a stale placeholder) had nothing to back
|
||||
off on: every load_logo_with_download() call retried the download, each with
|
||||
a 30s timeout, for as long as the host was down or the URL was dead.
|
||||
"""
|
||||
|
||||
import io
|
||||
import logging
|
||||
import time
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import pytest
|
||||
import requests
|
||||
from PIL import Image
|
||||
|
||||
from src.common.logo_helper import MISSING_LOGO_RECHECK_SECONDS, LogoHelper
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _no_real_chmod(monkeypatch):
|
||||
monkeypatch.setattr("src.common.logo_helper.ensure_directory_permissions", MagicMock())
|
||||
monkeypatch.setattr("src.logo_downloader.ensure_file_permissions", MagicMock())
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def helper():
|
||||
h = LogoHelper(display_width=64, display_height=32,
|
||||
logger=logging.getLogger("test.logo_backoff"))
|
||||
h.session.get = MagicMock(side_effect=requests.ConnectionError("down"))
|
||||
return h
|
||||
|
||||
|
||||
def _png_response():
|
||||
buf = io.BytesIO()
|
||||
Image.new("RGBA", (20, 20), (0, 128, 0, 255)).save(buf, "PNG")
|
||||
data = buf.getvalue()
|
||||
response = MagicMock()
|
||||
response.__enter__.return_value = response
|
||||
response.__exit__.return_value = False
|
||||
response.raise_for_status = MagicMock()
|
||||
response.headers = {"content-type": "image/png"}
|
||||
response.iter_content = lambda *a, **k: iter([data])
|
||||
return response
|
||||
|
||||
|
||||
def _load(helper, path):
|
||||
return helper.load_logo_with_download("PHI", path, "http://x/logo.png",
|
||||
max_width=20, max_height=20)
|
||||
|
||||
|
||||
def test_a_second_call_does_not_retry(helper, tmp_path):
|
||||
path = tmp_path / "PHI.png"
|
||||
assert _load(helper, path) is not None # placeholder
|
||||
assert _load(helper, path) is not None
|
||||
assert helper.session.get.call_count == 1
|
||||
|
||||
|
||||
def test_it_retries_after_the_recheck_window(helper, tmp_path):
|
||||
path = tmp_path / "PHI.png"
|
||||
_load(helper, path)
|
||||
helper._download_failures[str(path)] -= MISSING_LOGO_RECHECK_SECONDS + 1
|
||||
_load(helper, path)
|
||||
assert helper.session.get.call_count == 2
|
||||
|
||||
|
||||
def test_other_paths_are_not_held_back(helper, tmp_path):
|
||||
_load(helper, tmp_path / "PHI.png")
|
||||
_load(helper, tmp_path / "DAL.png")
|
||||
assert helper.session.get.call_count == 2
|
||||
|
||||
|
||||
def test_clear_cache_forgets_failures(helper, tmp_path):
|
||||
path = tmp_path / "PHI.png"
|
||||
_load(helper, path)
|
||||
helper.clear_cache()
|
||||
_load(helper, path)
|
||||
assert helper.session.get.call_count == 2
|
||||
|
||||
|
||||
def test_invalidate_forgets_the_failure(helper, tmp_path):
|
||||
path = tmp_path / "PHI.png"
|
||||
_load(helper, path)
|
||||
helper._invalidate_cached_logo("PHI", path)
|
||||
helper.session.get = MagicMock(return_value=_png_response())
|
||||
logo = _load(helper, path)
|
||||
helper.session.get.assert_called_once()
|
||||
assert path.exists()
|
||||
assert logo.getpixel((0, 0))[:3] == (0, 128, 0)
|
||||
@@ -0,0 +1,87 @@
|
||||
"""ScrollHelper: what a new scrolling image does on the way in.
|
||||
|
||||
- set_scrolling_image cached a non-RGB image as-is, and every frame is cut
|
||||
with Image.frombytes('RGB', ...): an RGBA strip came out as garbage and an
|
||||
L strip raised.
|
||||
- Only reset_scroll() reset last_update_time. In time-based mode the first
|
||||
update after a new image (e.g. the plugin coming back on screen) advanced
|
||||
the position by the whole idle gap, skipping the start of the content.
|
||||
"""
|
||||
|
||||
import time
|
||||
|
||||
import pytest
|
||||
from PIL import Image
|
||||
|
||||
from src.common.scroll_helper import ScrollHelper
|
||||
|
||||
W, H = 16, 8
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def helper():
|
||||
return ScrollHelper(display_width=W, display_height=H)
|
||||
|
||||
|
||||
class TestNonRgbImages:
|
||||
def test_rgba_is_composited_onto_black(self, helper):
|
||||
img = Image.new('RGBA', (40, H), (0, 0, 0, 0))
|
||||
for x in range(10):
|
||||
for y in range(H):
|
||||
img.putpixel((x, y), (255, 0, 0, 255))
|
||||
# Hidden colour under full transparency must not leak through.
|
||||
img.putpixel((20, 0), (0, 255, 0, 0))
|
||||
helper.set_scrolling_image(img)
|
||||
assert helper.cached_image.mode == 'RGB'
|
||||
frame = helper.get_visible_portion()
|
||||
assert frame.size == (W, H)
|
||||
assert frame.getpixel((0, 0)) == (255, 0, 0)
|
||||
assert frame.getpixel((12, 0)) == (0, 0, 0)
|
||||
helper.scroll_position = 20.0
|
||||
assert helper.get_visible_portion().getpixel((0, 0)) == (0, 0, 0)
|
||||
|
||||
def test_greyscale_does_not_raise(self, helper):
|
||||
helper.set_scrolling_image(Image.new('L', (40, H), 200))
|
||||
frame = helper.get_visible_portion()
|
||||
assert frame.getpixel((0, 0)) == (200, 200, 200)
|
||||
|
||||
def test_palette_with_transparency(self, helper):
|
||||
img = Image.new('P', (40, H), 1)
|
||||
img.putpalette([0, 0, 0, 0, 0, 255] + [0] * 762)
|
||||
img.info['transparency'] = 1
|
||||
helper.set_scrolling_image(img)
|
||||
assert helper.get_visible_portion().getpixel((0, 0)) == (0, 0, 0)
|
||||
|
||||
def test_an_rgb_image_is_kept_as_the_same_object(self, helper):
|
||||
img = Image.new('RGB', (40, H), (1, 2, 3))
|
||||
helper.set_scrolling_image(img)
|
||||
assert helper.cached_image is img
|
||||
|
||||
|
||||
class TestNoJumpAfterIdle:
|
||||
def _idle_then_new_image(self, helper, install):
|
||||
helper.scroll_speed = 100.0 # px/s, time-based
|
||||
install(helper)
|
||||
helper.update_scroll_position()
|
||||
# Off screen for 10 seconds, then a fresh image.
|
||||
helper.last_update_time = time.time() - 10.0
|
||||
install(helper)
|
||||
helper.update_scroll_position()
|
||||
return helper.scroll_position
|
||||
|
||||
def test_set_scrolling_image(self, helper):
|
||||
pos = self._idle_then_new_image(
|
||||
helper, lambda h: h.set_scrolling_image(Image.new('RGB', (5000, H))))
|
||||
assert pos < 50, f"jumped {pos}px on the first update of a new image"
|
||||
|
||||
def test_create_scrolling_image(self, helper):
|
||||
pos = self._idle_then_new_image(
|
||||
helper, lambda h: h.create_scrolling_image([Image.new('RGB', (5000, H))]))
|
||||
assert pos < 50, f"jumped {pos}px on the first update of a new image"
|
||||
|
||||
def test_fixed_step_mode_still_steps_exactly(self, helper):
|
||||
helper.set_pixels_per_frame(2)
|
||||
helper.set_scrolling_image(Image.new('RGB', (5000, H)))
|
||||
helper.update_scroll_position()
|
||||
helper.update_scroll_position()
|
||||
assert helper.scroll_position == 4.0
|
||||
@@ -242,6 +242,28 @@ class TestPlugins:
|
||||
assert is_valid is True
|
||||
assert errors == []
|
||||
|
||||
def test_a_null_plugin_block_does_not_abandon_the_checks(self, good_cache, tmp_path):
|
||||
# config.get(id, {}) returns None for "known": null, and .get() on it
|
||||
# raised -- caught as "Could not validate plugins", skipping every
|
||||
# plugin after it.
|
||||
pm = MagicMock()
|
||||
pm.discover_plugins.return_value = ['nulled', 'known']
|
||||
plugin_dir = tmp_path / "known"
|
||||
plugin_dir.mkdir() # no manifest.json
|
||||
pm.get_plugin_directory.return_value = str(plugin_dir)
|
||||
|
||||
config = dict(GOOD_CONFIG, nulled=None, known={'enabled': True})
|
||||
validator = StartupValidator(make_config_manager(config), pm)
|
||||
validator._validate_plugins()
|
||||
assert not any('Could not validate plugins' in w for w in validator.warnings)
|
||||
assert "Plugin 'known' manifest.json not found" in validator.errors
|
||||
|
||||
def test_a_passed_discovery_is_reused(self, good_cache):
|
||||
pm = MagicMock()
|
||||
validator = StartupValidator(make_config_manager(dict(GOOD_CONFIG)), pm)
|
||||
validator._validate_plugins(discovered_plugins=[])
|
||||
pm.discover_plugins.assert_not_called()
|
||||
|
||||
|
||||
class TestIdempotence:
|
||||
"""validate_all() resets error/warning state each run (the fixed bug)."""
|
||||
|
||||
@@ -0,0 +1,78 @@
|
||||
"""A rejected passphrase must reach the web UI as "wrong_password:".
|
||||
|
||||
_connect_nmcli prefixes the message with "wrong_password:" and the api_v3
|
||||
wifi route turns that into ``error_type == 'wrong_password'``, which the
|
||||
setup pages key their "incorrect password" prompt off. _connect_validated
|
||||
used to replace the message on every failure branch (restore the old
|
||||
network, or bring the setup AP up), so the prefix never survived.
|
||||
"""
|
||||
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).parent.parent))
|
||||
|
||||
from src.wifi_manager import WiFiManager # noqa: E402
|
||||
|
||||
|
||||
def _manager(connected_ssid=None):
|
||||
manager = WiFiManager.__new__(WiFiManager) # no __init__: no real host access
|
||||
manager.has_nmcli = True
|
||||
manager._wifi_interface = "wlan0"
|
||||
manager.get_wifi_status = MagicMock(return_value=SimpleNamespace(
|
||||
connected=bool(connected_ssid), ssid=connected_ssid, ip_address=None))
|
||||
manager._is_ap_mode_active = MagicMock(return_value=False)
|
||||
manager._ensure_wifi_radio_enabled = MagicMock(return_value=True)
|
||||
manager._show_led_message = MagicMock()
|
||||
manager.disconnect_from_network = MagicMock(return_value=(True, "ok"))
|
||||
manager._wait_for_device_idle = MagicMock(return_value=True)
|
||||
manager.enable_ap_mode = MagicMock(return_value=(True, "AP up"))
|
||||
return manager
|
||||
|
||||
|
||||
def _run(manager, nmcli_result):
|
||||
manager._connect_nmcli = MagicMock(return_value=nmcli_result)
|
||||
nmcli_out = SimpleNamespace(returncode=0, stdout="GENERAL.CONNECTION:OldNet\n")
|
||||
with patch("src.wifi_manager.subprocess.run", return_value=nmcli_out), \
|
||||
patch("src.wifi_manager.time.sleep"):
|
||||
return manager._connect_validated("HomeNet", "hunter22")
|
||||
|
||||
|
||||
WRONG = (False, "wrong_password: Secrets were required, but not provided")
|
||||
|
||||
|
||||
class TestWrongPasswordSurvivesRecovery:
|
||||
def test_when_the_setup_ap_comes_back_up(self):
|
||||
ok, message = _run(_manager(), WRONG)
|
||||
assert ok is False
|
||||
assert message.startswith("wrong_password:")
|
||||
|
||||
def test_when_the_setup_ap_fails_too(self):
|
||||
manager = _manager()
|
||||
manager.enable_ap_mode.return_value = (False, "hostapd missing")
|
||||
ok, message = _run(manager, WRONG)
|
||||
assert ok is False
|
||||
assert message.startswith("wrong_password:")
|
||||
assert "hostapd missing" in message
|
||||
|
||||
@pytest.mark.parametrize("restored", [True, False])
|
||||
def test_when_the_previous_network_is_restored_or_not(self, restored):
|
||||
manager = _manager(connected_ssid="OldNet")
|
||||
manager._restore_original_connection = MagicMock(return_value=restored)
|
||||
ok, message = _run(manager, WRONG)
|
||||
assert ok is False
|
||||
assert message.startswith("wrong_password:")
|
||||
|
||||
|
||||
class TestOtherFailuresAreUnchanged:
|
||||
def test_a_generic_failure_gets_no_prefix(self):
|
||||
ok, message = _run(_manager(), (False, "Connection timed out"))
|
||||
assert ok is False
|
||||
assert message == "Connection failed. AP mode enabled."
|
||||
|
||||
def test_success_passes_through(self):
|
||||
assert _run(_manager(), (True, "Connected to HomeNet")) == (True, "Connected to HomeNet")
|
||||
@@ -0,0 +1,51 @@
|
||||
"""/api/v3 JSON GETs must not be served from the browser cache.
|
||||
|
||||
They carried ``Cache-Control: private, max-age=5``, so a fetch() made right
|
||||
after an install, a toggle or a Wi-Fi connect could be answered with the copy
|
||||
from before it. Non-JSON files served through the API keep the short cache.
|
||||
"""
|
||||
|
||||
from flask import Flask, Response, jsonify
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client():
|
||||
import web_interface.app as web_app
|
||||
# The real after_request hook on a throwaway app, so the header logic is
|
||||
# tested without depending on any real endpoint's managers.
|
||||
app = Flask(__name__)
|
||||
app.after_request(web_app.add_security_headers)
|
||||
app.add_url_rule('/api/v3/probe/json', 'probe_json',
|
||||
lambda: jsonify({'status': 'success'}))
|
||||
app.add_url_rule('/api/v3/probe/file', 'probe_file',
|
||||
lambda: Response('body{}', mimetype='text/css'))
|
||||
app.add_url_rule('/static/v3/probe.css', 'probe_static',
|
||||
lambda: Response('body{}', mimetype='text/css'))
|
||||
with app.test_client() as c:
|
||||
yield c
|
||||
|
||||
|
||||
def test_json_gets_are_not_stored(client):
|
||||
resp = client.get('/api/v3/probe/json')
|
||||
assert resp.headers['Cache-Control'] == 'no-store'
|
||||
|
||||
|
||||
def test_non_json_files_keep_the_short_cache(client):
|
||||
resp = client.get('/api/v3/probe/file')
|
||||
assert resp.headers['Cache-Control'] == 'private, max-age=5, must-revalidate'
|
||||
|
||||
|
||||
def test_static_assets_are_still_long_cached(client):
|
||||
resp = client.get('/static/v3/probe.css')
|
||||
assert 'immutable' in resp.headers.get('Cache-Control', '')
|
||||
|
||||
|
||||
def test_the_real_app_marks_an_api_json_answer_no_store():
|
||||
import web_interface.app as web_app
|
||||
web_app.app.config['TESTING'] = True
|
||||
with web_app.app.test_client() as c:
|
||||
resp = c.get('/api/v3/no-such-endpoint-for-cache-test')
|
||||
assert resp.mimetype == 'application/json'
|
||||
assert resp.headers['Cache-Control'] == 'no-store'
|
||||
@@ -0,0 +1,116 @@
|
||||
"""POST /api/v3/starlark/apps/<app_id>/toggle shares _toggle_starlark_app.
|
||||
|
||||
The route carried its own copy of the toggle with the bugs the shared helper
|
||||
had already fixed: it changed the loaded app before the disk write (so a
|
||||
failed save left the two disagreeing), indexed ``manifest['apps'][app_id]``
|
||||
(a KeyError, answered as a 500, for a loaded app with no on-disk entry), and
|
||||
stored ``"false"`` -- a truthy string -- as the new state.
|
||||
"""
|
||||
|
||||
import contextlib
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
PKG = 'web_interface.blueprints.api_v3'
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client():
|
||||
from web_interface.app import app
|
||||
app.config['TESTING'] = True
|
||||
with app.test_client() as c:
|
||||
yield c
|
||||
|
||||
|
||||
def _loaded_plugin(enabled=True, save_ok=True, disk=None):
|
||||
app = MagicMock()
|
||||
app.manifest = {'enabled': enabled}
|
||||
app.is_enabled.return_value = enabled
|
||||
plugin = MagicMock()
|
||||
plugin.apps = {'clock': app}
|
||||
disk = {'apps': {}} if disk is None else disk
|
||||
|
||||
def _update(fn):
|
||||
if not save_ok:
|
||||
return False
|
||||
fn(disk)
|
||||
return True
|
||||
plugin._update_manifest_safe.side_effect = _update
|
||||
return plugin, app, disk
|
||||
|
||||
|
||||
def _post(client, body):
|
||||
return client.post('/api/v3/starlark/apps/clock/toggle', json=body)
|
||||
|
||||
|
||||
class TestLoadedApp:
|
||||
def test_a_missing_disk_entry_is_created_not_a_500(self, client):
|
||||
plugin, app, disk = _loaded_plugin(enabled=True)
|
||||
with patch(f'{PKG}._get_starlark_plugin', return_value=plugin):
|
||||
resp = _post(client, {'enabled': False})
|
||||
body = resp.get_json()
|
||||
assert resp.status_code == 200, body
|
||||
assert body == {'status': 'success', 'message': body['message'], 'enabled': False}
|
||||
assert disk['apps']['clock']['enabled'] is False
|
||||
assert app.manifest['enabled'] is False
|
||||
|
||||
def test_a_failed_save_leaves_the_loaded_app_alone(self, client):
|
||||
plugin, app, _ = _loaded_plugin(enabled=True, save_ok=False)
|
||||
with patch(f'{PKG}._get_starlark_plugin', return_value=plugin):
|
||||
resp = _post(client, {'enabled': False})
|
||||
assert resp.status_code == 500
|
||||
assert resp.get_json()['status'] == 'error'
|
||||
assert app.manifest['enabled'] is True
|
||||
|
||||
@pytest.mark.parametrize('sent', ['false', 'False', 0, '0'])
|
||||
def test_a_false_string_or_zero_disables(self, client, sent):
|
||||
plugin, app, disk = _loaded_plugin(enabled=True)
|
||||
with patch(f'{PKG}._get_starlark_plugin', return_value=plugin):
|
||||
resp = _post(client, {'enabled': sent})
|
||||
assert resp.get_json()['enabled'] is False
|
||||
assert disk['apps']['clock']['enabled'] is False
|
||||
|
||||
def test_an_unrecognised_value_is_refused(self, client):
|
||||
plugin, app, disk = _loaded_plugin(enabled=True)
|
||||
with patch(f'{PKG}._get_starlark_plugin', return_value=plugin):
|
||||
resp = _post(client, {'enabled': 'maybe'})
|
||||
assert resp.status_code == 400
|
||||
assert disk == {'apps': {}}
|
||||
|
||||
def test_no_enabled_flips_the_current_state(self, client):
|
||||
plugin, app, disk = _loaded_plugin(enabled=True)
|
||||
with patch(f'{PKG}._get_starlark_plugin', return_value=plugin):
|
||||
resp = _post(client, {})
|
||||
assert resp.get_json()['enabled'] is False
|
||||
|
||||
|
||||
class TestStandalone:
|
||||
def _run(self, client, body, manifest):
|
||||
written = {}
|
||||
# The real lock is fcntl-based (Linux only); the toggle logic is what
|
||||
# is under test here.
|
||||
with patch(f'{PKG}._get_starlark_plugin', return_value=None), \
|
||||
patch(f'{PKG}._starlark_manifest_lock', contextlib.nullcontext), \
|
||||
patch(f'{PKG}._read_starlark_manifest', return_value=manifest), \
|
||||
patch(f'{PKG}._write_starlark_manifest',
|
||||
side_effect=lambda m: written.update(m) or True):
|
||||
resp = _post(client, body)
|
||||
return resp, written
|
||||
|
||||
def test_toggle_is_written(self, client):
|
||||
resp, written = self._run(client, {'enabled': 'false'},
|
||||
{'apps': {'clock': {'enabled': True}}})
|
||||
assert resp.get_json()['enabled'] is False
|
||||
assert written['apps']['clock']['enabled'] is False
|
||||
|
||||
def test_no_enabled_flips_the_manifest_state(self, client):
|
||||
resp, _ = self._run(client, {}, {'apps': {'clock': {'enabled': False}}})
|
||||
assert resp.get_json()['enabled'] is True
|
||||
|
||||
@pytest.mark.parametrize('body', [{}, {'enabled': True}])
|
||||
def test_an_unknown_app_is_404(self, client, body):
|
||||
resp, written = self._run(client, body, {'apps': {}})
|
||||
assert resp.status_code == 404
|
||||
assert 'clock' in resp.get_json()['message']
|
||||
assert written == {}
|
||||
Reference in New Issue
Block a user