diff --git a/.python-version b/.python-version index e4fba21..24ee5b1 100644 --- a/.python-version +++ b/.python-version @@ -1 +1 @@ -3.12 +3.13 diff --git a/README.md b/README.md index 6697467..6d75dd3 100644 --- a/README.md +++ b/README.md @@ -25,17 +25,14 @@ Setup ----- The first time you use the utility, you will need to setup a connection to the TV. -With the TV on, run `alga setup [hostname/IP]`. +With the TV on, run `alga tv add [hostname/IP]`. This will bring up a prompt on the TV asking if you want to accept the pairing. When accepted, Alga will be ready to use. -If no hostname or IP address is provided to `alga setup`, it will be default try to connect to "lgwebostv" which should work. +If no hostname or IP address is provided to `alga tv add`, it will be default try to connect to "lgwebostv" which should work. The hostname, a key and MAC address will be written to `~/.config/alga/config.json` for future use. -Note: It's currently only possible to pair Alga with one TV at a time. -Let me know if this is a deal breaker for you. - Usage ----- diff --git a/src/alga/__init__.py b/src/alga/__init__.py index 3e8d9f9..55c7d83 100644 --- a/src/alga/__init__.py +++ b/src/alga/__init__.py @@ -1 +1,6 @@ +from alga.types import State + + __version__ = "1.4.0" + +state = State() diff --git a/src/alga/__main__.py b/src/alga/__main__.py index d69addc..4aac263 100644 --- a/src/alga/__main__.py +++ b/src/alga/__main__.py @@ -1,4 +1,6 @@ -from typer import Typer +from typing import Annotated, Optional + +from typer import Option, Typer from alga import ( cli_adhoc, @@ -8,14 +10,23 @@ cli_media, cli_power, cli_remote, - cli_setup, cli_sound_output, + cli_tv, cli_version, cli_volume, + state, ) -app = Typer(no_args_is_help=True) +def global_options( + tv: Annotated[ + Optional[str], Option(help="Specify which TV the command should be sent to") + ] = None, +) -> None: + state.tv_id = tv + + +app = Typer(no_args_is_help=True, callback=global_options) app.add_typer(cli_adhoc.app) app.add_typer(cli_app.app, name="app") app.add_typer(cli_channel.app, name="channel") @@ -23,8 +34,8 @@ app.add_typer(cli_media.app, name="media") app.add_typer(cli_power.app, name="power") app.add_typer(cli_remote.app, name="remote") -app.add_typer(cli_setup.app) app.add_typer(cli_sound_output.app, name="sound-output") +app.add_typer(cli_tv.app, name="tv") app.add_typer(cli_version.app) app.add_typer(cli_volume.app, name="volume") diff --git a/src/alga/cli_power.py b/src/alga/cli_power.py index ffff49b..0418e09 100644 --- a/src/alga/cli_power.py +++ b/src/alga/cli_power.py @@ -1,7 +1,7 @@ from typer import Typer from wakeonlan import send_magic_packet -from alga import client, config +from alga import client, config, state app = Typer(no_args_is_help=True, help="Turn TV (or screen) on and off") @@ -19,7 +19,11 @@ def on() -> None: """Turn TV on via Wake-on-LAN""" cfg = config.get() - send_magic_packet(cfg["mac"]) + + tv_id = state.tv_id or cfg["default_tv"] + tv = cfg["tvs"][tv_id] + + send_magic_packet(tv["mac"]) @app.command() diff --git a/src/alga/cli_setup.py b/src/alga/cli_setup.py deleted file mode 100644 index de0341e..0000000 --- a/src/alga/cli_setup.py +++ /dev/null @@ -1,72 +0,0 @@ -import json -from ipaddress import ip_address -from socket import gaierror, getaddrinfo -from typing import Annotated, Optional - -from getmac import get_mac_address -from rich import print -from rich.console import Console -from typer import Argument, Exit, Typer - -from alga import client, config -from alga.payloads import get_hello_data - - -app = Typer() - - -def _ip_from_hostname(hostname: str) -> Optional[str]: - try: - results = getaddrinfo(host=hostname, port=None) - # TODO: Do we want to handle receiving multiple IP addresses? - return results[0][4][0] - except gaierror: - return None - - -@app.command() -def setup( - hostname: Annotated[str, Argument()] = "lgwebostv", -) -> None: # pragma: no cover - """Pair a new TV""" - - # Check if we have been passed an IP address - ip: Optional[str] - try: - ip = ip_address(hostname).compressed - except ValueError: - ip = _ip_from_hostname(hostname) - if not ip: - print( - f"[red]Could not find any host by the name '{hostname}'.[/red] Is the TV on and connected to the network?" - ) - raise Exit(code=1) - - with client.new( - hostname=hostname, perform_handshake=False, timeout=60 - ) as connection: - connection.send(json.dumps(get_hello_data())) - response = json.loads(connection.recv()) - assert response == { - "id": "register_0", - "payload": {"pairingType": "PROMPT", "returnValue": True}, - "type": "response", - }, "Unexpected response received" - - console = Console() - with console.status("Please approve the connection request on the TV now..."): - response = json.loads(connection.recv()) - - if "client-key" not in response["payload"]: - print("[red]Setup failed![/red]") - raise Exit(code=1) - - mac_address = get_mac_address(ip=ip) - - cfg = config.get() - cfg.update( - hostname=hostname, key=response["payload"]["client-key"], mac=mac_address - ) - config.write(cfg) - - print("TV configured, Alga is ready to use") diff --git a/src/alga/cli_tv.py b/src/alga/cli_tv.py new file mode 100644 index 0000000..e8ab084 --- /dev/null +++ b/src/alga/cli_tv.py @@ -0,0 +1,155 @@ +import json +from ipaddress import ip_address +from socket import gaierror, getaddrinfo +from typing import Annotated, Optional + +from getmac import get_mac_address +from rich import print +from rich.console import Console +from rich.table import Table +from typer import Argument, Exit, Typer + +from alga import client, config +from alga.payloads import get_hello_data + + +def _ip_from_hostname(hostname: str) -> Optional[str]: + try: + results = getaddrinfo(host=hostname, port=None) + # TODO: Do we want to handle receiving multiple IP addresses? + return results[0][4][0] + except gaierror: + return None + + +app = Typer(no_args_is_help=True, help="Set up TVs to manage via Alga") + + +@app.command() +def add( + name: Annotated[str, Argument()], hostname: Annotated[str, Argument()] = "lgwebostv" +) -> None: # pragma: no cover + """Pair a new TV""" + + # Check if we have been passed an IP address + ip: Optional[str] + try: + ip = ip_address(hostname).compressed + except ValueError: + ip = _ip_from_hostname(hostname) + if not ip: + print( + f"[red]Could not find any host by the name '{hostname}'.[/red] Is the TV on and connected to the network?" + ) + raise Exit(code=1) + + with client.connect(hostname=hostname, timeout=60) as connection: + connection.send(json.dumps(get_hello_data())) + response = json.loads(connection.recv()) + assert response == { + "id": "register_0", + "payload": {"pairingType": "PROMPT", "returnValue": True}, + "type": "response", + }, "Unexpected response received" + + console = Console() + with console.status("Please approve the connection request on the TV now..."): + response = json.loads(connection.recv()) + + if "client-key" not in response["payload"]: + print("[red]Setup failed![/red]") + raise Exit(code=1) + + mac_address = get_mac_address(ip=ip) + + cfg = config.get() + cfg["tvs"][name] = { + "hostname": hostname, + "key": response["payload"]["client-key"], + "mac": mac_address, + } + + if not cfg.get("default_tv"): + cfg["default_tv"] = name + + config.write(cfg) + + print("TV configured, Alga is ready to use") + + +@app.command() +def list() -> None: + """List current TVs""" + + cfg = config.get() + + table = Table() + table.add_column("Default") + table.add_column("Name") + table.add_column("Hostname/IP") + table.add_column("MAC address") + + for name, tv in cfg["tvs"].items(): + default = "*" if cfg["default_tv"] == name else "" + table.add_row(default, name, tv["hostname"], tv["mac"]) + + console = Console() + console.print(table) + + +@app.command() +def remove(name: Annotated[str, Argument()]) -> None: + """Remove a TV""" + + cfg = config.get() + + try: + cfg["tvs"].pop(name) + + if cfg["default_tv"] == name: + cfg["default_tv"] = "" + + config.write(cfg) + except KeyError: + print( + f"[red]A TV with the name '{name}' was not found in the configuration[/red]" + ) + raise Exit(code=1) + + +@app.command() +def rename( + old_name: Annotated[str, Argument()], new_name: Annotated[str, Argument()] +) -> None: + """Change the identifier for a TV""" + + cfg = config.get() + + try: + cfg["tvs"][new_name] = cfg["tvs"].pop(old_name) + + if cfg["default_tv"] == old_name: + cfg["default_tv"] = new_name + + config.write(cfg) + except KeyError: + print( + f"[red]A TV with the name '{old_name}' was not found in the configuration[/red]" + ) + raise Exit(code=1) + + +@app.command() +def set_default(name: Annotated[str, Argument()]) -> None: + """Set the default TV""" + + cfg = config.get() + + if name in cfg["tvs"]: + cfg["default_tv"] = name + config.write(cfg) + else: + print( + f"[red]A TV with the name '{name}' was not found in the configuration[/red]" + ) + raise Exit(code=1) diff --git a/src/alga/client.py b/src/alga/client.py index c749e6c..8986165 100644 --- a/src/alga/client.py +++ b/src/alga/client.py @@ -8,48 +8,47 @@ from typer import Exit from websocket import WebSocket -from alga import config +from alga import config, state from alga.payloads import get_hello_data @contextmanager -def new( - hostname: Optional[str] = None, perform_handshake: bool = True, timeout: int = 3 -) -> Iterator[WebSocket]: # pragma: no cover - cfg = config.get() - - if hostname is None: - if "hostname" not in cfg: - print("[red]No connection configured, run 'alga setup' first[/red]") - raise Exit(code=1) - - hostname = cfg["hostname"] - +def connect(hostname: str, timeout: int = 10) -> Iterator[WebSocket]: connection = WebSocket(sslopt={"cert_reqs": ssl.CERT_NONE}) connection.connect(f"wss://{hostname}:3001/", suppress_origin=True, timeout=timeout) # type: ignore[no-untyped-call] - if perform_handshake: - if "key" not in cfg: - print("[red]No connection configured, run 'alga setup' first[/red]") - raise Exit(code=1) - - connection.send(json.dumps(get_hello_data(cfg["key"]))) - response = json.loads(connection.recv()) - if "client-key" not in response["payload"]: - raise Exception( - f"Something went wrong with performing a handshake. Response: {response}" - ) - try: yield connection finally: connection.close() -def request( - uri: str, data: Optional[dict[str, Any]] = None -) -> dict[str, Any]: # pragma: no cover - with new() as connection: +def do_handshake(connection: WebSocket, key: str) -> None: + connection.send(json.dumps(get_hello_data(key))) + response = json.loads(connection.recv()) + if "client-key" not in response["payload"]: + raise Exception( + f"Something went wrong with performing a handshake. Response: {response}" + ) + + +def request(uri: str, data: Optional[dict[str, Any]] = None) -> dict[str, Any]: + cfg = config.get() + tv_id = state.tv_id or cfg.get("default_tv") + + if not tv_id: + print("[red]No connection configured, run 'alga tv add' first[/red]") + raise Exit(code=1) + + if tv_id not in cfg["tvs"]: + print(f"[red]'{tv_id}' was not found in the configuration[/red]") + raise Exit(code=1) + + tv = cfg["tvs"][tv_id] + + with connect(tv["hostname"]) as connection: + do_handshake(connection, tv["key"]) + request: dict[str, Any] = {"type": "request", "uri": uri} if data: diff --git a/src/alga/config.py b/src/alga/config.py index c468f49..5942739 100644 --- a/src/alga/config.py +++ b/src/alga/config.py @@ -4,11 +4,35 @@ _config_file = App("alga").config.open("config.json") +_latest_version = 2 def get() -> dict[str, Any]: config = _config_file.contents - config.setdefault("version", 1) + + if config.setdefault("version", _latest_version) < _latest_version: + config = migrate(config) + write(config) + + config.setdefault("tvs", {}) + + return config + + +def migrate(config: dict[str, Any]) -> dict[str, Any]: + if config["version"] == 1: + config = { + "version": 2, + "default_tv": "default", + "tvs": { + "default": { + "hostname": config["hostname"], + "key": config["key"], + "mac": config["mac"], + } + }, + } + return config diff --git a/src/alga/types.py b/src/alga/types.py index 73a8fd9..eff3f03 100644 --- a/src/alga/types.py +++ b/src/alga/types.py @@ -1,5 +1,5 @@ from dataclasses import dataclass -from typing import Any +from typing import Any, Optional @dataclass @@ -50,3 +50,8 @@ class SoundOutputDevice: def __str__(self) -> str: return self.name + + +@dataclass +class State: + tv_id: Optional[str] = None diff --git a/tests/conftest.py b/tests/conftest.py index 2caa1f9..b1dd389 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,16 +1,28 @@ from collections.abc import Iterator -from unittest import mock +from unittest.mock import MagicMock, patch import pytest @pytest.fixture -def mock_request() -> Iterator[mock.MagicMock]: - with mock.patch("alga.client.request") as mocked: +def mock_request() -> Iterator[MagicMock]: + with patch("alga.client.request") as mocked: yield mocked @pytest.fixture -def mock_input() -> Iterator[mock.MagicMock]: - with mock.patch("alga.cli_remote._input_connection") as mocked: +def mock_input() -> Iterator[MagicMock]: + with patch("alga.cli_remote._input_connection") as mocked: yield mocked.return_value.__enter__.return_value.send + + +@pytest.fixture +def mock_config() -> Iterator[MagicMock]: + with patch("alga.config.get") as mocked: + yield mocked + + +@pytest.fixture +def mock_config_file() -> Iterator[MagicMock]: + with patch("alga.config._config_file") as mocked: + yield mocked diff --git a/tests/test_cli_power.py b/tests/test_cli_power.py index 74d2017..6b15f1a 100644 --- a/tests/test_cli_power.py +++ b/tests/test_cli_power.py @@ -1,4 +1,4 @@ -from unittest import mock +from unittest.mock import MagicMock, patch from faker import Faker from typer.testing import CliRunner @@ -9,7 +9,7 @@ runner = CliRunner() -def test_off(mock_request: mock.MagicMock) -> None: +def test_off(mock_request: MagicMock) -> None: result = runner.invoke(app, ["power", "off"]) mock_request.assert_called_once_with("ssap://system/turnOff") @@ -17,7 +17,7 @@ def test_off(mock_request: mock.MagicMock) -> None: assert result.stdout == "" -def test_screen_off(mock_request: mock.MagicMock) -> None: +def test_screen_off(mock_request: MagicMock) -> None: result = runner.invoke(app, ["power", "screen-off"]) mock_request.assert_called_once_with( @@ -27,7 +27,7 @@ def test_screen_off(mock_request: mock.MagicMock) -> None: assert result.stdout == "" -def test_screen_on(mock_request: mock.MagicMock) -> None: +def test_screen_on(mock_request: MagicMock) -> None: result = runner.invoke(app, ["power", "screen-on"]) mock_request.assert_called_once_with( @@ -37,17 +37,15 @@ def test_screen_on(mock_request: mock.MagicMock) -> None: assert result.stdout == "" -@mock.patch("alga.config.get") -@mock.patch("alga.cli_power.send_magic_packet") -def test_on( - mock_send_magic_packet: mock.MagicMock, - mock_config_get: mock.MagicMock, - faker: Faker, -) -> None: +def test_on(mock_config: MagicMock, faker: Faker) -> None: mac_address = faker.pystr() - mock_config_get.return_value = {"mac": mac_address} + mock_config.return_value = { + "default_tv": "default", + "tvs": {"default": {"mac": mac_address}}, + } - result = runner.invoke(app, ["power", "on"]) + with patch("alga.cli_power.send_magic_packet") as mock_send_magic_packet: + result = runner.invoke(app, ["power", "on"]) mock_send_magic_packet.assert_called_once_with(mac_address) assert result.exit_code == 0 diff --git a/tests/test_cli_setup.py b/tests/test_cli_setup.py deleted file mode 100644 index bfdac04..0000000 --- a/tests/test_cli_setup.py +++ /dev/null @@ -1,59 +0,0 @@ -from socket import AF_INET, SOCK_DGRAM, SOCK_STREAM, gaierror -from unittest import mock - -from faker import Faker -from typer.testing import CliRunner - -from alga.__main__ import app -from alga.cli_setup import _ip_from_hostname - - -runner = CliRunner() - - -@mock.patch("alga.cli_setup.getaddrinfo") -def test_ip_from_hostname(mock_getaddrinfo: mock.MagicMock, faker: Faker) -> None: - hostname = faker.hostname() - ip_address = faker.ipv4() - - mock_getaddrinfo.return_value = [ - (AF_INET, SOCK_DGRAM, 17, "", (ip_address, 0)), - (AF_INET, SOCK_STREAM, 6, "", (ip_address, 0)), - ] - - result = _ip_from_hostname(hostname) - - mock_getaddrinfo.assert_called_once_with(host=hostname, port=None) - assert result == ip_address - - -@mock.patch("alga.cli_setup.getaddrinfo") -def test_ip_from_hostname_not_found( - mock_getaddrinfo: mock.MagicMock, faker: Faker -) -> None: - hostname = faker.hostname() - mock_getaddrinfo.side_effect = gaierror( - "[Errno 8] nodename nor servname provided, or not known" - ) - - result = _ip_from_hostname(hostname) - - mock_getaddrinfo.assert_called_once_with(host=hostname, port=None) - assert result is None - - -@mock.patch("alga.cli_setup._ip_from_hostname") -def test_setup_ip_not_found( - mock_ip_from_hostname: mock.MagicMock, faker: Faker -) -> None: - hostname = faker.hostname() - mock_ip_from_hostname.return_value = None - - result = runner.invoke(app, ["setup", hostname]) - - mock_ip_from_hostname.assert_called_once_with(hostname) - assert result.exit_code == 1 - assert ( - result.stdout.replace("\n", "") - == f"Could not find any host by the name '{hostname}'. Is the TV on and connected to the network?" - ) diff --git a/tests/test_cli_tv.py b/tests/test_cli_tv.py new file mode 100644 index 0000000..324e4ef --- /dev/null +++ b/tests/test_cli_tv.py @@ -0,0 +1,126 @@ +from socket import AF_INET, SOCK_DGRAM, SOCK_STREAM, gaierror +from unittest.mock import MagicMock, patch + +from faker import Faker +from typer.testing import CliRunner + +from alga.__main__ import app +from alga.cli_tv import _ip_from_hostname + + +runner = CliRunner() + + +@patch("alga.cli_tv.getaddrinfo") +def test_ip_from_hostname(mock_getaddrinfo: MagicMock, faker: Faker) -> None: + hostname = faker.hostname() + ip_address = faker.ipv4() + + mock_getaddrinfo.return_value = [ + (AF_INET, SOCK_DGRAM, 17, "", (ip_address, 0)), + (AF_INET, SOCK_STREAM, 6, "", (ip_address, 0)), + ] + + result = _ip_from_hostname(hostname) + + mock_getaddrinfo.assert_called_once_with(host=hostname, port=None) + assert result == ip_address + + +@patch("alga.cli_tv.getaddrinfo") +def test_ip_from_hostname_not_found(mock_getaddrinfo: MagicMock, faker: Faker) -> None: + hostname = faker.hostname() + mock_getaddrinfo.side_effect = gaierror( + "[Errno 8] nodename nor servname provided, or not known" + ) + + result = _ip_from_hostname(hostname) + + mock_getaddrinfo.assert_called_once_with(host=hostname, port=None) + assert result is None + + +@patch("alga.cli_tv._ip_from_hostname") +def test_add_ip_not_found(mock_ip_from_hostname: MagicMock, faker: Faker) -> None: + hostname = faker.hostname() + mock_ip_from_hostname.return_value = None + + result = runner.invoke(app, ["tv", "add", "name", hostname]) + + mock_ip_from_hostname.assert_called_once_with(hostname) + assert result.exit_code == 1 + assert ( + result.stdout.replace("\n", "") + == f"Could not find any host by the name '{hostname}'. Is the TV on and connected to the network?" + ) + + +def test_list(faker: Faker, mock_config: MagicMock) -> None: + mock_config.return_value = { + "default_tv": faker.pystr(), + "tvs": {faker.pystr(): {"hostname": faker.pystr(), "mac": faker.pystr()}}, + } + + result = runner.invoke(app, ["tv", "list"]) + + assert result.exit_code == 0 + + +def test_remove(faker: Faker, mock_config: MagicMock) -> None: + name = faker.pystr() + mock_config.return_value = {"default_tv": name, "tvs": {name: {}}} + + with patch("alga.cli_tv.config.write") as mock_write: + result = runner.invoke(app, ["tv", "remove", name]) + + assert result.exit_code == 0 + mock_write.assert_called_once_with({"default_tv": "", "tvs": {}}) + + +def test_remove_not_found(faker: Faker, mock_config: MagicMock) -> None: + mock_config.return_value = {"default_tv": "", "tvs": {}} + name = faker.pystr() + + result = runner.invoke(app, ["tv", "remove", name]) + + assert result.exit_code == 1 + + +def test_rename(faker: Faker, mock_config: MagicMock) -> None: + old_name, new_name = faker.pystr(), faker.pystr() + mock_config.return_value = {"default_tv": old_name, "tvs": {old_name: {}}} + + with patch("alga.cli_tv.config.write") as mock_write: + result = runner.invoke(app, ["tv", "rename", old_name, new_name]) + + assert result.exit_code == 0 + mock_write.assert_called_once_with({"default_tv": new_name, "tvs": {new_name: {}}}) + + +def test_rename_not_found(faker: Faker, mock_config: MagicMock) -> None: + mock_config.return_value = {"default_tv": "", "tvs": {}} + old_name, new_name = faker.pystr(), faker.pystr() + + result = runner.invoke(app, ["tv", "rename", old_name, new_name]) + + assert result.exit_code == 1 + + +def test_set_default(faker: Faker, mock_config: MagicMock) -> None: + name = faker.pystr() + mock_config.return_value = {"default_tv": "", "tvs": {name: {}}} + + with patch("alga.cli_tv.config.write") as mock_write: + result = runner.invoke(app, ["tv", "set-default", name]) + + assert result.exit_code == 0 + mock_write.assert_called_once_with({"default_tv": name, "tvs": {name: {}}}) + + +def test_set_default_not_found(faker: Faker, mock_config: MagicMock) -> None: + name = faker.pystr() + mock_config.return_value = {"default_tv": "", "tvs": {}} + + result = runner.invoke(app, ["tv", "set-default", name]) + + assert result.exit_code == 1 diff --git a/tests/test_client.py b/tests/test_client.py new file mode 100644 index 0000000..c6d075a --- /dev/null +++ b/tests/test_client.py @@ -0,0 +1,130 @@ +import json +import ssl +from unittest.mock import ANY, MagicMock, call, patch + +import pytest +from faker import Faker +from typer import Exit + +from alga import client +from alga.payloads import get_hello_data + + +def test_connect(faker: Faker) -> None: + hostname = faker.pystr() + timeout = faker.pyint() + + with patch("alga.client.WebSocket") as mock_websocket: + mock_websocket + + with client.connect(hostname, timeout): + pass + + mock_websocket.assert_has_calls( + [ + call(sslopt={"cert_reqs": ssl.CERT_NONE}), + call().connect( + f"wss://{hostname}:3001/", suppress_origin=True, timeout=timeout + ), + call().close(), + ] + ) + + +def test_do_handshake(faker: Faker) -> None: + key = faker.pystr() + mock_connection = MagicMock() + + mock_connection.recv.return_value = json.dumps( + {"payload": {"client-key": faker.pystr()}} + ) + + client.do_handshake(mock_connection, key) + + mock_connection.send.assert_called_once_with(json.dumps(get_hello_data(key))) + + +def test_do_handshake_error(faker: Faker) -> None: + key = faker.pystr() + mock_connection = MagicMock() + + mock_connection.recv.return_value = json.dumps({"payload": {}}) + + with pytest.raises( + Exception, match="Something went wrong with performing a handshake" + ): + client.do_handshake(mock_connection, key) + + +def test_request_no_config(faker: Faker, mock_config: MagicMock) -> None: + mock_config.return_value = {} + + with pytest.raises(Exit) as exc_info: + client.request(faker.pystr()) + + assert exc_info.value.exit_code == 1 + + +def test_request_tv_id_not_in_config(faker: Faker, mock_config: MagicMock) -> None: + mock_config.return_value = {"default_tv": faker.pystr(), "tvs": {}} + + with pytest.raises(Exit) as exc_info: + client.request(faker.pystr()) + + assert exc_info.value.exit_code == 1 + + +def test_request_no_data(faker: Faker, mock_config: MagicMock) -> None: + name, hostname, key = faker.pystr(), faker.pystr(), faker.pystr() + mock_config.return_value = { + "default_tv": name, + "tvs": {name: {"hostname": hostname, "key": key}}, + } + + uri = faker.pystr() + payload = {"returnValue": True} | faker.pydict() + + with patch("alga.client.connect") as mock_connect: + mock_connect().__enter__().recv.side_effect = [ + json.dumps({"payload": {"client-key": faker.pystr()}}), + json.dumps({"payload": payload}), + ] + + response = client.request(uri) + + assert response == payload + + mock_connect().__enter__().send.assert_has_calls( + [ + call(ANY), # Handshake + call(json.dumps({"type": "request", "uri": uri})), + ] + ) + + +def test_request_with_data(faker: Faker, mock_config: MagicMock) -> None: + name, hostname, key = faker.pystr(), faker.pystr(), faker.pystr() + mock_config.return_value = { + "default_tv": name, + "tvs": {name: {"hostname": hostname, "key": key}}, + } + + uri, data = faker.pystr(), faker.pydict(allowed_types=[str, float, int]) + payload = {"returnValue": True} | faker.pydict() + + with patch("alga.client.connect") as mock_connect: + mock_connect().__enter__().recv.side_effect = [ + json.dumps({"payload": {"client-key": faker.pystr()}}), + json.dumps({"payload": payload}), + ] + + response = client.request(uri, data) + + assert response == payload + + mock_connect().__enter__().send.assert_has_calls( + [ + call(ANY), # Handshake + call(json.dumps({"type": "request", "uri": uri, "payload": data})), + ] + ) diff --git a/tests/test_config.py b/tests/test_config.py index 1331db5..59cc7b9 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -1,34 +1,31 @@ -from unittest import mock +from unittest.mock import MagicMock, patch from faker import Faker from alga import config -@mock.patch("alga.config._config_file") -def test_get_sets_version(mock_config_file: mock.MagicMock) -> None: +def test_get_sets_defaults(mock_config_file: MagicMock) -> None: mock_config_file.contents = {} cfg = config.get() - assert cfg == {"version": 1} + assert cfg == {"version": 2, "tvs": {}} -@mock.patch("alga.config._config_file") def test_get_does_not_override_version( - mock_config_file: mock.MagicMock, faker: Faker + mock_config_file: MagicMock, faker: Faker ) -> None: version = faker.pyint() mock_config_file.contents = {"version": version} cfg = config.get() - assert cfg == {"version": version} + assert cfg == {"version": version, "tvs": {}} -@mock.patch("alga.config._config_file") -def test_get_returns_data(mock_config_file: mock.MagicMock, faker: Faker) -> None: - data = {"version": faker.pyint()} | faker.pydict() +def test_get_returns_data(mock_config_file: MagicMock, faker: Faker) -> None: + data = {"version": faker.pyint(), "tvs": {}} | faker.pydict() mock_config_file.contents = data cfg = config.get() @@ -36,11 +33,30 @@ def test_get_returns_data(mock_config_file: mock.MagicMock, faker: Faker) -> Non assert cfg == data -@mock.patch("alga.config._config_file") -def test_write(mock_config_file: mock.MagicMock, faker: Faker) -> None: +def test_get_calls_migrate(mock_config_file: MagicMock) -> None: + mock_config_file.contents = {"version": 1} + + with patch("alga.config.migrate") as mock_migrate: + config.get() + + mock_migrate.assert_called_once() + + +def test_write(mock_config_file: MagicMock, faker: Faker) -> None: data = faker.pydict() config.write(data) mock_config_file.write.assert_called_once() assert mock_config_file.contents == data + + +def test_migrate_v1_to_v2(faker: Faker) -> None: + hostname, key, mac = faker.pystr(), faker.pystr(), faker.pystr() + v1_config = {"version": 1, "hostname": hostname, "key": key, "mac": mac} + + assert config.migrate(v1_config) == { + "version": 2, + "default_tv": "default", + "tvs": {"default": {"hostname": hostname, "key": key, "mac": mac}}, + } diff --git a/usage.md b/usage.md index d1937a4..05078e5 100644 --- a/usage.md +++ b/usage.md @@ -8,6 +8,7 @@ $ alga [OPTIONS] COMMAND [ARGS]... **Options**: +* `--tv TEXT`: Specify which TV the command should be sent to * `--install-completion`: Install completion for the current shell. * `--show-completion`: Show completion for the current shell, to copy it or customize the installation. * `--help`: Show this message and exit. @@ -21,8 +22,8 @@ $ alga [OPTIONS] COMMAND [ARGS]... * `media`: Control the playing media * `power`: Turn TV (or screen) on and off * `remote`: Remote control button presses -* `setup`: Pair a new TV * `sound-output`: Audio output device +* `tv`: Set up TVs to manage via Alga * `version`: Print Alga version * `volume`: Audio volume @@ -547,24 +548,6 @@ $ alga remote send [OPTIONS] BUTTON * `--help`: Show this message and exit. -## `alga setup` - -Pair a new TV - -**Usage**: - -```console -$ alga setup [OPTIONS] [HOSTNAME] -``` - -**Arguments**: - -* `[HOSTNAME]`: [default: lgwebostv] - -**Options**: - -* `--help`: Show this message and exit. - ## `alga sound-output` Audio output device @@ -631,6 +614,116 @@ $ alga sound-output set [OPTIONS] VALUE * `--help`: Show this message and exit. +## `alga tv` + +Set up TVs to manage via Alga + +**Usage**: + +```console +$ alga tv [OPTIONS] COMMAND [ARGS]... +``` + +**Options**: + +* `--help`: Show this message and exit. + +**Commands**: + +* `add`: Pair a new TV +* `list`: List current TVs +* `remove`: Remove a TV +* `rename`: Change the identifier for a TV +* `set-default`: Set the default TV + +### `alga tv add` + +Pair a new TV + +**Usage**: + +```console +$ alga tv add [OPTIONS] NAME [HOSTNAME] +``` + +**Arguments**: + +* `NAME`: [required] +* `[HOSTNAME]`: [default: lgwebostv] + +**Options**: + +* `--help`: Show this message and exit. + +### `alga tv list` + +List current TVs + +**Usage**: + +```console +$ alga tv list [OPTIONS] +``` + +**Options**: + +* `--help`: Show this message and exit. + +### `alga tv remove` + +Remove a TV + +**Usage**: + +```console +$ alga tv remove [OPTIONS] NAME +``` + +**Arguments**: + +* `NAME`: [required] + +**Options**: + +* `--help`: Show this message and exit. + +### `alga tv rename` + +Change the identifier for a TV + +**Usage**: + +```console +$ alga tv rename [OPTIONS] OLD_NAME NEW_NAME +``` + +**Arguments**: + +* `OLD_NAME`: [required] +* `NEW_NAME`: [required] + +**Options**: + +* `--help`: Show this message and exit. + +### `alga tv set-default` + +Set the default TV + +**Usage**: + +```console +$ alga tv set-default [OPTIONS] NAME +``` + +**Arguments**: + +* `NAME`: [required] + +**Options**: + +* `--help`: Show this message and exit. + ## `alga version` Print Alga version