import pytest from fastapi.testclient import TestClient from app.main import create_app from app.mobile.session import RpcError def test_identity_persistence_and_rename(tmp_path): app = create_app(project_root=tmp_path, driver_kind='mock') mobile = app.state.mobile_control before = mobile.identity() renamed = mobile.dispatch('device.rename', {'name': '客厅小屏幕'}) assert renamed['device_id'] == before['device_id'] assert renamed['device_name'] == '客厅小屏幕' other = create_app(project_root=tmp_path, driver_kind='mock') assert other.state.mobile_control.identity() == renamed def test_web_change_invalidates_mobile_revision(tmp_path): app = create_app(project_root=tmp_path, driver_kind='mock') mobile = app.state.mobile_control settings = mobile.dispatch('settings.get', {}) client = TestClient(app) assert client.put('/api/config', json={'brightness': 37}).status_code == 200 with pytest.raises(RpcError, match='CONFLICT'): mobile.dispatch('settings.patch', {'expected_revision': settings['revision'], 'changes': {'brightness': 20}}) revision = mobile.settings()['revision'] mobile.dispatch('settings.patch', {'expected_revision': revision, 'changes': {'brightness': 25}}) assert client.get('/api/config').json()['brightness'] == 25 def test_library_and_frame_readback(tmp_path): app = create_app(project_root=tmp_path, driver_kind='mock') mobile = app.state.mobile_control first = mobile.dispatch('library.list', {'limit': 1}) assert len(first['items']) == 1 and first['next_cursor'] second = mobile.dispatch('library.list', {'cursor': first['next_cursor'], 'limit': 1}) assert second['items'][0]['id'] != first['items'][0]['id'] item = first['items'][0] assert mobile.dispatch('library.thumbnail', item)['mime'] == 'image/png' mobile.dispatch('content.play', item) frame = mobile.dispatch('frame.get', {}) assert mobile.dispatch('frame.get', {'known_revision': frame['frame_revision']})['unchanged'] with pytest.raises(RpcError, match='CONFLICT'): mobile.dispatch('content.play', {**item, 'revision': 'old'}) def test_progressive_preview_and_revision_cleanup(tmp_path): import base64 from io import BytesIO from PIL import Image app = create_app(project_root=tmp_path, driver_kind='mock') mobile = app.state.mobile_control assert mobile.identity()['protocol_minor'] == 3 assert 'wifi_transport' in mobile.identity()['capabilities'] assert 'library_progressive' in mobile.identity()['capabilities'] created = mobile.templates.create('预览清理', { 'version': 1, 'width': 64, 'height': 64, 'pixelRgb': base64.b64encode(bytes(64 * 64 * 3)).decode('ascii'), 'elements': [], }) summary = mobile.dispatch('library.summary', {}) items = mobile.dispatch('library.list', {})['items'] assert summary == {'library_revision': mobile.dispatch('library.list', {})['library_revision'], 'item_count': len(items)} item = next(item for item in items if item['id'] == created['id']) small = base64.b64decode(mobile.dispatch('library.preview', item)['data_base64']) large = base64.b64decode(mobile.dispatch('library.thumbnail', item)['data_base64']) assert len(small) < len(large) with Image.open(BytesIO(small)) as image: assert image.format == 'PNG' and image.size == (16, 16) with Image.open(BytesIO(large)) as image: assert image.format == 'PNG' and image.size == (64, 64) old_file = mobile.preview_cache.root / mobile.preview_cache.name(item) assert old_file.read_bytes() == small mobile.templates.rename(created['id'], '预览清理新版', created['revision']) assert mobile.dispatch('library.summary', {})['library_revision'] != summary['library_revision'] assert not old_file.exists() with pytest.raises(RpcError, match='CONFLICT'): mobile.dispatch('library.preview', item) empty = mobile.animations.create('空动图') empty_item = next(row for row in mobile.dispatch('library.list', {})['items'] if row['id'] == empty['id']) with pytest.raises(RpcError, match='CONFLICT'): mobile.dispatch('library.preview', empty_item) def test_user_animation_preview_reads_persisted_first_frame(tmp_path): import base64 from io import BytesIO from PIL import Image mobile = create_app(project_root=tmp_path, driver_kind='mock').state.mobile_control scene = {'version': 1, 'width': 64, 'height': 64, 'pixelRgb': base64.b64encode(bytes((12, 34, 56)) * (64 * 64)).decode('ascii'), 'elements': []} created = mobile.animations.create_from_frames('用户动图', [{'scene': scene, 'duration_ms': 100}]) item = next(row for row in mobile.dispatch('library.list', {})['items'] if row['id'] == created['id']) small = base64.b64decode(mobile.dispatch('library.preview', item)['data_base64']) large = base64.b64decode(mobile.dispatch('library.thumbnail', item)['data_base64']) with Image.open(BytesIO(small)) as image: assert image.size == (16, 16) and image.getpixel((0, 0)) == (12, 34, 56) with Image.open(BytesIO(large)) as image: assert image.size == (64, 64) and image.getpixel((0, 0)) == (12, 34, 56) def test_forbidden_operation_and_settings_fields(tmp_path): mobile = create_app(project_root=tmp_path, driver_kind='mock').state.mobile_control with pytest.raises(RpcError, match='UNSUPPORTED_CAPABILITY'): mobile.dispatch('shell.run', {'command': 'anything'}) with pytest.raises(RpcError, match='BAD_REQUEST'): mobile.dispatch('settings.patch', {'expected_revision': mobile.settings()['revision'], 'changes': {'low_voltage_protection_enabled': False}}) def test_wifi_open_network_and_password_not_returned(tmp_path): mobile = create_app(project_root=tmp_path, driver_kind='mock').state.mobile_control revision = mobile.dispatch('wifi.get', {})['revision'] result = mobile.dispatch('wifi.set', dict(expected_revision=revision, ssid='Test Open', security='open', password_action='none', ipv4_mode='dhcp', dns_servers=[], activation='next_boot')) assert mobile.dispatch('task.get', {'task_id': result['task_id']})['state'] == 'succeeded' wifi = mobile.dispatch('wifi.get', {}) assert 'password' not in wifi['saved'] assert not wifi['saved']['password_configured'] assert wifi['saved']['security'] == 'open' assert wifi['revision'] != revision mobile.dispatch('wifi.set', dict(expected_revision=wifi['revision'], ssid='Test Private', security='wpa-psk', password_action='replace', password='public-test-fixture', ipv4_mode='dhcp', activation='next_boot')) wifi = mobile.dispatch('wifi.get', {}) assert 'password' not in wifi['saved'] and wifi['saved']['password_configured'] def test_scan_escaped_ssid_deduplicates_and_marks_unsupported(): from app.network.manager import NmcliNetworkManager class Stub(NmcliNetworkManager): def _run(self, args, *, timeout=10): assert timeout == 15 and '--rescan' in args return '*:Test\\:WiFi:WPA2:42\n:Test\\:WiFi:WPA2:81\n:Open:--:37\n:EAP:WPA2 802.1X:80' rows = Stub().scan() assert rows[0] == dict(ssid='Test:WiFi', security='wpa-psk', signal_percent=81, connected=True) assert rows[1]['security'] == 'unsupported' assert rows[2]['security'] == 'open' def test_mobile_default_and_animation_controls_share_web_state(tmp_path): app = create_app(project_root=tmp_path, driver_kind='mock') with TestClient(app) as client: mobile = app.state.mobile_control items = mobile.dispatch('library.list', {})['items'] animation = next(item for item in items if item['type'] == 'animation' and item['playable']) mobile.dispatch('content.default.set', animation) assert mobile.dispatch('content.default.get', {})['id'] == animation['id'] assert client.get('/api/display/default-content').json()['id'] == animation['id'] playback = client.get('/api/status').json()['state']['animation_playback'] controlled = mobile.dispatch('playback.patch', dict( session_id=playback['session_id'], paused=True, position_ms=0, speed=1.5)) assert controlled['animation_playback']['paused'] current = client.get('/api/status').json()['state']['animation_playback'] assert current['position_ms'] == 0 and current['speed'] == 1.5 mobile.dispatch('content.play', animation) with pytest.raises(RpcError, match='CONFLICT'): mobile.dispatch('playback.patch', dict(session_id=playback['session_id'], paused=False)) def test_mobile_library_uses_persistent_mixed_order_and_invalidates_cursor(tmp_path): from app.demo_library import demo_template app = create_app(project_root=tmp_path, driver_kind='mock') mobile = app.state.mobile_control templates, animations = mobile.templates, mobile.animations import base64 t = templates.create('user-template', {'version': 1, 'width': 64, 'height': 64, 'pixelRgb': base64.b64encode(bytes(64*64*3)).decode('ascii'), 'elements': []}) # Empty animations are valid library items, even though not playable. a = animations.create('user-animation') defaults = [{'type': 'template', 'id': t['id']}, {'type': 'animation', 'id': a['id']}] store = app.state.library_order_store current = store.get(defaults) first = mobile.dispatch('library.list', {'limit': 1}) order = list(reversed(defaults)) store.update(order, default_items=defaults, expected_revision=current['revision']) items = mobile.dispatch('library.list', {})['items'] assert [{'type': i['type'], 'id': i['id']} for i in items if not i['is_demo']] == order with pytest.raises(RpcError, match='CONFLICT'): mobile.dispatch('library.list', {'cursor': first['next_cursor']}) @pytest.mark.parametrize('reason,expected', [(53, 'NETWORK_UNAVAILABLE'), (5, 'IP_CONFIG_FAILED'), (11, 'UNKNOWN'), (7, 'UNKNOWN')]) def test_wifi_failure_classification_does_not_guess_authentication(monkeypatch, reason, expected): import subprocess from app.network.manager import NmcliNetworkManager, NetworkManagerError class Backend(NmcliNetworkManager): def _run(self, args, *, timeout=10): if 'up' in args: raise NetworkManagerError('sanitized') return str(reason) + ' (reason)' monkeypatch.setattr(subprocess, 'run', lambda *a, **k: subprocess.CompletedProcess(a, 0, '', '')) with pytest.raises(NetworkManagerError) as error: Backend().activate('public-test-id') assert error.value.error_code == expected def test_wifi_auth_requires_explicit_wrong_key_event(monkeypatch): import subprocess from app.network.manager import NmcliNetworkManager, NetworkManagerError class Backend(NmcliNetworkManager): def _run(self, args, *, timeout=10): if 'up' in args: raise NetworkManagerError('sanitized') return '7 (no-secrets)' monkeypatch.setattr(subprocess, 'run', lambda *a, **k: subprocess.CompletedProcess(a, 0, 'wlan0: CTRL-EVENT-SSID-TEMP-DISABLED id=0 ssid="fixture" auth_failures=1 duration=10 reason=WRONG_KEY', '')) with pytest.raises(NetworkManagerError) as error: Backend().activate('public-test-id') assert error.value.error_code == 'AUTH_FAILED' assert str(error.value) == 'WiFi activation failed' @pytest.mark.parametrize('failure', ['AUTH_FAILED', 'NETWORK_UNAVAILABLE', 'IP_CONFIG_FAILED', 'UNKNOWN']) def test_task_returns_only_sanitized_failure_code(tmp_path, failure): from app.network.manager import NetworkManagerError mobile = create_app(project_root=tmp_path, driver_kind='mock').state.mobile_control def fail(*args, **kwargs): raise NetworkManagerError('private backend detail must not escape', error_code=failure) mobile.dispatch('wifi.set', dict(expected_revision=mobile.wifi()['revision'], ssid='Test Open', security='open', password_action='none', ipv4_mode='dhcp', dns_servers=[], activation='next_boot')) mobile.network.backend.activate = fail task = 'public-test-task' mobile._activate(task) result = mobile.dispatch('task.get', {'task_id': task}) assert result == {'state': 'failed', 'stage': 'finished', 'error_code': failure} assert 'private' not in str(result) def test_animation_chunks_and_read_only(tmp_path): import base64 import hashlib from io import BytesIO from PIL import Image mobile = create_app(project_root=tmp_path, driver_kind='mock').state.mobile_control item = next(i for i in mobile.dispatch('library.list', {})['items'] if i['type'] == 'animation') info = mobile.dispatch('library.animation.info', {**item, 'limit': 1}) assert info['frames'][0]['index'] == 0 frames = list(info['frames']) while info['next_offset'] is not None: info = mobile.dispatch('library.animation.info', {**item, 'limit': 1, 'offset': info['next_offset']}) frames += info['frames'] assert len(frames) == info['frame_count'] assert all(f['duration_ms'] > 0 for f in frames) before = mobile.dispatch('frame.get', {})['frame_revision'] for tier, size in [('preview', 16), ('full', 64)]: for frame in frames: offset, data = 0, bytearray() while True: result = mobile.dispatch('library.animation.chunk', {**item, 'tier': tier, 'frame_index': frame['index'], 'offset': offset}) chunk = base64.b64decode(result['data_base64'], validate=True) assert 0 < len(chunk) <= 4096 and result['offset'] == offset data.extend(chunk) if result['next_offset'] is None: break assert result['next_offset'] == offset + len(chunk) offset = result['next_offset'] assert len(data) == result['total_bytes'] assert hashlib.sha256(data).hexdigest() == result['sha256'] with Image.open(BytesIO(data)) as image: assert image.size == (size, size) assert mobile.dispatch('frame.get', {})['frame_revision'] == before for bad in [{'offset': -1}, {'limit': 51}, {'offset': True}]: with pytest.raises(RpcError, match='BAD_REQUEST'): mobile.dispatch('library.animation.info', {**item, **bad}) for bad in [{'frame_index': -1}, {'frame_index': True}, {'tier': 'invalid'}, {'offset': 999999}]: with pytest.raises(RpcError, match='BAD_REQUEST'): mobile.dispatch('library.animation.chunk', {**item, 'tier': 'full', 'frame_index': 0, **bad}) with pytest.raises(RpcError, match='CONFLICT'): mobile.dispatch('library.animation.info', {**item, 'revision': 'stale'})