homeassistant_espscreen/tests/test_renamed_inbox.py

"""Firmware 0.2.34 gave the screen entities English names. Home Assistant removes a renamed ESPHome entity
and registers it under a new entity id, so the manager moves a screen's layout and update history to the
new inbox id instead of orphaning them (app 0.2.40)."""
import asyncio
from datetime import timezone
import importlib.util
import json
from pathlib import Path
import sys
import tempfile
import unittest

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / 'screen_manager/app'))
from core import FIRMWARE_VERSION, device_prefixes, discover_screens, entity_slug, inbox_prefix

HAS_AIOHTTP = importlib.util.find_spec('aiohttp') is not None

DUTCH = {'text': 'Tegelinstellingen', 'firmware': 'Schermfirmware', 'node': 'Apparaatnaam', 'ip': 'IP-adres'}
ENGLISH = {'text': 'Tile settings', 'firmware': 'Screen firmware', 'node': 'Device name', 'ip': 'IP address'}


def entities(prefix, device, names, firmware, node, ip='10.0.0.5'):
    """Registry entries and states one screen creates; 'Node Status' is a name the translation kept."""
    ids = {'text': f'text.{prefix}_{entity_slug(names["text"])}', 'firmware': f'sensor.{prefix}_{entity_slug(names["firmware"])}',
           'node': f'sensor.{prefix}_{entity_slug(names["node"])}', 'ip': f'sensor.{prefix}_{entity_slug(names["ip"])}',
           'status': f'binary_sensor.{prefix}_node_status'}
    registry = [{'entity_id': ids[key], 'platform': 'esphome', 'original_name': names[key], 'device_id': device} for key in names]
    registry.append({'entity_id': ids['status'], 'platform': 'esphome', 'original_name': 'Node Status', 'device_id': device})
    states = {ids['text']: {'state': 'Synced'}, ids['firmware']: {'state': firmware}, ids['node']: {'state': node},
              ids['ip']: {'state': ip}, ids['status']: {'state': 'on'}}
    return registry, states


class PrefixTests(unittest.TestCase):
    def test_slugs_and_prefixes(self):
        self.assertEqual([entity_slug(n) for n in ('Tile settings', 'IP-adres', 'Node Status', 'Guition schermtype')],
                         ['tile_settings', 'ip_adres', 'node_status', 'guition_schermtype'])
        self.assertEqual(inbox_prefix('text.office_1_tegelinstellingen'), 'office_1')
        self.assertEqual(inbox_prefix('text.office_1_tile_settings'), 'office_1')
        for other in ('text.tile_settings', 'sensor.office_1_tile_settings', 'text.office_1_tile_settings_2', 'text.my_inbox', None):
            self.assertIsNone(inbox_prefix(other), other)
        registry, _ = entities('office_1', 'd1', ENGLISH, '0.2.34', 'office-1')
        registry.append({'entity_id': 'text.renamed_by_hand', 'platform': 'esphome', 'original_name': 'Tile settings', 'device_id': 'd1'})
        registry.append({'entity_id': 'light.office_1_lamp', 'platform': 'hue', 'original_name': 'Lamp', 'device_id': 'd1'})
        self.assertEqual(device_prefixes(registry, 'd1'), {'office_1'})
        self.assertEqual(device_prefixes(registry, 'd2'), set())

    def test_both_languages_are_discovered(self):
        for names, inbox in ((DUTCH, 'text.office_1_tegelinstellingen'), (ENGLISH, 'text.office_1_tile_settings')):
            registry, states = entities('office_1', 'd1', names, '0.2.33', 'office-1')
            [screen] = discover_screens(registry, states, [{'id': 'd1', 'name': 'Office 1'}], [])
            self.assertEqual((screen['id'], screen['firmware'], screen['node'], screen['ip']), (inbox, '0.2.33', 'office-1', '10.0.0.5'))


class FakeFirmware:
    """The ESPHome CLI stand-in: an install reflashes the screen with English entity names."""
    def __init__(self, ha):
        self.ha, self.task, self.job, self.calls = ha, None, None, []

    def profile_names(self):
        return {'office-1.yaml': {'node': 'office-1', 'friendly': 'Office 1'}}

    def start(self, data):
        self.calls.append(data)
        self.job = {'state': 'running', **data}
        self.task = asyncio.create_task(self.run())
        return self.job

    async def run(self):
        await asyncio.sleep(0)
        registry, states = entities('office_1', 'd1', ENGLISH, FIRMWARE_VERSION, 'office-1')
        self.ha.registry = registry + [item for item in self.ha.registry if item.get('device_id') != 'd1']
        self.ha.states = {**{k: v for k, v in self.ha.states.items() if 'office_1' not in k}, **states}
        self.job['state'] = 'success'


