diff --git a/README.md b/README.md index 0be6d35..0461de6 100644 --- a/README.md +++ b/README.md @@ -4,7 +4,7 @@ Show what's playing on your Plex Media Server — movies, TV episodes and music ![Plex Now Playing Display](./docs/board-display.png) -**→ [Setup Guide](./docs/SETUP.md)** - Finding your Plex token and configuration +**→ [Setup Guide](./docs/SETUP.md)** - Signing in with Plex (or pasting a token) and configuration ## Overview @@ -95,8 +95,9 @@ Center all three lines: | Setting | Type | Required | Default | Description | |---------|------|----------|---------|-------------| -| `server_url` | string | Yes | - | Plex server address, e.g. `http://192.168.1.100:32400` | -| `token` | string | Yes | - | Your X-Plex-Token | +| `server_url` | string | No | - | Plex server address, e.g. `http://192.168.1.100:32400`. Empty: found through your Plex account | +| `token` | string | No | - | An X-Plex-Token to use instead of **Sign in with Plex**. When set, it wins | +| `server_name` | string | No | - | Which server to use when your account has several and `server_url` is empty | | `plex_user` | string | No | - | Only show this Plex user's streams | | `show_accents` | boolean | No | `true` | Frame the year or season/episode line with yellow tiles | | `refresh_seconds` | integer | No | `30` | How often to ask Plex what's playing (10-600) | @@ -104,6 +105,12 @@ Center all three lines: `PLEX_URL` and `PLEX_TOKEN` environment variables can be used instead of the UI settings. +### Sign in with Plex + +Requires FiestaBoard 9.11.0 or later. Click **Sign in with Plex** in the plugin's settings and approve FiestaBoard on Plex's page; nothing to copy. A pasted token (or `PLEX_TOKEN`) still works and takes priority over the sign-in. If Plex stops accepting the sign-in, the settings ask you to sign in again. + +With no `server_url`, the plugin asks plex.tv for the servers on your account (`/api/v2/resources`), prefers one you own, and tries its home-network address first, then its remote address, then Plex's relay. This works with a pasted token too. + ## Features - Movies, TV episodes, and music @@ -113,7 +120,8 @@ Center all three lines: - Filter to a single Plex user - Every active stream available as `sessions`, plus a stream count - Playback state color-coded: green playing, yellow paused, orange buffering -- Talks to your server directly on your network; no Plex account API or cloud relay +- Sign in with Plex, or paste a token +- Finds your server through your Plex account, or talks to the address you give it directly ## Author diff --git a/__init__.py b/__init__.py index 076eb76..e23785c 100644 --- a/__init__.py +++ b/__init__.py @@ -11,6 +11,7 @@ import math import os import textwrap +import uuid from typing import Any, Dict, List, Optional, Tuple import requests @@ -23,6 +24,13 @@ DEFAULT_ROWS = 6 ACCENT = "{65}{65}" # two yellow tiles, Plex's brand colour REQUEST_TIMEOUT = 10 +PROBE_TIMEOUT = 3 +PLEX_RESOURCES_URL = "https://plex.tv/api/v2/resources" + +NEEDS_TOKEN = "Sign in with Plex, or paste a Plex token, in the plugin settings" +SIGN_IN_REJECTED = "Plex stopped accepting the sign-in. Sign in with Plex again." +TOKEN_REJECTED = "Plex rejected the token (401 Unauthorized)" +SET_SERVER_URL = "set the Plex Server URL in the plugin settings" STATE_LABELS = { "playing": "Playing", @@ -199,9 +207,42 @@ def pick_primary(sessions: List[Dict[str, Any]]) -> Dict[str, Any]: return min(sessions, key=lambda s: STATE_PRIORITY.index(s["state"])) +def candidate_uris(resource: Dict[str, Any]) -> List[str]: + """A server's connection addresses, best first: on the home network, then remote, then Plex's relay.""" + connections = [c for c in resource.get("connections") or [] if isinstance(c, dict) and c.get("uri")] + connections.sort(key=lambda c: (bool(c.get("relay")), not c.get("local"))) + return [c["uri"].rstrip("/") for c in connections] + + +def pick_servers(resources: Any, server_name: str) -> List[Dict[str, Any]]: + """The Plex Media Servers in a plex.tv resources answer, owned ones first. + + Raises LookupError when there is none (or none with *server_name*). + """ + if not isinstance(resources, list): + raise ValueError("plex.tv gave an unexpected answer") + servers = [ + r for r in resources if isinstance(r, dict) and "server" in str(r.get("provides") or "").split(",") + ] + if server_name: + servers = [s for s in servers if str(s.get("name") or "").strip().lower() == server_name.lower()] + if not servers: + raise LookupError(f"No Plex server named {server_name} on this Plex account") + if not servers: + raise LookupError(f"No Plex server found on this Plex account; {SET_SERVER_URL}") + servers.sort(key=lambda s: (not s.get("owned"), not s.get("presence", True))) + return servers + + class PlexPlugin(PluginBase): """Plex Now Playing plugin.""" + def __init__(self, *args: Any, **kwargs: Any) -> None: + super().__init__(*args, **kwargs) + self._client_identifier = str(uuid.uuid4()) + # ((token, server_name), server_url, server_token) from the last plex.tv lookup + self._discovered: Optional[Tuple[Tuple[str, str], str, str]] = None + @property def plugin_id(self) -> str: return "plex" @@ -215,49 +256,134 @@ def _token(config: Dict[str, Any]) -> str: return (config.get("token") or os.getenv("PLEX_TOKEN", "")).strip() def validate_config(self, config: Dict[str, Any]) -> List[str]: + # Both are optional: Sign in with Plex supplies a token, and plex.tv finds the server. errors = [] server_url = self._server_url(config) - if not server_url: - errors.append("Plex server URL is required") - elif not server_url.startswith(("http://", "https://")): + if server_url and not server_url.startswith(("http://", "https://")): errors.append("Plex server URL must start with http:// or https://") - if not self._token(config): - errors.append("Plex token is required") - errors.extend(self._validate_refresh_seconds(config)) return errors + def _signed_in_token(self) -> str: + """The token from Sign in with Plex, or "" (not signed in, or a core without sign-in).""" + getter = getattr(self, "get_oauth_token", None) + if not callable(getter): + return "" + try: + return (getter() or "").strip() + except Exception as e: # never let the sign-in store break a fetch + logger.warning("Could not read the Plex sign-in: %s", e) + return "" + + def _report_rejected(self) -> None: + report = getattr(self, "report_oauth_rejected", None) + if callable(report): + report() + + def _headers(self, token: str) -> Dict[str, str]: + """Headers for the Plex server: exactly what 1.0.x sent.""" + return { + "X-Plex-Token": token, + "X-Plex-Product": "FiestaBoard", + "Accept": "application/json", + } + + def _plex_tv_headers(self, token: str) -> Dict[str, str]: + # plex.tv's v2 API needs a client identifier; the server does not, and would list each new one as a device. + return {**self._headers(token), "X-Plex-Client-Identifier": self._client_identifier} + + def _reachable(self, uri: str, token: str) -> bool: + try: + return requests.get(f"{uri}/identity", headers=self._headers(token), timeout=PROBE_TIMEOUT).status_code == 200 + except requests.exceptions.RequestException: + return False + + def _discover_server(self, token: str) -> Tuple[str, str]: + """(server_url, server_token) for the account's server, found through plex.tv and cached.""" + server_name = (self.config.get("server_name") or "").strip() + key = (token, server_name) + if self._discovered and self._discovered[0] == key: + return self._discovered[1], self._discovered[2] + + response = requests.get( + PLEX_RESOURCES_URL, + params={"includeHttps": 1, "includeRelay": 1}, + headers=self._plex_tv_headers(token), + timeout=REQUEST_TIMEOUT, + ) + if response.status_code == 401: + raise PermissionError(TOKEN_REJECTED) + response.raise_for_status() + for resource in pick_servers(response.json(), server_name): + server_token = resource.get("accessToken") or token + for uri in candidate_uris(resource): + if self._reachable(uri, server_token): + self._discovered = (key, uri, server_token) + return uri, server_token + raise LookupError(f"Found your Plex server but the board cannot reach it; {SET_SERVER_URL}") + def _fetch_sessions(self, server_url: str, token: str) -> List[Dict[str, Any]]: response = requests.get( f"{server_url}/status/sessions", - headers={ - "X-Plex-Token": token, - "X-Plex-Product": "FiestaBoard", - "Accept": "application/json", - }, + headers=self._headers(token), timeout=REQUEST_TIMEOUT, ) if response.status_code == 401: - raise PermissionError("Plex rejected the token (401 Unauthorized)") + raise PermissionError(TOKEN_REJECTED) response.raise_for_status() container = response.json().get("MediaContainer") or {} return container.get("Metadata") or [] + def _rejected(self, signed_in: bool) -> PluginResult: + if signed_in: + self._report_rejected() + return PluginResult(available=False, error=SIGN_IN_REJECTED) + return PluginResult(available=False, error=TOKEN_REJECTED) + + def _locate_server(self, token: str, signed_in: bool) -> Tuple[Optional[Tuple[str, str]], Optional[PluginResult]]: + """The server to ask (from settings, or found through plex.tv), or the result explaining why not.""" + server_url = self._server_url(self.config) + if server_url: + return (server_url, token), None + try: + return self._discover_server(token), None + except PermissionError: + return None, self._rejected(signed_in) + except LookupError as e: + return None, PluginResult(available=False, error=str(e)) + except (requests.exceptions.RequestException, ValueError) as e: + logger.warning("Could not look up the Plex server at plex.tv: %s", e) + return None, PluginResult(available=False, error=f"Could not find your Plex server through plex.tv: {e}") + def fetch_data(self) -> PluginResult: """Fetch the active Plex sessions and build the template variables.""" - server_url = self._server_url(self.config) + # A pasted token (or PLEX_TOKEN) wins over Sign in with Plex. token = self._token(self.config) - if not server_url or not token: - return PluginResult(available=False, error="Plex server URL or token not configured") + signed_in = False + if not token: + token = self._signed_in_token() + signed_in = bool(token) + if not token: + return PluginResult(available=False, error=NEEDS_TOKEN) + + located, failure = self._locate_server(token, signed_in) + if failure: + return failure + server_url, server_token = located try: - items = self._fetch_sessions(server_url, token) + items = self._fetch_sessions(server_url, server_token) except requests.exceptions.RequestException as e: + self._discovered = None logger.warning("Could not reach Plex at %s: %s", server_url, e) return PluginResult(available=False, error=f"Could not reach Plex server: {e}") - except (PermissionError, ValueError) as e: + except PermissionError: + self._discovered = None + # Only the account token itself proves the sign-in is dead; a server's shared token can be revoked alone. + return self._rejected(signed_in and server_token == token) + except ValueError as e: logger.warning("Plex request failed: %s", e) return PluginResult(available=False, error=str(e)) diff --git a/docs/SETUP.md b/docs/SETUP.md index 3cd926c..9e537b7 100644 --- a/docs/SETUP.md +++ b/docs/SETUP.md @@ -8,23 +8,17 @@ Connect FiestaBoard to your Plex Media Server so your board shows what's playing **Prerequisites:** -- A Plex Media Server that FiestaBoard can reach on your network -- The server's address (usually `http://:32400`) -- A Plex token (X-Plex-Token) for an account with access to the server +- A Plex Media Server that FiestaBoard can reach +- A Plex account with access to the server +- FiestaBoard 9.11.0 or later for **Sign in with Plex** (on older versions, paste a token instead) ## Quick Setup -1. **Find your Plex token** - 1. Open Plex Web (app.plex.tv) and sign in with an account on your server. - 2. Open any movie or episode in your library. - 3. Click the **⋯** menu and choose **Get Info**, then **View XML**. - 4. The XML opens in a new tab. Copy the value after `X-Plex-Token=` at the end of its address. - - Treat the token like a password: it grants access to your Plex account. +1. **Enable** — In FiestaBoard, go to **Integrations**, find **Plex Now Playing**, and turn it on. -2. **Enable** — In FiestaBoard, go to **Integrations**, find **Plex Now Playing**, and turn it on. +2. **Sign in** — In the plugin's settings, click **Sign in with Plex**. Plex's sign-in page opens in a new tab; sign in and approve **FiestaBoard**. The settings show you are connected once Plex confirms, usually within a few seconds. -3. **Configure** — Enter your **Plex Server URL** (for example `http://192.168.1.100:32400`) and paste your **Plex Token**. Optionally set a **Plex User** to only show that person's streams. +3. **Server (usually nothing to do)** — Leave **Plex Server URL** empty and the plugin finds your server through your Plex account, trying its home-network address first. If your account can reach several servers, enter the one you want in **Plex Server Name**. If the board cannot reach the server that way, enter its address in **Plex Server URL** (for example `http://192.168.1.100:32400`). Optionally set a **Plex User** to only show that person's streams. 4. **Template** — Create a page from the plugin's demo, or add the display lines to your own page: @@ -38,6 +32,20 @@ Connect FiestaBoard to your Plex Media Server so your board shows what's playing 5. **View** — Start playing something in Plex. The board updates on the next refresh (30 seconds by default). +### Instead of signing in: paste a token + +This is how the plugin worked before 1.1.0, and it still does. A pasted token is used instead of the sign-in whenever it is set. + +1. **Find your Plex token** + 1. Open Plex Web (app.plex.tv) and sign in with an account on your server. + 2. Open any movie or episode in your library. + 3. Click the **⋯** menu and choose **Get Info**, then **View XML**. + 4. The XML opens in a new tab. Copy the value after `X-Plex-Token=` at the end of its address. + + Treat the token like a password: it grants access to your Plex account. + +2. Paste it into **Plex Token**. **Plex Server URL** can stay empty, as above, or hold your server's address. + ## Template Variables | Variable | Description | @@ -65,8 +73,9 @@ Connect FiestaBoard to your Plex Media Server so your board shows what's playing | Setting | Required | Default | Description | |---------|----------|---------|-------------| -| Plex Server URL | Yes | - | Your server's address, including the port | -| Plex Token | Yes | - | Your X-Plex-Token | +| Plex Server URL | No | - | Your server's address, including the port. Empty: found through your Plex account | +| Plex Token | No | - | An X-Plex-Token, used instead of Sign in with Plex when set | +| Plex Server Name | No | - | Which server to use when the URL is empty and your account has several | | Plex User | No | - | Only show streams from this Plex user (case-insensitive) | | Plex yellow accents | No | On | Frame the year or season/episode line with two yellow tiles on each side | | Refresh Interval | No | 30 seconds | How often to check Plex (10-600 seconds) | @@ -86,6 +95,23 @@ Connect FiestaBoard to your Plex Media Server so your board shows what's playing - FiestaBoard must be able to reach the server. If FiestaBoard runs in Docker, use the server's LAN IP, not `localhost`. - Using `https://` with a plain IP address fails the certificate check. Use `http://` on your local network. If your server's **Secure connections** setting is **Required**, change it to **Preferred**. +### "Sign in with Plex, or paste a Plex token, in the plugin settings" + +- Click **Sign in with Plex** in the settings, or paste a token. + +### "Plex stopped accepting the sign-in. Sign in with Plex again." + +- Plex refused the signed-in token, for example after **Sign out of all devices** or removing FiestaBoard from your Plex devices. Click **Sign in with Plex** again. + +### "No Plex server found on this Plex account" or "the board cannot reach it" + +- Make sure the server is signed in to the same Plex account (or shared with it) and online. +- If the board still cannot reach it, enter the server's address in **Plex Server URL**. + +### "Could not find your Plex server through plex.tv" + +- The board could not reach plex.tv. Check its internet connection, or enter **Plex Server URL** to talk to the server directly. + ### "Plex rejected the token (401 Unauthorized)" - The token is wrong or has expired. Signing out of all devices in Plex resets tokens, so find it again and paste the new one. diff --git a/manifest.json b/manifest.json index f7c0487..3fbe73b 100644 --- a/manifest.json +++ b/manifest.json @@ -1,7 +1,7 @@ { "id": "plex", "name": "Plex Now Playing", - "version": "1.0.1", + "version": "1.1.0", "description": "Show what's playing on your Plex Media Server: movies, TV episodes and music, laid out to fit any board", "author": "FiestaBoard Team", "repository": "https://github.com/Fiestaboard/fiestaboard-plugin--plex", @@ -19,15 +19,20 @@ }, "server_url": { "type": "string", - "title": "Plex Server URL", - "description": "Address of your Plex Media Server, e.g. http://192.168.1.100:32400" + "title": "Plex Server URL (optional)", + "description": "Address of your Plex Media Server, e.g. http://192.168.1.100:32400. Leave empty to find it through your Plex account." }, "token": { "type": "string", - "title": "Plex Token", - "description": "Your X-Plex-Token (see the setup guide for how to find it)", + "title": "Plex Token (optional)", + "description": "Paste an X-Plex-Token instead of signing in with Plex (see the setup guide). When set, it is used instead of the sign-in.", "ui:widget": "password" }, + "server_name": { + "type": "string", + "title": "Plex Server Name (optional)", + "description": "When your Plex account can reach more than one server and the Plex Server URL is empty, use the server with this name." + }, "plex_user": { "type": "string", "title": "Plex User (optional)", @@ -47,11 +52,14 @@ "minimum": 10, "maximum": 600 } - }, - "required": [ - "server_url", - "token" - ] + } + }, + "oauth": { + "provider_name": "Plex", + "flows": [ + "plex_pin" + ], + "plex_product": "FiestaBoard" }, "env_vars": [ { @@ -242,7 +250,7 @@ ] } }, - "fiestaboard_version": ">=2.10.0", + "fiestaboard_version": ">=9.11.0", "teaser": "{65} BREAKING BAD", "previews": [ { diff --git a/run_tests.sh b/run_tests.sh new file mode 100755 index 0000000..cba4981 --- /dev/null +++ b/run_tests.sh @@ -0,0 +1,22 @@ +#!/usr/bin/env bash +# Run this plugin's tests the same way CI does. +# Usage: ./run_tests.sh [path-to-FiestaBoard-core] (default: ../FiestaBoard) +set -euo pipefail + +CORE="${1:-../FiestaBoard}" +if [ ! -d "$CORE/src/plugins" ]; then + echo "FiestaBoard core not found at: $CORE" >&2 + echo "Pass the path to your FiestaBoard checkout: ./run_tests.sh /path/to/FiestaBoard" >&2 + exit 1 +fi +CORE_ABS="$(cd "$CORE" && pwd)" +PLUGIN_ID=$(python3 -c "import json; print(json.load(open('manifest.json'))['id'])") + +# Recreate the import scaffold CI builds (ignored by git). +mkdir -p plugins +touch plugins/__init__.py +[ -e "plugins/$PLUGIN_ID" ] || ln -s .. "plugins/$PLUGIN_ID" +[ -e "$PLUGIN_ID" ] || ln -s . "$PLUGIN_ID" + +PYTHONPATH="$(pwd):$CORE_ABS" pytest tests/ -v \ + --cov=. --cov-report=term-missing --cov-fail-under=80 --ignore="$CORE_ABS" diff --git a/tests/test_plugin.py b/tests/test_plugin.py index 4ce60cb..920cffe 100644 --- a/tests/test_plugin.py +++ b/tests/test_plugin.py @@ -115,12 +115,11 @@ class TestValidateConfig: def test_valid_config(self): assert make_plugin().validate_config({"server_url": "http://plex.local:32400", "token": "t"}) == [] - def test_missing_url_and_token(self, monkeypatch): + def test_url_and_token_optional(self, monkeypatch): + """Sign in with Plex supplies the token, and plex.tv finds the server.""" monkeypatch.delenv("PLEX_URL", raising=False) monkeypatch.delenv("PLEX_TOKEN", raising=False) - errors = make_plugin().validate_config({}) - assert "Plex server URL is required" in errors - assert "Plex token is required" in errors + assert make_plugin().validate_config({}) == [] def test_url_without_scheme(self): errors = make_plugin().validate_config({"server_url": "192.168.1.100:32400", "token": "t"}) @@ -363,9 +362,10 @@ def test_missing_config(self, monkeypatch): monkeypatch.delenv("PLEX_TOKEN", raising=False) plugin = PlexPlugin(manifest=MANIFEST) plugin.config = {} + plugin.get_oauth_token = lambda: None result = plugin.fetch_data() assert result.available is False - assert result.error == "Plex server URL or token not configured" + assert result.error == "Sign in with Plex, or paste a Plex token, in the plugin settings" def test_request(self): plugin = make_plugin(server_url="http://192.168.1.100:32400/") @@ -376,6 +376,15 @@ def test_request(self): assert kwargs["headers"]["Accept"] == "application/json" assert kwargs["timeout"] == 10 + def test_server_request_headers_unchanged(self): + """A pasted token plus server URL sends exactly what 1.0.x sent (no per-restart client id the server would list as a new device).""" + _, get = fetch(make_plugin()) + assert get.call_args.kwargs["headers"] == { + "X-Plex-Token": "test_token", + "X-Plex-Product": "FiestaBoard", + "Accept": "application/json", + } + def test_env_vars_used_when_settings_empty(self, monkeypatch): monkeypatch.setenv("PLEX_URL", "http://plex.local:32400") monkeypatch.setenv("PLEX_TOKEN", "env_token") @@ -518,7 +527,10 @@ def render(template_entry, data, board): registry = Mock(get_manifest=Mock(return_value=SimpleNamespace(color_rules_schema=MANIFEST["color_rules_schema"]))) with patch("src.templates.engine.get_plugin_registry", return_value=registry): engine = TemplateEngine() - engine._config_manager = Mock(get_color_rules=Mock(return_value=None)) + engine._config_manager = Mock( + get_color_rules=Mock(return_value=None), # cores before 9.10 + get_effective_color_rules=Mock(return_value=None), # 9.10+: per-instance rules + ) notes_wide = board.cols // 15 if board.device_type == "note_array" else 1 notes_tall = board.rows // 3 if board.device_type == "note_array" else 1 text = engine.render_lines( diff --git a/tests/test_sign_in.py b/tests/test_sign_in.py new file mode 100644 index 0000000..d57c573 --- /dev/null +++ b/tests/test_sign_in.py @@ -0,0 +1,305 @@ +"""Sign in with Plex (core's plex_pin flow) and server discovery through plex.tv.""" + +import json +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest +import requests + +from plugins.plex import PLEX_RESOURCES_URL, PlexPlugin +from src.oauth.provider import parse_provider_block, validate_provider_block + +MANIFEST = json.loads((Path(__file__).resolve().parent.parent / "manifest.json").read_text()) + +LAN = "https://192-168-1-5.abc.plex.direct:32400" +REMOTE = "https://203-0-113-9.abc.plex.direct:32400" +RELAY = "https://10-0-0-1.abc.plex.direct:8443" + + +def response(status_code=200, body=None): + resp = MagicMock() + resp.status_code = status_code + resp.json.return_value = body if body is not None else {"MediaContainer": {"size": 0}} + if status_code >= 400: + resp.raise_for_status.side_effect = requests.HTTPError(f"{status_code} Error") + else: + resp.raise_for_status.return_value = None + return resp + + +def server(name="Home", owned=True, access_token="test_server_token", connections=None): + return { + "name": name, + "provides": "server", + "owned": owned, + "presence": True, + "accessToken": access_token, + "connections": connections + if connections is not None + else [ + {"uri": RELAY, "local": False, "relay": True}, + {"uri": REMOTE, "local": False, "relay": False}, + {"uri": LAN, "local": True, "relay": False}, + ], + } + + +class Router: + """Answers requests.get by URL; records every call.""" + + def __init__(self, resources=None, resources_status=200, reachable=(LAN, REMOTE, RELAY), sessions_status=200): + self.resources = resources if resources is not None else [server()] + self.resources_status = resources_status + self.reachable = set(reachable) + self.sessions_status = sessions_status + self.calls = [] + + def __call__(self, url, **kwargs): + self.calls.append((url, kwargs)) + if url == PLEX_RESOURCES_URL: + return response(self.resources_status, self.resources) + base = url.rsplit("/", 2)[0] if url.endswith("/status/sessions") else url.rsplit("/", 1)[0] + if base not in self.reachable: + raise requests.ConnectionError("unreachable") + if url.endswith("/status/sessions"): + return response(self.sessions_status) + return response(200, {}) + + def urls(self): + return [url for url, _ in self.calls] + + +def make_plugin(signed_in_token="test_signed_in_token", **config): + plugin = PlexPlugin(manifest=MANIFEST) + plugin.config = config + plugin.get_oauth_token = MagicMock(return_value=signed_in_token) + plugin.report_oauth_rejected = MagicMock(return_value=None) + return plugin + + +@pytest.fixture(autouse=True) +def no_env(monkeypatch): + monkeypatch.delenv("PLEX_URL", raising=False) + monkeypatch.delenv("PLEX_TOKEN", raising=False) + + +def run(plugin, router): + with patch("plugins.plex.requests.get", side_effect=router): + return plugin.fetch_data() + + +class TestManifest: + def test_oauth_block_is_plex_pin(self): + assert validate_provider_block(MANIFEST["oauth"]) == [] + provider = parse_provider_block(MANIFEST["oauth"], MANIFEST["name"]) + assert provider.flows == ("plex_pin",) + assert provider.plex_product == "FiestaBoard" + + def test_requires_core_with_plex_pin(self): + assert MANIFEST["fiestaboard_version"] == ">=9.11.0" + + def test_existing_settings_keys_kept_and_optional(self): + props = MANIFEST["settings_schema"]["properties"] + for key in ("server_url", "token", "plex_user", "show_accents", "refresh_seconds"): + assert key in props + assert "token" not in MANIFEST["settings_schema"].get("required", []) + assert "server_url" not in MANIFEST["settings_schema"].get("required", []) + + def test_minor_version_bump(self): + assert MANIFEST["version"] == "1.1.0" + + +class TestTokenChoice: + def test_pasted_token_wins_over_sign_in(self): + plugin = make_plugin(server_url="http://plex.local:32400", token="test_pasted") + router = Router(reachable=("http://plex.local:32400",)) + result = run(plugin, router) + assert result.available is True + assert router.calls[-1][1]["headers"]["X-Plex-Token"] == "test_pasted" + plugin.get_oauth_token.assert_not_called() + + def test_env_token_wins_over_sign_in(self, monkeypatch): + monkeypatch.setenv("PLEX_TOKEN", "test_env") + plugin = make_plugin(server_url="http://plex.local:32400") + router = Router(reachable=("http://plex.local:32400",)) + run(plugin, router) + assert router.calls[-1][1]["headers"]["X-Plex-Token"] == "test_env" + + def test_signed_in_token_used_with_server_url(self): + plugin = make_plugin(server_url="http://plex.local:32400") + router = Router(reachable=("http://plex.local:32400",)) + result = run(plugin, router) + assert result.available is True + assert router.urls() == ["http://plex.local:32400/status/sessions"] + assert router.calls[0][1]["headers"]["X-Plex-Token"] == "test_signed_in_token" + + def test_neither_token_asks_to_sign_in(self): + plugin = make_plugin(signed_in_token=None, server_url="http://plex.local:32400") + result = run(plugin, Router()) + assert result.available is False + assert "Sign in with Plex" in result.error + + def test_old_core_without_sign_in(self): + """A core without get_oauth_token still works with a pasted token, and explains otherwise.""" + plugin = PlexPlugin(manifest=MANIFEST) + plugin.config = {} + with patch.object(PlexPlugin, "get_oauth_token", None, create=True): + result = plugin.fetch_data() + assert result.available is False + assert "Plex token" in result.error + + def test_sign_in_lookup_failure_is_not_fatal(self): + plugin = make_plugin(server_url="http://plex.local:32400") + plugin.get_oauth_token.side_effect = RuntimeError("oauth store unreadable") + result = run(plugin, Router()) + assert result.available is False + assert "Sign in with Plex" in result.error + + +class TestDiscovery: + def test_finds_server_and_prefers_local_connection(self): + plugin = make_plugin() + router = Router() + result = run(plugin, router) + assert result.available is True + url, kwargs = router.calls[0] + assert url == PLEX_RESOURCES_URL + assert kwargs["headers"]["X-Plex-Token"] == "test_signed_in_token" + assert kwargs["headers"]["X-Plex-Client-Identifier"] + assert kwargs["headers"]["Accept"] == "application/json" + assert router.urls()[1:] == [f"{LAN}/identity", f"{LAN}/status/sessions"] + assert router.calls[-1][1]["headers"]["X-Plex-Token"] == "test_server_token" + + def test_falls_back_to_remote_then_relay(self): + router = Router(reachable=(RELAY,)) + result = run(make_plugin(), router) + assert result.available is True + assert router.urls()[1:] == [f"{LAN}/identity", f"{REMOTE}/identity", f"{RELAY}/identity", f"{RELAY}/status/sessions"] + + def test_discovery_is_cached(self): + plugin = make_plugin() + router = Router() + run(plugin, router) + run(plugin, router) + assert router.urls().count(PLEX_RESOURCES_URL) == 1 + + def test_cache_dropped_when_server_unreachable(self): + plugin = make_plugin() + router = Router() + run(plugin, router) + router.reachable = {REMOTE} + result = run(plugin, router) + assert result.available is False + assert result.error.startswith("Could not reach Plex server") + run(plugin, router) + assert router.urls().count(PLEX_RESOURCES_URL) == 2 + assert router.urls()[-1] == f"{REMOTE}/status/sessions" + + def test_works_with_pasted_token_too(self): + plugin = make_plugin(signed_in_token=None, token="test_pasted") + router = Router() + assert run(plugin, router).available is True + assert router.calls[0][1]["headers"]["X-Plex-Token"] == "test_pasted" + + def test_owned_server_preferred(self): + friend = server(name="Friend", owned=False, access_token="test_friend", + connections=[{"uri": REMOTE, "local": False, "relay": False}]) + router = Router(resources=[friend, server()]) + run(make_plugin(), router) + assert router.urls()[1] == f"{LAN}/identity" + + def test_server_name_picks_server(self): + friend = server(name="Friend", owned=False, access_token="test_friend", + connections=[{"uri": REMOTE, "local": False, "relay": False}]) + router = Router(resources=[server(), friend]) + result = run(make_plugin(server_name="friend"), router) + assert result.available is True + assert router.calls[-1][0] == f"{REMOTE}/status/sessions" + assert router.calls[-1][1]["headers"]["X-Plex-Token"] == "test_friend" + + def test_unknown_server_name(self): + result = run(make_plugin(server_name="Cabin"), Router()) + assert result.available is False + assert "Cabin" in result.error + + def test_ignores_non_servers_and_missing_access_token(self): + player = {"name": "Phone", "provides": "client,player", "connections": [{"uri": REMOTE}]} + router = Router(resources=[player, server(access_token=None)]) + run(make_plugin(), router) + assert router.calls[-1][1]["headers"]["X-Plex-Token"] == "test_signed_in_token" + + def test_no_server_on_account(self): + result = run(make_plugin(), Router(resources=[])) + assert result.available is False + assert "Plex Server URL" in result.error + + def test_server_not_reachable_from_board(self): + result = run(make_plugin(), Router(reachable=())) + assert result.available is False + assert "Plex Server URL" in result.error + + def test_plex_tv_unreachable(self): + with patch("plugins.plex.requests.get", side_effect=requests.ConnectionError("dns")): + result = make_plugin().fetch_data() + assert result.available is False + assert "plex.tv" in result.error + + def test_plex_tv_bad_answer(self): + result = run(make_plugin(), Router(resources={"error": "x"})) + assert result.available is False + assert "plex.tv" in result.error + + def test_plex_tv_server_error(self): + result = run(make_plugin(), Router(resources_status=503)) + assert result.available is False + assert "plex.tv" in result.error + + +class TestRejection: + def test_plex_tv_rejects_signed_in_token(self): + plugin = make_plugin() + result = run(plugin, Router(resources_status=401)) + assert result.available is False + assert "Sign in with Plex again" in result.error + plugin.report_oauth_rejected.assert_called_once() + + def test_plex_tv_rejects_pasted_token(self): + plugin = make_plugin(signed_in_token=None, token="test_pasted") + result = run(plugin, Router(resources_status=401)) + assert result.error == "Plex rejected the token (401 Unauthorized)" + plugin.report_oauth_rejected.assert_not_called() + + def test_server_rejects_signed_in_token(self): + plugin = make_plugin(server_url="http://plex.local:32400") + result = run(plugin, Router(reachable=("http://plex.local:32400",), sessions_status=401)) + assert "Sign in with Plex again" in result.error + plugin.report_oauth_rejected.assert_called_once() + + def test_discovered_server_rejects_shared_token_drops_cache(self): + plugin = make_plugin() + router = Router() + run(plugin, router) + router.sessions_status = 401 + result = run(plugin, router) + assert result.error == "Plex rejected the token (401 Unauthorized)" + plugin.report_oauth_rejected.assert_not_called() + router.sessions_status = 200 + run(plugin, router) + assert router.urls().count(PLEX_RESOURCES_URL) == 2 + + def test_old_core_without_report(self): + plugin = make_plugin(server_url="http://plex.local:32400") + plugin.report_oauth_rejected = None + result = run(plugin, Router(reachable=("http://plex.local:32400",), sessions_status=401)) + assert "Sign in with Plex again" in result.error + + +class TestValidateConfig: + def test_nothing_required_now(self): + assert make_plugin().validate_config({}) == [] + + def test_bad_url_still_rejected(self): + assert make_plugin().validate_config({"server_url": "plex.local"}) == [ + "Plex server URL must start with http:// or https://" + ]