diff --git a/.env.example b/.env.example deleted file mode 100644 index a823ab8..0000000 --- a/.env.example +++ /dev/null @@ -1,17 +0,0 @@ -#!/usr/bin/env -# ========================================== -# Matrix PreMiD Configuration -# ========================================== - -# The base URL of your Matrix homeserver (e.g., https://matrix.org) -HOMESERVER="https://matrix.org" - -# Your full Matrix user ID (e.g., @shane:matrix.org) -USERNAME="@xyz:some.domain" - -# A valid access token for your account -# (You can get this from Help & About -> Access Token in Element) -ACCESS_TOKEN="" - -# (Optional) A specific Device ID to bind this session to -DEVICE_ID="" diff --git a/.envrc b/.envrc new file mode 100644 index 0000000..98b28c8 --- /dev/null +++ b/.envrc @@ -0,0 +1,9 @@ +#!/bin/bash + +if [ -f .venv/bin/activate ]; then + source .venv/bin/activate +fi + +if [ -f .env ]; then + dotenv .env +fi diff --git a/.flake8 b/.flake8 index 8dd399a..68d6bb8 100644 --- a/.flake8 +++ b/.flake8 @@ -1,3 +1,9 @@ [flake8] max-line-length = 88 +exclude = .venv, .git, __pycache__, build, dist extend-ignore = E203 + +[flake8:local-plugins] +# Flake8 doesn't support per-directory line length natively in the config file +# without plugins, but we can use the --max-line-length flag in the Makefile +# or just keep this here for general use. diff --git a/.github/workflows/integration.yml b/.github/workflows/integration.yml index 06c0ecd..d8eb766 100644 --- a/.github/workflows/integration.yml +++ b/.github/workflows/integration.yml @@ -10,20 +10,6 @@ on: jobs: integration: runs-on: ubuntu-latest - services: - conduwuit: - image: ghcr.io/continuwuity/continuwuity:latest - ports: - - 8008:8008 - env: - CONDUWUIT_SERVER_NAME: localhost - CONDUWUIT_ALLOW_REGISTRATION: "true" - CONDUWUIT_DATABASE_PATH: /tmp/conduwuit - CONDUWUIT_DATABASE_BACKEND: sqlite - CONDUWUIT_ADDRESS: "0.0.0.0" - CONDUWUIT_PORT: 8008 - CONDUWUIT_REGISTRATION_TOKEN: "ci_test_token_123" - steps: - uses: actions/checkout@v4 @@ -37,13 +23,22 @@ jobs: run: | make init make deps + . .venv/bin/activate + pip install . echo "$PWD/.venv/bin" >> $GITHUB_PATH + - name: Start Conduwuit Configured Homeserver + run: | + docker run -d --name conduwuit \ + -p 8008:8008 \ + -v $PWD/docker/conduwuit.toml:/etc/conduwuit.toml:ro \ + -e CONDUWUIT_CONFIG=/etc/conduwuit.toml \ + ghcr.io/continuwuity/continuwuity:latest - name: Wait for Conduwuit to be ready run: | echo "Waiting for Conduwuit to be ready..." for i in {1..30}; do - if curl -s http://localhost:8008/_matrix/client/versions > /dev/null; then + if curl -fsS http://localhost:8008/_matrix/client/versions > /dev/null 2>&1; then echo "Conduwuit Ready!" exit 0 fi @@ -53,7 +48,6 @@ jobs: exit 1 - name: Run CI Integration Tests - env: - CONDUWUIT_CONTAINER_ID: ${{ job.services.conduwuit.id }} run: | - python docker/test_integration.py + export PYTHONPATH="$PWD/src:${PYTHONPATH}" + PYTHONPATH=src python docker/test_integration.py diff --git a/.gitignore b/.gitignore index 42f20c5..3667515 100644 --- a/.gitignore +++ b/.gitignore @@ -214,3 +214,5 @@ __marimo__/ # Streamlit .streamlit/secrets.toml +config.json +_version.py diff --git a/.prettierrc b/.prettierrc new file mode 100644 index 0000000..d744ff2 --- /dev/null +++ b/.prettierrc @@ -0,0 +1,11 @@ +{ + "overrides": [ + { + "files": ["*.json", ".prettierrc"], + "options": { + "tabWidth": 4, + "useTabs": false + } + } + ] +} diff --git a/Makefile b/Makefile index 3290a1f..94299d6 100644 --- a/Makefile +++ b/Makefile @@ -11,60 +11,50 @@ PYTHON=$(VENV)/bin/python PIP=$(VENV)/bin/pip .PHONY: init -init: +init: ##H Initialize .venv virtual dev env python3 -m venv $(VENV) -direnv allow .PHONY: deps -deps: ##H Install standard and dev dependencies +deps: ##H Install standard and dev dependencies $(VENV)/bin/pip install -r requirements.txt -r requirements-dev.txt -# Default install location -INSTALL_DIR ?= /opt/matrix-premid - .PHONY: install -install: ##H Install dependencies, env, binary, and systemd service to /opt (requires sudo) - @echo "Installing globally to $(INSTALL_DIR)..." - sudo mkdir -p $(INSTALL_DIR) - sudo cp matrix_premid.py requirements.txt $(INSTALL_DIR)/ - if [ -f .env ]; then \ - sudo cp .env $(INSTALL_DIR)/.env; \ - sudo chown $$(id -un):$$(id -gn) $(INSTALL_DIR)/.env; \ - sudo chmod 600 $(INSTALL_DIR)/.env; \ - fi - sudo python3 -m venv $(INSTALL_DIR)/.venv - sudo $(INSTALL_DIR)/.venv/bin/pip install -r $(INSTALL_DIR)/requirements.txt - sudo ln -sf $(INSTALL_DIR)/matrix_premid.py /usr/local/bin/matrix_premid - sudo chmod +x $(INSTALL_DIR)/matrix_premid.py - sudo cp etc/matrix-premid.service /etc/systemd/system/matrix-premid.service - sudo systemctl daemon-reload - sudo systemctl enable matrix-premid.service - @echo "Installed to $(INSTALL_DIR) and service created." - +install: ##H Install locally and setup systemd user service + sudo mkdir -p /opt/matrix-premid + sudo python3 -m venv /opt/matrix-premid/venv + sudo /opt/matrix-premid/venv/bin/pip install . + sudo ln -sf /opt/matrix-premid/venv/bin/matrix-premid /usr/local/bin/matrix-premid + @echo "Setting up systemd service..." + if [ -n "$$SUDO_USER" ]; then sudo -u "$$SUDO_USER" /usr/local/bin/matrix-premid install-service; else /usr/local/bin/matrix-premid install-service; fi # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ # Unit tests and local running # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ .PHONY: test -test: deps ##H Run unit tests with coverage - PYTHONPATH=. $(VENV)/bin/python -m pytest --cov=matrix_premid --cov-report=term-missing tests/ +test: ##H Run unit tests with coverage + PYTHONPATH=src $(VENV)/bin/python -m pytest --cov=matrix_premid --cov-report=term-missing tests/ .PHONY: run -run: deps ##H Run the application locally - $(PYTHON) matrix_premid.py +run: ##H Run the application locally + PYTHONPATH=src $(PYTHON) -m matrix_premid --debug + +.PHONY: start +start: ##H Start the background systemd service + systemctl --user start matrix-premid.service .PHONY: restart restart: ##H Restart the background systemd service - sudo systemctl restart matrix-premid.service + systemctl --user restart matrix-premid.service .PHONY: log log: ##H Watch journalctl logs of installed/running service - sudo journalctl -fu matrix-premid + journalctl --user -fu matrix-premid .PHONY: stop stop: ##H Stop the background systemd service - sudo systemctl stop matrix-premid.service + systemctl --user stop matrix-premid.service # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ @@ -72,30 +62,41 @@ stop: ##H Stop the background systemd service # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ -LINT_LOCS_PY = $$(git ls-files '*.py') +LINT_LOCS_PY = $$(git ls-files 'src/**/*.py' 'tests/*.py') .PHONY: format format: ##H Format the code using Black - $(VENV)/bin/black $(LINT_LOCS_PY) - $(VENV)/bin/isort $(LINT_LOCS_PY) + $(VENV)/bin/black src/ tests/ + $(VENV)/bin/isort src/ tests/ -prettier -w . -pre-commit run --all-files .PHONY: lint lint: ##H Lint the code using Flake8 - flake8 $(LINT_LOCS_PY) - pylint $(LINT_LOCS_PY) - ruff check $(LINT_LOCS_PY) + flake8 src/ + flake8 --max-line-length=100 tests/ + pylint src/ tests/ + ruff check src/ tests/ + +.PHONY: build +build: ##H Build the package (requires hatch) + $(VENV)/bin/pip install hatch + $(VENV)/bin/hatch build + +.PHONY: publish +publish: build ##H Upload the package to PyPI using twine + $(VENV)/bin/pip install twine + $(VENV)/bin/twine upload dist/* .PHONY: clean clean: ##H Clean the virtual environment and caches - rm -rf $(VENV) + #rm -rf $(VENV) find . -type f -name '*.pyc' -delete find . -type d -name '__pycache__' -exec rm -rf {} + rm -rf .mypy_cache .PHONY: _help _help: ##H Show this help, list available targets - @grep -hE '^[a-zA-Z0-9_\/-]+:.*?##H .*$$' $(MAKEFILE_LIST) \ - | awk 'BEGIN {FS = ":.*?##H "}; {printf "$(STYLE_CYAN)%-15s$(STYLE_RESET) %s\n", $$1, $$2}' + @grep -hE '^[a-zA-Z0-9_\/-]+:[[:space:]]*##H .*$$' $(MAKEFILE_LIST) \ + | awk 'BEGIN {FS = ":[[:space:]]*##H "}; {printf "$(STYLE_CYAN)%-15s$(STYLE_RESET) %s\n", $$1, $$2}' diff --git a/README.rst b/README.rst index d146637..3dcc131 100644 --- a/README.rst +++ b/README.rst @@ -23,59 +23,98 @@ If you want to run this constantly in the background as a Linux service, indepen git clone https://github.com/user/matrix-premid cd matrix-premid -2. Configure your credentials locally (or edit later): +2. Install the script, systemd user service, and dependencies globally to ``/opt``: .. code-block:: bash - cp .env.example .env - nano .env + make install + + This creates the directory ``/opt/matrix-premid``, sets up an isolated Python virtual environment exclusively for the service, symlinks the script to ``/usr/local/bin/matrix-premid``, and automatically initializes a systemd user service and config template for your user account. - *(Note: If you populate the ``.env`` file locally before installing, the installer will automatically copy and use it for the background service.)* +3. Edit your configuration file at ``~/.config/matrix-premid/config.json`` to add your homeserver and username. -3. Install the script, systemd service, and dependencies globally to ``/opt``: +4. Store your Matrix access token in the system keyring: .. code-block:: bash - make install + python -m keyring set matrix-premid @username:domain.com + +User Installation +----------------- + +Alternatively, you can install the package to your user site-packages: - This creates the directory ``/opt/matrix-premid``, copies the script and ``.env`` there, sets up an isolated Python virtual environment exclusively for the service, and symlinks the script to ``/usr/local/bin/matrix_premid``. The systemd service is placed in ``/etc/systemd/system/``. +.. code-block:: bash + + pip install --user . + +This will install the ``matrix-premid`` command to your ``~/.local/bin`` (make sure ``~/.local/bin`` is on your ``PATH``). -4. (Optional) Edit credentials after installation: +Basic Usage +----------- + +1. **Install dependencies**: ``pip install .`` (or use installation methods above). +2. **Setup configuration**: Create your configuration file at ``~/.config/matrix-premid/config.json``. You can use ``matrix-premid install-service`` to generate a default template, or copy ``config.template.json`` from this repository to that location. +3. **Store credentials**: For security, store your Matrix access token in your system keyring rather than putting it in plain text inside the config file: .. code-block:: bash - sudo nano /opt/matrix-premid/.env - sudo systemctl restart matrix-premid.service + python -m keyring set matrix-premid @username:domain.com + + *(Alternatively, for non-interactive setups like CI, you can set the ``access_token`` directly in the ``accounts`` section of ``config.json``).* -5. Start and enable the background service: +4. **Run the script**: ``matrix-premid`` + +Command-line Options +-------------------- + +* ``--unset`` or ``--clear``: Manually set Matrix presence to ``offline``, clear the status message/account data, and exit. +* ``--debug``: Enable verbose debug logging. +* ``--help``: Show all available options. + +Shell Completion +---------------- + +This script supports bash/zsh completion via ``argcomplete``. To enable it: + +1. Install ``argcomplete`` (included in requirements). +2. Register the script: .. code-block:: bash - sudo systemctl daemon-reload - sudo systemctl enable --now matrix-premid.service + eval "$(register-python-argcomplete matrix-premid)" + + (Add this to your ``.bashrc`` or ``.zshrc`` for persistence). Development / Local Running --------------------------- If you want to run the script locally from the folder (for testing or development) without installing it system-wide: -1. Clone the repository and configure your environment: +1. Clone the repository and configure your configuration: .. code-block:: bash git clone https://github.com/user/matrix-premid cd matrix-premid - cp .env.example .env + mkdir -p ~/.config/matrix-premid + cp config.template.json ~/.config/matrix-premid/config.json + + Edit the ``~/.config/matrix-premid/config.json`` file to set your Matrix username and homeserver. + +2. Store your Matrix access token in the system keyring: + + .. code-block:: bash - Edit the ``.env`` file and fill in your Matrix credentials. Make sure to export them to your shell (e.g., using ``direnv allow`` or sourcing the file) because the script reads directly from ``os.environ``. + python -m keyring set matrix-premid @your_username:homeserver.com -2. Install development dependencies: +3. Install development dependencies: .. code-block:: bash make deps -3. Run the script directly: +4. Run the script directly: .. code-block:: bash diff --git a/config.template.json b/config.template.json new file mode 100644 index 0000000..c65967c --- /dev/null +++ b/config.template.json @@ -0,0 +1,76 @@ +{ + "accounts": [ + { + "enabled": true, + "homeserver": "https://matrix.org", + "username": "@user:matrix.org", + "device_id": "" + } + ], + "idle_timeout": 15, + "poll_interval": 5, + "providers": { + "Methstreams": { + "regex": "(methstreams)", + "priority": 95, + "enabled": true, + "is_video": true, + "template": "Watching: {title} | {provider}" + }, + "YouTube": { + "regex": "(youtube)", + "priority": 90, + "enabled": true, + "is_video": true + }, + "YouTube Music": { + "regex": "(music\\.youtube\\.com|yt music|youtube music)", + "priority": 100, + "enabled": true, + "is_video": false, + "template": "{prefix} {title} - {artist} [{time}] | {provider}" + }, + "Apple Music": { + "regex": "(apple music|music\\.apple\\.com)", + "priority": 90, + "enabled": true, + "is_video": false + }, + "Spotify": { + "regex": "(spotify)", + "priority": 80, + "enabled": true, + "is_video": false + }, + "Last.fm": { + "regex": "(last\\.fm)", + "priority": 70, + "enabled": true, + "is_video": false + }, + "Netflix": { + "regex": "(netflix)", + "priority": 80, + "enabled": true, + "is_video": true + }, + "Plex": { + "regex": "(plex)", + "priority": 80, + "enabled": true, + "is_video": true + }, + "Twitch": { + "regex": "(twitch)", + "priority": 70, + "enabled": true, + "is_video": true + }, + "SoundCloud": { + "regex": "(soundcloud)", + "priority": 70, + "enabled": true, + "is_video": false + } + } +} diff --git a/docker/test_integration.py b/docker/test_integration.py deleted file mode 100644 index 93f008c..0000000 --- a/docker/test_integration.py +++ /dev/null @@ -1,149 +0,0 @@ -"""Longer running integration test best performed locally or in docker.""" - -import json -import os -import subprocess -import sys -import tempfile -import time -import urllib.parse -import urllib.request - -SEP_STR = "_||_" - -# pylint: disable=missing-docstring, too-many-locals -# pylint: disable=too-many-statements, consider-using-with - - -def test_integration(): - print("[i] Starting Matrix PreMiD Integration Test...") - - print("[*] Registering dummy user via matrix client API (UIA Handshake)...") - req = urllib.request.Request( - "http://localhost:8008/_matrix/client/v3/register", - data=json.dumps({"username": "ci_user", "password": "ci_password"}).encode( - "utf-8" - ), - headers={"Content-Type": "application/json"}, - ) - session = None - try: - with urllib.request.urlopen(req) as resp: - pass # Expected to fail with 401 Unauthorized for UIA - except urllib.error.HTTPError as e: - if e.code == 401: - res = json.loads(e.read().decode()) - session = res.get("session") - else: - print("Initial registration handshake failed non-401:", e.read().decode()) - sys.exit(1) - - if not session: - print("[!] Failed to obtain registration UIA session!") - sys.exit(1) - - print(f"[*] Proceeding with UIA session: {session} ...") - req = urllib.request.Request( - "http://localhost:8008/_matrix/client/v3/register", - data=json.dumps( - { - "username": "ci_user", - "password": "ci_password", - "auth": { - "type": "m.login.registration_token", - "token": "ci_test_token_123", - "session": session, - }, - } - ).encode("utf-8"), - headers={"Content-Type": "application/json"}, - ) - try: - with urllib.request.urlopen(req) as resp: - res = json.loads(resp.read().decode()) - token = res["access_token"] - device_id = res["device_id"] - user_id = res["user_id"] - print(f"[✓] Registered user: {user_id}") - except urllib.error.HTTPError as e: - print("Registration failed:", e.read().decode()) - sys.exit(1) - - # 2. Setup mock playerctl - print("[*] Setting up mocked playerctl...") - mock_dir = tempfile.mkdtemp() - mock_script = os.path.join(mock_dir, "playerctl") - with open(mock_script, "w", encoding="utf-8") as f: - f.write("#!/bin/bash\n") - # Echo our predetermined test string to spoof what Linux MPRIS outputs - f.write( - f"echo 'Playing{SEP_STR}GitHub Actions Song" - f"{SEP_STR}Integration Tests{SEP_STR}firefox'\n" - ) - os.chmod(mock_script, 0o755) - - # 3. Write .env inside repo root - # Note: tests are run from the project root in CI - print("[*] Writing local configuration .env file...") - with open(".env", "w", encoding="utf-8") as f: - f.write("HOMESERVER=http://localhost:8008\n") - f.write(f"USERNAME={user_id}\n") - f.write(f"ACCESS_TOKEN={token}\n") - f.write(f"DEVICE_ID={device_id}\n") - - # 4. Start matrix_premid.py - print("[i] Starting matrix_premid.py background daemon...") - env = os.environ.copy() - # Prepend mock_dir so subprocess picks up our fake playerctl instead of real one - env["PATH"] = f"{mock_dir}:{env['PATH']}" - proc = subprocess.Popen([sys.executable, "matrix_premid.py"], env=env) - - # 5. Wait for loop to pick up playerctl, parse, and send to homeserver - print("[~] Waiting 10 seconds for service to update presence...") - time.sleep(10) - - # 6. Verify Matrix Status updates - try: - print("[*] Verifying Presence API state...") - expected = "Listening to: GitHub Actions Song - Integration Tests" - - encoded_user = urllib.parse.quote(user_id) - url = f"http://localhost:8008/_matrix/client/v3/presence/{encoded_user}/status" - req = urllib.request.Request( - url, - headers={"Authorization": f"Bearer {token}"}, - ) - with urllib.request.urlopen(req) as resp: - pres = json.loads(resp.read().decode()) - assert pres["presence"] == "online", f"Presence not online: {pres}" - assert pres["status_msg"] == expected, f"Status MSG mismatch: {pres}" - print(f"[✓] Presence successfully verified: {pres['status_msg']}") - - print("[*] Verifying Account Data API state (im.vector.user_status)...") - encoded_user = urllib.parse.quote(user_id) - acc_url = ( - f"http://localhost:8008/_matrix/client/v3/user/" - f"{encoded_user}/account_data/im.vector.user_status" - ) - req = urllib.request.Request( - acc_url, - headers={"Authorization": f"Bearer {token}"}, - ) - with urllib.request.urlopen(req) as resp: - acc = json.loads(resp.read().decode()) - assert acc["status"] == expected, f"Account Data mismatch: {acc}" - print(f"[✓] Account Data successfully verified: {acc['status']}") - - print("*** ALL INTEGRATION TESTS PASSED ***") - - except AssertionError as e: - print(f"[X] TEST FAILED: {e}") - sys.exit(1) - finally: - # Cleanup - proc.terminate() - proc.wait() - - -if __name__ == "__main__": - test_integration() diff --git a/etc/matrix-premid.service b/etc/matrix-premid.service deleted file mode 100644 index 57a2b96..0000000 --- a/etc/matrix-premid.service +++ /dev/null @@ -1,21 +0,0 @@ -[Unit] -Description=Matrix Presence Updater -After=network.target - -[Service] -Type=simple -User=shane -SyslogIdentifier=matrix-premid -WorkingDirectory=/opt/matrix-premid -Environment=PYTHONUNBUFFERED=1 -Environment=DBUS_SESSION_BUS_ADDRESS=unix:path=/run/user/1000/bus - -# Load environment variables -EnvironmentFile=/opt/matrix-premid/.env - -ExecStart=/opt/matrix-premid/.venv/bin/python /opt/matrix-premid/matrix_premid.py -Restart=on-failure -RestartSec=120 - -[Install] -WantedBy=multi-user.target diff --git a/etc/matrix-premid.service.template b/etc/matrix-premid.service.template deleted file mode 100644 index bff3118..0000000 --- a/etc/matrix-premid.service.template +++ /dev/null @@ -1,21 +0,0 @@ -[Unit] -Description=Matrix Presence Updater -After=network.target - -[Service] -Type=simple -User=%u -SyslogIdentifier=matrix-premid -WorkingDirectory={{WORKING_DIR}} -Environment=PYTHONUNBUFFERED=1 -Environment=DBUS_SESSION_BUS_ADDRESS=unix:path=/run/user/1000/bus - -# Load environment variables -EnvironmentFile=/etc/matrix-premid.env - -ExecStart={{WORKING_DIR}}/.venv/bin/python {{WORKING_DIR}}/matrix_premid.py -Restart=on-failure -RestartSec=120 - -[Install] -WantedBy=multi-user.target diff --git a/matrix_premid.py b/matrix_premid.py deleted file mode 100755 index b7a7f20..0000000 --- a/matrix_premid.py +++ /dev/null @@ -1,344 +0,0 @@ -#!/usr/bin/env python3 -""" -Matrix Presence Updater. - -A robust script to update Matrix presence and Element status based on -native Linux MPRIS (playerctl) events. -""" - -import asyncio -import fcntl -import html -import logging -import os -import sys - -from dotenv import load_dotenv -from nio import Api, AsyncClient -from nio.responses import EmptyResponse, PresenceSetResponse - -PROVIDERS = { - "youtube music": "YouTube Music", - "yt music": "YouTube Music", - "music.youtube.com": "YouTube Music", - "spotify": "Spotify", - "netflix": "Netflix", - "plex": "Plex", - "soundcloud": "SoundCloud", - "last.fm": "Last.fm", - "twitch": "Twitch", - "apple music": "Apple Music", -} - -SEP_STR = "_||_" - -# Load environment variables from .env if present -load_dotenv() - -# Lock file to prevent multiple instances -LOCK_FILE = "/tmp/matrix-premid.lock" - - -def acquire_lock(): - """Ensure only one instance runs. Returns the file descriptor.""" - try: - # We need to keep the file open for the duration of the process - # pylint: disable=consider-using-with - lock_fd = open(LOCK_FILE, "w", encoding="utf-8") - fcntl.lockf(lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB) - return lock_fd - except OSError: - print("ERROR: Another instance is already running.", file=sys.stderr) - sys.exit(1) - - -# Suppress noisy matrix-nio validation errors -logging.getLogger("nio").setLevel(logging.CRITICAL) - -# --- CONFIGURATION --- -HOMESERVER = os.environ.get("HOMESERVER", "") -USERNAME = os.environ.get("USERNAME", "") -ACCESS_TOKEN = os.environ.get("ACCESS_TOKEN", "") -DEVICE_ID = os.environ.get("DEVICE_ID", "") - - -class MatrixStatusUpdater: - """Manages Matrix status updates with state tracking and error handling.""" - - def __init__(self, homeserver, username, access_token, device_id=None): - self.client = AsyncClient(homeserver, username) - self.client.access_token = access_token - self.client.user_id = username - if device_id: - self.client.device_id = device_id - - self.last_activity = "" - self.last_title = "" - self.last_quality = 0 # 0: Idle, 1: Basic, 2: Full (Artist) - self.lock = asyncio.Lock() - - async def close(self): - """Close the Matrix client session.""" - await self.client.close() - - async def update(self, activity: str, title: str = "", force: bool = False): - """Update Matrix presence with metadata quality filtering.""" - if not activity: - activity = "Idle" - - # Determine metadata quality - quality = 0 - if activity.startswith("Listening to:"): - quality = 20 if " - " in activity else 10 - if "YT Music" in activity: - quality += 1 - elif activity.startswith("Paused:"): - quality = 6 if " - " in activity else 4 - if "YT Music" in activity: - quality += 1 - elif activity != "Idle" and not activity.startswith("Idle"): - quality = 10 - - async with self.lock: - # If same song but lower quality metadata, ignore it - # This prevents Firefox (no artist) from overriding Plasma (artist) - if not force and title and title == self.last_title: - if quality < self.last_quality: - return - - is_new = activity != self.last_activity - if not force and not is_new: - return - - if is_new: - print(f"Matrix Status -> {activity}", flush=True) - self.last_activity = activity - self.last_title = title - self.last_quality = quality - - try: - # 1. Standard Presence - await self.client.set_presence(presence="online", status_msg=activity) - - path = ["presence", self.client.user_id, "status"] - # pylint: disable=protected-access - full_path = Api._build_path( - path, {"access_token": self.client.access_token} - ) - - await self.client._send( - PresenceSetResponse, - "PUT", - full_path, - data=Api.to_json( - { - "presence": "online", - "status_msg": activity, - "currently_active": True, - } - ), - ) - - # 2. Element Custom Status - path = [ - "user", - self.client.user_id, - "account_data", - "im.vector.user_status", - ] - # pylint: disable=protected-access - full_path = Api._build_path( - path, {"access_token": self.client.access_token} - ) - content = {"status": activity} if activity != "Idle" else {} - await self.client._send( - EmptyResponse, "PUT", full_path, data=Api.to_json(content) - ) - - except (asyncio.TimeoutError, OSError) as e: - print(f"ERROR: Matrix update failed: {e}", file=sys.stderr) - - -def parse_mpris_data(data: str, global_provider: str = "") -> tuple[str, str]: - """Parse playerctl data into (activity_string, normalized_title).""" - data = html.unescape(data) - parts = [p.strip() for p in data.split(SEP_STR)] - if not parts or not parts[0]: - return "Idle", "" - - status = parts[0] - title = parts[1] if len(parts) > 1 else "Unknown Title" - artist = parts[2] if len(parts) > 2 else "" - - if status not in ("Playing", "Paused"): - return "Idle", "" - - norm_title = title.strip() - - for provider in set(PROVIDERS.values()): - for suffix in [ - f" - {provider}", - f" | {provider}", - f" - {provider.lower()}", - f" | {provider.lower()}", - ]: - if norm_title.endswith(suffix): - norm_title = norm_title[: -len(suffix)].strip() - if artist.endswith(suffix): - artist = artist[: -len(suffix)].strip() - - if title == "YouTube Music" and not artist: - return "Idle (YouTube Music)", norm_title - - banned = {"plasma-browser-integration", "firefox", "chrome", "chromium"} - is_banned = artist.lower() in banned - clean_artist = "" if is_banned else artist - - prefix = "Listening to:" if status == "Playing" else "Paused:" - - if clean_artist: - activity = f"{prefix} {norm_title} - {clean_artist}" - else: - activity = f"{prefix} {norm_title}" - - if global_provider and global_provider not in activity: - activity += f" | {global_provider}" - - return activity, norm_title - - -def _get_best_mpris_activity(lines: list[str]) -> tuple[str, str]: - """Parse multiple player lines and extract the best metadata.""" - best_activity = "Idle" - best_title = "" - best_quality = 0 - - global_provider = "" - for line in lines: - lower_line = line.lower() - for key, name in PROVIDERS.items(): - if key in lower_line: - if ( - name == "Last.fm" - and global_provider - and global_provider != "Last.fm" - ): - continue - global_provider = name - - for raw in lines: - raw = raw.strip() - if not raw: - continue - - activity, title = parse_mpris_data(raw, global_provider) - - quality = 0 - if activity.startswith("Listening to:"): - quality = 20 if " - " in activity else 10 - if global_provider and f"| {global_provider}" in activity: - quality += 1 - elif activity.startswith("Paused:"): - quality = -1 - elif activity != "Idle" and not activity.startswith("Idle"): - quality = 10 - - if quality > best_quality: - best_activity = activity - best_title = title - best_quality = quality - - return best_activity, best_title - - -async def monitor_mpris(updater: MatrixStatusUpdater): - """Monitor MPRIS events via playerctl by polling all players.""" - while True: - try: - # We poll playerctl instead of --follow to avoid holding a persistent - # D-Bus connection. This allows apps like Firefox to close without - # warning about an active media control lock. - # Using --all-players ensures we don't accidentally poll a paused tab - # when another tab is actively playing. - process = await asyncio.create_subprocess_exec( - "playerctl", - "--all-players", - "metadata", - "--format", - f"{{{{status}}}}{SEP_STR}{{{{title}}}}{SEP_STR}" - f"{{{{artist}}}}{SEP_STR}{{{{playerName}}}}", - stdout=asyncio.subprocess.PIPE, - stderr=asyncio.subprocess.DEVNULL, - ) - - stdout, _ = await process.communicate() - if stdout: - lines = stdout.decode("utf-8").strip().splitlines() - activity, title = _get_best_mpris_activity(lines) - await updater.update(activity, title=title) - - except asyncio.CancelledError: - break - except (OSError, ValueError) as e: - print(f"MPRIS Monitor Error: {e}", file=sys.stderr) - - await asyncio.sleep(5) - - -async def main(): - """Start the Matrix updater.""" - # pylint: disable=unused-variable - lock_fd = acquire_lock() # noqa: F841 - if not all([HOMESERVER, USERNAME, ACCESS_TOKEN]): - print("ERROR: Missing configuration in .env", file=sys.stderr) - sys.exit(1) - - updater = MatrixStatusUpdater(HOMESERVER, USERNAME, ACCESS_TOKEN) - print(f"Matrix User: {USERNAME} on {HOMESERVER}", flush=True) - - async def keep_alive(): - """Keep status online.""" - - async def sync_loop(): - while True: - try: - await updater.client.sync(timeout=30, set_presence="online") - except asyncio.CancelledError: - break - except (asyncio.TimeoutError, OSError): - await asyncio.sleep(5) - - async def update_loop(): - while True: - try: - await updater.update(updater.last_activity, force=True) - await asyncio.sleep(15) - except asyncio.CancelledError: - break - except (asyncio.TimeoutError, OSError): - await asyncio.sleep(5) - - await asyncio.gather(sync_loop(), update_loop()) - - try: - print("Listening for MPRIS events...", flush=True) - await asyncio.gather(monitor_mpris(updater), keep_alive()) - except asyncio.CancelledError: - pass - finally: - print("Clearing Matrix status before exit...", flush=True) - try: - await asyncio.wait_for(updater.update("Idle", force=True), timeout=3.0) - except ( # pylint: disable=broad-exception-caught - Exception, - asyncio.CancelledError, - ): - pass - await updater.close() - - -if __name__ == "__main__": - try: - asyncio.run(main()) - except (KeyboardInterrupt, asyncio.CancelledError): - pass diff --git a/pyproject.toml b/pyproject.toml index 115fe66..dea4721 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,3 +1,35 @@ +[build-system] +requires = ["hatchling", "hatch-vcs"] +build-backend = "hatchling.build" + +[project] +name = "matrix-premid" +dynamic = ["version"] +description = "Matrix Presence Updater based on MPRIS events" +readme = "README.rst" +requires-python = ">=3.10" +license = {text = "MIT"} +authors = [ + {name = "Shane Jaroch", email = "chown_tee@proton.me"}, +] +dependencies = [ + "aiohttp>=3.8.0", + "argcomplete>=3.5.3", + "keyring>=24.0.0" +] + +[tool.hatch.version] +source = "vcs" + +[tool.hatch.build.hooks.vcs] +version-file = "src/matrix_premid/_version.py" + +[project.scripts] +matrix-premid = "matrix_premid.__main__:cli" + +[tool.hatch.build.targets.wheel] +packages = ["src/matrix_premid"] + [tool.black] line-length = 88 @@ -7,21 +39,21 @@ line_length = 88 [tool.pytest.ini_options] testpaths = ["tests"] -addopts = "--cov=matrix_premid --cov-report=term-missing" +pythonpath = ["src"] +addopts = "--cov=matrix_premid --cov-report=term-missing tests/" asyncio_mode = "strict" asyncio_default_fixture_loop_scope = "function" -filterwarnings = [ - "ignore::DeprecationWarning", - "ignore::pytest.PytestDeprecationWarning" -] [tool.coverage.run] -command_line = "-m pytest -svv" -source = ["matrix_premid.py"] +source = ["matrix_premid"] +branch = true +omit = ["src/matrix_premid/_version.py"] [tool.coverage.report] -fail_under = 85.00 +fail_under = 80.00 precision = 2 show_missing = true skip_empty = true skip_covered = true +[tool.pylint.master] +ignore-paths = ["^src/matrix_premid/_version.py$"] diff --git a/requirements-dev.txt b/requirements-dev.txt index 8cbb02f..7c1e00b 100644 --- a/requirements-dev.txt +++ b/requirements-dev.txt @@ -3,6 +3,6 @@ flake8==7.3.0 isort==8.0.1 pylint==4.0.5 pytest==8.3.3 -pytest-asyncio==0.24.0 +pytest-asyncio==1.3.0 pytest-cov==6.0.0 ruff==0.15.7 diff --git a/requirements.txt b/requirements.txt index abef831..c1e55fa 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,2 +1,4 @@ -matrix-nio==0.25.2 +argcomplete==3.5.3 +aiohttp>=3.8.0 +keyring==25.7.0 python-dotenv==1.2.2 diff --git a/src/matrix_premid/__init__.py b/src/matrix_premid/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/matrix_premid/__main__.py b/src/matrix_premid/__main__.py new file mode 100755 index 0000000..0000506 --- /dev/null +++ b/src/matrix_premid/__main__.py @@ -0,0 +1,1217 @@ +#!/usr/bin/env python3 +# PYTHON_ARGCOMPLETE_OK +""" +Matrix Presence Updater. + +A robust script to update Matrix presence and Element status based on +native Linux MPRIS (playerctl) events. +""" + +import argparse +import asyncio +import fcntl +import html +import json +import logging +import os +import random +import re +import shutil +import signal +import subprocess +import sys +import urllib.parse +from dataclasses import dataclass +from typing import Dict, Optional + +import aiohttp +import argcomplete +import keyring + +# pylint: disable=too-many-lines + +try: + from matrix_premid._version import __version__ +except ImportError: # pragma: no cover + __version__ = "unknown" + + +@dataclass +class ProviderMetadata: + """Data class storing parsed regex patterns and priorities for media players.""" + + name: str + raw_regex: str + priority: int + is_video: bool + enabled: bool + template: str = "" + pattern: Optional[re.Pattern] = None + + def __post_init__(self): + if self.enabled and self.raw_regex: + try: + self.pattern = re.compile(self.raw_regex, re.IGNORECASE) + except re.error as e: + print( + f"ERROR: Invalid regex '{self.raw_regex}' " + f"for provider '{self.name}': {e}", + file=sys.stderr, + ) + self.enabled = False + + +class ProviderConfig: + """Manager class that handles parsing and mapping media applications.""" + + def __init__(self): + self.providers: Dict[str, ProviderMetadata] = {} + + def load(self, custom_providers: dict): + """Load default schemas and overwrite with user config.""" + defaults = { + "YouTube Music": { + "regex": "(music\\.youtube\\.com|yt music|youtube music)", + "priority": 100, + "is_video": False, + "enabled": True, + }, + "Methstreams": { + "regex": "(methstreams)", + "priority": 95, + "is_video": True, + "enabled": True, + "template": "Watching: {title} | {provider}", + }, + "YouTube": { + "regex": "(youtube)", + "priority": 90, + "is_video": True, + "enabled": True, + }, + "Apple Music": { + "regex": "(apple music|music\\.apple\\.com)", + "priority": 90, + "is_video": False, + "enabled": True, + }, + "Spotify": { + "regex": "(spotify)", + "priority": 80, + "is_video": False, + "enabled": True, + }, + "Netflix": { + "regex": "(netflix)", + "priority": 80, + "is_video": True, + "enabled": True, + }, + "Plex": { + "regex": "(plex)", + "priority": 80, + "is_video": True, + "enabled": True, + }, + "SoundCloud": { + "regex": "(soundcloud)", + "priority": 70, + "is_video": False, + "enabled": True, + }, + "Twitch": { + "regex": "(twitch)", + "priority": 70, + "is_video": True, + "enabled": True, + }, + "Last.fm": { + "regex": "(last\\.fm)", + "priority": 70, + "is_video": False, + "enabled": True, + }, + } + + merged = defaults.copy() + for name, data in custom_providers.items(): + if name in merged: + merged[name].update(data) + else: + merged[name] = data + + self.providers = {} + for name, data in merged.items(): + self.providers[name] = ProviderMetadata( + name=name, + raw_regex=data.get("regex", ""), + priority=data.get("priority", 50), + is_video=data.get("is_video", False), + enabled=data.get("enabled", True), + template=data.get("template", ""), + ) + + def match_provider(self, text: str) -> Optional[ProviderMetadata]: + """Find the highest-priority enabled provider matching the text.""" + if not text: + return None + matches = [] + for p in self.providers.values(): + if p.enabled and p.pattern and p.pattern.search(text): + matches.append(p) + if not matches: + return None + matches.sort(key=lambda p: (p.priority, len(p.raw_regex)), reverse=True) + return matches[0] + + +GLOBAL_PROVIDERS = ProviderConfig() +GLOBAL_PROVIDERS.load({}) + +SEP_STR = "_||_" + +# Lock file to prevent multiple instances +LOCK_FILE = os.environ.get("PREMID_LOCK_FILE", "/tmp/matrix-premid.lock") + + +def acquire_lock(): + """Ensure only one instance runs. Returns the file descriptor.""" + try: + # We need to keep the file open for the duration of the process + # pylint: disable=consider-using-with + lock_fd = open(LOCK_FILE, "a+", encoding="utf-8") + lock_fd.seek(0) + fcntl.lockf(lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB) + lock_fd.truncate() + lock_fd.write(str(os.getpid())) + lock_fd.flush() + return lock_fd + except OSError: + print("ERROR: Another instance is already running.", file=sys.stderr) + sys.exit(1) + + +class MatrixStatusUpdater: + """Manages Matrix status updates with state tracking and error handling.""" + + # pylint: disable=too-many-instance-attributes + + def __init__( + self, + homeserver, + username, + access_token, + device_id=None, + enabled=True, + idle_timeout=15, + poll_interval=5, + verbose=False, + session=None, + ): + # pylint: disable=too-many-arguments,too-many-positional-arguments + self.homeserver = homeserver.rstrip("/") + self.username = username + self.access_token = access_token + self.device_id = device_id + self.enabled = enabled + + self.idle_timeout = idle_timeout + self.poll_interval = poll_interval + self.verbose = verbose + self.last_activity = "" + self.last_title = "" + self.last_quality = 0 # 0: Idle, 1: Basic, 2: Full (Artist) + self.current_presence = "online" # Default fallback + self.idle_strikes = 0 + self.lock = asyncio.Lock() + self._update_task = None + self._session = session + + async def _get_session(self): + """Create or return existing aiohttp ClientSession.""" + if self._session is None or self._session.closed: + self._session = aiohttp.ClientSession() + return self._session + + async def close(self): + """Close the Matrix client session.""" + if self._session and not self._session.closed: + await self._session.close() + + async def update( + self, activity: str, title: str = "", force: bool = False, is_exit: bool = False + ): + """Update Matrix presence with metadata quality filtering.""" + # pylint: disable=too-many-branches, too-many-statements, too-many-locals + if not activity and not is_exit: + # If no activity detected, we don't force 'Idle' immediately. + # We preserve the last activity unless it's explicitly 'Idle' + # or we are exiting. + return + + if activity.startswith("Idle"): + activity = "Idle" + + # Determine metadata quality + quality = 0 + if activity.startswith("Listening to:") or activity.startswith("Watching:"): + quality = 20 if " - " in activity else 10 + if "YT Music" in activity or "YouTube Music" in activity: + quality += 1 + elif activity.startswith("Paused:"): + quality = 6 if " - " in activity else 4 + if "YT Music" in activity or "YouTube Music" in activity: + quality += 1 + elif activity != "Idle" and not activity.startswith("Idle") and activity != "": + quality = 10 + + async with self.lock: + # If same song but lower quality metadata, ignore it + if not force and not is_exit and title and title == self.last_title: + if quality < self.last_quality: + return None + + is_new = activity != self.last_activity + if not force and not is_new and not is_exit: + # Reset strikes if we are consistently playing the same non-idle song + if activity != "Idle": + self.idle_strikes = 0 + return None + + is_idle = activity == "Idle" or activity.startswith("Idle") + if is_idle and not is_exit: + self.idle_strikes += 1 + max_strikes = max(1, self.idle_timeout // self.poll_interval) + if self.idle_strikes < max_strikes and not force: + # Debounce idle state + return None + else: + self.idle_strikes = 0 + + log_res = None + if is_new or is_exit or force: + if not self.enabled: + log_res = (None, f"[{self.username}] (Disabled)", True) + else: + activity_str = "Offline" if is_exit else activity + user_str = f"[{self.username}] [{self.current_presence}]" + log_res = (activity_str, user_str, False) + + if not is_exit: + self.last_activity = activity + self.last_title = title + self.last_quality = quality + + if self._update_task and not self._update_task.done(): + self._update_task.cancel() + + if is_exit: + await self.send_update(activity, is_exit) + return log_res + + async def debounced_send(): + if not force: + # Wait 2 seconds to absorb rapid metadata shifts + await asyncio.sleep(2.0) + await self.send_update(activity, is_exit) + + self._update_task = asyncio.create_task(debounced_send()) + return log_res + + async def send_update(self, activity: str, is_exit: bool = False): + """Send presence and status update to Matrix.""" + if not self.enabled: + return + try: + session = await self._get_session() + headers = {"Authorization": f"Bearer {self.access_token}"} + + encoded_username = urllib.parse.quote(self.username) + + # 1. Presence Payload + url_p = ( + f"{self.homeserver}/_matrix/client/v3/presence/" + f"{encoded_username}/status" + ) + + if is_exit: + payload_p = { + "presence": "offline", + "status_msg": "", + } + else: + payload_p = { + "presence": self.current_presence, + "status_msg": "", + } + if activity and activity != "Idle": + payload_p["status_msg"] = activity + + # 2. Element Status Payload + url_s = ( + f"{self.homeserver}/_matrix/client/v3/user/{encoded_username}/" + "account_data/im.vector.user_status" + ) + + if is_exit or activity == "Idle" or not activity: + payload_s = {} + else: + payload_s = {"status": activity} + + async def send_presence(): + if self.verbose: + print(f"DEBUG [{self.username}]: Req 1/2 (presence)") + try: + async with session.put( + url_p, + json=payload_p, + headers=headers, + timeout=aiohttp.ClientTimeout(total=5.0 if is_exit else 10.0), + ) as resp: + if resp.status >= 400: + data = await resp.text() + print( + f"ERROR [{self.username}]: presence failed " + f"({resp.status}): {data}", + file=sys.stderr, + ) + if resp.status == 401: + print( + f"HINT [{self.username}]: Token may be invalid. " + f"Update it using: {sys.executable} -m keyring set " + f"matrix-premid {self.username}", + file=sys.stderr, + ) + except Exception as e: # pylint: disable=broad-exception-caught + if self.verbose: + print( + f"DEBUG [{self.username}]: presence error: {e}", + file=sys.stderr, + ) + + async def send_status(): + if self.verbose: + print(f"DEBUG [{self.username}]: Req 2/2 (account_data)") + try: + async with session.put( + url_s, + json=payload_s, + headers=headers, + timeout=aiohttp.ClientTimeout(total=5.0 if is_exit else 10.0), + ) as resp: + if resp.status >= 400: + data = await resp.text() + print( + f"ERROR [{self.username}]: account_data failed " + f"({resp.status}): {data}", + file=sys.stderr, + ) + if resp.status == 401: + print( + f"HINT [{self.username}]: Token may be invalid. " + f"Update it using: {sys.executable} -m keyring set " + f"matrix-premid {self.username}", + file=sys.stderr, + ) + except Exception as e: # pylint: disable=broad-exception-caught + if self.verbose: + print( + f"DEBUG [{self.username}]: account_data error: {e}", + file=sys.stderr, + ) + + await asyncio.gather(send_presence(), send_status()) + + except asyncio.CancelledError: + pass + except Exception as e: # pylint: disable=broad-exception-caught + print(f"ERROR: Matrix update exception: {e}", file=sys.stderr) + + +def _detect_provider_from_url(url: str) -> str: + """Detect the provider based on the xesam:url metadata.""" + if not url: + return "" + url_lower = url.lower() + if "music.youtube.com" in url_lower: + return "YouTube Music" + if "youtube.com/watch" in url_lower or "youtube.com/v/" in url_lower: + return "YouTube" + if "netflix.com" in url_lower: + return "Netflix" + if "twitch.tv" in url_lower: + return "Twitch" + return "" + + +def _clean_suffixes(title: str, artist: str) -> tuple[str, str]: + """Remove provider suffixes from title and artist.""" + norm_title = title.strip() + norm_artist = artist.strip() + # Check longest providers first to prevent substring bugs + # (e.g. YouTube vs YouTube Music) + for provider in sorted(GLOBAL_PROVIDERS.providers.keys(), key=len, reverse=True): + for suffix in [ + f" - {provider}", + f" | {provider}", + f" - {provider.lower()}", + f" | {provider.lower()}", + ]: + if norm_title.endswith(suffix): + norm_title = norm_title[: -len(suffix)].strip() + if norm_artist.endswith(suffix): + norm_artist = norm_artist[: -len(suffix)].strip() + return norm_title, norm_artist + + +def _format_duration(microseconds: str) -> str: + """Format microseconds into M:SS or H:MM:SS.""" + try: + if not microseconds or microseconds == "0": + return "" + total_seconds = int(float(microseconds)) // 1_000_000 + if total_seconds == 0: + return "" + + is_negative = total_seconds < 0 + abs_seconds = abs(total_seconds) + + minutes, seconds = divmod(abs_seconds, 60) + hours, minutes = divmod(minutes, 60) + + sign = "- " if is_negative else "" + if hours > 0: + return f"{sign}{hours}:{minutes:02d}:{seconds:02d}" + return f"{sign}{minutes}:{seconds:02d}" + except (ValueError, TypeError, OverflowError): + return "" + + +def parse_mpris_data( + data: str, global_provider: str = "", url: str = "" +) -> tuple[str, str]: + """Parse playerctl data into (activity_string, normalized_title).""" + # pylint: disable=too-many-branches,too-many-statements,too-many-locals + # Browsers often double-escape MPRIS metadata, so we unescape aggressively + data = html.unescape(html.unescape(data)) + data = data.replace(""", '"').replace("'", "'").replace("'", "'") + + parts = [p.strip() for p in data.split(SEP_STR)] + if not parts or not parts[0]: + return "", "" + + def _deep_clean(text: str) -> str: + import ast # pylint: disable=import-outside-toplevel + + text = html.unescape(html.unescape(text)) + cleaned = ( + text.replace(""", '"') + .replace("'", "'") + .replace("'", "'") + .replace("&", "&") + .strip() + ) + + if cleaned.startswith("[") and cleaned.endswith("]"): + try: + parsed = ast.literal_eval(cleaned) + if isinstance(parsed, list): # pragma: no branch + return ", ".join(str(x) for x in parsed) + except (ValueError, SyntaxError): # pragma: no cover + pass + + return cleaned + + title = _deep_clean(parts[1]) if len(parts) > 1 else "Unknown Title" + artist = _deep_clean(parts[2]) if len(parts) > 2 else "" + + if parts[0] not in ("Playing", "Paused"): + return "", "" + + # Use URL to explicitly set/refine the provider if available + url_provider = _detect_provider_from_url(url) + if url_provider: + global_provider = url_provider + + norm_title, norm_artist = _clean_suffixes(title, artist) + + if title == "YouTube Music" and not artist: + return "Idle (YouTube Music)", norm_title + + banned = {"plasma-browser-integration", "firefox", "chrome", "chromium"} + is_banned = norm_artist.lower() in banned + clean_artist = "" if is_banned else norm_artist + + if parts[0] == "Playing": + is_video = False + if global_provider and global_provider in GLOBAL_PROVIDERS.providers: + is_video = GLOBAL_PROVIDERS.providers[global_provider].is_video + prefix = "Watching:" if is_video else "Listening to:" + else: + prefix = "Paused:" + + activity = f"{prefix} {norm_title}" + if clean_artist: + activity += f" - {clean_artist}" + + # Playback timestamp (e.g. [0:36 / 5:44]) + pos_raw = parts[5] if len(parts) > 5 else "" + len_raw = parts[6] if len(parts) > 6 else "" + pos_str = _format_duration(pos_raw) + len_str = _format_duration(len_raw) + + timestamp = "" + if pos_str and len_str: + timestamp = f" [{pos_str} / {len_str}]" + elif pos_str: + timestamp = f" [{pos_str}]" + + if global_provider and global_provider in GLOBAL_PROVIDERS.providers: + tpl = GLOBAL_PROVIDERS.providers[global_provider].template + if tpl: + raw_time_str = "" + if pos_str and len_str: + raw_time_str = f"{pos_str} / {len_str}" + elif pos_str: + raw_time_str = pos_str + + activity = ( + tpl.replace("{prefix}", prefix) + .replace("{title}", norm_title) + .replace("{artist}", clean_artist) + .replace("{time}", raw_time_str) + .replace("{provider}", global_provider) + ).strip() + + # Clean up artifacts if {artist} or {time} were empty + if not clean_artist: + activity = activity.replace(" - ", " ") + if not raw_time_str: + activity = activity.replace(" []", "") + activity = activity.replace(" ", " ").strip() + + return activity, norm_title + + if timestamp: + activity += timestamp + + if global_provider and global_provider not in activity: + activity += f" | {global_provider}" + + return activity, norm_title + + +def _get_line_provider(raw: str) -> str: + """Detect provider for a single line.""" + match = GLOBAL_PROVIDERS.match_provider(raw) + if match: + return match.name + return "" + + +def _apply_provider_inheritance(parsed_lines: list[dict]): + """Second pass: Inheritance for players without provider.""" + providers = {p["provider"] for p in parsed_lines if p["provider"]} + for item in parsed_lines: + if item["provider"]: + continue + raw_parts = item["raw"].split(SEP_STR) + raw_title = raw_parts[1] if len(raw_parts) > 1 else "" + for other in parsed_lines: + if not other["provider"]: + continue + # Inherit if titles match OR if there is only one provider in the batch + if raw_title in other["raw"] or len(providers) == 1: + item["provider"] = other["provider"] + break + + +def _get_best_mpris_activity(lines: list[str]) -> tuple[str, str]: + """Parse multiple player lines and extract the best metadata.""" + # pylint: disable=too-many-branches,too-many-locals,too-many-statements + best_activity, best_title, best_quality = "", "", 0 + + # First pass: Parse each line and detect its own provider + parsed_lines = [] + for raw in lines: + raw = raw.strip() + if not raw or SEP_STR not in raw: + continue + parts = raw.split(SEP_STR) + url = parts[4] if len(parts) > 4 else "" + parsed_lines.append( + {"raw": raw, "provider": _get_line_provider(raw), "url": url} + ) + + _apply_provider_inheritance(parsed_lines) + + # Third pass: Evaluate quality + for item in parsed_lines: + # User defined disabled providers completely drop here + if item["provider"] and item["provider"] in GLOBAL_PROVIDERS.providers: + if not GLOBAL_PROVIDERS.providers[item["provider"]].enabled: + continue + + activity, title = parse_mpris_data(item["raw"], item["provider"], item["url"]) + if not activity: + continue + + raw_parts = item["raw"].split(SEP_STR) + status = raw_parts[0].strip() if raw_parts else "" + raw_artist = raw_parts[2].strip() if len(raw_parts) > 2 else "" + raw_pos = raw_parts[5].strip() if len(raw_parts) > 5 else "" + raw_len = raw_parts[6].strip() if len(raw_parts) > 6 else "" + + banned_artists = {"plasma-browser-integration", "firefox", "chrome", "chromium"} + _, clean_raw_artist = _clean_suffixes("", raw_artist) + has_real_artist = bool( + clean_raw_artist and clean_raw_artist.lower() not in banned_artists + ) + has_time = False + try: + if raw_pos and raw_len: + pos_val = float(raw_pos) + len_val = float(raw_len) + if pos_val >= 0 and len_val > 0: + has_time = True + except (ValueError, TypeError): + pass + + quality = 0 + if status == "Paused": + quality = 500 + elif ( + status == "Playing" + and activity not in ("", "Idle") + and not activity.startswith("Idle") + ): + quality = 1000 + if has_real_artist: + quality += 1000 + if has_time: + quality += 500 + + if item["provider"] and f"| {item['provider']}" in activity: + provider_meta = GLOBAL_PROVIDERS.providers.get(item["provider"]) + if provider_meta: + quality += provider_meta.priority + else: + quality += 1 + + # Add fractional priority based on elapsed position to break ties + # (e.g., between two streams, pick the one with a larger elapsed time) + try: + if raw_pos: + # Add tiny fraction (max 0.999) based on hours of playback + pos_seconds = abs(int(float(raw_pos))) / 1_000_000 + quality += min(0.999, pos_seconds / 3600.0) + except (ValueError, TypeError): + pass + + if quality > best_quality: + best_activity, best_title, best_quality = activity, title, quality + + return best_activity, best_title + + +async def monitor_mpris( + updaters: list[MatrixStatusUpdater], poll_interval: int, config_file: str +): + """Monitor MPRIS events via playerctl by polling all players.""" + + # pylint: disable=too-many-locals,too-many-branches + + last_mtime = 0.0 + + while True: + try: + if os.path.exists(config_file): # pragma: no cover + current_mtime = os.stat(config_file).st_mtime + if current_mtime > last_mtime: + try: + with open(config_file, "r", encoding="utf-8") as f: + hot_config = json.load(f) + GLOBAL_PROVIDERS.load(hot_config.get("providers", {})) + poll_interval = hot_config.get( + "poll_interval", poll_interval + ) + for u in updaters: + u.poll_interval = poll_interval + u.idle_timeout = hot_config.get( + "idle_timeout", u.idle_timeout + ) + if updaters and updaters[0].verbose: + print("DEBUG: Reloaded config successfully.") + last_mtime = current_mtime + except Exception as e: # pylint: disable=broad-exception-caught + if updaters and updaters[0].verbose: + print( + f"DEBUG: Failed to hot-reload config: {e}", + file=sys.stderr, + ) + # Avoid repeatedly trying to parse a broken file + # by updating mtime anyway + last_mtime = current_mtime + # We poll playerctl instead of --follow to avoid holding a persistent + # D-Bus connection. + process = await asyncio.create_subprocess_exec( + "playerctl", + "--all-players", + "metadata", + "--format", + f"{{{{status}}}}{SEP_STR}{{{{title}}}}{SEP_STR}" + f"{{{{artist}}}}{SEP_STR}{{{{playerName}}}}" + f"{SEP_STR}{{{{xesam:url}}}}{SEP_STR}" + f"{{{{position}}}}{SEP_STR}{{{{mpris:length}}}}", + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.DEVNULL, + ) + stdout, _ = await process.communicate() + lines = stdout.decode("utf-8").strip().splitlines() if stdout else [] + if updaters and updaters[0].verbose: # pragma: no cover + print(f"DEBUG: raw playerctl lines: {lines}", flush=True) + activity, title = _get_best_mpris_activity(lines) + if not activity: + activity = "Idle" + + updates_by_status = {} + disabled_users = [] + + for updater in updaters: + log_res = await updater.update(activity, title=title) + if log_res: + act_str, usr_str, is_disabled = log_res + if is_disabled: + disabled_users.append(usr_str) + else: + updates_by_status.setdefault(act_str, []).append(usr_str) + + for act_str, users in updates_by_status.items(): + print(f"Matrix Status -> {act_str}", flush=True) + for user in users: + print(f" {user}", flush=True) + + for user in disabled_users: + print(f" {user}", flush=True) + + except asyncio.CancelledError: + break + except (OSError, ValueError) as e: + print(f"MPRIS Monitor Error: {e}", file=sys.stderr) + + # Use a random normal distribution around the poll_interval + # (e.g. sigma = 0.1 * poll_interval) to prevent tight sync loops + # Minimum wait time of 1s as a fallback + jittered_interval = max(1.0, random.gauss(poll_interval, poll_interval * 0.1)) + await asyncio.sleep(jittered_interval) + + +def install_service(): + """Install the systemd user service.""" + executable = shutil.which("matrix-premid") + if not executable: + # Fallback if not in PATH + executable = sys.executable + " -m matrix_premid" + + service_content = f"""[Unit] +Description=Matrix Presence Updater +After=network.target + +[Service] +Type=simple +Environment=PYTHONUNBUFFERED=1 +ExecStart={executable} +Restart=on-failure +RestartSec=120 + +[Install] +WantedBy=default.target +""" + config_dir = os.path.expanduser("~/.config/systemd/user") + os.makedirs(config_dir, exist_ok=True) + service_file = os.path.join(config_dir, "matrix-premid.service") + + with open(service_file, "w", encoding="utf-8") as f: + f.write(service_content) + + print(f"Created systemd user service at {service_file}") + + app_config_dir = os.path.expanduser("~/.config/matrix-premid") + os.makedirs(app_config_dir, exist_ok=True) + config_file = os.path.join(app_config_dir, "config.json") + if not os.path.exists(config_file): + sample_config = { + "accounts": [ + { + "enabled": True, + "homeserver": "https://matrix.org", + "username": "@user:matrix.org", + "device_id": "", + } + ], + "idle_timeout": 15, + "poll_interval": 5, + } + with open(config_file, "w", encoding="utf-8") as f: + json.dump(sample_config, f, indent=4) + print(f"Created empty config at {config_file} (Please edit!)") + print( + "Note: Store your access token using keyring: " + "python -m keyring set matrix-premid @user:matrix.org" + ) + else: + print(f"Config already exists at: {config_file}") + + try: + subprocess.run(["systemctl", "--user", "daemon-reload"], check=True) + subprocess.run( + ["systemctl", "--user", "enable", "matrix-premid.service"], + check=True, + ) + print( + "Service enabled successfully. Start it with: " + "systemctl --user start matrix-premid.service" + ) + except Exception as e: # pylint: disable=broad-exception-caught + print(f"Failed to enable service: {e}", file=sys.stderr) + + +def parse_args(args=None): + """Parse command line arguments.""" + parser = argparse.ArgumentParser( + description="Matrix Presence/PreMiD Updater", + formatter_class=argparse.RawDescriptionHelpFormatter, + epilog="""\ +configuration: + %(prog)s install-service create config and systemd user service + config file ~/.config/matrix-premid/config.json + +token management (via keyring): + add/update python -m keyring set matrix-premid @user:example.com + view python -m keyring get matrix-premid @user:example.com + remove python -m keyring del matrix-premid @user:example.com +""", + ) + parser.add_argument( + "command", + nargs="?", + choices=["install-service", "daemon", "shutdown", "set"], + help="Optional command (e.g., install-service, daemon, shutdown, set)", + ) + parser.add_argument( + "status_args", + nargs="*", + help="Status message for the 'set' command", + ) + parser.add_argument( + "--debug", action="store_true", help="Enable verbose debug logging" + ) + parser.add_argument( + "--version", + action="version", + version=f"%(prog)s {__version__}", + help="Show program's version number and exit", + ) + parser.add_argument( + "--unset", + "--clear", + action="store_true", + help="Manually clear status (Offline) and exit", + ) + parser.add_argument( + "--config", + default=os.environ.get( + "PREMID_CONFIG", + os.path.expanduser("~/.config/matrix-premid/config.json"), + ), + help="Path to config.json (default: ~/.config/matrix-premid/config.json, " + "or PREMID_CONFIG env var)", + ) + argcomplete.autocomplete(parser) + return parser.parse_args(args) + + +async def main(args=None): + """Start the Matrix updater.""" + # pylint: disable=too-many-statements,too-many-locals,too-many-branches + if args is None: + args = parse_args() + + if args.command == "install-service": # pragma: no cover + install_service() + return + + if args.debug: + logging.basicConfig(level=logging.DEBUG) + else: + logging.basicConfig(level=logging.CRITICAL) + + if not args.unset and not shutil.which("playerctl"): + print("ERROR: playerctl command not found. Please install it.", file=sys.stderr) + sys.exit(1) + + accounts = [] + idle_timeout = 15 + poll_interval = 5 + + # Load configuration + config_file = args.config + if not os.path.exists(config_file): + print( + "ERROR: Missing configuration. Please run 'matrix-premid install-service' " + "to create a config.json template.", + file=sys.stderr, + ) + sys.exit(1) + + with open(config_file, "r", encoding="utf-8") as f: + config = json.load(f) + GLOBAL_PROVIDERS.load(config.get("providers", {})) + accounts = config.get("accounts", []) + idle_timeout = config.get("idle_timeout", 15) + poll_interval = config.get("poll_interval", 5) + + if not accounts: + print("ERROR: No accounts defined in config.json.", file=sys.stderr) + sys.exit(1) + + # Resolve tokens from keyring if missing + for account in accounts: + if not account.get("access_token"): + token = keyring.get_password("matrix-premid", account["username"]) + if token: + account["access_token"] = token + else: + print( + f"ERROR: Missing access token for {account['username']}. " + f"Set it using: {sys.executable} -m keyring set matrix-premid " + f"{account['username']}", + file=sys.stderr, + ) + sys.exit(1) + + updaters = [] + is_debug = args.debug + for account in accounts: + updaters.append( + MatrixStatusUpdater( + account["homeserver"], + account["username"], + account["access_token"], + device_id=account.get("device_id", ""), + enabled=account.get("enabled", True), + idle_timeout=idle_timeout, + poll_interval=poll_interval, + verbose=is_debug, + ) + ) + + if args.command == "set": + status_msg = " ".join(args.status_args) + if not status_msg: + print("ERROR: 'set' command requires a status message.", file=sys.stderr) + for u in updaters: + await u.close() + sys.exit(1) + + print( + f"Setting status to: '{status_msg}' for {len(updaters)} accounts...", + flush=True, + ) + try: + # We call send_update directly to ensure it finishes before we exit + await asyncio.wait_for( + asyncio.gather(*(u.send_update(status_msg) for u in updaters)), + timeout=10.0, + ) + print("Successfully updated status.") + except Exception as e: # pylint: disable=broad-exception-caught + print(f"ERROR: manual set failed: {e}", file=sys.stderr) + + for u in updaters: + await u.close() + return + + if args.unset: + print( + f"Manual status clear requested (Offline) for {len(updaters)} accounts...", + flush=True, + ) + try: + # Update all accounts to idle concurrently + await asyncio.wait_for( + asyncio.gather( + *(u.update("", force=True, is_exit=True) for u in updaters) + ), + timeout=10.0, + ) + print(f"Successfully cleared status for {len(updaters)} accounts.") + except asyncio.CancelledError: + pass + except Exception as e: # pylint: disable=broad-exception-caught + print(f"ERROR: Manual clear failed: {e}", file=sys.stderr) + + for u in updaters: + await u.close() + return + + # pylint: disable=unused-variable + lock_fd = acquire_lock() # noqa: F841 + + users_str = ", ".join([a["username"] for a in accounts]) + print(f"Matrix Users: {users_str}", flush=True) + + shutdown_event = asyncio.Event() + loop = asyncio.get_running_loop() + + def signal_handler(): + print("\nInitiating graceful shutdown...", flush=True) + shutdown_event.set() + + try: + loop.add_signal_handler(signal.SIGINT, signal_handler) + loop.add_signal_handler(signal.SIGTERM, signal_handler) + except NotImplementedError: # pragma: no cover + pass + + async def keep_alive(updater: MatrixStatusUpdater): + """Keep status online via periodic presence updates.""" + backoff = 5 + while not shutdown_event.is_set(): + try: + # Instead of syncing, we just push a presence update periodically + # to stay 'online' in the eyes of the server. + await updater.send_update(updater.last_activity) + + # Wait for a while before the next refresh + # typically Matrix presence expires in 5-15 minutes if not refreshed + # but we'll be more aggressive to ensure status visibility. + for _ in range(20): # 20 seconds + if shutdown_event.is_set(): + break + await asyncio.sleep(1) + backoff = 5 + except asyncio.CancelledError: + break + except Exception as e: # pylint: disable=broad-exception-caught + print( + f"ERROR: keep-alive exception ({updater.username}): {e}", + file=sys.stderr, + ) + await asyncio.sleep(backoff) + backoff = min(backoff * 2, 300) + + print("Listening for MPRIS events...", flush=True) + + # Run tasks in background + tasks = [asyncio.create_task(monitor_mpris(updaters, poll_interval, config_file))] + for u in updaters: + tasks.append(asyncio.create_task(keep_alive(u))) + + # Wait for a shutdown signal + await shutdown_event.wait() + + # Cancel background tasks + for t in tasks: + t.cancel() + + # Wait for tasks to acknowledge cancellation + await asyncio.gather(*tasks, return_exceptions=True) + + print( + f"Clearing Matrix status before exit for {len(updaters)} accounts...", + flush=True, + ) + try: + await asyncio.wait_for( + asyncio.gather(*(u.update("", force=True, is_exit=True) for u in updaters)), + timeout=5.0, + ) + except ( # pylint: disable=broad-exception-caught + Exception, + asyncio.CancelledError, + ): + pass + + for u in updaters: + await u.close() + + # Clean up lock file so next start works cleanly + try: + os.unlink(LOCK_FILE) + except OSError: + pass + + print("Done.") + + +def daemonize(): # pragma: no cover + """Fork the process to run in the background.""" + try: + pid = os.fork() + if pid > 0: + sys.exit(0) + except OSError as e: + print(f"Fork #1 failed: {e}", file=sys.stderr) + sys.exit(1) + + os.chdir("/") + os.setsid() + os.umask(0) + + try: + pid = os.fork() + if pid > 0: + sys.exit(0) + except OSError as e: + print(f"Fork #2 failed: {e}", file=sys.stderr) + sys.exit(1) + + sys.stdout.flush() + sys.stderr.flush() + try: + nul_r = os.open(os.devnull, os.O_RDWR) + os.dup2(nul_r, sys.stdin.fileno()) + os.dup2(nul_r, sys.stdout.fileno()) + os.dup2(nul_r, sys.stderr.fileno()) + except Exception: # pylint: disable=broad-exception-caught + pass + + +def shutdown_daemon(): # pragma: no cover + """Send SIGTERM to the running background daemon.""" + try: + with open(LOCK_FILE, "r", encoding="utf-8") as f: + pid = int(f.read().strip()) + os.kill(pid, signal.SIGTERM) + print(f"Sent shutdown signal to matrix-premid daemon (PID {pid}).") + except FileNotFoundError: + print("No matrix-premid daemon is currently running (lock file not found).") + except ProcessLookupError: + print("Daemon is not running (PID not found).") + except ValueError: + print("Invalid PID in lock file.") + except Exception as e: # pylint: disable=broad-exception-caught + print(f"Failed to shutdown daemon: {e}") + + +def cli(): + """Synchronous entry point for the package.""" + args = parse_args() + + if args.command == "shutdown": # pragma: no cover + shutdown_daemon() + return + + if args.command == "daemon": # pragma: no cover + print("Starting matrix-premid in the background...") + daemonize() + + try: + asyncio.run(main(args)) + except (KeyboardInterrupt, SystemExit, asyncio.CancelledError): + pass + + +if __name__ == "__main__": + cli() diff --git a/tests/test_parser.py b/tests/test_parser.py index 2a31c23..d5c9e9f 100644 --- a/tests/test_parser.py +++ b/tests/test_parser.py @@ -1,9 +1,21 @@ -"""Unit tests for the project.""" +# -*- coding: utf-8 -*- +""" +Created on Sun Mar 29 00:33:45 2026 -from matrix_premid import SEP_STR, _get_best_mpris_activity, parse_mpris_data +@author: shane +Unit tests for the project. +""" # pylint: disable=missing-docstring,line-too-long +from matrix_premid.__main__ import SEP_STR, _get_best_mpris_activity, parse_mpris_data + + +def test_get_best_mpris_activity_idle(): + assert _get_best_mpris_activity([]) == ("", "") + assert _get_best_mpris_activity(["", " "]) == ("", "") + assert _get_best_mpris_activity(["Invalid Line"]) == ("", "") + def test_parse_mpris_data_playing_song_with_artist(): raw = f"Playing{SEP_STR}Sea Of Feelings{SEP_STR}LIONE{SEP_STR}firefox" @@ -13,13 +25,16 @@ def test_parse_mpris_data_playing_song_with_artist(): def test_parse_mpris_data_playing_youtube_music_suffix(): - raw = f"Playing{SEP_STR}Sea Of Feelings - YouTube Music{SEP_STR}{SEP_STR}firefox" - activity, title = parse_mpris_data(raw, "YouTube Music") + url = "https://music.youtube.com/watch?v=123" + content = f"Sea Of Feelings - YouTube Music{SEP_STR}{SEP_STR}firefox{SEP_STR}" + raw = f"Playing{SEP_STR}{content}{url}" + activity, title = parse_mpris_data(raw, "YouTube Music", url) assert activity == "Listening to: Sea Of Feelings | YouTube Music" assert title == "Sea Of Feelings" def test_parse_mpris_data_paused_song(): + raw = f"Paused{SEP_STR}Sea Of Feelings{SEP_STR}LIONE{SEP_STR}firefox" activity, title = parse_mpris_data(raw) assert activity == "Paused: Sea Of Feelings - LIONE" @@ -33,37 +48,176 @@ def test_parse_mpris_data_html_entities(): assert title == "Princess Chelsea & Friends" -def test_get_best_mpris_activity_ignores_idle_youtube_music_when_paused(): +def test_parse_mpris_data_youtube_url_detection(): + # Test that even without "YouTube" in title, URL detects it + url = "https://www.youtube.com/watch?v=abc" + raw = ( + f"Playing{SEP_STR}Cool Video{SEP_STR}Cool Channel{SEP_STR}firefox{SEP_STR}{url}" + ) + activity, title = parse_mpris_data(raw, url=url) + assert activity == "Watching: Cool Video - Cool Channel | YouTube" + assert title == "Cool Video" + + +def test_get_best_mpris_activity_prevents_poisoning(): + # YouTube Music is Paused, YouTube is Playing + # In the old code, both would be identified as "YouTube Music" + url_yt = "https://youtube.com/watch?v=1" + url_ym = "https://music.youtube.com/watch?v=2" + lines = [ + f"Playing{SEP_STR}Cool Video - YouTube{SEP_STR}{SEP_STR}firefox{SEP_STR}{url_yt}", + f"Paused{SEP_STR}Some Song{SEP_STR}Artist{SEP_STR}firefox{SEP_STR}{url_ym}", + ] + activity, title = _get_best_mpris_activity(lines) + assert activity == "Watching: Cool Video | YouTube" + assert title == "Cool Video" + + +def test_get_best_mpris_activity_prioritizes_paused_over_idle(): + url = "https://music.youtube.com/watch?v=3" lines = [ - f"Playing{SEP_STR}YouTube Music{SEP_STR}{SEP_STR}firefox", - f"Paused{SEP_STR}Awesome Song{SEP_STR}Awesome Artist{SEP_STR}plasma-browser-integration", # noqa: E501 + f"Playing{SEP_STR}YouTube Music{SEP_STR}{SEP_STR}firefox{SEP_STR}{url}", + ( + f"Paused{SEP_STR}Awesome Song{SEP_STR}Awesome Artist" + f"{SEP_STR}plasma-integration{SEP_STR}" + ), # noqa: E501 ] activity, title = _get_best_mpris_activity(lines) - # The new behavior properly drops Paused songs so we get a clean Idle state - assert activity == "Idle" - assert title == "" + # The new behavior properly boosts Paused songs over empty Idle + assert activity == "Paused: Awesome Song - Awesome Artist | YouTube Music" + assert title == "Awesome Song" def test_get_best_mpris_activity_picks_highest_quality(): lines = [ - f"Playing{SEP_STR}YouTube Music{SEP_STR}{SEP_STR}firefox", - f"Playing{SEP_STR}Awesome Song{SEP_STR}Awesome Artist{SEP_STR}firefox", - f"Playing{SEP_STR}Basic Song Without Artist{SEP_STR}{SEP_STR}firefox", + f"Playing{SEP_STR}YouTube Music{SEP_STR}{SEP_STR}firefox{SEP_STR}", + f"Playing{SEP_STR}Awesome Song{SEP_STR}Awesome Artist{SEP_STR}firefox{SEP_STR}", + f"Playing{SEP_STR}Basic Song Without Artist{SEP_STR}{SEP_STR}firefox{SEP_STR}", ] activity, title = _get_best_mpris_activity(lines) - # The Awesome Song has an artist, giving it quality=20+1=21 - assert activity == "Listening to: Awesome Song - Awesome Artist | YouTube Music" + # The Awesome Song has an artist, giving it quality=20 + assert "Awesome Song" in activity + assert "Awesome Artist" in activity assert title == "Awesome Song" def test_get_best_mpris_activity_inherits_youtube_music_across_players(): """Test that rich players without YT Music inherit the tag from other tabs.""" lines = [ - f"Playing{SEP_STR}Eyes on Fire (Zeds Dead remix) | YouTube Music{SEP_STR}{SEP_STR}firefox", # noqa: E501 - f"Playing{SEP_STR}Eyes on Fire (Zeds Dead remix){SEP_STR}Blue Foundation{SEP_STR}plasma-browser-integration", # noqa: E501 + f"Playing{SEP_STR}Eyes on Fire (Zeds Dead remix) | YouTube Music{SEP_STR}{SEP_STR}firefox{SEP_STR}", # noqa: E501 + f"Playing{SEP_STR}Eyes on Fire (Zeds Dead remix){SEP_STR}Blue Foundation{SEP_STR}plasma-browser-integration{SEP_STR}", # noqa: E501 ] activity, title = _get_best_mpris_activity(lines) assert activity == ( "Listening to: Eyes on Fire (Zeds Dead remix) - Blue Foundation | YouTube Music" # noqa: E501 ) assert title == "Eyes on Fire (Zeds Dead remix)" + + +def test_parse_mpris_data_youtube(): + url = "https://youtube.com/watch?v=v" + raw = f"Playing{SEP_STR}Some Video{SEP_STR}{SEP_STR}firefox{SEP_STR}{url}" + act, title = parse_mpris_data(raw, "YouTube", url=url) + assert act == "Watching: Some Video | YouTube" + assert title == "Some Video" + + +def test_parse_mpris_data_netflix_url_detection(): + url = "https://www.netflix.com/watch/123" + raw = f"Playing{SEP_STR}Stranger Things{SEP_STR}{SEP_STR}firefox{SEP_STR}{url}" + activity, title = parse_mpris_data(raw, url=url) + assert activity == "Watching: Stranger Things | Netflix" + assert title == "Stranger Things" + + +def test_parse_mpris_data_twitch_url_detection(): + url = "https://www.twitch.tv/some_streamer" + raw = f"Playing{SEP_STR}Some Stream{SEP_STR}{SEP_STR}firefox{SEP_STR}{url}" + activity, title = parse_mpris_data(raw, url=url) + assert activity == "Watching: Some Stream | Twitch" + assert title == "Some Stream" + + +def test_parse_mpris_data_netflix(): + raw = f"Playing{SEP_STR}Stranger Things{SEP_STR}{SEP_STR}firefox" + act, title = parse_mpris_data(raw, "Netflix") + assert act == "Watching: Stranger Things | Netflix" + assert title == "Stranger Things" + + +def test_parse_mpris_data_array_artist(): + raw = f"Playing{SEP_STR}Song Title{SEP_STR}['Artist 1', 'Artist 2']{SEP_STR}firefox" + act, title = parse_mpris_data(raw) + assert act == "Listening to: Song Title - Artist 1, Artist 2" + assert title == "Song Title" + + +def test_parse_mpris_data_with_timestamp(): + # 36 seconds / 5 minutes 44 seconds + pos = str(36 * 1_000_000) + length = str((5 * 60 + 44) * 1_000_000) + raw = ( + f"Playing{SEP_STR}Song Title{SEP_STR}Artist{SEP_STR}player" + f"{SEP_STR}url{SEP_STR}{pos}{SEP_STR}{length}" + ) + activity, _ = parse_mpris_data(raw) + assert activity == "Listening to: Song Title - Artist [0:36 / 5:44]" + + +def test_parse_mpris_data_with_position_only(): + # 1 minute 23 seconds, no length (e.g. live stream) + pos = str(83 * 1_000_000) + raw = ( + f"Playing{SEP_STR}Live Stream{SEP_STR}Broadcaster{SEP_STR}player" + f"{SEP_STR}url{SEP_STR}{pos}" + ) + activity, _ = parse_mpris_data(raw) + assert activity == "Listening to: Live Stream - Broadcaster [1:23]" + + +def test_parse_mpris_data_with_hours(): + # 1 hour 2 minutes 3 seconds / 2 hours + pos = str((1 * 3600 + 2 * 60 + 3) * 1_000_000) + length = str(2 * 3600 * 1_000_000) + raw = ( + f"Playing{SEP_STR}Long Podcast{SEP_STR}Host{SEP_STR}player" + f"{SEP_STR}url{SEP_STR}{pos}{SEP_STR}{length}" + ) + activity, _ = parse_mpris_data(raw) + assert activity == "Listening to: Long Podcast - Host [1:02:03 / 2:00:00]" + + +def test_parse_mpris_data_with_timestamp_and_provider(): + pos = str(36 * 1_000_000) + length = str((5 * 60 + 44) * 1_000_000) + raw = ( + f"Playing{SEP_STR}Song Title{SEP_STR}Artist{SEP_STR}spotify" + f"{SEP_STR}https://open.spotify.com/track/123{SEP_STR}{pos}{SEP_STR}{length}" + ) + # The provider "Spotify" is detected from url or raw playerName + activity, _ = parse_mpris_data(raw, global_provider="Spotify") + assert activity == "Listening to: Song Title - Artist [0:36 / 5:44] | Spotify" + + +def test_parse_mpris_data_invalid_timestamp(): + raw = ( + f"Playing{SEP_STR}Song{SEP_STR}Artist{SEP_STR}player{SEP_STR}url" + f"{SEP_STR}invalid{SEP_STR}data" + ) + activity, _ = parse_mpris_data(raw) + assert activity == "Listening to: Song - Artist" + + +def test_get_best_mpris_activity_quality_scoring(): + """Test that time and artist presence correctly calculate quality score.""" + lines = [ + f"Paused{SEP_STR}Song{SEP_STR}Artist{SEP_STR}firefox{SEP_STR}", + f"Playing{SEP_STR}Song{SEP_STR}Artist{SEP_STR}firefox" + f"{SEP_STR}url{SEP_STR}1000000{SEP_STR}5000000", + f"Playing{SEP_STR}Song{SEP_STR}{SEP_STR}firefox{SEP_STR}", + f"Unknown{SEP_STR}Song{SEP_STR}{SEP_STR}firefox{SEP_STR}", + ] + activity, title = _get_best_mpris_activity(lines) + assert title == "Song" + assert "Artist" in activity + assert "[0:01 / 0:05]" in activity diff --git a/tests/test_provider_config.py b/tests/test_provider_config.py new file mode 100644 index 0000000..0c5f479 --- /dev/null +++ b/tests/test_provider_config.py @@ -0,0 +1,78 @@ +# -*- coding: utf-8 -*- +""" +Created on Sun Mar 29 00:33:45 2026 + +@author: shane +Test the configuration parsing logic. +""" + +# pylint: disable=missing-function-docstring + +from matrix_premid.__main__ import ProviderConfig + + +def test_load_custom_providers(): + pc = ProviderConfig() + custom_cfg = { + "Custom": { + "regex": "custom_app", + "priority": 150, + "is_video": False, + "enabled": True, + "template": "Hello {title}", + }, + "YouTube Music": {"enabled": False}, + } + pc.load(custom_cfg) + + # Custom exists + assert "Custom" in pc.providers + custom_prov = pc.providers["Custom"] + assert custom_prov.priority == 150 + assert custom_prov.pattern is not None + assert custom_prov.template == "Hello {title}" + + # YouTube Music was overridden + assert pc.providers["YouTube Music"].enabled is False + + +def test_load_invalid_regex_disables_provider(): + pc = ProviderConfig() + custom_cfg = {"Broken": {"regex": "[unclosed bracket", "enabled": True}} + pc.load(custom_cfg) + + assert "Broken" in pc.providers + assert pc.providers["Broken"].enabled is False + assert pc.providers["Broken"].pattern is None + + +def test_match_provider_highest_priority(): + pc = ProviderConfig() + custom_cfg = { + "Low Priority": {"regex": "shared", "priority": 10, "enabled": True}, + "High Priority": {"regex": "shared", "priority": 100, "enabled": True}, + "Disabled High Priority": { + "regex": "shared", + "priority": 200, + "enabled": False, + }, + } + pc.load(custom_cfg) + + match = pc.match_provider("testing shared player") + assert match is not None + assert match.name == "High Priority" + + +def test_match_provider_no_match(): + pc = ProviderConfig() + pc.load({}) + match = pc.match_provider("something completely random") + assert match is None + + +def test_match_provider_empty_string(): + pc = ProviderConfig() + pc.load({}) + match = pc.match_provider("") + assert match is None diff --git a/tests/test_updater.py b/tests/test_updater.py index 046667c..be5bc93 100644 --- a/tests/test_updater.py +++ b/tests/test_updater.py @@ -1,177 +1,564 @@ -"""Tests for MatrixStatusUpdater and monitor_mpris.""" +# -*- coding: utf-8 -*- +""" +Created on Sun Mar 29 00:33:45 2026 + +@author: shane +Tests for MatrixStatusUpdater and monitor_mpris. +""" + +# pylint: disable=protected-access,no-member,redefined-outer-name,broad-exception-caught import asyncio -from unittest.mock import AsyncMock, patch +import logging +from unittest.mock import AsyncMock, MagicMock, patch import pytest +import pytest_asyncio # pylint: disable=import-error + +from matrix_premid.__main__ import ( + SEP_STR, + MatrixStatusUpdater, + install_service, + main, + monitor_mpris, +) + + +class FakeResponse: + """Fake aiohttp response.""" + + def __init__(self, status=200, text="OK", side_effect=None): + self.status = status + self._text = text + self._side_effect = side_effect + + async def text(self): + """Mock text() method.""" + if self._side_effect: + raise self._side_effect + return self._text + + async def __aenter__(self): + return self + + async def __aexit__(self, exc_type, exc_val, exc_tb): + pass + + +class FakeSession: + """Fake aiohttp session.""" + + def __init__(self): + self.puts = [] + self.closed = False + self.put_side_effect = None + + def put(self, url, **kwargs): + """Mock put() method (synchronous returning an async context manager).""" + if self.put_side_effect: + if isinstance(self.put_side_effect, Exception): + raise self.put_side_effect + return self.put_side_effect + self.puts.append((url, kwargs)) + return FakeResponse() -from matrix_premid import SEP_STR, MatrixStatusUpdater, main, monitor_mpris + async def close(self): + """Mock close() method.""" + self.closed = True @pytest.fixture(autouse=True) def patch_sleep(): - """Bypass asyncio.sleep delays globally for fast test execution.""" - with patch("asyncio.sleep", new_callable=AsyncMock) as m: - yield m + """Bypass asyncio.sleep delays globally.""" + + async def dummy_sleep(_delay, *_args, **_kwargs): + return + + with patch("matrix_premid.__main__.asyncio.sleep", side_effect=dummy_sleep): + yield + + +@pytest_asyncio.fixture +async def matrix_updater_obj(): + """Fixture to provide a MatrixStatusUpdater with a FakeSession.""" + session = FakeSession() + u = MatrixStatusUpdater("http://mock", "@test:mock", "tok", "dev", session=session) + try: + yield u + finally: + if u._update_task and not u._update_task.done(): + u._update_task.cancel() + try: + await u._update_task + except (asyncio.CancelledError, Exception): + pass + await u.close() def test_updater_init(): - """Test the updater initializes the nio client correctly.""" - with patch("matrix_premid.AsyncClient") as mock_client: - mock_client.return_value = AsyncMock() - updater = MatrixStatusUpdater("http://mock", "@test:mock", "tok", "dev") - mock_client.assert_called_with("http://mock", "@test:mock") - assert updater.client.access_token == "tok" + """Test the updater initializes correctly.""" + u = MatrixStatusUpdater("http://mock", "@test:mock", "tok", "dev") + assert u.homeserver == "http://mock" + assert u.username == "@test:mock" + assert u.access_token == "tok" + assert u.device_id == "dev" @pytest.mark.asyncio -async def test_updater_update(): +async def test_updater_update(matrix_updater_obj): """Test pushing presence state to the Matrix room.""" - with patch("matrix_premid.AsyncClient") as mock_client: - mock_client.return_value = AsyncMock() - updater = MatrixStatusUpdater("http://mock", "@test:mock", "tok", "dev") - await updater.update("Listening to: Song | YT Music") - updater.client.set_presence.assert_awaited_with( - presence="online", status_msg="Listening to: Song | YT Music" - ) + await matrix_updater_obj.update("Listening to: Song | YT Music") + if matrix_updater_obj._update_task: + await matrix_updater_obj._update_task + + assert len(matrix_updater_obj._session.puts) >= 1 + # Check if presence URL was used + urls = [p[0] for p in matrix_updater_obj._session.puts] + assert any("presence" in url for url in urls) @pytest.mark.asyncio -async def test_updater_update_paused(): +async def test_updater_update_paused(matrix_updater_obj): """Test a paused song yields presence correctly.""" - with patch("matrix_premid.AsyncClient") as mock_client: - mock_client.return_value = AsyncMock() - updater = MatrixStatusUpdater("mock", "mock", "mock") - await updater.update("Paused: Song - Artist | YT Music") - updater.client.set_presence.assert_awaited() + await matrix_updater_obj.update("Paused: Song - Artist | YT Music") + if matrix_updater_obj._update_task: + await matrix_updater_obj._update_task + assert len(matrix_updater_obj._session.puts) >= 1 @pytest.mark.asyncio -async def test_updater_update_empty(): - """Test empty string correctly defaults to Idle status.""" - with patch("matrix_premid.AsyncClient") as mock_client: - mock_client.return_value = AsyncMock() - updater = MatrixStatusUpdater("http://mock", "mock", "mock") - await updater.update("") - updater.client.set_presence.assert_awaited_with( - presence="online", status_msg="Idle" - ) +async def test_updater_update_empty(matrix_updater_obj): + """Test empty string is ignored by default (new behavior).""" + await matrix_updater_obj.update("") + assert matrix_updater_obj._update_task is None @pytest.mark.asyncio -async def test_updater_update_other(): +async def test_updater_update_other(matrix_updater_obj): """Test non-music activities receive correct base quality attributes.""" - with patch("matrix_premid.AsyncClient") as mock_client: - mock_client.return_value = AsyncMock() - updater = MatrixStatusUpdater("http://mock", "mock", "mock") - await updater.update("Watching: Movie", title="Movie") - updater.client.set_presence.assert_awaited() + await matrix_updater_obj.update("Watching: Movie", title="Movie") + if matrix_updater_obj._update_task: + await matrix_updater_obj._update_task + assert len(matrix_updater_obj._session.puts) >= 1 @pytest.mark.asyncio -async def test_updater_update_exception(): - """Test updating surviving network timeouts bounds.""" - with patch("matrix_premid.AsyncClient") as mock_client: - mock_client.return_value = AsyncMock() - updater = MatrixStatusUpdater("mock", "mock", "mock") - updater.client.set_presence.side_effect = asyncio.TimeoutError() - await updater.update("Listening to: Song") - updater.client.set_presence.assert_awaited() +async def test_updater_update_exception(matrix_updater_obj, capsys): + """Test updating surviving network errors.""" + matrix_updater_obj._session.put_side_effect = Exception("Network Error") + matrix_updater_obj.verbose = True + + await matrix_updater_obj.update("Listening to: Song") + if matrix_updater_obj._update_task: + await matrix_updater_obj._update_task + + _, err = capsys.readouterr() + assert "presence error: Network Error" in err @pytest.mark.asyncio -async def test_updater_update_same_song_ignored(): +async def test_updater_update_same_song_ignored(matrix_updater_obj): """Test ignoring unchanged song status strings.""" - with patch("matrix_premid.AsyncClient") as mock_client: - mock_client.return_value = AsyncMock() - updater = MatrixStatusUpdater("http://mock", "@test:mock", "tok", "dev") - updater.last_activity = "Listening to: Song" - updater.last_title = "Song" - await updater.update("Listening to: Song", title="Song") - updater.client.set_presence.assert_not_called() + matrix_updater_obj.last_activity = "Listening to: Song" + matrix_updater_obj.last_title = "Song" + await matrix_updater_obj.update("Listening to: Song", title="Song") + assert matrix_updater_obj._update_task is None @pytest.mark.asyncio -@patch("matrix_premid.asyncio.create_subprocess_exec") -async def test_monitor_mpris_picks_best_activity(mock_exec): +@patch("matrix_premid.__main__.asyncio.create_subprocess_exec") +async def test_monitor_mpris_picks_best_activity(mock_exec, matrix_updater_obj): """Test MPRIS subprocess parsing defaults best output cleanly.""" - mock_proc = AsyncMock() - mock_proc.communicate.side_effect = [ - ( - f"Playing{SEP_STR}Awesome Song{SEP_STR}" - f"Awesome Artist{SEP_STR}firefox\n".encode("utf-8"), - b"", - ), - Exception("Break loop"), - ] + mock_proc = MagicMock() + mock_proc.communicate = AsyncMock( + side_effect=[ + ( + f"Playing{SEP_STR}Awesome Song{SEP_STR}" + f"Awesome Artist{SEP_STR}firefox\n".encode("utf-8"), + b"", + ), + # Break the loop + Exception("Break loop"), + ] + ) mock_exec.return_value = mock_proc - with patch("matrix_premid.AsyncClient") as mock_client: - mock_client.return_value = AsyncMock() - updater = MatrixStatusUpdater("mock", "mock", "mock") - updater.update = AsyncMock() - try: - await monitor_mpris(updater) - except Exception: # pylint: disable=broad-exception-caught - pass - updater.update.assert_awaited_with( - "Listening to: Awesome Song - Awesome Artist", title="Awesome Song" - ) + matrix_updater_obj.update = AsyncMock() + try: + await monitor_mpris([matrix_updater_obj], 5, "mock_config.json") + except Exception: + pass + # Get the actual calls + calls = matrix_updater_obj.update.await_args_list + assert len(calls) > 0 + # ensure it was called with roughly the title + args, _kwargs = calls[0] + assert "Awesome Song" in args[0] @pytest.mark.asyncio -@patch("matrix_premid.sys.exit") -@patch("matrix_premid.acquire_lock") -@patch("matrix_premid.HOMESERVER", None) -@patch("matrix_premid.USERNAME", None) -@patch("matrix_premid.ACCESS_TOKEN", None) -@patch("matrix_premid.AsyncClient") -async def test_main_missing_env(_mock_client, _mock_lock, mock_exit): - """Test main script breaks when Env details are lacking.""" - mock_exit.side_effect = SystemExit() - try: - await main() - except SystemExit: - pass +@patch("matrix_premid.__main__.sys.exit", side_effect=SystemExit) +@patch("matrix_premid.__main__.shutil.which", return_value="/usr/bin/playerctl") +@patch("matrix_premid.__main__.os.path.exists", return_value=False) +async def test_main_missing_env(_mock_exists, _mock_which, mock_exit): + """Test main script breaks when config is lacking.""" + with patch("sys.argv", ["matrix_premid.py"]): + with pytest.raises(SystemExit): + await main() mock_exit.assert_called_with(1) @pytest.mark.asyncio -@patch("matrix_premid.sys.exit") -@patch("matrix_premid.acquire_lock") -@patch("matrix_premid.HOMESERVER", "mock") -@patch("matrix_premid.USERNAME", "@user") -@patch("matrix_premid.ACCESS_TOKEN", "tok") -@patch("matrix_premid.DEVICE_ID", "dev") -async def test_main_execution_mocked_gather(_mock_lock, mock_exit): +@patch("matrix_premid.__main__.sys.exit", side_effect=SystemExit) +@patch("matrix_premid.__main__.shutil.which", return_value="/usr/bin/playerctl") +@patch("matrix_premid.__main__.acquire_lock") +@patch("matrix_premid.__main__.os.path.exists", return_value=True) +@patch("matrix_premid.__main__.json.load") +@patch("matrix_premid.__main__.keyring.get_password", return_value="mock_token") +@patch("matrix_premid.__main__.open", new_callable=MagicMock) +async def test_main_execution_mocked_gather( + _mock_open, + _mock_keyring, + mock_json, + _mock_exists, + _mock_lock, + _mock_which, + mock_exit, +): """Test main entrypoint setups everything cleanly resolving without errors.""" - with patch("matrix_premid.AsyncClient") as mock_client: - mock_instance = AsyncMock() - mock_client.return_value = mock_instance - # Raise CancelledError to gracefully exit the infinite sync loop natively - mock_instance.sync.side_effect = asyncio.CancelledError() - - with patch( - "matrix_premid.MatrixStatusUpdater.update", new_callable=AsyncMock - ) as mock_update: - mock_update.side_effect = asyncio.CancelledError() - - with patch("matrix_premid.asyncio.create_subprocess_exec") as mock_exec: - mock_proc = AsyncMock() - mock_proc.communicate.side_effect = asyncio.CancelledError() + mock_json.return_value = { + "accounts": [{"homeserver": "mock", "username": "@user", "device_id": "dev"}] + } + + loop = asyncio.get_running_loop() + + def mock_create_task_side_effect(coro): + # We must return a real task or future bound to the loop + coro.close() + f = loop.create_future() + f.set_result(None) + return f + + # Mock the background tasks to return immediately + async def mock_bg_task(*_args, **_kwargs): + return + + with ( + patch( + "matrix_premid.__main__.asyncio.Event.wait", AsyncMock(return_value=None) + ), + patch( + "matrix_premid.__main__.asyncio.create_task", + side_effect=mock_create_task_side_effect, + ), + patch("matrix_premid.__main__.monitor_mpris", side_effect=mock_bg_task), + ): + # Mock MatrixStatusUpdater methods to ensure no real network or hangs + with ( + patch( + "matrix_premid.__main__.MatrixStatusUpdater.update", + AsyncMock(return_value=None), + ) as mock_update, + patch( + "matrix_premid.__main__.MatrixStatusUpdater.send_update", + AsyncMock(return_value=None), + ), + patch( + "matrix_premid.__main__.MatrixStatusUpdater.close", + AsyncMock(return_value=None), + ), + ): + with patch( + "matrix_premid.__main__.asyncio.create_subprocess_exec" + ) as mock_exec: + mock_proc = MagicMock() + mock_proc.communicate = AsyncMock(side_effect=asyncio.CancelledError()) mock_exec.return_value = mock_proc - await main() - # Use unused vars assertions to cleanly please pylint explicitly - mock_exit.assert_not_called() + with patch("sys.argv", ["matrix_premid.py"]): + await main() + assert mock_update.called + + mock_exit.assert_not_called() @pytest.mark.asyncio async def test_updater_close(): """Test shutdown cleanup of updater client sockets.""" - with patch("matrix_premid.AsyncClient") as mock_client: - mock_client.return_value = AsyncMock() - updater = MatrixStatusUpdater("http://mock", "@test:mock", "tok", "dev") - await updater.close() - # pylint: disable=no-member - updater.client.close.assert_awaited() + session = FakeSession() + u = MatrixStatusUpdater("http://mock", "@test:mock", "tok", "dev", session=session) + await u.close() + assert session.closed + + +@pytest.mark.asyncio +@patch("matrix_premid.__main__.sys.exit", side_effect=SystemExit) +@patch("matrix_premid.__main__.shutil.which", return_value="/usr/bin/playerctl") +@patch("matrix_premid.__main__.MatrixStatusUpdater") +@patch("matrix_premid.__main__.logging.basicConfig") +@patch("matrix_premid.__main__.acquire_lock") +@patch("matrix_premid.__main__.os.path.exists", return_value=True) +@patch("matrix_premid.__main__.open", new_callable=MagicMock) +@patch("matrix_premid.__main__.json.load") +@patch("matrix_premid.__main__.keyring.get_password", return_value="mock_token") +async def test_main_debug_flag( + _mock_keyring, + mock_json, + _mock_open, + _mock_exists, + _mock_lock, + mock_config_logger, + mock_updater_class, + _mock_which, + _mock_exit, +): + """Test the --debug flag in main sets log level.""" + mock_json.return_value = {"accounts": [{"homeserver": "mock", "username": "@user"}]} + + mock_updater = MagicMock() + mock_updater.send_update = AsyncMock() + mock_updater.update = AsyncMock(return_value=None) + mock_updater.close = AsyncMock() + mock_updater_class.return_value = mock_updater + _mock_lock.return_value = MagicMock() + + loop = asyncio.get_running_loop() + + def mock_create_task_side_effect(coro): + coro.close() + f = loop.create_future() + f.set_result(None) + return f + + # Mocking wait/create_task to exit immediately + with ( + patch( + "matrix_premid.__main__.asyncio.Event.wait", AsyncMock(return_value=None) + ), + patch( + "matrix_premid.__main__.asyncio.create_task", + side_effect=mock_create_task_side_effect, + ), + ): + with patch("sys.argv", ["matrix_premid.py", "--debug"]): + await main() + + mock_config_logger.assert_called_with(level=logging.DEBUG) + + +@pytest.mark.asyncio +@patch("matrix_premid.__main__.sys.exit", side_effect=SystemExit) +@patch("matrix_premid.__main__.shutil.which", return_value="/usr/bin/playerctl") +@patch("matrix_premid.__main__.MatrixStatusUpdater") +@patch("matrix_premid.__main__.os.path.exists", return_value=True) +@patch("matrix_premid.__main__.open", new_callable=MagicMock) +@patch("matrix_premid.__main__.json.load") +@patch("matrix_premid.__main__.keyring.get_password", return_value="mock_token") +async def test_main_set_command( + _mock_keyring, + mock_json, + _mock_open, + _mock_exists, + mock_updater_class, + _mock_which, + _mock_exit, +): + """Test the manual 'set' command in main.""" + mock_json.return_value = {"accounts": [{"homeserver": "mock", "username": "@user"}]} + + mock_updater = MagicMock() + mock_updater.send_update = AsyncMock() + mock_updater.close = AsyncMock() + mock_updater_class.return_value = mock_updater + + with patch("sys.argv", ["matrix_premid.py", "set", "Working", "Hard"]): + await main() + + mock_updater.send_update.assert_awaited_with("Working Hard") + mock_updater.close.assert_awaited() + + +@pytest.mark.asyncio +@patch("matrix_premid.__main__.sys.exit", side_effect=SystemExit) +@patch("matrix_premid.__main__.shutil.which", return_value="/usr/bin/playerctl") +@patch("matrix_premid.__main__.MatrixStatusUpdater") +@patch("matrix_premid.__main__.os.path.exists", return_value=True) +@patch("matrix_premid.__main__.open", new_callable=MagicMock) +@patch("matrix_premid.__main__.json.load") +@patch("matrix_premid.__main__.keyring.get_password", return_value="mock_token") +async def test_main_set_command_no_args( + _mock_keyring, + mock_json, + _mock_open, + _mock_exists, + mock_updater_class, + _mock_which, + mock_exit, + capsys, +): + """Test the 'set' command with no status message (error case).""" + mock_json.return_value = {"accounts": [{"homeserver": "mock", "username": "@user"}]} + + mock_updater = MagicMock() + mock_updater.close = AsyncMock() + mock_updater_class.return_value = mock_updater + + with patch("sys.argv", ["matrix_premid.py", "set"]): + with pytest.raises(SystemExit): + await main() + + mock_exit.assert_called_with(1) + _, err = capsys.readouterr() + assert "ERROR: 'set' command requires a status message." in err + + +@pytest.mark.asyncio +@patch("matrix_premid.__main__.sys.exit", side_effect=SystemExit) +@patch("matrix_premid.__main__.shutil.which", return_value="/usr/bin/playerctl") +@patch("matrix_premid.__main__.MatrixStatusUpdater") +@patch("matrix_premid.__main__.os.path.exists", return_value=True) +@patch("matrix_premid.__main__.open", new_callable=MagicMock) +@patch("matrix_premid.__main__.json.load") +@patch("matrix_premid.__main__.keyring.get_password", return_value="mock_token") +async def test_main_unset_flag( + _mock_keyring, + mock_json, + _mock_open, + _mock_exists, + mock_updater_class, + _mock_which, + _mock_exit, +): + """Test the manual --unset flag in main.""" + mock_json.return_value = {"accounts": [{"homeserver": "mock", "username": "@user"}]} + + mock_updater = MagicMock() + mock_updater.update = AsyncMock() + mock_updater.close = AsyncMock() + mock_updater_class.return_value = mock_updater + + with patch("sys.argv", ["matrix_premid.py", "--unset"]): + await main() + + mock_updater.update.assert_awaited_with("", force=True, is_exit=True) + mock_updater.close.assert_awaited() + + +@pytest.mark.asyncio +async def test_updater_update_lower_quality_ignored(matrix_updater_obj): + """Test that lower quality metadata is ignored for the same song.""" + matrix_updater_obj.last_title = "Song" + matrix_updater_obj.last_quality = 20 # High quality + await matrix_updater_obj.update( + "Listening to: Song", title="Song" # Low quality (10) + ) + assert matrix_updater_obj._update_task is None + + +@pytest.mark.asyncio +async def test_updater_update_resets_idle_strikes(matrix_updater_obj): + """Test that non-idle activity resets idle strikes.""" + matrix_updater_obj.idle_strikes = 5 + with patch.object(matrix_updater_obj, "send_update", AsyncMock()): + await matrix_updater_obj.update("Listening to: Song") + if matrix_updater_obj._update_task: + await matrix_updater_obj._update_task + assert matrix_updater_obj.idle_strikes == 0 + + +@pytest.mark.asyncio +async def test_updater_update_cancels_existing_task(matrix_updater_obj): + """Test that starting a new update cancels any pending debounced update.""" + mock_task = MagicMock() + mock_task.done.return_value = False + matrix_updater_obj._update_task = mock_task + + with patch.object(matrix_updater_obj, "send_update", AsyncMock()): + await matrix_updater_obj.update("Listening to: Song") + mock_task.cancel.assert_called_once() + + +@pytest.mark.asyncio +async def test_updater_update_exit_sends_immediately(matrix_updater_obj): + """Test that exiting sends the update immediately without debouncing.""" + with patch.object(matrix_updater_obj, "send_update", AsyncMock()) as mock_send: + await matrix_updater_obj.update("", is_exit=True) + mock_send.assert_awaited_once() + assert matrix_updater_obj._update_task is None + + +@pytest.mark.asyncio +async def test_send_update_failure_logs_error(matrix_updater_obj, capsys): + """Test that failed API requests log errors to stderr.""" + matrix_updater_obj.verbose = True + matrix_updater_obj._session.put_side_effect = FakeResponse( + status=400, text="Bad Request" + ) + + await matrix_updater_obj.send_update("Activity") + + _, err = capsys.readouterr() + assert "presence failed (400): Bad Request" in err + assert "account_data failed (400): Bad Request" in err + + +@pytest.mark.asyncio +async def test_send_update_exception_logs_debug(matrix_updater_obj, capsys): + """Test that exceptions during API requests log to stderr if verbose.""" + matrix_updater_obj.verbose = True + matrix_updater_obj._session.put_side_effect = Exception("Crash") + + await matrix_updater_obj.send_update("Activity") + + _, err = capsys.readouterr() + assert "presence error: Crash" in err + assert "account_data error: Crash" in err + + +@pytest.mark.asyncio +@patch("matrix_premid.__main__.asyncio.create_subprocess_exec") +async def test_monitor_mpris_error_handling(mock_exec, capsys): + """Test that monitor_mpris handles subprocess errors gracefully.""" + mock_exec.side_effect = OSError("Subprocess Error") + + async def dummy_sleep_cancel(*_args, **_kwargs): + raise asyncio.CancelledError() + + with patch("matrix_premid.__main__.asyncio.sleep", side_effect=dummy_sleep_cancel): + try: + await monitor_mpris([], 1, "mock_config.json") + except (asyncio.CancelledError, Exception): + pass + + _, err = capsys.readouterr() + assert "MPRIS Monitor Error: Subprocess Error" in err + + +@patch("matrix_premid.__main__.shutil.which", return_value="/usr/bin/matrix-premid") +@patch("matrix_premid.__main__.os.makedirs") +@patch("matrix_premid.__main__.open", new_callable=MagicMock) +@patch("matrix_premid.__main__.subprocess.run") +def test_install_service_flow(mock_run, _mock_open, _mock_makedirs, _mock_which): + """Test the installation flow for systemd service.""" + # Patch the open in the target module specifically + with patch("matrix_premid.__main__.open", new_callable=MagicMock): + install_service() + assert _mock_makedirs.called + assert mock_run.called + + +@pytest.mark.asyncio +async def test_send_update_uses_custom_presence(): + """Test that non-default current_presence propagates to the presence payload.""" + session = FakeSession() + u = MatrixStatusUpdater("http://mock", "@test:mock", "tok", "dev", session=session) + u.current_presence = "unavailable" + + await u.send_update("Listening to: Song") + + # Find the presence PUT call and verify the payload + presence_puts = [(url, kwargs) for url, kwargs in session.puts if "presence" in url] + assert len(presence_puts) == 1 + _, kwargs = presence_puts[0] + assert kwargs["json"]["presence"] == "unavailable" + await u.close()