def fake_ha():
    class HA:
        online = True
        time_zone = timezone.utc

        def __init__(self):
            office, office_states = entities('office_1', 'd1', DUTCH, '0.2.30', 'office-1')
            kitchen, kitchen_states = entities('kitchen', 'd2', ENGLISH, FIRMWARE_VERSION, 'kitchen', '10.0.0.6')
            self.registry = office + kitchen + [{'entity_id': 'light.a', 'platform': 'hue'}]
            self.states = {**office_states, **kitchen_states, 'light.a': {'state': 'on', 'attributes': {'friendly_name': 'Lamp'}}}
            self.devices, self.areas = [{'id': 'd1', 'name': 'Office 1'}, {'id': 'd2', 'name': 'Kitchen'}], []
            self.changed, self.dirty, self.relevant, self.messages, self.calls = asyncio.Event(), set(), None, [], []

        async def send(self, inbox, message, action=None):
            self.messages.append((inbox, message, action))

        async def request(self, kind, **data):
            self.calls.append((kind, data))
    return HA()


OLD, NEW, KITCHEN = 'text.office_1_tegelinstellingen', 'text.office_1_tile_settings', 'text.kitchen_tile_settings'
LAYOUT = {'title': 'Office', 'tiles': [{'entity': 'light.a', 'name': 'Lamp', 'slot': 0}]}


