"""tests/test_intents_registration.py — t17: the registered-intent feature. Four things are pinned here, in the order the user meets them: 1. THE MODEL AND THE COMMAND (`media_player.py`): what a registered intent is, that an unusable row never becomes a button, and — the part that matters — that the ``am start`` command is BUILT from validated data with plain text substitution and quoted values, so nothing a variable contains can become shell syntax. 2. THE SCREEN (`config_flow.py`): the Intents sub-menu, the selector with its Add-new sentinel, the per-intent form, removal, and the scan that offers a discovered candidate — plus the seed validator, which must produce a row IDENTICAL to one typed into the form. 3. THE BUTTON (`button.py`): one entity per registration, sending the declared defaults, failing loudly (the uninstalled-app case is the device's own message, not a swallowed no-op). 4. THE ACTION (`media_player._async_send_intent_service`): the intent lookup, the variables, and the output returned under ``intent_response``. The flow and the entity modules are loaded in ONE stub environment on purpose: point 1's model exists twice (the runtime's reader and the screen's), and the test at the bottom of ``TestTheModel`` drives both over the same rows so they cannot drift into meaning different things. """ from __future__ import annotations import asyncio import importlib import shlex import sys from pathlib import Path from types import SimpleNamespace import pytest sys.path.insert(0, str(Path(__file__).resolve().parent)) import entity_stublib # noqa: E402 from entity_stublib import load_entity_modules # noqa: E402 from nsh.const import ( # noqa: E402 CONF_REGISTERED_INTENTS, INTENT_FIELD_ACTION, INTENT_FIELD_ACTIVITY, INTENT_FIELD_CANDIDATE, INTENT_FIELD_CATEGORIES, INTENT_FIELD_CHOICE, INTENT_FIELD_DATA, INTENT_FIELD_DECLARED, INTENT_FIELD_LIMIT, INTENT_FIELD_NAME, INTENT_FIELD_PACKAGE, INTENT_FIELD_SCAN_PACKAGE, INTENT_NEW_SENTINEL, INTENT_REGISTRATION_LIMIT, INTENT_RESPONSE_KEY, ) from test_standalone_setup import FakeHass, _form_module, _install_stubs # noqa: E402 #: A registration as the plan describes it: a component, a data template with a #: declared placeholder, and a required field the caller must supply. NETFLIX_INTENT = { "id": "netflix_title", "name": "Netflix: open a title", "package": "com.netflix.ninja", "activity": "com.netflix.ninja/.MainActivity", "action": "android.intent.action.VIEW", "categories": ["android.intent.category.DEFAULT"], "data": "https://www.netflix.com/watch/{id}", "fields": [{"key": "id", "type": "string", "required": True}], } @pytest.fixture(autouse=True) def scoped_stubs(): """Keep the import stubs and the loaded modules inside this file (t7).""" snapshot = entity_stublib.snapshot_modules() _install_stubs() yield entity_stublib.restore_modules(snapshot) @pytest.fixture() def modules(tmp_path): """The flow module AND the entity modules, in one environment.""" flow = _form_module() media = load_entity_modules() return SimpleNamespace(flow=flow, media=media["media_player"]) class _RecordingPoller: """A poller that answers with what a test puts on the wire.""" def __init__(self, answer='Starting: Intent { act=android.intent.action.VIEW }'): self.answer = answer self.commands: list[str] = [] async def shell(self, command): self.commands.append(command) if isinstance(self.answer, Exception): raise self.answer return self.answer def _manager(poller, *, connected=True): return SimpleNamespace(adb_poller=poller, connected=connected, entry=None) # ── 1. the model and the command ─────────────────────────────────────────── class TestTheCommand: """The `am start` command: built, quoted, and never evaluated.""" def test_the_plan_s_mapping_is_exactly_what_is_sent(self, modules): command = modules.media.intent_am_start_command( NETFLIX_INTENT, {"id": "80100172"} ) assert command == ( "am start -n com.netflix.ninja/.MainActivity " "-a android.intent.action.VIEW -c android.intent.category.DEFAULT " "-d https://www.netflix.com/watch/80100172 --es id 80100172" ) def test_every_declared_type_becomes_its_own_flag(self, modules): intent = { **NETFLIX_INTENT, "data": "", "fields": [ {"key": "title", "type": "string", "required": True}, {"key": "season", "type": "int", "required": True}, {"key": "resume", "type": "bool", "required": True}, {"key": "extra", "type": "raw", "required": True}, ], } command = modules.media.intent_am_start_command( intent, {"title": "Sky", "season": "2", "resume": "YES", "extra": "--user 0"}, ) assert "--es title Sky" in command assert "--ei season 2" in command assert "--ez resume true" in command # normalised to the device's boolean assert "--es extra '--user 0'" in command, "raw is quoted like everything else" def test_a_hostile_variable_is_one_literal_argument(self, modules): hostile = '"; rm -rf /"' command = modules.media.intent_am_start_command( {**NETFLIX_INTENT, "data": ""}, {"id": hostile} ) tokens = shlex.split(command) # the shell sees the value as ONE argument, exactly as typed assert tokens == ["am", "start", "-n", NETFLIX_INTENT["activity"], "-a", NETFLIX_INTENT["action"], "-c", "android.intent.category.DEFAULT", "--es", "id", hostile] assert "rm -rf" not in command.split("--es")[0] def test_a_hostile_data_value_is_quoted_too(self, modules): command = modules.media.intent_am_start_command( NETFLIX_INTENT, {"id": "a; reboot"} ) assert shlex.split(command)[-1] == "a; reboot" assert "https://www.netflix.com/watch/a; reboot" in shlex.split(command) def test_substitution_is_plain_text_and_runs_once(self, modules): """A value that LOOKS like a placeholder is data, never re-substituted.""" intent = { **NETFLIX_INTENT, "data": "https://example.test/{id}", "fields": [{"key": "id", "type": "string", "required": True}], } command = modules.media.intent_am_start_command(intent, {"id": "{name}"}) assert "https://example.test/{name}" in command assert shlex.split(command)[-1] == "{name}" def test_attribute_style_placeholders_are_left_alone(self, modules): """No eval, no format(): ``{id.__class__}`` is not a placeholder at all.""" intent = {**NETFLIX_INTENT, "data": "https://example.test/{id.__class__}"} command = modules.media.intent_am_start_command(intent, {"id": "1"}) assert "https://example.test/{id.__class__}" in command def test_a_missing_required_field_names_the_field(self, modules): err = pytest.raises( modules.media.ServiceValidationError if hasattr(modules.media, "ServiceValidationError") else Exception ) with err as caught: modules.media.intent_am_start_command(NETFLIX_INTENT, {}) assert caught.value.translation_key == "missing_intent_field" assert caught.value.translation_placeholders["field"] == "id" assert "id" in str(caught.value) def test_an_undeclared_variable_is_refused_not_dropped(self, modules): import homeassistant.exceptions as ha_exceptions with pytest.raises(ha_exceptions.ServiceValidationError) as caught: modules.media.intent_am_start_command( NETFLIX_INTENT, {"id": "1", "sneaky": "x"} ) assert caught.value.translation_key == "unknown_intent_variable" assert caught.value.translation_placeholders["fields"] == "sneaky" def test_a_value_that_cannot_become_its_type_names_field_and_value(self, modules): import homeassistant.exceptions as ha_exceptions intent = { **NETFLIX_INTENT, "data": "", "fields": [{"key": "season", "type": "int", "required": True}], } with pytest.raises(ha_exceptions.ServiceValidationError) as caught: modules.media.intent_am_start_command(intent, {"season": "winter"}) assert caught.value.translation_key == "invalid_intent_field" assert caught.value.translation_placeholders["field"] == "season" assert "winter" in str(caught.value) def test_an_unfilled_placeholder_raises_instead_of_sending_braces(self, modules): import homeassistant.exceptions as ha_exceptions intent = {**NETFLIX_INTENT, "fields": []} # nothing declares {id} with pytest.raises(ha_exceptions.ServiceValidationError) as caught: modules.media.intent_am_start_command(intent, {}) assert caught.value.translation_key == "unresolved_intent_placeholder" assert caught.value.translation_placeholders["placeholder"] == "id" def test_defaults_satisfy_the_template(self, modules): intent = { **NETFLIX_INTENT, "fields": [{"key": "id", "type": "string", "default": "42"}], } assert "watch/42" in modules.media.intent_am_start_command(intent, {}) assert modules.media.intent_default_variables(intent) == {"id": "42"} def test_an_optional_field_without_a_value_is_simply_not_sent(self, modules): intent = { **NETFLIX_INTENT, "data": "", "fields": [{"key": "quality", "type": "string"}], } command = modules.media.intent_am_start_command(intent, {}) assert "--es" not in command assert "quality" not in command class TestRunningAnIntent: """What actually reaches the device, and what happens when it refuses.""" async def test_the_device_gets_exactly_one_built_command(self, modules): poller = _RecordingPoller() manager = _manager(poller) output = await modules.media.run_registered_intent( manager, "media_player.shield", NETFLIX_INTENT, {"id": "1"} ) assert output.startswith("Starting: Intent") assert poller.commands == [ modules.media.intent_am_start_command(NETFLIX_INTENT, {"id": "1"}) ] async def test_adb_down_says_the_protocol_cannot_carry_an_intent(self, modules): import homeassistant.exceptions as ha_exceptions for manager in (_manager(None), _manager(_RecordingPoller(), connected=False)): with pytest.raises(ha_exceptions.ServiceValidationError) as caught: await modules.media.run_registered_intent( manager, "media_player.shield", NETFLIX_INTENT, {"id": "1"} ) assert caught.value.translation_key == "intents_need_adb" assert "protocol" in str(caught.value) async def test_a_transport_that_never_answered_is_an_error(self, modules): import homeassistant.exceptions as ha_exceptions with pytest.raises(ha_exceptions.ServiceValidationError) as caught: await modules.media.run_registered_intent( _manager(_RecordingPoller(None)), "media_player.shield", NETFLIX_INTENT, {"id": "1"}, ) assert caught.value.translation_key == "intent_no_answer" async def test_a_refused_intent_keeps_the_device_s_own_words(self, modules): import homeassistant.exceptions as ha_exceptions refusal = ( "Starting: Intent { act=android.intent.action.VIEW }\n" "Error: Activity class {com.netflix.ninja/.MainActivity} does not exist." ) with pytest.raises(ha_exceptions.ServiceValidationError) as caught: await modules.media.run_registered_intent( _manager(_RecordingPoller(refusal)), "media_player.shield", NETFLIX_INTENT, {"id": "1"}, ) assert caught.value.translation_key == "intent_failed" assert "does not exist" in str(caught.value) assert "com.netflix.ninja/.MainActivity" in str(caught.value) def test_the_failure_detector_knows_a_success_from_a_refusal(self, modules): detect = modules.media.intent_failure_message assert detect("Starting: Intent { cmp=com.a/.B }") is None assert detect(None) is None assert "unable to resolve" in detect( "Error: Unable to resolve Intent { act=x }" ).lower() assert detect("Permission Denial: starting Intent requires permission") is not None # ── 2. the model as READ, and the screen's copy of it ────────────────────── class TestTheModel: """What counts as a usable registration, and the two readers agreeing.""" def test_a_row_that_names_nothing_to_launch_is_dropped(self, modules): rows = modules.media.registered_intents( { CONF_REGISTERED_INTENTS: [ NETFLIX_INTENT, {"id": "broken", "name": "Broken", "package": "com.x"}, {"id": "nolabel", "package": "com.x", "action": "a.B"}, ] } ) assert [row["id"] for row in rows] == ["netflix_title"] def test_categories_and_fields_are_normalised(self, modules): rows = modules.media.registered_intents( { CONF_REGISTERED_INTENTS: [ { **NETFLIX_INTENT, "categories": ["", " android.intent.category.DEFAULT "], "fields": [ {"key": "id", "type": "STRING", "required": 1}, {"key": "9bad", "type": "string"}, {"key": "ok", "type": "nope"}, ], } ] } ) assert rows[0]["categories"] == ["android.intent.category.DEFAULT"] assert rows[0]["fields"] == [{"key": "id", "type": "string", "required": True}] def test_duplicate_ids_keep_the_first_and_the_cap_holds(self, modules): rows = modules.media.registered_intents( { CONF_REGISTERED_INTENTS: [ {**NETFLIX_INTENT, "name": "First"}, {**NETFLIX_INTENT, "name": "Second"}, ] + [ {**NETFLIX_INTENT, "id": f"i{n}", "name": f"n{n}"} for n in range(INTENT_REGISTRATION_LIMIT + 5) ] } ) assert rows[0]["name"] == "First" assert len(rows) == INTENT_REGISTRATION_LIMIT def test_a_mapping_shaped_blob_still_reads(self, modules): rows = modules.media.registered_intents( {CONF_REGISTERED_INTENTS: {"netflix_title": NETFLIX_INTENT}} ) assert [row["id"] for row in rows] == ["netflix_title"] def test_lookup_by_id_and_by_name(self, modules): options = {CONF_REGISTERED_INTENTS: [NETFLIX_INTENT]} assert modules.media.intent_by_id(options, "netflix_title")["name"] == ( NETFLIX_INTENT["name"] ) assert modules.media.intent_by_id(options, NETFLIX_INTENT["name"]) is not None assert modules.media.intent_by_id(options, "nope") is None def test_the_screen_and_the_runtime_read_the_same_rows(self, modules): """The anti-drift pin: the flow's reader and the runtime's must agree. They live in different modules because the options flow must stay importable without Home Assistant's media_player platform. This drives both over one battery — valid rows, junk, a duplicate id, an over-long list, a mapping-shaped blob — and requires the same ids, names, packages, components, actions, categories and fields from each. """ battery = [ NETFLIX_INTENT, {**NETFLIX_INTENT, "id": "second"}, {"id": "junk", "name": "Junk", "package": "com.x"}, "not-a-dict", {**NETFLIX_INTENT}, {**NETFLIX_INTENT, "activity": "no-slash", "action": "a.B"}, {**NETFLIX_INTENT, "id": "fields", "fields": [{"key": "bad key"}]}, ] runtime = modules.media.registered_intents({CONF_REGISTERED_INTENTS: battery}) screen = modules.flow.registered_intent_rows({CONF_REGISTERED_INTENTS: battery}) assert runtime == screen # ── 3. the screen ────────────────────────────────────────────────────────── def _flow(modules, tmp_path, options=None): flow = modules.flow.NvidiaShieldOptionsFlow() flow.config_entry = SimpleNamespace( options=dict(options or {}), entry_id="e1", title="Shield" ) flow.hass = FakeHass(str(tmp_path)) flow.async_show_form = lambda **kwargs: {"type": "form", **kwargs} flow.async_show_menu = lambda **kwargs: {"type": "menu", **kwargs} flow.async_create_entry = lambda **kwargs: {"type": "create_entry", **kwargs} return flow def _form_payload(**overrides): payload = { INTENT_FIELD_NAME: NETFLIX_INTENT["name"], INTENT_FIELD_PACKAGE: NETFLIX_INTENT["package"], INTENT_FIELD_ACTIVITY: NETFLIX_INTENT["activity"], INTENT_FIELD_ACTION: NETFLIX_INTENT["action"], INTENT_FIELD_CATEGORIES: "android.intent.category.DEFAULT", INTENT_FIELD_DATA: NETFLIX_INTENT["data"], INTENT_FIELD_DECLARED: "id:string:required", } payload.update(overrides) return payload class TestTheIntentsScreen: """The menu, the sentinel, the form, removal — and nothing half-stored.""" async def test_the_menu_offers_the_screen_and_labels_it(self, modules, tmp_path): flow = _flow(modules, tmp_path) menu = await flow.async_step_intents(None) assert menu["type"] == "menu" assert list(menu["menu_options"]) == ["intents_edit", "intents_scan", "menu_again"] assert "no intents registered yet" in menu["description_placeholders"]["intents"] async def test_remove_appears_only_with_something_to_remove(self, modules, tmp_path): flow = _flow(modules, tmp_path) menu = await flow.async_step_intents(None) assert "intents_remove" not in menu["menu_options"] flow._options_dict[CONF_REGISTERED_INTENTS] = [NETFLIX_INTENT] menu = await flow.async_step_intents(None) assert "intents_remove" in menu["menu_options"] assert "1 intent registered" in menu["description_placeholders"]["intents"] async def test_the_selector_lists_every_intent_and_the_add_new_sentinel( self, modules, tmp_path ): flow = _flow(modules, tmp_path, {CONF_REGISTERED_INTENTS: [NETFLIX_INTENT]}) result = await flow.async_step_intents_edit(None) assert result["type"] == "form" and result["step_id"] == "intents_edit" # the selector is the field the form declares; its options carry the # registered intent AND the explicit Add-new row choices = flow._intent_choices() assert [row["value"] for row in choices] == ["netflix_title", INTENT_NEW_SENTINEL] assert choices[-1]["label"].startswith("+ Add a new intent") async def test_add_new_opens_a_blank_form(self, modules, tmp_path): flow = _flow(modules, tmp_path, {CONF_REGISTERED_INTENTS: [NETFLIX_INTENT]}) form = await flow.async_step_intents_edit( {INTENT_FIELD_CHOICE: INTENT_NEW_SENTINEL} ) assert form["type"] == "form" and form["step_id"] == "intents_form" assert "new intent" in form["description_placeholders"]["intent_note"] async def test_an_unknown_choice_is_refused(self, modules, tmp_path): flow = _flow(modules, tmp_path, {CONF_REGISTERED_INTENTS: [NETFLIX_INTENT]}) result = await flow.async_step_intents_edit({INTENT_FIELD_CHOICE: "ghost"}) assert result["errors"] == {"base": "unknown_intent_row"} async def test_a_new_intent_is_stored_with_a_slug_id(self, modules, tmp_path): flow = _flow(modules, tmp_path) result = await flow.async_step_intents_form(_form_payload()) assert result["type"] == "menu" # back at the Intents editor stored = flow._options_dict[CONF_REGISTERED_INTENTS] assert len(stored) == 1 assert stored[0]["id"] == "netflix_open_a_title" assert stored[0]["fields"] == [{"key": "id", "type": "string", "required": True}] assert stored[0]["categories"] == ["android.intent.category.DEFAULT"] async def test_stored_rows_are_json_safe(self, modules, tmp_path): """Options are persisted as JSON: a set or a tuple here would break storage.""" import json flow = _flow(modules, tmp_path) await flow.async_step_intents_form(_form_payload()) json.dumps(flow._options_dict) # raises if a row is not serialisable async def test_editing_one_row_leaves_the_others_alone(self, modules, tmp_path): other = {**NETFLIX_INTENT, "id": "other", "name": "Another"} flow = _flow( modules, tmp_path, {CONF_REGISTERED_INTENTS: [NETFLIX_INTENT, other]} ) result = await flow.async_step_intents_edit( {INTENT_FIELD_CHOICE: "netflix_title"} ) assert result["step_id"] == "intents_form" assert result["description_placeholders"]["intent_note"].startswith("editing ") await flow.async_step_intents_form( _form_payload(**{INTENT_FIELD_NAME: "Renamed"}) ) stored = flow._options_dict[CONF_REGISTERED_INTENTS] assert [row["id"] for row in stored] == ["netflix_title", "other"] assert stored[0]["name"] == "Renamed" # the rename landed on THAT row assert stored[1] == other # …and this one is byte-identical to before async def test_a_refused_submission_stores_nothing_and_says_why( self, modules, tmp_path ): flow = _flow(modules, tmp_path, {CONF_REGISTERED_INTENTS: [NETFLIX_INTENT]}) result = await flow.async_step_intents_form( _form_payload( **{ INTENT_FIELD_NAME: "Nothing to launch", INTENT_FIELD_ACTIVITY: "", INTENT_FIELD_ACTION: "", } ) ) assert result["type"] == "form" assert result["errors"] == {"base": "invalid_intent_row"} note = result["description_placeholders"]["intent_note"] assert "component" in note and "action" in note assert len(flow._options_dict[CONF_REGISTERED_INTENTS]) == 1 # unchanged async def test_each_refusal_names_its_own_problem(self, modules, tmp_path): cases = [ ({INTENT_FIELD_PACKAGE: "has spaces"}, "not a usable package id"), ({INTENT_FIELD_ACTIVITY: "com.x.MainActivity"}, "not a component"), ({INTENT_FIELD_DATA: "https://x/{nope}"}, "no declared field"), ({INTENT_FIELD_DECLARED: "id:weird"}, "is not one of"), ({INTENT_FIELD_NAME: NETFLIX_INTENT["name"]}, "already called"), ] for overrides, expected in cases: flow = _flow(modules, tmp_path, {CONF_REGISTERED_INTENTS: [NETFLIX_INTENT]}) payload = _form_payload(**overrides) if INTENT_FIELD_NAME in overrides: payload[INTENT_FIELD_CHOICE] = INTENT_NEW_SENTINEL else: flow._options_dict[CONF_REGISTERED_INTENTS] = [NETFLIX_INTENT] payload[INTENT_FIELD_NAME] = "New name" result = await flow.async_step_intents_form(payload) assert result["errors"] == {"base": "invalid_intent_row"}, overrides assert expected in result["description_placeholders"]["intent_note"], overrides async def test_removal_takes_a_confirmation_and_drops_only_that_row( self, modules, tmp_path ): other = {**NETFLIX_INTENT, "id": "other", "name": "Another"} flow = _flow( modules, tmp_path, {CONF_REGISTERED_INTENTS: [NETFLIX_INTENT, other]} ) confirm = await flow.async_step_intents_remove( {INTENT_FIELD_CHOICE: "netflix_title"} ) assert confirm["step_id"] == "intents_remove_confirm" assert confirm["description_placeholders"]["intent"] == ( "Netflix: open a title (com.netflix.ninja)" ) assert confirm["description_placeholders"]["button"] == NETFLIX_INTENT["name"] assert "intents_remove_yes" in confirm["menu_options"], "it takes a second act" # nothing is dropped until the user confirms assert len(flow._options_dict[CONF_REGISTERED_INTENTS]) == 2 await flow.async_step_intents_remove_yes(None) stored = flow._options_dict[CONF_REGISTERED_INTENTS] assert [row["id"] for row in stored] == ["other"] async def test_cancelling_a_removal_changes_nothing(self, modules, tmp_path): flow = _flow(modules, tmp_path, {CONF_REGISTERED_INTENTS: [NETFLIX_INTENT]}) await flow.async_step_intents_remove({INTENT_FIELD_CHOICE: "netflix_title"}) result = await flow.async_step_intents_remove_no(None) assert result["step_id"] == "intents" assert flow._options_dict[CONF_REGISTERED_INTENTS] == [NETFLIX_INTENT] async def test_a_registration_survives_a_new_flow_on_the_same_options( self, modules, tmp_path ): """The options ARE the store: a fresh flow reads what the last one wrote.""" options = {} first = _flow(modules, tmp_path, options) await first.async_step_intents_form(_form_payload()) options.update(first._options_dict) second = _flow(modules, tmp_path, options) assert [row["id"] for row in second._registered_intents()] == [ "netflix_open_a_title" ] class TestTheDeclaredFieldsText: """The `key:type[:required][=value]` grammar, one line at a time.""" def test_the_grammar_reads_every_shape(self, modules): fields, problems = modules.flow.intent_fields_from_text( "# a comment\nid:string:required\nseason:int:required\n" "quality:string=high\nresume:bool\nblob:raw\n" ) assert problems == [] assert fields == [ {"key": "id", "type": "string", "required": True}, {"key": "season", "type": "int", "required": True}, {"key": "quality", "type": "string", "required": False, "default": "high"}, {"key": "resume", "type": "bool", "required": False}, {"key": "blob", "type": "raw", "required": False}, ] def test_every_problem_names_its_line(self, modules): fields, problems = modules.flow.intent_fields_from_text( "id:string\nnoline\n9bad:string\nid:string:maybe\nboth:int:required=1\nid:string\n" ) assert fields is None assert [problem.split(":")[0] for problem in problems] == [ "fields line 2", "fields line 3", "fields line 4", "fields line 5", "fields line 6", ] def test_the_text_round_trips_through_the_form(self, modules): fields = [ {"key": "id", "type": "string", "required": True}, {"key": "quality", "type": "string", "required": False, "default": "high"}, ] text = modules.flow.intent_fields_to_text(fields) assert text == "id:string:required\nquality:string=high" assert modules.flow.intent_fields_from_text(text)[0] == fields def test_the_cap_is_refused_rather_than_truncated(self, modules): text = "\n".join(f"k{n}:string" for n in range(INTENT_FIELD_LIMIT + 2)) fields, problems = modules.flow.intent_fields_from_text(text) assert fields is None assert any("cap" in problem for problem in problems) class _FakeScan: """A stand-in for `manager.async_scan_intents` (t16's API).""" def __init__(self, result=None, exc=None): self.result = result self.exc = exc self.packages: list[str] = [] async def __call__(self, package): self.packages.append(package) if self.exc is not None: raise self.exc return self.result def _candidate(**overrides): """One discovered candidate, with the dedup KEY the real one carries. ``candidate_value`` is built from ``IntentCandidate.key`` (t16), so the fake has to have one — and two fakes differing only in a field must differ in it, exactly as the real candidates do. """ base = { "component": "com.netflix.ninja/.MainActivity", "action": "android.intent.action.VIEW", "categories": ("android.intent.category.DEFAULT",), "scheme": "https", "authority": "www.netflix.com", "path": "/browse", "path_kind": "LITERAL", "likely_internal": False, } base.update(overrides) base["key"] = ( base["component"], base["action"], base["categories"], base["scheme"], base["authority"], base["path"], base["path_kind"], ) return SimpleNamespace(**base) def _scan_result(candidates=(), *, status="fresh", package="com.netflix.ninja"): return SimpleNamespace( package=package, status=status, candidates=tuple(candidates), entry_state="candidates" if candidates else "empty", age_seconds=0.0, truncated=False, ) class TestTheScanStep: """The scan answers honestly, and a candidate prefills only what it can.""" async def test_no_manager_anywhere_is_reported_not_crashed(self, modules, tmp_path): flow = _flow(modules, tmp_path) flow.config_entry = SimpleNamespace(options={}, entry_id="e1", title="Shield") flow.hass.data = {} # no runtime_data, no domain store result = await flow.async_step_intents_scan( {INTENT_FIELD_SCAN_PACKAGE: "com.netflix.ninja"} ) assert result["step_id"] == "intents_scan_result" note = result["description_placeholders"]["scan_note"] assert "no intent discovery" in note assert "Nothing was registered" in note async def test_a_bad_package_never_reaches_the_manager(self, modules, tmp_path): flow = _flow(modules, tmp_path) scan = _FakeScan(_scan_result()) flow.config_entry.runtime_data = SimpleNamespace(async_scan_intents=scan) result = await flow.async_step_intents_scan({INTENT_FIELD_SCAN_PACKAGE: "not one"}) assert result["errors"] == {"base": "invalid_intent_package"} assert scan.packages == [] async def test_one_package_is_scanned_and_its_candidates_offered( self, modules, tmp_path ): flow = _flow(modules, tmp_path) scan = _FakeScan( _scan_result( [_candidate(), _candidate(path="/title.*", path_kind="GLOB")] ) ) flow.config_entry.runtime_data = SimpleNamespace(async_scan_intents=scan) result = await flow.async_step_intents_scan( {INTENT_FIELD_SCAN_PACKAGE: "com.netflix.ninja"} ) assert scan.packages == ["com.netflix.ninja"], "ONE package per request" note = result["description_placeholders"]["scan_note"] assert "2 candidate intent(s)" in note assert "Nothing has been registered yet" in note async def test_each_status_reads_differently(self, modules, tmp_path): cases = { "fresh_empty": "advertises no intents", "adb_disabled": "switched OFF", "adb_unavailable": "did not answer", "unreadable": "did not answer", "cached": "cache", } for status, expected in cases.items(): note = modules.flow.intent_scan_note( _scan_result(status=status), "com.netflix.ninja" ) assert expected in note, status async def test_a_literal_candidate_prefills_a_real_url(self, modules, tmp_path): candidate = _candidate() # https + authority + LITERAL /browse flow = _flow(modules, tmp_path) scan = _FakeScan(_scan_result([candidate])) flow.config_entry.runtime_data = SimpleNamespace(async_scan_intents=scan) await flow.async_step_intents_scan( {INTENT_FIELD_SCAN_PACKAGE: "com.netflix.ninja"} ) form = await flow.async_step_intents_scan_result( {INTENT_FIELD_CANDIDATE: modules.flow.candidate_value(candidate)} ) assert form["step_id"] == "intents_form" assert form["type"] == "form" # the schema is rendered, not submitted assert form["description_placeholders"]["intent_note"] == "new intent" assert flow._intent_prefill[INTENT_FIELD_DATA] == "https://www.netflix.com/browse" assert flow._intent_prefill[INTENT_FIELD_ACTIVITY] == candidate.component assert flow._intent_prefill[INTENT_FIELD_ACTION] == candidate.action async def test_a_path_pattern_prefills_no_url_and_says_why(self, modules, tmp_path): candidate = _candidate(path="/title.*", path_kind="GLOB") flow = _flow(modules, tmp_path) scan = _FakeScan(_scan_result([candidate])) flow.config_entry.runtime_data = SimpleNamespace(async_scan_intents=scan) await flow.async_step_intents_scan( {INTENT_FIELD_SCAN_PACKAGE: "com.netflix.ninja"} ) form = await flow.async_step_intents_scan_result( {INTENT_FIELD_CANDIDATE: modules.flow.candidate_value(candidate)} ) assert flow._intent_prefill[INTENT_FIELD_DATA] == "" assert "pattern" in form["description_placeholders"]["intent_note"] async def test_an_internal_candidate_is_offered_and_marked(self, modules, tmp_path): candidate = _candidate(action="com.netflix.action.DEBUG_FORCE_EXIT", likely_internal=True) assert "[internal-looking]" in modules.flow.candidate_label(candidate) flow = _flow(modules, tmp_path) flow.config_entry.runtime_data = SimpleNamespace( async_scan_intents=_FakeScan(_scan_result([candidate])) ) await flow.async_step_intents_scan( {INTENT_FIELD_SCAN_PACKAGE: "com.netflix.ninja"} ) form = await flow.async_step_intents_scan_result( {INTENT_FIELD_CANDIDATE: modules.flow.candidate_value(candidate)} ) assert "internal hook" in form["description_placeholders"]["intent_note"] async def test_a_scan_that_cannot_run_registers_nothing(self, modules, tmp_path): flow = _flow(modules, tmp_path) flow.config_entry.runtime_data = SimpleNamespace( async_scan_intents=_FakeScan(exc=RuntimeError("adb died")) ) result = await flow.async_step_intents_scan( {INTENT_FIELD_SCAN_PACKAGE: "com.netflix.ninja"} ) assert result["step_id"] == "intents_scan_result" assert CONF_REGISTERED_INTENTS not in flow._options_dict class TestTheSeed: """`registered_intents:` in configuration.yaml — validated, or named and dropped.""" def test_a_valid_seed_produces_the_form_s_row(self, modules): seeded = modules.flow._seeded_intent_rows([NETFLIX_INTENT]) assert len(seeded) == 1 # …and it is the SAME row the form would have built from those values typed, problems = modules.flow.intent_row_from_form( { INTENT_FIELD_NAME: NETFLIX_INTENT["name"], INTENT_FIELD_PACKAGE: NETFLIX_INTENT["package"], INTENT_FIELD_ACTIVITY: NETFLIX_INTENT["activity"], INTENT_FIELD_ACTION: NETFLIX_INTENT["action"], INTENT_FIELD_CATEGORIES: ", ".join(NETFLIX_INTENT["categories"]), INTENT_FIELD_DATA: NETFLIX_INTENT["data"], INTENT_FIELD_DECLARED: "id:string:required", }, existing_id=NETFLIX_INTENT["id"], taken_ids=set(), taken_names=set(), ) assert problems == [] assert seeded[0] == typed def test_a_broken_row_is_dropped_with_a_warning(self, modules, caplog): rows = [ NETFLIX_INTENT, {"name": "No target", "package": "com.x"}, {"name": "Bad package", "package": "has spaces", "action": "a.B"}, "not a mapping", ] with caplog.at_level("WARNING"): seeded = modules.flow._seeded_intent_rows(rows) assert [row["id"] for row in seeded] == [NETFLIX_INTENT["id"]] assert len(caplog.records) == 3, [r.message for r in caplog.records] assert all("registered_intents row" in r.message for r in caplog.records) def test_an_explicit_id_is_kept_and_a_duplicate_is_skipped(self, modules, caplog): with caplog.at_level("WARNING"): seeded = modules.flow._seeded_intent_rows( [NETFLIX_INTENT, {**NETFLIX_INTENT, "name": "Duplicate"}] ) assert [row["id"] for row in seeded] == [NETFLIX_INTENT["id"]] assert "repeats the id" in caplog.records[0].message def test_every_documented_row_key_is_honoured(self, modules): """The README's documented keys and the seed validator must agree (t17). A key the docs show but the validator drops would be worse than a typo: it would look configured and do nothing. """ from test_readme import INTENT_ROW_KEYS row = { "id": "example", "name": "Example", "package": "com.example.app", "activity": "com.example.app/.MainActivity", "action": "android.intent.action.VIEW", "categories": ["android.intent.category.DEFAULT"], "data": "https://example.test/{slug}", "fields": [{"key": "slug", "type": "string"}], } assert set(row) == INTENT_ROW_KEYS, "the fixture must cover the docs" seeded = modules.flow._seeded_intent_rows([row]) assert set(seeded[0]) == INTENT_ROW_KEYS, "every documented key survives" def test_a_non_list_block_is_ignored_with_a_warning(self, modules, caplog): with caplog.at_level("WARNING"): seeded = modules.flow._seeded_intent_rows("nope") assert seeded == [] assert "must be a LIST" in caplog.records[0].message # ── 4. the buttons ───────────────────────────────────────────────────────── class TestTheButtons: """One entity per registration, sending the defaults, failing loudly.""" async def _setup(self, modules, options): entry = SimpleNamespace( entry_id="e1", options=options, title="Shield", runtime_data=None ) added: list = [] button = importlib.import_module("nsh.button") await button.async_setup_entry(None, entry, added.extend) return entry, added async def test_one_button_per_registered_intent(self, modules): _entry, added = await self._setup( modules, {CONF_REGISTERED_INTENTS: [NETFLIX_INTENT]} ) assert len(added) == 1 assert added[0]._attr_name == NETFLIX_INTENT["name"] assert added[0]._attr_unique_id.endswith("_intent_netflix_title") async def test_no_registration_means_no_button(self, modules): _entry, added = await self._setup(modules, {}) assert added == [] async def test_a_rename_keeps_the_entity_identity(self, modules): _entry, added = await self._setup( modules, {CONF_REGISTERED_INTENTS: [{**NETFLIX_INTENT, "name": "Renamed"}]} ) assert added[0]._attr_name == "Renamed" assert added[0]._attr_unique_id.endswith("_intent_netflix_title") async def test_the_button_shows_what_it_will_send(self, modules): _entry, added = await self._setup( modules, {CONF_REGISTERED_INTENTS: [NETFLIX_INTENT]} ) attrs = added[0].extra_state_attributes assert attrs["intent_id"] == "netflix_title" assert attrs["activity"] == NETFLIX_INTENT["activity"] assert attrs["fields"] == NETFLIX_INTENT["fields"] async def test_pressing_sends_the_declared_defaults(self, modules): entry, added = await self._setup( modules, {CONF_REGISTERED_INTENTS: [NETFLIX_INTENT]} ) poller = _RecordingPoller() entry.runtime_data = _manager(poller) button = added[0] button.entity_id = "button.shield_netflix_open_a_title" # with no default for the required field, the press is an honest error import homeassistant.exceptions as ha_exceptions with pytest.raises(ha_exceptions.ServiceValidationError) as caught: await button.async_press() assert caught.value.translation_key == "missing_intent_field" assert poller.commands == [], "nothing was sent" # give the field a default and the same press works intent = {**NETFLIX_INTENT, "fields": [{"key": "id", "type": "string", "default": "7"}]} _entry, added = await self._setup(modules, {CONF_REGISTERED_INTENTS: [intent]}) added[0]._manager = _manager(poller) added[0].entity_id = "button.shield_netflix_open_a_title" await added[0].async_press() assert poller.commands == [ modules.media.intent_am_start_command(intent, {"id": "7"}) ] async def test_an_uninstalled_app_fails_the_press_honestly(self, modules): """The plan's acceptance: a package that is gone must not press silently. The device is the authority here — its own sentence is what the error carries, so the user learns WHICH component vanished instead of seeing a button that did nothing. """ import homeassistant.exceptions as ha_exceptions intent = { **NETFLIX_INTENT, "fields": [{"key": "id", "type": "string", "default": "7"}], } entry, added = await self._setup( modules, {CONF_REGISTERED_INTENTS: [intent]} ) poller = _RecordingPoller( "Starting: Intent { act=android.intent.action.VIEW }\n" "Error: Activity class {com.netflix.ninja/.MainActivity} does not exist." ) entry.runtime_data = _manager(poller) added[0]._manager = _manager(poller) added[0].entity_id = "button.shield_netflix_open_a_title" with pytest.raises(ha_exceptions.ServiceValidationError) as caught: await added[0].async_press() assert caught.value.translation_key == "intent_failed" assert "does not exist" in str(caught.value) assert "com.netflix.ninja/.MainActivity" in str(caught.value) assert len(poller.commands) == 1, "attempted once, never retried" async def test_a_button_with_no_manager_fails_loudly(self, modules): """A press that cannot reach the Shield RAISES — it never no-ops. The field gets a default here on purpose: that leaves ADB as the only thing wrong, so the assertion can pin WHICH error the user sees. """ _entry, added = await self._setup( modules, { CONF_REGISTERED_INTENTS: [ { **NETFLIX_INTENT, "fields": [{"key": "id", "type": "string", "default": "7"}], } ] }, ) import homeassistant.exceptions as ha_exceptions with pytest.raises(ha_exceptions.ServiceValidationError) as caught: await added[0].async_press() assert caught.value.translation_key == "intents_need_adb" # ── 5. the action ────────────────────────────────────────────────────────── class _Call: """The shape `ServiceCall` has for the handler (data + hass).""" def __init__(self, data, hass=None): self.data = data self.hass = hass class TestTheAction: """`nvidia_shield.send_intent`: the lookup, the variables, the response.""" def _wire(self, modules, options, poller=None, entity_id="media_player.shield"): manager = _manager(poller or _RecordingPoller()) manager.entry = SimpleNamespace(options=options) modules.media._INTENT_TARGETS[entity_id] = manager return manager def _cleanup(self, modules, entity_id="media_player.shield"): modules.media._INTENT_TARGETS.pop(entity_id, None) async def test_the_output_comes_back_under_the_response_key(self, modules): poller = _RecordingPoller() self._wire(modules, {CONF_REGISTERED_INTENTS: [NETFLIX_INTENT]}, poller) try: result = await modules.media._async_send_intent_service( _Call( { "intent": NETFLIX_INTENT["name"], "variables": {"id": "1"}, "entity_id": "media_player.shield", } ) ) finally: self._cleanup(modules) assert set(result) == {INTENT_RESPONSE_KEY} assert result[INTENT_RESPONSE_KEY].startswith("Starting: Intent") assert poller.commands == [ modules.media.intent_am_start_command(NETFLIX_INTENT, {"id": "1"}) ] async def test_an_unknown_intent_lists_the_known_names(self, modules): import homeassistant.exceptions as ha_exceptions self._wire(modules, {CONF_REGISTERED_INTENTS: [NETFLIX_INTENT]}) try: with pytest.raises(ha_exceptions.ServiceValidationError) as caught: await modules.media._async_send_intent_service( _Call({"intent": "nope", "entity_id": "media_player.shield"}) ) finally: self._cleanup(modules) assert caught.value.translation_key == "unknown_intent" assert caught.value.translation_placeholders["known"] == NETFLIX_INTENT["name"] async def test_no_registered_intents_yet_is_said_plainly(self, modules): import homeassistant.exceptions as ha_exceptions self._wire(modules, {}) try: with pytest.raises(ha_exceptions.ServiceValidationError) as caught: await modules.media._async_send_intent_service( _Call({"intent": "anything", "entity_id": "media_player.shield"}) ) finally: self._cleanup(modules) assert caught.value.translation_placeholders["known"] == "none yet" async def test_variables_must_be_a_mapping(self, modules): import homeassistant.exceptions as ha_exceptions self._wire(modules, {CONF_REGISTERED_INTENTS: [NETFLIX_INTENT]}) try: with pytest.raises(ha_exceptions.ServiceValidationError) as caught: await modules.media._async_send_intent_service( _Call( { "intent": NETFLIX_INTENT["name"], "variables": "id=1", "entity_id": "media_player.shield", } ) ) finally: self._cleanup(modules) assert caught.value.translation_key == "invalid_intent_variables" async def test_the_target_errors_are_the_shared_ones(self, modules): import homeassistant.exceptions as ha_exceptions self._wire(modules, {CONF_REGISTERED_INTENTS: [NETFLIX_INTENT]}) try: # nothing selected with pytest.raises(ha_exceptions.ServiceValidationError) as caught: await modules.media._async_send_intent_service( _Call({"intent": NETFLIX_INTENT["name"]}) ) assert caught.value.translation_key == "no_target" assert "send_intent" in caught.value.translation_placeholders["service"] # an entity that does not participate in THIS action with pytest.raises(ha_exceptions.ServiceValidationError) as caught: await modules.media._async_send_intent_service( _Call({"intent": "x", "entity_id": "media_player.other"}) ) assert caught.value.translation_key == "unsupported_target" finally: self._cleanup(modules) async def test_the_key_a_template_reads_is_not_adb_response(self, modules): from nsh.const import INTENT_RESPONSE_KEY assert INTENT_RESPONSE_KEY == "intent_response" class TestTheServiceDescription: """The dropdown the frontend renders comes from the runtime description.""" def test_the_intent_field_is_a_dropdown_of_registered_names(self, modules): description = modules.media.send_intent_service_description( ["Netflix: open a title", "TiviMate: channel"] ) intent_field = description["fields"]["intent"] assert intent_field["required"] is True selector = intent_field["selector"]["select"] assert sorted(selector["options"]) == [ "Netflix: open a title", "TiviMate: channel", ] assert selector["custom_value"] is True assert selector["mode"] == "dropdown" def test_the_description_declares_the_target_and_the_variables_field(self, modules): description = modules.media.send_intent_service_description([]) assert description["target"] == {"entity": [{"domain": "media_player"}]} assert "variables" in description["fields"] assert description["fields"]["variables"]["selector"] == {"object": {}} assert description["fields"]["intent"]["selector"]["select"]["options"] == [] def test_the_strings_files_carry_the_same_service_and_exception_keys(self): """The runtime description and the translation cannot disagree.""" import json base = Path(__file__).resolve().parent.parent / "custom_components" shield = base / "nvidia_shield" for name in ("strings.json", "translations/en.json"): data = json.loads((shield / name).read_text(encoding="utf-8")) assert set(data["services"]["send_intent"]["fields"]) == { "intent", "variables", "entity_id", } for key in ( "unknown_intent", "missing_intent_field", "unknown_intent_variable", "invalid_intent_field", "invalid_intent_variables", "unresolved_intent_placeholder", "intents_need_adb", "intent_no_answer", "intent_failed", ): assert key in data["exceptions"], (name, key) class TestNothingRunsByItself: """The plan's non-negotiable: registration is inert until someone calls it.""" async def test_loading_the_options_screen_touches_no_device(self, modules, tmp_path): poller = _RecordingPoller() flow = _flow(modules, tmp_path, {CONF_REGISTERED_INTENTS: [NETFLIX_INTENT]}) flow.config_entry.runtime_data = SimpleNamespace( async_scan_intents=_FakeScan(_scan_result()), adb_poller=poller, ) for step in ("intents", "intents_edit", "intents_remove", "intents_scan"): result = await getattr(flow, f"async_step_{step}")(None) assert result["type"] in ("form", "menu") assert poller.commands == [], "rendering a screen never runs anything" async def test_a_scan_never_registers_anything(self, modules, tmp_path): flow = _flow(modules, tmp_path) flow.config_entry.runtime_data = SimpleNamespace( async_scan_intents=_FakeScan(_scan_result([_candidate()])) ) await flow.async_step_intents_scan( {INTENT_FIELD_SCAN_PACKAGE: "com.netflix.ninja"} ) assert flow._options_dict.get(CONF_REGISTERED_INTENTS) in (None, []) async def test_a_registered_intent_is_inert_until_pressed(self, modules): _entry, added = await self._setup_intents(modules) assert added, "the button exists" assert added[0]._manager is None or added[0]._manager.adb_poller is not None async def _setup_intents(self, modules): entry = SimpleNamespace( entry_id="e1", options={CONF_REGISTERED_INTENTS: [NETFLIX_INTENT]}, title="Shield", runtime_data=None, ) added: list = [] button = importlib.import_module("nsh.button") await button.async_setup_entry(None, entry, added.extend) return entry, added def test_the_text_helpers_do_not_run_in_a_thread(): """A guard against a future rewrite reaching for `eval`/`exec` (t17).""" source = ( Path(__file__).resolve().parent.parent / "custom_components" / "nvidia_shield" / "media_player.py" ).read_text(encoding="utf-8") assert "eval(" not in source assert "exec(" not in source assert ".format(" not in source.split("def intent_am_start_command")[1].split("\ndef ")[0] def test_asyncio_is_still_needed_by_the_module_under_test(): assert asyncio is not None # keeps the import honest if the tests shrink