@unittest.skipUnless(HAS_AIOHTTP, 'Run using .venv-portal/bin/python for server tests')
class RenamedInboxTests(unittest.IsolatedAsyncioTestCase):
    def manager(self, path, stored=None, updates=None, ha=None):
        from server import Manager
        if stored is not None:
            (Path(path) / 'screens.json').write_text(json.dumps({'version': 1, 'screens': stored}))
        if updates is not None:
            (Path(path) / 'updates.json').write_text(json.dumps({'version': 1, 'auto': False, **updates}))
        m = Manager(ha or fake_ha(), Path(path) / 'screens.json')
        m.firmware = FakeFirmware(m.ha)
        for attr in ('verify_timeout', 'settle_seconds', 'pause_seconds', 'poll_seconds'):
            setattr(m.updates, attr, 1 if attr == 'verify_timeout' else 0)
        return m

    def reflash(self, m):
        registry, states = entities('office_1', 'd1', ENGLISH, FIRMWARE_VERSION, 'office-1')
        m.ha.registry = registry + [item for item in m.ha.registry if item.get('device_id') != 'd1']
        m.ha.states = {**{k: v for k, v in m.ha.states.items() if 'office_1' not in k}, **states}

    async def test_reflash_moves_layout_and_update_history(self):
        with tempfile.TemporaryDirectory() as tmp:
            kitchen = {'title': 'Kitchen', 'tiles': []}
            m = self.manager(tmp, {OLD: LAYOUT, KITCHEN: kitchen},
                             {'hosts': {OLD: '10.0.0.5'}, 'results': {OLD: {'state': 'success', 'message': 'Updated.'}}})
            self.assertEqual(sorted(s['id'] for s in m.screens()), [KITCHEN, OLD])
            self.assertEqual(m.aliases, {})
            self.reflash(m)
            self.assertEqual(sorted(s['id'] for s in m.screens()), [KITCHEN, NEW])
            self.assertEqual(m.layouts[NEW]['title'], 'Office')
            self.assertNotIn(OLD, m.layouts)
            self.assertEqual(m.layouts[KITCHEN], kitchen)
            self.assertEqual(json.loads((Path(tmp) / 'screens.json').read_text())['screens'].keys(), {NEW, KITCHEN})
            self.assertEqual((m.updates.hosts, list(m.updates.results)), ({NEW: '10.0.0.5'}, [NEW]))
            self.assertEqual(json.loads((Path(tmp) / 'updates.json').read_text())['hosts'], {NEW: '10.0.0.5'})
            self.assertEqual(m.aliases, {OLD: NEW})
            self.assertEqual(m.screen(OLD)['id'], NEW, 'lookups by the old id follow the screen')
            # A page opened before the rename still saves to the right screen.
            m.save(OLD, {**LAYOUT, 'title': 'Office 2'})
            self.assertEqual((m.layouts[NEW]['title'], OLD in m.layouts), ('Office 2', False))
            # The sync loop sends the layout to the new inbox.
            m.ha.changed.set()
            runner = asyncio.create_task(m.run())
            for _ in range(50):
                await asyncio.sleep(0.02)
                if any(inbox == NEW for inbox, _, _ in m.ha.messages):
                    break
            runner.cancel()
            self.assertTrue(any(inbox == NEW and msg['op'] == 'layout' and msg['inbox'] == NEW for inbox, msg, _ in m.ha.messages))

    async def test_pages_opened_before_the_rename_keep_working(self):
        from aiohttp.test_utils import TestClient, TestServer
        from server import create_app
        with tempfile.TemporaryDirectory() as tmp:
            m = self.manager(tmp, {OLD: LAYOUT})
            async with TestClient(TestServer(create_app(m, True))) as client:
                csrf = (await (await client.get('/api/inventory?light=1')).json())['csrf']
                self.reflash(m)
                light = await (await client.get('/api/inventory?light=1')).json()
                office = next(s for s in light['screens'] if s['node'] == 'office-1')
                self.assertEqual((office['id'], office['layout']['title']), (NEW, 'Office'))
                inspected = await (await client.get(f'/api/screens/{OLD}/inspect')).json()
                self.assertEqual((inspected['screen']['id'], [t['entity'] for t in inspected['tiles']]), (NEW, ['light.a']))
                saved = await client.put(f'/api/screens/{OLD}', headers={'X-Screen-CSRF': csrf}, json={**LAYOUT, 'title': 'Office 3'})
                self.assertEqual(saved.status, 200)
                self.assertEqual((m.layouts[NEW]['title'], OLD in m.layouts), ('Office 3', False))

    async def test_restart_after_reflash_adopts_by_device_prefix(self):
        with tempfile.TemporaryDirectory() as tmp:
            ha = fake_ha()
            registry, states = entities('office_1', 'd1', ENGLISH, FIRMWARE_VERSION, 'office-1')
            ha.registry = registry + [item for item in ha.registry if item.get('device_id') != 'd1']
            ha.states = {**{k: v for k, v in ha.states.items() if 'office_1' not in k}, **states}
            m = self.manager(tmp, {OLD: LAYOUT}, ha=ha)
            self.assertEqual(sorted(s['id'] for s in m.screens()), [KITCHEN, NEW])
            self.assertEqual(list(m.layouts), [NEW])
            self.assertIn(NEW, [screen['id'] for screen in m.inventory()[0]])

    async def test_orphaned_layouts_of_other_screens_stay_put(self):
        with tempfile.TemporaryDirectory() as tmp:
            hall = 'text.hall_tegelinstellingen'
            m = self.manager(tmp, {hall: LAYOUT})
            self.reflash(m)
            m.screens()
            self.assertEqual(list(m.layouts), [hall], 'a removed screen keeps its layout; nobody else takes it')
            self.assertEqual(m.aliases, {})

    async def test_a_layout_on_the_new_id_is_never_overwritten(self):
        with tempfile.TemporaryDirectory() as tmp:
            newer = {'title': 'New', 'tiles': []}
            m = self.manager(tmp, {OLD: LAYOUT, NEW: newer})
            self.reflash(m)
            m.screens()
            self.assertEqual((m.layouts[NEW], m.layouts[OLD]), (newer, LAYOUT))
            self.assertEqual(m.aliases, {OLD: NEW})
            m.ha.registry = list(m.ha.registry)  # a later registry refresh does not try again
            before = dict(m.layouts)
            m.screens()
            self.assertEqual((m.layouts, m.aliases), (before, {OLD: NEW}))

    async def test_update_round_follows_the_screen_through_the_rename(self):
        with tempfile.TemporaryDirectory() as tmp:
            m = self.manager(tmp, {OLD: LAYOUT})
            [screen] = [s for s in m.screens() if s['id'] == OLD]
            self.assertTrue(m.updates.state_for(screen)['available'])
            m.updates.start(OLD)
            await m.updates.task
            self.assertEqual(m.firmware.calls, [{'file': 'office-1.yaml', 'action': 'install', 'target': '10.0.0.5'}])
            self.assertEqual(list(m.updates.results), [NEW])
            self.assertEqual(m.updates.results[NEW]['state'], 'success', m.updates.results[NEW]['message'])
            self.assertEqual((m.updates.current, m.updates.queue), (None, []))
            self.assertEqual(list(m.layouts), [NEW])
            [after] = [s for s in m.screens() if s['id'] == NEW]
            self.assertFalse(m.updates.state_for(after)['available'])

    async def test_nightly_round_reports_through_the_new_id(self):
        with tempfile.TemporaryDirectory() as tmp:
            m = self.manager(tmp, {OLD: LAYOUT})
            async def fails_after_rename():
                await asyncio.sleep(0)
                self.reflash(m)
                m.firmware.job['state'] = 'failed'
            m.firmware.run = fails_after_rename
            await m.updates.run_round([OLD], automatic=True)
            self.assertEqual(m.updates.results[NEW]['state'], 'failed')
            self.assertEqual(m.ha.calls[0][1]['domain'], 'persistent_notification')
            self.assertIn('Office 1', m.ha.calls[0][1]['service_data']['message'])


if __name__ == '__main__':
    unittest.main()