diff --git a/CHANGELOG.md b/CHANGELOG.md index 8ef9131..a66b0f1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,6 +10,26 @@ The format is inspired by Keep a Changelog, and this project uses semantic versi ### Added +- Nothing yet. + +### Changed + +- Nothing yet. + +### Fixed + +- Nothing yet. + + +## [1.2.0] - 2026-09-02 + +### Added + +- Added optional radio stream recording to the Textual TUI and GTK4 desktop GUI. +- Added FFmpeg-based stream-copy recording with Matroska audio output and no transcoding. +- Added profile-scoped recording metadata persistence in SQLite, including station, + timestamps, logical duration, file size, status and output path. +- Added safe, unique recording filenames under the FluxTuner data directory. - Added automatic Linux system tray integration for the GTK frontend using StatusNotifierItem and DBusMenu over Gio/D-Bus, without a new runtime dependency. - Added tray actions for restoring the main window, showing the current station, @@ -18,16 +38,29 @@ The format is inspired by Keep a Changelog, and this project uses semantic versi ### Changed -- GTK closing behavior now hides the main window and keeps playback running - when a compatible Linux StatusNotifierWatcher is available. +- Recording and playback are independent: a selected station can be recorded without + being played, and playback changes do not interrupt an active recording. +- FFmpeg recording input is paced in real time so live/HLS sources are not consumed + faster than wall-clock time. +- Active recordings are finalized cleanly when the TUI or GTK application exits. +- GTK closing behavior now hides the main window and keeps playback and recording + running when a compatible Linux StatusNotifierWatcher is available. - GTK falls back to the traditional close-and-exit behavior when tray registration is unavailable. ### Fixed -- Nothing yet. +- Fixed live streams that could produce substantially more recorded audio than the + user-requested wall-clock recording time. + +### Documentation + +- Documented local recording requirements, storage location and current TUI/GTK scope. ### Internal +- Added recording lifecycle, FFmpeg backend, persistence, path-generation, TUI and GTK + recording coverage. +- Validated stream-copy recording with MP3, AAC, Opus and HLS sources. - Added isolated Linux tray backend coverage, packaged-icon checks and GTK close-to-tray lifecycle coverage. - Tray lifetime uses `GApplication.hold()` / `release()` and a dedicated D-Bus connection so notifier removal is clean and does not affect GTK's shared bus. diff --git a/README.md b/README.md index 2fffaef..f781798 100644 --- a/README.md +++ b/README.md @@ -42,6 +42,7 @@ Run the Web/server mode on your own infrastructure and keep accounts, favorites, - Switch built-in TUI themes with live preview. - Run the default Textual TUI, GTK4 desktop GUI, legacy numbered CLI or browser-based web/server mode. - On Linux, keep the GTK GUI available from a StatusNotifierItem-compatible system tray while the main window is hidden. +- Record the currently selected station from the TUI or GTK GUI with FFmpeg stream copy, independently from playback. - Use Web/server accounts with first-run admin setup, pending account requests, authenticated profiles, CSRF-protected mutations, dashboard metrics and admin user management. - Store library data in a local SQLite database, with XDG-style config, data and cache locations. @@ -96,6 +97,7 @@ Run FluxTuner on your own server and access your radio library from any modern b - Python 3.11+ - `mpv` recommended, `ffmpeg` / `ffplay` as broad fallback, or optional lightweight `mpg123` / `ogg123` backends +- `ffmpeg` is required for local stream recording - Optional GUI dependencies: GTK4 and PyGObject ### Run the Web platform with Docker Compose @@ -203,6 +205,8 @@ Then launch the GUI mode using the documented FluxTuner GUI option. On Linux, FluxTuner GTK automatically registers a StatusNotifierItem when the desktop session provides a compatible watcher. When available, closing the window hides FluxTuner while playback continues; use the tray icon to restore the window, stop playback or quit the application. No additional Python dependency is required. If no compatible tray watcher is available, FluxTuner keeps the normal close-window behavior. +Local recording is available from the Textual TUI and GTK GUI when `ffmpeg` is available. Recording is independent from playback and writes Matroska audio (`.mka`) files under `~/.local/share/fluxtuner/recordings/` using stream copy, preserving the source codec without transcoding. The Web/server interface does not expose recording in this release. + This method is useful for testing a release quickly. For regular use, prefer the packaged installation methods when available. ### Launch modes diff --git a/SMOKE_TEST.md b/SMOKE_TEST.md index 4dc38d4..5bb8f86 100644 --- a/SMOKE_TEST.md +++ b/SMOKE_TEST.md @@ -104,6 +104,9 @@ Expected: - minimum bitrate filter works - playback starts with selected station - playback stops cleanly +- if `ffmpeg` is available, `R` starts/stops recording of the selected station +- recording continues when playback starts, stops or changes station +- the completed `.mka` is playable and has approximately the requested wall-clock duration - data usage updates while streaming Optional explicit backend tests: @@ -129,7 +132,12 @@ Expected: - backend is displayed in the side panel - playback starts with selected station - playback stops cleanly -- closing the window stops playback +- if `ffmpeg` is available, Record / Stop recording records the selected station +- recording continues independently while playback starts, stops or changes station +- with a compatible StatusNotifierWatcher, closing the window hides GTK and keeps playback and recording running +- restoring the window from the tray returns to the same application state +- actual Quit finalizes an active recording and stops playback cleanly +- without a compatible tray watcher, closing the window performs the traditional application shutdown - data usage updates while streaming - favorites controls work - tag playlist controls work @@ -143,7 +151,45 @@ python -m fluxtuner --gui --player ffplay --- -# 7. Live metadata +# 7. Recording persistence + +After a short TUI or GTK recording, inspect the latest recording row: + +```bash +sqlite3 ~/.local/share/fluxtuner/fluxtuner.db \ + 'SELECT id, profile_id, station_name, duration_seconds, file_size, status, file_path + FROM recordings + ORDER BY id DESC + LIMIT 1;' +``` + +Expected: + +- `status` is `completed` +- `duration_seconds` is close to the measured wall-clock recording time +- `file_size` is greater than zero +- `file_path` points inside the FluxTuner data directory + +Inspect the media file with: + +```bash +ffprobe -v error \ + -show_entries format=start_time,duration,size \ + -show_entries stream=codec_name,start_time,duration \ + -of default=noprint_wrappers=1 \ + /path/to/recording.mka +``` + +Expected: + +- the file is a valid Matroska audio recording +- stream copy preserves the source codec +- media duration is reasonably close to the requested recording duration +- a live/HLS source is not recorded substantially faster than wall clock + +--- + +# 8. Live metadata Run a known stream that exposes ICY metadata. @@ -177,7 +223,7 @@ Notes: --- -# 8. Web/server mode +# 9. Web/server mode Use an isolated data directory so the smoke test does not touch your regular FluxTuner library: @@ -242,7 +288,7 @@ UI checks: --- -# 9. macOS GTK notes +# 10. macOS GTK notes Install dependencies: @@ -284,7 +330,7 @@ python -m fluxtuner --gui --- -# 10. Linux notes +# 11. Linux notes ## CRUX @@ -313,7 +359,7 @@ sudo dnf install mpv ffmpeg python3-gobject gtk4 --- -# 11. Documentation checks +# 12. Documentation checks ```bash python -m fluxtuner --help @@ -338,7 +384,7 @@ Review visually: --- -# 12. Pre-commit checklist +# 13. Pre-commit checklist Before committing: @@ -363,7 +409,7 @@ Recommended manual checks: --- -# 13. Suggested release smoke test +# 14. Suggested release smoke test Before a release candidate, run the canonical gate and smoke-test the generated wheel rather than only the source checkout: @@ -401,5 +447,6 @@ Expected: - no crashes - correct backend selection - playback works -- GUI closes cleanly +- GUI tray/close behavior matches watcher availability +- a short TUI/GTK recording finalizes cleanly when `ffmpeg` is available - metadata appears when available diff --git a/docs/architecture.md b/docs/architecture.md index bf150e3..7ea3451 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -6,8 +6,8 @@ The primary local library store is SQLite: ~/.local/share/fluxtuner/fluxtuner.db -The database stores normalized stations, profiles, favorites, playback history -and manual playlists. +The database stores normalized stations, profiles, favorites, playback history, +manual playlists and profile-scoped recording metadata. FluxTuner local interfaces use profile-scoped library data. Profiles are context-level separation inside the same FluxTuner installation. They are useful for contexts @@ -23,15 +23,18 @@ Current model: ├── default │ ├── favorites │ ├── playback history - │ └── manual playlists + │ ├── manual playlists + │ └── recordings metadata ├── work │ ├── favorites │ ├── playback history - │ └── manual playlists + │ ├── manual playlists + │ └── recordings metadata └── terrace ├── favorites ├── playback history - └── manual playlists + ├── manual playlists + └── recordings metadata Profile resolution order: @@ -50,6 +53,10 @@ The Web user/account model adds ownership above profiles: ├── playback history └── manual playlists +Recording metadata uses the same profile-scoped SQLite model for local TUI/GTK +recordings. Web/server recording is intentionally not wired into the Web user +model in this release. + FluxTuner is organized as a multi-interface platform with frontends that share core services, user data and playback backends. For completed refactor milestones and larger internal cleanup plans, see [`docs/refactor-roadmap.md`](refactor-roadmap.md). @@ -83,6 +90,13 @@ flowchart LR Core --> Usage["Data usage tracking"] Core --> Config["Config and XDG storage"] Core --> Compatibility["Station compatibility"] + Core --> Recording["RecordingManager"] + Recording --> RecorderStore["SqliteRecordingStore"] + Recording --> FfmpegRecorder["FfmpegRecorder"] + FfmpegRecorder --> FFMPEG["ffmpeg"] + FFMPEG --> RecordingStreams["Online radio streams"] + FFMPEG --> RecordingFiles["XDG data recordings/*.mka"] + RecorderStore --> Library Compatibility --> Capabilities["PlayerCapabilities"] Capabilities --> Registry["Player registry"] @@ -244,6 +258,8 @@ fluxtuner/core/ playlists.py Tag playlists and playlist persistence helpers profiles.py Profile persistence and effective-profile resolution public_stats.py Public activity statistics + recording.py Recording lifecycle contracts and manager + recordings.py Recording paths and SQLite persistence search_service.py Shared station search service stations.py Station normalization and persistence helpers storage.py Atomic JSON writes for remaining JSON files @@ -304,6 +320,44 @@ Current backends: `mpv` and `ffplay` are treated as broadly compatible backends. `mpg123` and `ogg123` are specialized backends, so FluxTuner uses declared `PlayerCapabilities` plus station metadata to filter unsupported stations where possible. +## Recording layer + +Local recording is a separate lifecycle from playback and is currently exposed by +the Textual TUI and GTK4 GUI only. + +```mermaid +flowchart LR + TUI["Textual TUI"] --> Manager["RecordingManager"] + GTK["GTK4 GUI"] --> Manager + + Manager --> Backend["FfmpegRecorder"] + Manager --> Store["SqliteRecordingStore"] + + Backend --> FFmpeg["ffmpeg -readrate 1 -c copy"] + FFmpeg --> Stream["Online radio stream"] + FFmpeg --> Media["XDG data recordings/*.mka"] + + Store --> DB["SQLite recordings table"] + DB --> Profiles["Profile-scoped recording metadata"] +``` + +The recording manager owns the single active recording session for an interface. +Playback and recording are intentionally independent: changing or stopping playback +does not stop an active recording. + +FFmpeg records with stream copy into Matroska audio and uses real-time input pacing +so live/HLS sources are not consumed faster than wall-clock time. Logical duration +is derived from the manager's start/stop timestamps rather than container timestamps, +which can inherit timing from the source stream. + +Completed recording metadata is persisted in SQLite. Media files are stored under +the FluxTuner XDG data directory in `recordings/`. Closing GTK to a compatible +system tray keeps an active recording running; actual application shutdown finalizes +the FFmpeg process and persists the completed session. + +Web/server recording is outside the current local recording boundary and requires +separate storage, quota and long-running-worker design before being exposed. + ## Player capabilities and station compatibility Each backend declares static capabilities through `PlayerCapabilities`. @@ -334,6 +388,7 @@ The SQLite database stores the profile-scoped library: - favorites - playback history - manual playlists +- recording metadata Favorites are the canonical saved-station library. Manual playlists store station references and resolve them through the saved station/favorites model rather than @@ -358,6 +413,7 @@ Other local files remain JSON-based: ```text ~/.config/fluxtuner/config.json ~/.local/share/fluxtuner/usage.json +~/.local/share/fluxtuner/recordings/*.mka ~/.cache/fluxtuner/search_cache.json ``` diff --git a/docs/development.md b/docs/development.md index 228aa7b..fd785ab 100644 --- a/docs/development.md +++ b/docs/development.md @@ -12,6 +12,7 @@ Recommended local tools: - `pip`. - `venv`. - `mpv` and/or `ffplay` for broad manual playback testing. +- `ffmpeg` for manual recording testing. - Optional lightweight players: `mpg123` and `ogg123`. - Git. - Node.js for Web JavaScript syntax checks. @@ -136,7 +137,8 @@ sudo apt install ffmpeg `ffplay` is usually provided by FFmpeg packages. -The automated test suite mocks player execution and should not require external player binaries to be installed. +The automated test suite mocks player and recorder execution and should not require external player binaries to be installed. +Real recording smoke tests require `ffmpeg` and should verify a short `.mka` file with `ffprobe` or a media player. ## Project structure @@ -170,6 +172,8 @@ fluxtuner/core/ playlists.py Tag playlists and persistence helpers profiles.py Profile persistence and resolution public_stats.py Public activity statistics + recording.py Recording lifecycle contracts and manager + recordings.py Recording paths and SQLite persistence search_service.py Shared station search service stations.py Station normalization helpers storage.py Atomic JSON writes @@ -187,6 +191,9 @@ fluxtuner/web/ static/js/ Focused Web browser controllers/helpers static/styles.css Web visual system +fluxtuner/recorders/ + ffmpeg.py FFmpeg stream-copy recording backend + fluxtuner/players/ base.py Player interface and errors capabilities.py Backend capability declarations @@ -274,7 +281,7 @@ That is expected because the local project itself is not a published PyPI depend bandit -r fluxtuner -c pyproject.toml ``` -Bandit scans production code only. Subprocess-related skips are limited to the external player backends, which intentionally launch `mpv` or `ffplay`. +Bandit scans production code only. Subprocess-related skips are limited to the external player/recorder integrations, which intentionally launch trusted executables such as `mpv`, `ffplay` and `ffmpeg` after explicit executable and URL validation. ## Full local check @@ -340,6 +347,7 @@ Good candidates for tests: - Atomic persistence. - Error handling and fallback behavior. - Player command construction without requiring real player binaries. +- Recorder command construction, lifecycle and persistence without requiring real FFmpeg execution. - Radio Browser API failure handling. - ICY metadata parsing limits. - Runtime theme parsing and application. @@ -395,7 +403,7 @@ except OSError: Security-sensitive areas include: -- Player subprocess execution. +- Player and recorder subprocess execution. - Stream URL validation. - Imported JSON files. - Local user data persistence. diff --git a/docs/release.md b/docs/release.md index 3f41fa2..9c2c4af 100644 --- a/docs/release.md +++ b/docs/release.md @@ -10,6 +10,8 @@ A release should ensure that: - The version number is correct. - The changelog is updated. +- User-facing documentation and architecture diagrams match the shipped behavior. +- Flatpak/AppStream release metadata is updated when the release affects the local GTK/TUI package. - The package builds successfully. - CI is green on `main`. - Security checks pass. @@ -113,11 +115,21 @@ Create a release branch: git switch -c release/x.y.z ``` -Update files: +Update the canonical release metadata and review user-facing documentation: ```text pyproject.toml CHANGELOG.md +README.md +docs/usage.md +docs/architecture.md +SMOKE_TEST.md +``` + +When the release affects the local GTK/TUI package, also update: + +```text +flatpak/io.github.pitill0.Fluxtuner.metainfo.xml ``` Then run the complete release-quality gate and build from a clean artifact @@ -154,7 +166,8 @@ The PR should include: - Version bump. - Changelog update. -- Any release documentation updates. +- User-facing documentation and architecture updates. +- AppStream/Flatpak release metadata when applicable. - Validation checklist. Suggested PR title: diff --git a/docs/usage.md b/docs/usage.md index 7743b28..6570ef0 100644 --- a/docs/usage.md +++ b/docs/usage.md @@ -13,6 +13,7 @@ FluxTuner requires: - `mpg123` as an optional lightweight MP3/MPEG backend. - `ogg123`, provided by vorbis-tools, as an optional lightweight Ogg/Vorbis/Opus-style backend. - A terminal emulator with good Unicode support for the TUI. +- FFmpeg if you want to use local stream recording. The GTK desktop GUI also requires GTK4 and PyGObject. @@ -148,6 +149,28 @@ Backend notes: - `mpg123` is a specialized lightweight backend for MP3/MPEG streams. - `ogg123` is a specialized lightweight backend for Ogg/Vorbis/Opus-style streams, depending on the local `ogg123` build. +## Recording + +Local stream recording is available in the Textual TUI and GTK4 desktop GUI when +`ffmpeg` is installed and available in `PATH`. + +Recording is independent from playback: FluxTuner records the currently selected +station, which does not need to be playing. Only one recording can be active at a time. + +Recordings are saved as Matroska audio (`.mka`) files using FFmpeg stream copy, so the +original audio codec is preserved without transcoding. Files are stored under the +FluxTuner data directory: + +```text +~/.local/share/fluxtuner/recordings/ +``` + +The TUI uses `R` to start or stop recording. GTK provides a Record / Stop recording +button. + +Recording metadata is stored in the FluxTuner SQLite database and scoped to the active +profile. Web/server recording is not included in this release. + ## TUI Themes List available themes: diff --git a/flatpak/README.md b/flatpak/README.md index 1cd941b..3659134 100644 --- a/flatpak/README.md +++ b/flatpak/README.md @@ -35,6 +35,20 @@ flatpak-builder --user --install --force-clean \ build-dir flatpak/io.github.pitill0.Fluxtuner.yml ``` +### CRUX Linux + +On CRUX, disable `rofiles-fuse` when running `flatpak-builder`: + +```bash +flatpak-builder \ + --disable-rofiles-fuse \ + --user \ + --install \ + --force-clean \ + build-dir \ + flatpak/io.github.pitill0.Fluxtuner.yml +``` + ## Run ### GUI @@ -84,6 +98,9 @@ flatpak install --user ./fluxtuner.flatpak - `ogg123` lightweight Ogg/Vorbis/Opus/FLAC-style fallback - The local development manifest does not currently bundle player binaries explicitly. Use `flatpak run io.github.pitill0.Fluxtuner --list-players` to confirm which backends are available inside the sandbox. +- Local stream recording is available in the TUI and GTK GUI when `ffmpeg` is + available inside the sandbox. The GNOME runtime currently exposes `ffmpeg`, + but this should still be verified when changing runtime versions. - Python dependencies are installed from the checked-in `python3-requirements.json` module using pinned source URLs and checksums. Flatpak builds should not resolve application dependencies from the network @@ -108,7 +125,7 @@ Wayland and Ubuntu Xorg. See `docs/validation/permission-share-ipc.md`. CRUX Xorg, CRUX Wayland and Ubuntu Xorg. See `docs/validation/permission-device-dri.md`. -The manifest does not request broad filesystem access. FluxTuner should store its configuration, cache and SQLite library database through Flatpak-managed application data paths. The library database contains favorites, playback history and manual playlists. +The manifest does not request broad filesystem access. FluxTuner should store its configuration, cache, SQLite library database and local recordings through Flatpak-managed application data paths. The library database contains favorites, playback history, manual playlists and recording metadata. Recorded media files are stored under the app-specific Flatpak data directory in `fluxtuner/recordings/`. Reviewed permissions and environment overrides: diff --git a/flatpak/io.github.pitill0.Fluxtuner.metainfo.xml b/flatpak/io.github.pitill0.Fluxtuner.metainfo.xml index d5f3346..89bf882 100644 --- a/flatpak/io.github.pitill0.Fluxtuner.metainfo.xml +++ b/flatpak/io.github.pitill0.Fluxtuner.metainfo.xml @@ -33,6 +33,8 @@
  • Live stream metadata
  • Theme support
  • Estimated data usage tracking
    • +
  • Optional local stream recording when FFmpeg is available
    • +
  • Linux system tray integration on compatible desktop sessions
  • Multiple playback backends with automatic detection and compatibility checks
    • @@ -81,6 +83,29 @@ + + +
      +
    • Added optional local stream recording to the Textual TUI and GTK4 interface when FFmpeg is available.
      • +
    • Added profile-scoped recording metadata and Matroska audio output using stream copy.
      • +
    • Added Linux system tray integration for hiding and restoring the GTK window on compatible desktops.
      • +
    • Kept active playback and recording running while the GTK window is hidden to the tray.
      • +
    • Improved FFmpeg live-stream recording pacing so recorded duration follows wall-clock time.
      • +
    +     +     + + + +
      +
    • Added GTK System, Dark and Light appearance modes with persistent preferences.
      • +
    • Added runtime appearance switching without restarting FluxTuner.
      • +
    • Added dedicated GTK light and dark palettes.
      • +
    • Improved System appearance integration with the desktop color-scheme preference where supported.
      • +
    +     +     +
      diff --git a/fluxtuner/core/db.py b/fluxtuner/core/db.py index ca02c43..96d10b2 100644 --- a/fluxtuner/core/db.py +++ b/fluxtuner/core/db.py @@ -20,7 +20,7 @@ DB_FILE = data_file("fluxtuner.db") DEFAULT_USER_NAME = "default" DEFAULT_PROFILE_NAME = "default" -SCHEMA_MIGRATION_NAME = "schema_v5" +SCHEMA_MIGRATION_NAME = "schema_v6" APPROVAL_APPROVED = "approved" APPROVAL_PENDING = "pending" APPROVAL_REJECTED = "rejected" @@ -242,6 +242,30 @@ def create_schema(conn: sqlite3.Connection) -> None: FOREIGN KEY (station_id) REFERENCES stations(id) ON DELETE CASCADE ); + CREATE TABLE IF NOT EXISTS recordings ( + id INTEGER PRIMARY KEY, + profile_id INTEGER NOT NULL, + station_name TEXT NOT NULL, + source_url TEXT NOT NULL, + file_path TEXT NOT NULL UNIQUE, + started_at TEXT NOT NULL, + stopped_at TEXT NOT NULL, + duration_seconds REAL NOT NULL, + file_size INTEGER NOT NULL, + status TEXT NOT NULL DEFAULT 'completed', + created_at TEXT NOT NULL, + FOREIGN KEY (profile_id) REFERENCES profiles(id) ON DELETE CASCADE, + CHECK (length(trim(station_name)) > 0), + CHECK (length(trim(source_url)) > 0), + CHECK (length(trim(file_path)) > 0), + CHECK (duration_seconds >= 0), + CHECK (file_size >= 0), + CHECK (length(trim(status)) > 0) + ); + + CREATE INDEX IF NOT EXISTS idx_recordings_profile_started_at + ON recordings(profile_id, started_at DESC); + CREATE INDEX IF NOT EXISTS idx_favorites_profile_id ON favorites(profile_id); diff --git a/fluxtuner/core/recording.py b/fluxtuner/core/recording.py new file mode 100644 index 0000000..2b3ee06 --- /dev/null +++ b/fluxtuner/core/recording.py @@ -0,0 +1,121 @@ +from __future__ import annotations + +from collections.abc import Callable +from dataclasses import dataclass, replace +from datetime import UTC, datetime +from pathlib import Path +from typing import Protocol + + +class RecordingError(RuntimeError): + """Raised when a recording operation cannot be completed.""" + + +class RecordingBackend(Protocol): + """Minimal backend contract used by RecordingManager.""" + + def start(self, source_url: str, output_path: Path) -> None: ... + + def stop(self) -> None: ... + + def is_recording(self) -> bool: ... + + +class RecordingStore(Protocol): + """Persistence contract for completed recording sessions.""" + + def save(self, session: RecordingSession) -> int: ... + + +@dataclass(frozen=True) +class RecordingRequest: + """Input required to start one recording session.""" + + station_name: str + source_url: str + output_path: Path + + +@dataclass(frozen=True) +class RecordingSession: + """Snapshot of one recording session.""" + + station_name: str + source_url: str + output_path: Path + started_at: str + stopped_at: str | None = None + recording_id: int | None = None + + @property + def active(self) -> bool: + return self.stopped_at is None + + +def _utc_now() -> str: + return datetime.now(UTC).isoformat() + + +class RecordingManager: + """Coordinate one active recording independently from any UI.""" + + def __init__( + self, + backend: RecordingBackend, + *, + clock: Callable[[], str] = _utc_now, + store: RecordingStore | None = None, + ) -> None: + self._backend = backend + self._clock = clock + self._store = store + self._active_session: RecordingSession | None = None + + @property + def active_session(self) -> RecordingSession | None: + return self._active_session + + def is_recording(self) -> bool: + return self._active_session is not None and self._backend.is_recording() + + def start(self, request: RecordingRequest) -> RecordingSession: + """Start one recording session.""" + if self._active_session is not None: + raise RecordingError("A recording is already active.") + + station_name = request.station_name.strip() + source_url = request.source_url.strip() + + if not station_name: + raise RecordingError("Recording station name is required.") + if not source_url: + raise RecordingError("Recording source URL is required.") + + request.output_path.parent.mkdir(parents=True, exist_ok=True) + self._backend.start(source_url, request.output_path) + + session = RecordingSession( + station_name=station_name, + source_url=source_url, + output_path=request.output_path, + started_at=self._clock(), + ) + self._active_session = session + return session + + def stop(self) -> RecordingSession | None: + """Stop the active recording and return its completed session.""" + session = self._active_session + if session is None: + return None + + stopped_at = self._clock() + self._backend.stop() + completed = replace(session, stopped_at=stopped_at) + self._active_session = None + + if self._store is not None: + recording_id = self._store.save(completed) + completed = replace(completed, recording_id=recording_id) + + return completed diff --git a/fluxtuner/core/recordings.py b/fluxtuner/core/recordings.py new file mode 100644 index 0000000..9f9c704 --- /dev/null +++ b/fluxtuner/core/recordings.py @@ -0,0 +1,280 @@ +# SPDX-License-Identifier: MIT + +from __future__ import annotations + +import re +import secrets +import sqlite3 +import unicodedata +from dataclasses import dataclass +from datetime import UTC, datetime +from pathlib import Path +from typing import Any + +from fluxtuner.core import db +from fluxtuner.core.profiles import resolve_profile_id +from fluxtuner.core.recording import RecordingError, RecordingSession +from fluxtuner.paths import data_file + +RECORDINGS_DIR = data_file("recordings") +STATUS_COMPLETED = "completed" + + +def ensure_recordings_dir() -> Path: + """Return the recordings directory, creating it when needed.""" + RECORDINGS_DIR.mkdir(parents=True, exist_ok=True) + return RECORDINGS_DIR + + +def recording_output_path( + station_name: str, + *, + timestamp: datetime | None = None, + suffix: str | None = None, +) -> Path: + """Return a unique, filesystem-safe Matroska path for one recording.""" + normalized = ( + unicodedata.normalize("NFKD", station_name) + .encode("ascii", "ignore") + .decode("ascii") + .lower() + ) + slug = re.sub(r"[^a-z0-9]+", "-", normalized).strip("-") + slug = slug[:48].rstrip("-") or "station" + + started_at = timestamp or datetime.now(UTC) + stamp = started_at.strftime("%Y%m%d-%H%M%S") + unique_suffix = suffix or secrets.token_hex(4) + + return ensure_recordings_dir() / f"{stamp}-{slug}-{unique_suffix}.mka" + + +def _resolve_recording_profile_id( + conn: sqlite3.Connection, + *, + profile_id: int | None = None, + profile_name: str | None = None, +) -> int: + """Resolve recording ownership, falling back to the default profile.""" + resolved_profile_id = resolve_profile_id( + conn, + profile_id=profile_id, + profile_name=profile_name, + ) + if resolved_profile_id is not None: + return resolved_profile_id + + return db.ensure_default_profile(conn) + + +def recording_from_row(row: sqlite3.Row) -> dict[str, Any]: + """Return a public recording dictionary from a SQLite row.""" + return { + "id": int(row["id"]), + "profile_id": int(row["profile_id"]), + "station_name": str(row["station_name"]), + "source_url": str(row["source_url"]), + "file_path": str(row["file_path"]), + "started_at": str(row["started_at"]), + "stopped_at": str(row["stopped_at"]), + "duration_seconds": float(row["duration_seconds"]), + "file_size": int(row["file_size"]), + "status": str(row["status"]), + "created_at": str(row["created_at"]), + } + + +def add_recording( + conn: sqlite3.Connection, + *, + station_name: str, + source_url: str, + file_path: Path, + started_at: str, + stopped_at: str, + duration_seconds: float, + file_size: int, + status: str = STATUS_COMPLETED, + profile_id: int | None = None, + profile_name: str | None = None, +) -> int: + """Persist one completed recording and return its database id.""" + active_profile_id = _resolve_recording_profile_id( + conn, + profile_id=profile_id, + profile_name=profile_name, + ) + + clean_station_name = station_name.strip() + clean_source_url = source_url.strip() + clean_status = status.strip() + + if not clean_station_name: + raise ValueError("Recording station name is required.") + if not clean_source_url: + raise ValueError("Recording source URL is required.") + if not clean_status: + raise ValueError("Recording status is required.") + if duration_seconds < 0: + raise ValueError("Recording duration cannot be negative.") + if file_size < 0: + raise ValueError("Recording file size cannot be negative.") + + cursor = conn.execute( + """ + INSERT INTO recordings ( + profile_id, + station_name, + source_url, + file_path, + started_at, + stopped_at, + duration_seconds, + file_size, + status, + created_at + ) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + active_profile_id, + clean_station_name, + clean_source_url, + str(file_path), + started_at, + stopped_at, + float(duration_seconds), + int(file_size), + clean_status, + db.utc_now(), + ), + ) + recording_id = cursor.lastrowid + if recording_id is None: + raise RuntimeError("Could not create recording.") + + return int(recording_id) + + +def list_recordings( + conn: sqlite3.Connection, + *, + profile_id: int | None = None, + profile_name: str | None = None, +) -> list[dict[str, Any]]: + """Return recordings for one profile, newest first.""" + active_profile_id = _resolve_recording_profile_id( + conn, + profile_id=profile_id, + profile_name=profile_name, + ) + + rows = conn.execute( + """ + SELECT * + FROM recordings + WHERE profile_id = ? + ORDER BY started_at DESC, id DESC + """, + (active_profile_id,), + ).fetchall() + return [recording_from_row(row) for row in rows] + + +def get_recording( + conn: sqlite3.Connection, + recording_id: int, + *, + profile_id: int | None = None, + profile_name: str | None = None, +) -> dict[str, Any] | None: + """Return one recording owned by the selected profile.""" + active_profile_id = _resolve_recording_profile_id( + conn, + profile_id=profile_id, + profile_name=profile_name, + ) + + row = conn.execute( + """ + SELECT * + FROM recordings + WHERE id = ? + AND profile_id = ? + """, + (recording_id, active_profile_id), + ).fetchone() + return recording_from_row(row) if row is not None else None + + +def delete_recording( + conn: sqlite3.Connection, + recording_id: int, + *, + profile_id: int | None = None, + profile_name: str | None = None, +) -> bool: + """Delete one recording metadata row owned by the selected profile. + + File deletion is intentionally not performed here. Storage cleanup will be + coordinated separately so metadata and filesystem error handling stay + explicit. + """ + active_profile_id = _resolve_recording_profile_id( + conn, + profile_id=profile_id, + profile_name=profile_name, + ) + + cursor = conn.execute( + """ + DELETE FROM recordings + WHERE id = ? + AND profile_id = ? + """, + (recording_id, active_profile_id), + ) + return cursor.rowcount > 0 + + +@dataclass(frozen=True) +class SqliteRecordingStore: + """Persist completed recording sessions in FluxTuner SQLite storage.""" + + db_path: Path | None = None + profile_id: int | None = None + profile_name: str | None = None + + def save(self, session: RecordingSession) -> int: + """Persist a completed session and return its recording id.""" + if session.stopped_at is None: + raise RecordingError("Cannot persist an active recording session.") + + output_path = session.output_path + if not output_path.is_file(): + raise RecordingError(f"Recording output file does not exist: {output_path}") + + try: + started_at = datetime.fromisoformat(session.started_at) + stopped_at = datetime.fromisoformat(session.stopped_at) + duration_seconds = max(0.0, (stopped_at - started_at).total_seconds()) + except ValueError as exc: + raise RecordingError("Recording session contains invalid timestamps.") from exc + + db.init_db(self.db_path) + with db.connect(self.db_path) as conn: + recording_id = add_recording( + conn, + station_name=session.station_name, + source_url=session.source_url, + file_path=output_path, + started_at=session.started_at, + stopped_at=session.stopped_at, + duration_seconds=duration_seconds, + file_size=output_path.stat().st_size, + profile_id=self.profile_id, + profile_name=self.profile_name, + ) + conn.commit() + + return recording_id diff --git a/fluxtuner/gui/window.py b/fluxtuner/gui/window.py index 93f95ef..1eb952b 100644 --- a/fluxtuner/gui/window.py +++ b/fluxtuner/gui/window.py @@ -34,6 +34,11 @@ ) from fluxtuner.core.history import add_history, load_history # noqa: E402 from fluxtuner.core.profiles import resolve_effective_profile_name # noqa: E402 +from fluxtuner.core.recording import RecordingManager, RecordingRequest # noqa: E402 +from fluxtuner.core.recordings import ( # noqa: E402 + SqliteRecordingStore, + recording_output_path, +) from fluxtuner.core.search_service import SearchRequest, SearchService # noqa: E402 from fluxtuner.core.stations import ( # noqa: E402 same_station, @@ -59,6 +64,7 @@ from fluxtuner.gui.gtk_search import SearchLifecycle # noqa: E402 from fluxtuner.gui.gtk_view_state import ViewState # noqa: E402 from fluxtuner.players import create_player, selected_player_name # noqa: E402 +from fluxtuner.recorders.ffmpeg import FfmpegRecorder # noqa: E402 class MainWindow(Gtk.ApplicationWindow): @@ -84,6 +90,12 @@ def __init__( self.profile_name = resolve_effective_profile_name() self.player = create_player(self.player_backend_name) self.player_capabilities = self.player.capabilities() + self.recording_backend = FfmpegRecorder() + self.recording_available = self.recording_backend.is_available() + self.recording_manager = RecordingManager( + self.recording_backend, + store=SqliteRecordingStore(profile_name=self.profile_name), + ) self.search_service = SearchService(capabilities=self.player_capabilities) self.usage_tracker = DataUsageTracker() self._usage_timer_id: int | None = None @@ -359,6 +371,14 @@ def _build_bottom_playback_bar(self, root: Gtk.Box) -> None: self.play_button.connect("clicked", self.on_play_clicked) playback_bar.append(self.play_button) + self.record_button = Gtk.Button(label="● Record") + self.record_button.set_size_request(140, -1) + self.record_button.set_hexpand(False) + self.record_button.set_tooltip_text("Record selected station") + self.record_button.connect("clicked", self.on_record_clicked) + self.record_button.set_sensitive(self.recording_available) + playback_bar.append(self.record_button) + self.mute_button = Gtk.Button(label="Mute") self.mute_button.set_tooltip_text("Mute / unmute") self.mute_button.connect("clicked", self.on_mute_clicked) @@ -735,6 +755,82 @@ def on_play_clicked(self, _button: Gtk.Button) -> None: return self.play_selected_station() + def on_record_clicked(self, _button: Gtk.Button) -> None: + if self.recording_manager.active_session is not None: + self.stop_recording() + return + self.start_selected_recording() + + def start_selected_recording(self) -> None: + if not self.recording_available: + self.status_label.set_text("Recording unavailable: ffmpeg was not found in PATH.") + self._update_record_button() + return + + if not self.selected_station: + self.status_label.set_text("Select a station first.") + self._update_record_button() + return + + source_url = self._station_url(self.selected_station) + if not source_url: + self.status_label.set_text("Selected station has no stream URL.") + self._update_record_button() + return + + station_display_name = self._station_display_name(self.selected_station) + output_path = recording_output_path(station_display_name) + + try: + self.recording_manager.start( + RecordingRequest( + station_name=station_display_name, + source_url=source_url, + output_path=output_path, + ) + ) + except Exception as exc: # noqa: BLE001 - user-facing status in GTK GUI. + self.status_label.set_text(f"Could not start recording: {exc}") + self._update_record_button() + return + + self._update_record_button() + self.status_label.set_text(f"Recording: {station_display_name} → {output_path.name}") + + def stop_recording(self) -> None: + if self.recording_manager.active_session is None: + self.status_label.set_text("No recording is active.") + self._update_record_button() + return + + try: + completed = self.recording_manager.stop() + except Exception as exc: # noqa: BLE001 - user-facing status in GTK GUI. + self.status_label.set_text(f"Could not finalize recording: {exc}") + self._update_record_button() + return + + self._update_record_button() + if completed is None: + self.status_label.set_text("Recording stopped.") + return + + self.status_label.set_text( + f"Saved recording #{completed.recording_id}: " + f"{completed.station_name} → {completed.output_path.name}" + ) + + def _update_record_button(self) -> None: + if not hasattr(self, "record_button"): + return + + active = self.recording_manager.active_session is not None + self.record_button.set_label("■ Stop recording" if active else "● Record") + self.record_button.set_tooltip_text( + "Stop recording" if active else "Record selected station" + ) + self.record_button.set_sensitive(self.recording_available or active) + def play_selected_station(self) -> None: if not self.selected_station: self.status_label.set_text("Select a station first.") @@ -1379,11 +1475,14 @@ def _refresh_data_usage(self) -> bool: return True def shutdown(self) -> None: - """Stop GTK runtime work and the player before exiting the application.""" + """Stop GTK runtime work, recording and the player before exiting.""" self._stop_usage_timer() self._stop_player_state_timer() self._stop_metadata_polling() self.usage_tracker.stop() + with suppress(Exception): + if self.recording_manager.active_session is not None: + self.recording_manager.stop() with suppress(Exception): self.player.stop() diff --git a/fluxtuner/recorders/__init__.py b/fluxtuner/recorders/__init__.py new file mode 100644 index 0000000..5f02003 --- /dev/null +++ b/fluxtuner/recorders/__init__.py @@ -0,0 +1,3 @@ +from fluxtuner.recorders.ffmpeg import FfmpegRecorder + +__all__ = ["FfmpegRecorder"] diff --git a/fluxtuner/recorders/ffmpeg.py b/fluxtuner/recorders/ffmpeg.py new file mode 100644 index 0000000..380662b --- /dev/null +++ b/fluxtuner/recorders/ffmpeg.py @@ -0,0 +1,112 @@ +from __future__ import annotations + +import os +import signal +import subprocess # nosec B404 +from pathlib import Path +from typing import Any + +from fluxtuner.core.recording import RecordingError +from fluxtuner.logging_config import get_logger +from fluxtuner.players.base import PlayerError +from fluxtuner.players.security import resolve_executable, validate_stream_url + +logger = get_logger(__name__) + + +class FfmpegRecorder: + """Record one radio stream with FFmpeg stream copy.""" + + def __init__(self) -> None: + self.process: subprocess.Popen[Any] | None = None + + @classmethod + def is_available(cls) -> bool: + try: + resolve_executable("ffmpeg") + return True + except PlayerError: + logger.debug("ffmpeg recorder is not available") + return False + + def start(self, source_url: str, output_path: Path) -> None: + """Start recording a stream into a Matroska audio file.""" + if self.is_recording(): + raise RecordingError("FFmpeg recording is already active.") + + try: + ffmpeg_path = resolve_executable("ffmpeg") + safe_url = validate_stream_url(source_url) + except PlayerError as exc: + raise RecordingError(str(exc)) from exc + + output_path = output_path.expanduser() + output_path.parent.mkdir(parents=True, exist_ok=True) + + command = [ + ffmpeg_path, + "-nostdin", + "-hide_banner", + "-loglevel", + "warning", + "-readrate", + "1", + "-i", + safe_url, + "-map", + "0:a:0", + "-c", + "copy", + "-f", + "matroska", + "-y", + str(output_path), + ] + + logger.debug("Starting FFmpeg recording") + try: + self.process = subprocess.Popen( # noqa: S603 # nosec B603 + command, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + start_new_session=True, + ) + except OSError as exc: + raise RecordingError(f"Could not start ffmpeg: {exc}") from exc + + logger.debug("FFmpeg recording process started") + + def stop(self) -> None: + """Stop recording gracefully so FFmpeg can finalize the container.""" + if self.process is None: + logger.debug("FFmpeg stop requested without active recording") + return + + process = self.process + + try: + if process.poll() is None: + logger.debug("Stopping FFmpeg recording process") + try: + os.killpg(os.getpgid(process.pid), signal.SIGINT) + process.wait(timeout=5) + except Exception: # noqa: BLE001 + logger.debug( + "Graceful FFmpeg stop failed; killing process group", + exc_info=True, + ) + try: + os.killpg(os.getpgid(process.pid), signal.SIGKILL) + except Exception: # noqa: BLE001 + logger.debug( + "Could not kill FFmpeg process group; killing process", + exc_info=True, + ) + process.kill() + process.wait(timeout=3) + finally: + self.process = None + logger.debug("FFmpeg recording process stopped") + + def is_recording(self) -> bool: + return self.process is not None and self.process.poll() is None diff --git a/fluxtuner/tui.py b/fluxtuner/tui.py index 32b2716..a919d82 100644 --- a/fluxtuner/tui.py +++ b/fluxtuner/tui.py @@ -42,6 +42,8 @@ ) from fluxtuner.core.playlists import get_by_tag, get_tag_counts from fluxtuner.core.profiles import resolve_effective_profile_name +from fluxtuner.core.recording import RecordingManager, RecordingRequest +from fluxtuner.core.recordings import SqliteRecordingStore, recording_output_path from fluxtuner.core.search_service import SearchRequest, SearchService from fluxtuner.core.stations import ( station_bitrate, @@ -58,6 +60,7 @@ from fluxtuner.core.stream_metadata import fetch_stream_metadata from fluxtuner.logging_config import get_logger from fluxtuner.players import create_player, selected_player_name +from fluxtuner.recorders.ffmpeg import FfmpegRecorder from fluxtuner.theme_runtime import apply_theme_runtime from fluxtuner.themes import DEFAULT_THEME, get_theme_path, list_themes, theme_exists from fluxtuner.tui_details import ( @@ -102,6 +105,7 @@ [b]Playback[/b] Space Play / Stop x Stop + R Record / Stop recording l Play last station r Random favorite + / - Volume up / down @@ -185,6 +189,7 @@ class FluxTunerTUI(App[None]): Binding("minus", "volume_down", "Vol-", show=False), Binding("m", "toggle_mute", "Mute", show=False), Binding("x", "stop", "Stop", show=False), + Binding("R", "toggle_recording", "Record", key_display="R", show=False), Binding("t", "show_themes", "Themes", show=False), Binding("p", "show_playlists", "Playlists", show=False), Binding("y", "save_theme", "Save theme", show=False), @@ -198,6 +203,12 @@ def __init__(self, theme: str | None = None, player_name: str = "mpv") -> None: self.profile_name = resolve_effective_profile_name() self.player = create_player(self.player_backend_name) self.player_capabilities = self.player.capabilities() + self.recording_backend = FfmpegRecorder() + self.recording_available = self.recording_backend.is_available() + self.recording_manager = RecordingManager( + self.recording_backend, + store=SqliteRecordingStore(profile_name=self.profile_name), + ) self.search_service = SearchService(capabilities=self.player_capabilities) self.usage_tracker = DataUsageTracker() self.current_artist = "—" @@ -251,6 +262,7 @@ def compose(self) -> ComposeResult: yield Static("[b]Player[/b]\nBackend: auto\nState: stopped", id="player-state") yield Static("No station selected.", id="details") yield Button("Play", id="play", classes="side-button success-button") + yield Button("● Record", id="record", classes="side-button warning-button") yield Button("Add fav", id="add-favorite", classes="side-button primary-button") yield Button( "Remove fav", id="remove-favorite", classes="side-button warning-button" @@ -277,6 +289,7 @@ def on_mount(self) -> None: self.query_one("#stations", DataTable).focus() self.update_now_playing() self.update_play_button() + self.update_record_button() self.set_interval(1.5, self.update_now_playing) self.set_status(f"Ready. Player backend: {self.player_backend_name}.") if self.last_station: @@ -292,6 +305,9 @@ def on_unmount(self) -> None: self._cancel_metadata_request() with suppress(Exception): self.usage_tracker.stop() + with suppress(Exception): + if self.recording_manager.active_session is not None: + self.recording_manager.stop() self.player.stop() def action_focus_search(self) -> None: @@ -411,6 +427,9 @@ def action_play_stop(self) -> None: def action_stop(self) -> None: self.stop_playback() + def action_toggle_recording(self) -> None: + self.toggle_recording() + def action_toggle_mute(self) -> None: if not self.player.supports_mute(): self.set_status(f"{self.player_backend_name} does not support live mute.") @@ -520,6 +539,10 @@ def stop_from_button(self) -> None: def play_from_button(self) -> None: self.action_play_stop() + @on(Button.Pressed, "#record") + def record_from_button(self) -> None: + self.toggle_recording() + @on(Button.Pressed, "#add-favorite") def add_favorite_from_button(self) -> None: self.add_selected_to_favorites() @@ -1053,6 +1076,75 @@ def update_play_button(self) -> None: button = self.query_one("#play", Button) button.label = "■ Stop" if self.player.is_playing() else "▶ Play" + def update_record_button(self) -> None: + button = self.query_one("#record", Button) + active = self.recording_manager.active_session is not None + button.label = "■ Stop recording" if active else "● Record" + button.disabled = not self.recording_available and not active + + def toggle_recording(self) -> None: + if self.recording_manager.active_session is not None: + self.stop_recording() + return + + if not self.recording_available: + self.set_status("Recording unavailable: ffmpeg was not found in PATH.") + return + + if not self.selected_station: + self.set_status("No station selected to record.") + return + + source_url = core_station_url(self.selected_station) + if not source_url: + self.set_status("Selected station has no stream URL.") + return + + station_name = favorite_display_name(self.selected_station) + output_path = recording_output_path(station_name) + + try: + self.recording_manager.start( + RecordingRequest( + station_name=station_name, + source_url=source_url, + output_path=output_path, + ) + ) + except Exception as exc: # noqa: BLE001 + self.update_record_button() + self.set_status(f"Could not start recording: {exc}") + self.notify(f"Could not start recording: {exc}", severity="error") + return + + self.update_record_button() + self.set_status(f"Recording: {station_name} → {output_path.name}") + + def stop_recording(self) -> None: + session = self.recording_manager.active_session + if session is None: + self.update_record_button() + self.set_status("No recording is active.") + return + + try: + completed = self.recording_manager.stop() + except Exception as exc: # noqa: BLE001 + self.update_record_button() + self.set_status(f"Could not finalize recording: {exc}") + self.notify(f"Could not finalize recording: {exc}", severity="error") + return + + self.update_record_button() + if completed is None: + self.set_status("Recording stopped.") + return + + self.set_status( + f"Saved recording #{completed.recording_id}: " + f"{completed.station_name} → {completed.output_path.name}" + ) + def stop_playback(self) -> None: result = coordinate_playback_stop( player=self.player, diff --git a/pyproject.toml b/pyproject.toml index 62edd26..1fa4ccb 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "fluxtuner" -version = "1.1.0" +version = "1.2.0" description = "Modern internet radio platform for terminal, desktop and web, powered by Textual, GTK and browser playback." readme = "README.md" requires-python = ">=3.11" @@ -87,6 +87,7 @@ packages = [ "fluxtuner.core", "fluxtuner.gui", "fluxtuner.players", + "fluxtuner.recorders", "fluxtuner.themes", "fluxtuner.web.routes", "fluxtuner.web.metadata", diff --git a/tests/test_db.py b/tests/test_db.py index e109f6d..2abfbe5 100644 --- a/tests/test_db.py +++ b/tests/test_db.py @@ -37,6 +37,7 @@ def test_init_db_creates_expected_tables( "playlist_stations", "playlists", "profiles", + "recordings", "schema_migrations", "stations", "users", diff --git a/tests/test_gtk_recording.py b/tests/test_gtk_recording.py new file mode 100644 index 0000000..b2f42f8 --- /dev/null +++ b/tests/test_gtk_recording.py @@ -0,0 +1,189 @@ +from __future__ import annotations + +import sys +from pathlib import Path +from types import ModuleType, SimpleNamespace +from unittest.mock import Mock + + +def _import_window_module(): + try: + from fluxtuner.gui import window + except ImportError as exc: + if exc.name != "gi": + raise + + gi = ModuleType("gi") + gi.require_version = lambda *_args: None # type: ignore[attr-defined] + + repository = ModuleType("gi.repository") + + class ApplicationWindow: + pass + + gtk = SimpleNamespace(ApplicationWindow=ApplicationWindow) + repository.GLib = SimpleNamespace() # type: ignore[attr-defined] + repository.Gtk = gtk # type: ignore[attr-defined] + repository.Pango = SimpleNamespace() # type: ignore[attr-defined] + + sys.modules["gi"] = gi + sys.modules["gi.repository"] = repository + + from fluxtuner.gui import window + + return window + + +window = _import_window_module() + + +class FakeRecordingManager: + def __init__(self) -> None: + self.active_session = None + self.started_request = None + self.stop_calls = 0 + + def start(self, request): + self.started_request = request + self.active_session = SimpleNamespace( + station_name=request.station_name, + output_path=request.output_path, + ) + return self.active_session + + def stop(self): + self.stop_calls += 1 + active = self.active_session + self.active_session = None + if active is None: + return None + return SimpleNamespace( + recording_id=9, + station_name=active.station_name, + output_path=active.output_path, + ) + + +def _recording_harness() -> SimpleNamespace: + return SimpleNamespace( + recording_available=True, + recording_manager=FakeRecordingManager(), + selected_station={ + "name": "Flux FM", + "url_resolved": "https://radio.example/stream", + }, + status_label=Mock(), + record_button=Mock(), + _station_url=Mock(return_value="https://radio.example/stream"), + _station_display_name=Mock(return_value="Flux FM"), + _update_record_button=Mock(), + ) + + +def test_gtk_recording_starts_selected_station(monkeypatch, tmp_path: Path) -> None: + harness = _recording_harness() + output = tmp_path / "flux-fm.mka" + monkeypatch.setattr(window, "recording_output_path", lambda _name: output) + + window.MainWindow.start_selected_recording(harness) + + request = harness.recording_manager.started_request + assert request is not None + assert request.station_name == "Flux FM" + assert request.source_url == "https://radio.example/stream" + assert request.output_path == output + harness._update_record_button.assert_called_once_with() + harness.status_label.set_text.assert_called_once_with(f"Recording: Flux FM → {output.name}") + + +def test_gtk_record_button_stops_active_recording(monkeypatch, tmp_path: Path) -> None: + harness = _recording_harness() + output = tmp_path / "flux-fm.mka" + monkeypatch.setattr(window, "recording_output_path", lambda _name: output) + + window.MainWindow.start_selected_recording(harness) + harness._update_record_button.reset_mock() + harness.status_label.set_text.reset_mock() + + harness.stop_recording = lambda: window.MainWindow.stop_recording(harness) + window.MainWindow.on_record_clicked(harness, Mock()) + + assert harness.recording_manager.stop_calls == 1 + assert harness.recording_manager.active_session is None + harness._update_record_button.assert_called_once_with() + harness.status_label.set_text.assert_called_once_with( + f"Saved recording #9: Flux FM → {output.name}" + ) + + +def test_gtk_recording_does_not_depend_on_player(monkeypatch, tmp_path: Path) -> None: + harness = _recording_harness() + output = tmp_path / "independent.mka" + monkeypatch.setattr(window, "recording_output_path", lambda _name: output) + + # The harness deliberately has no player/current_station attributes. + window.MainWindow.start_selected_recording(harness) + + assert harness.recording_manager.started_request is not None + + +def test_gtk_recording_requires_selection() -> None: + harness = _recording_harness() + harness.selected_station = None + + window.MainWindow.start_selected_recording(harness) + + assert harness.recording_manager.started_request is None + harness.status_label.set_text.assert_called_once_with("Select a station first.") + + +def test_gtk_recording_reports_missing_ffmpeg() -> None: + harness = _recording_harness() + harness.recording_available = False + + window.MainWindow.start_selected_recording(harness) + + assert harness.recording_manager.started_request is None + harness.status_label.set_text.assert_called_once_with( + "Recording unavailable: ffmpeg was not found in PATH." + ) + + +def test_gtk_shutdown_finalizes_active_recording() -> None: + manager = FakeRecordingManager() + manager.active_session = SimpleNamespace( + station_name="Flux FM", + output_path=Path("/tmp/flux-fm.mka"), + ) + harness = SimpleNamespace( + recording_manager=manager, + player=Mock(), + usage_tracker=Mock(), + _stop_usage_timer=Mock(), + _stop_player_state_timer=Mock(), + _stop_metadata_polling=Mock(), + ) + + window.MainWindow.shutdown(harness) + + assert manager.stop_calls == 1 + harness.player.stop.assert_called_once_with() + + +def test_gtk_record_button_projection() -> None: + harness = _recording_harness() + + window.MainWindow._update_record_button(harness) + + harness.record_button.set_label.assert_called_once_with("● Record") + harness.record_button.set_tooltip_text.assert_called_once_with("Record selected station") + harness.record_button.set_sensitive.assert_called_once_with(True) + + harness.record_button.reset_mock() + harness.recording_manager.active_session = object() + + window.MainWindow._update_record_button(harness) + + harness.record_button.set_label.assert_called_once_with("■ Stop recording") + harness.record_button.set_tooltip_text.assert_called_once_with("Stop recording") + harness.record_button.set_sensitive.assert_called_once_with(True) diff --git a/tests/test_recorders_ffmpeg.py b/tests/test_recorders_ffmpeg.py new file mode 100644 index 0000000..a7c91b4 --- /dev/null +++ b/tests/test_recorders_ffmpeg.py @@ -0,0 +1,216 @@ +from pathlib import Path + +import pytest + +from fluxtuner.core.recording import RecordingError +from fluxtuner.recorders.ffmpeg import FfmpegRecorder + + +def test_ffmpeg_recorder_reports_availability(monkeypatch) -> None: + monkeypatch.setattr( + "fluxtuner.recorders.ffmpeg.resolve_executable", + lambda _name: "/usr/bin/ffmpeg", + ) + + assert FfmpegRecorder.is_available() is True + + +def test_ffmpeg_recorder_reports_missing_executable(monkeypatch) -> None: + from fluxtuner.players.base import PlayerError + + def missing(_name: str) -> str: + raise PlayerError("ffmpeg missing") + + monkeypatch.setattr("fluxtuner.recorders.ffmpeg.resolve_executable", missing) + + assert FfmpegRecorder.is_available() is False + + +def test_ffmpeg_recorder_start_uses_stream_copy_and_matroska( + monkeypatch, + tmp_path: Path, +) -> None: + created_commands: list[list[str]] = [] + + class FakePopen: + pid = 123 + + def __init__(self, command, **kwargs): + created_commands.append(command) + self.kwargs = kwargs + + def poll(self): + return None + + monkeypatch.setattr( + "fluxtuner.recorders.ffmpeg.resolve_executable", + lambda _name: "/usr/bin/ffmpeg", + ) + monkeypatch.setattr("subprocess.Popen", FakePopen) + + output = tmp_path / "session.mka" + recorder = FfmpegRecorder() + recorder.start("https://example.com/stream", output) + + assert created_commands == [ + [ + "/usr/bin/ffmpeg", + "-nostdin", + "-hide_banner", + "-loglevel", + "warning", + "-readrate", + "1", + "-i", + "https://example.com/stream", + "-map", + "0:a:0", + "-c", + "copy", + "-f", + "matroska", + "-y", + str(output), + ] + ] + assert recorder.is_recording() is True + + +def test_ffmpeg_recorder_rejects_invalid_stream_url( + monkeypatch, + tmp_path: Path, +) -> None: + monkeypatch.setattr( + "fluxtuner.recorders.ffmpeg.resolve_executable", + lambda _name: "/usr/bin/ffmpeg", + ) + recorder = FfmpegRecorder() + + with pytest.raises(RecordingError, match="Unsupported or invalid stream URL"): + recorder.start("file:///tmp/source.mp3", tmp_path / "session.mka") + + +def test_ffmpeg_recorder_rejects_parallel_start( + monkeypatch, + tmp_path: Path, +) -> None: + class RunningProcess: + pid = 123 + + def poll(self): + return None + + recorder = FfmpegRecorder() + recorder.process = RunningProcess() # type: ignore[assignment] + + with pytest.raises(RecordingError, match="already active"): + recorder.start("https://example.com/stream", tmp_path / "session.mka") + + +class FinishedProcess: + pid = 123 + + def poll(self): + return 0 + + +def test_ffmpeg_recorder_stop_clears_finished_process() -> None: + recorder = FfmpegRecorder() + recorder.process = FinishedProcess() # type: ignore[assignment] + + recorder.stop() + + assert recorder.process is None + + +def test_ffmpeg_recorder_stop_sends_sigint_before_kill( + monkeypatch, +) -> None: + events: list[tuple[str, object]] = [] + + class RunningProcess: + pid = 123 + + def poll(self): + return None + + def wait(self, timeout=None): + events.append(("wait", timeout)) + return 0 + + def kill(self): + events.append(("kill", None)) + + recorder = FfmpegRecorder() + recorder.process = RunningProcess() # type: ignore[assignment] + + monkeypatch.setattr("os.getpgid", lambda _pid: 456) + monkeypatch.setattr( + "os.killpg", + lambda pgid, sig: events.append(("signal", (pgid, sig))), + ) + + recorder.stop() + + import signal + + assert events == [ + ("signal", (456, signal.SIGINT)), + ("wait", 5), + ] + assert recorder.process is None + + +def test_ffmpeg_recorder_start_failure_leaves_recorder_idle( + monkeypatch, + tmp_path: Path, +) -> None: + monkeypatch.setattr( + "fluxtuner.recorders.ffmpeg.resolve_executable", + lambda _name: "/usr/bin/ffmpeg", + ) + + def fail_popen(*_args, **_kwargs): + raise OSError("cannot execute") + + monkeypatch.setattr("subprocess.Popen", fail_popen) + + recorder = FfmpegRecorder() + + with pytest.raises(RecordingError, match="Could not start ffmpeg"): + recorder.start("https://example.com/stream", tmp_path / "session.mka") + + assert recorder.process is None + + +def test_ffmpeg_recorder_logs_without_stream_url( + monkeypatch, + caplog, + tmp_path: Path, +) -> None: + import logging + + class FakePopen: + pid = 123 + + def __init__(self, _command, **_kwargs): + pass + + def poll(self): + return None + + monkeypatch.setattr( + "fluxtuner.recorders.ffmpeg.resolve_executable", + lambda _name: "/usr/bin/ffmpeg", + ) + monkeypatch.setattr("subprocess.Popen", FakePopen) + + recorder = FfmpegRecorder() + with caplog.at_level(logging.DEBUG): + recorder.start( + "https://example.com/private-stream", + tmp_path / "session.mka", + ) + + assert "Starting FFmpeg recording" in caplog.text + assert "https://example.com/private-stream" not in caplog.text diff --git a/tests/test_recording.py b/tests/test_recording.py new file mode 100644 index 0000000..3480047 --- /dev/null +++ b/tests/test_recording.py @@ -0,0 +1,262 @@ +from __future__ import annotations + +from pathlib import Path +from unittest.mock import Mock + +import pytest + +from fluxtuner.core.recording import ( + RecordingError, + RecordingManager, + RecordingRequest, + RecordingSession, +) + + +class FakeRecordingBackend: + def __init__(self) -> None: + self.recording = False + self.start_calls: list[tuple[str, Path]] = [] + self.stop_calls = 0 + + def start(self, source_url: str, output_path: Path) -> None: + self.start_calls.append((source_url, output_path)) + self.recording = True + + def stop(self) -> None: + self.stop_calls += 1 + self.recording = False + + def is_recording(self) -> bool: + return self.recording + + +def _request(tmp_path: Path) -> RecordingRequest: + return RecordingRequest( + station_name="Flux FM", + source_url="https://radio.example/stream", + output_path=tmp_path / "recordings" / "flux-fm.mp3", + ) + + +def test_recording_manager_starts_session(tmp_path: Path) -> None: + backend = FakeRecordingBackend() + manager = RecordingManager( + backend, + clock=lambda: "2026-09-02T12:00:00+00:00", + ) + request = _request(tmp_path) + + session = manager.start(request) + + assert backend.start_calls == [(request.source_url, request.output_path)] + assert request.output_path.parent.is_dir() + assert session.station_name == "Flux FM" + assert session.started_at == "2026-09-02T12:00:00+00:00" + assert session.active is True + assert manager.active_session is session + assert manager.is_recording() is True + + +def test_recording_manager_rejects_parallel_session(tmp_path: Path) -> None: + backend = FakeRecordingBackend() + manager = RecordingManager(backend) + manager.start(_request(tmp_path)) + + with pytest.raises(RecordingError, match="already active"): + manager.start( + RecordingRequest( + station_name="Other station", + source_url="https://radio.example/other", + output_path=tmp_path / "other.mp3", + ) + ) + + assert len(backend.start_calls) == 1 + + +def test_recording_manager_stop_returns_completed_session(tmp_path: Path) -> None: + backend = FakeRecordingBackend() + timestamps = iter( + [ + "2026-09-02T12:00:00+00:00", + "2026-09-02T12:10:00+00:00", + ] + ) + manager = RecordingManager(backend, clock=lambda: next(timestamps)) + manager.start(_request(tmp_path)) + + completed = manager.stop() + + assert completed is not None + assert completed.active is False + assert completed.stopped_at == "2026-09-02T12:10:00+00:00" + assert backend.stop_calls == 1 + assert manager.active_session is None + assert manager.is_recording() is False + + +def test_recording_manager_stop_is_noop_when_idle() -> None: + backend = FakeRecordingBackend() + manager = RecordingManager(backend) + + assert manager.stop() is None + assert backend.stop_calls == 0 + + +def test_recording_manager_start_failure_leaves_manager_idle( + tmp_path: Path, +) -> None: + backend = FakeRecordingBackend() + backend.start = Mock(side_effect=RuntimeError("recorder failed")) + manager = RecordingManager(backend) + + with pytest.raises(RuntimeError, match="recorder failed"): + manager.start(_request(tmp_path)) + + assert manager.active_session is None + + +@pytest.mark.parametrize( + ("station_name", "source_url", "message"), + [ + ("", "https://radio.example/stream", "station name"), + ("Flux FM", "", "source URL"), + ], +) +def test_recording_manager_rejects_incomplete_request( + tmp_path: Path, + station_name: str, + source_url: str, + message: str, +) -> None: + backend = FakeRecordingBackend() + manager = RecordingManager(backend) + + with pytest.raises(RecordingError, match=message): + manager.start( + RecordingRequest( + station_name=station_name, + source_url=source_url, + output_path=tmp_path / "recording.mp3", + ) + ) + + assert backend.start_calls == [] + + +class FakeRecordingStore: + def __init__(self) -> None: + self.saved_sessions = [] + + def save(self, session) -> int: + self.saved_sessions.append(session) + return 42 + + +def test_stop_persists_completed_session_and_returns_recording_id(tmp_path: Path) -> None: + backend = FakeRecordingBackend() + store = FakeRecordingStore() + timestamps = iter( + [ + "2026-09-02T12:00:00+00:00", + "2026-09-02T12:00:20+00:00", + ] + ) + manager = RecordingManager( + backend, + clock=lambda: next(timestamps), + store=store, + ) + + manager.start( + RecordingRequest( + station_name="Flux FM", + source_url="https://radio.example/stream", + output_path=tmp_path / "recording.mka", + ) + ) + completed = manager.stop() + + assert completed is not None + assert completed.recording_id == 42 + assert completed.stopped_at == "2026-09-02T12:00:20+00:00" + assert store.saved_sessions == [ + RecordingSession( + station_name="Flux FM", + source_url="https://radio.example/stream", + output_path=tmp_path / "recording.mka", + started_at="2026-09-02T12:00:00+00:00", + stopped_at="2026-09-02T12:00:20+00:00", + ) + ] + + +def test_persistence_failure_leaves_manager_idle(tmp_path: Path) -> None: + backend = FakeRecordingBackend() + + class FailingStore: + def save(self, _session) -> int: + raise RecordingError("database unavailable") + + timestamps = iter( + [ + "2026-09-02T12:00:00+00:00", + "2026-09-02T12:00:20+00:00", + ] + ) + manager = RecordingManager( + backend, + clock=lambda: next(timestamps), + store=FailingStore(), + ) + + manager.start( + RecordingRequest( + station_name="Flux FM", + source_url="https://radio.example/stream", + output_path=tmp_path / "recording.mka", + ) + ) + + with pytest.raises(RecordingError, match="database unavailable"): + manager.stop() + + assert manager.active_session is None + assert manager.is_recording() is False + + +def test_stop_captures_timestamp_before_backend_shutdown(tmp_path: Path) -> None: + events: list[str] = [] + + class TimingBackend(FakeRecordingBackend): + def stop(self) -> None: + events.append("backend-stop") + super().stop() + + timestamps = iter( + [ + "2026-09-02T12:00:00+00:00", + "2026-09-02T12:00:20+00:00", + ] + ) + + def clock() -> str: + events.append("clock") + return next(timestamps) + + manager = RecordingManager(TimingBackend(), clock=clock) + manager.start( + RecordingRequest( + station_name="Flux FM", + source_url="https://radio.example/stream", + output_path=tmp_path / "recording.mka", + ) + ) + + events.clear() + completed = manager.stop() + + assert completed is not None + assert completed.stopped_at == "2026-09-02T12:00:20+00:00" + assert events == ["clock", "backend-stop"] diff --git a/tests/test_recording_paths.py b/tests/test_recording_paths.py new file mode 100644 index 0000000..3afce5d --- /dev/null +++ b/tests/test_recording_paths.py @@ -0,0 +1,29 @@ +from datetime import UTC, datetime + +from fluxtuner.core import recordings + + +def test_recording_output_path_is_safe_and_deterministic(tmp_path, monkeypatch) -> None: + monkeypatch.setattr(recordings, "RECORDINGS_DIR", tmp_path) + + path = recordings.recording_output_path( + "Rádio Música / 80's", + timestamp=datetime(2026, 9, 2, 15, 55, 1, tzinfo=UTC), + suffix="deadbeef", + ) + + assert path.parent == tmp_path + assert path.name == "20260902-155501-radio-musica-80-s-deadbeef.mka" + assert tmp_path.is_dir() + + +def test_recording_output_path_falls_back_for_non_ascii_name(tmp_path, monkeypatch) -> None: + monkeypatch.setattr(recordings, "RECORDINGS_DIR", tmp_path) + + path = recordings.recording_output_path( + "東京", + timestamp=datetime(2026, 9, 2, 15, 55, 1, tzinfo=UTC), + suffix="cafebabe", + ) + + assert path.name == "20260902-155501-station-cafebabe.mka" diff --git a/tests/test_recordings.py b/tests/test_recordings.py new file mode 100644 index 0000000..5f1a0d1 --- /dev/null +++ b/tests/test_recordings.py @@ -0,0 +1,236 @@ +# SPDX-License-Identifier: MIT + +from pathlib import Path + +import pytest + +from fluxtuner.core import db +from fluxtuner.core.recording import RecordingError, RecordingSession +from fluxtuner.core.recordings import ( + SqliteRecordingStore, + add_recording, + delete_recording, + ensure_recordings_dir, + get_recording, + list_recordings, +) + + +def _add_sample_recording( + conn, + tmp_path: Path, + *, + station_name: str = "Flux FM", + started_at: str = "2026-09-02T12:00:00+00:00", + file_name: str = "recording.mka", +) -> int: + return add_recording( + conn, + station_name=station_name, + source_url="https://radio.example/stream", + file_path=tmp_path / file_name, + started_at=started_at, + stopped_at="2026-09-02T12:00:20+00:00", + duration_seconds=20.0, + file_size=640_000, + ) + + +def test_ensure_recordings_dir_creates_directory( + tmp_path: Path, + monkeypatch, +) -> None: + from fluxtuner.core import recordings + + target = tmp_path / "recordings" + monkeypatch.setattr(recordings, "RECORDINGS_DIR", target) + + assert ensure_recordings_dir() == target + assert target.is_dir() + + +def test_add_and_get_recording(tmp_path: Path) -> None: + db_file = tmp_path / "fluxtuner.db" + db.init_db(db_file) + + with db.connect(db_file) as conn: + recording_id = _add_sample_recording(conn, tmp_path) + conn.commit() + + recording = get_recording(conn, recording_id) + + assert recording is not None + assert recording["station_name"] == "Flux FM" + assert recording["source_url"] == "https://radio.example/stream" + assert recording["file_path"] == str(tmp_path / "recording.mka") + assert recording["duration_seconds"] == 20.0 + assert recording["file_size"] == 640_000 + assert recording["status"] == "completed" + + +def test_list_recordings_returns_newest_first(tmp_path: Path) -> None: + db_file = tmp_path / "fluxtuner.db" + db.init_db(db_file) + + with db.connect(db_file) as conn: + _add_sample_recording( + conn, + tmp_path, + station_name="Older", + started_at="2026-09-02T11:00:00+00:00", + file_name="older.mka", + ) + _add_sample_recording( + conn, + tmp_path, + station_name="Newer", + started_at="2026-09-02T12:00:00+00:00", + file_name="newer.mka", + ) + conn.commit() + + recordings = list_recordings(conn) + + assert [item["station_name"] for item in recordings] == ["Newer", "Older"] + + +def test_recordings_are_profile_scoped(tmp_path: Path) -> None: + db_file = tmp_path / "fluxtuner.db" + db.init_db(db_file) + + with db.connect(db_file) as conn: + default_user_id = db.ensure_default_user(conn) + other_profile_id = db.get_or_create_profile( + conn, + "other", + user_id=default_user_id, + ) + + default_id = _add_sample_recording(conn, tmp_path, station_name="Default") + other_id = add_recording( + conn, + station_name="Other", + source_url="https://radio.example/other", + file_path=tmp_path / "other.mka", + started_at="2026-09-02T13:00:00+00:00", + stopped_at="2026-09-02T13:00:20+00:00", + duration_seconds=20.0, + file_size=123, + profile_id=other_profile_id, + ) + conn.commit() + + assert get_recording(conn, other_id) is None + assert get_recording(conn, default_id, profile_id=other_profile_id) is None + assert [item["id"] for item in list_recordings(conn)] == [default_id] + assert [item["id"] for item in list_recordings(conn, profile_id=other_profile_id)] == [ + other_id + ] + + +def test_delete_recording_removes_metadata_only(tmp_path: Path) -> None: + db_file = tmp_path / "fluxtuner.db" + db.init_db(db_file) + media_file = tmp_path / "recording.mka" + media_file.write_bytes(b"media") + + with db.connect(db_file) as conn: + recording_id = add_recording( + conn, + station_name="Flux FM", + source_url="https://radio.example/stream", + file_path=media_file, + started_at="2026-09-02T12:00:00+00:00", + stopped_at="2026-09-02T12:00:20+00:00", + duration_seconds=20.0, + file_size=media_file.stat().st_size, + ) + conn.commit() + + assert delete_recording(conn, recording_id) is True + conn.commit() + assert get_recording(conn, recording_id) is None + + assert media_file.exists() + + +@pytest.mark.parametrize( + ("duration_seconds", "file_size", "message"), + [ + (-1.0, 1, "duration"), + (1.0, -1, "file size"), + ], +) +def test_add_recording_rejects_negative_values( + tmp_path: Path, + duration_seconds: float, + file_size: int, + message: str, +) -> None: + db_file = tmp_path / "fluxtuner.db" + db.init_db(db_file) + + with db.connect(db_file) as conn, pytest.raises(ValueError, match=message): + add_recording( + conn, + station_name="Flux FM", + source_url="https://radio.example/stream", + file_path=tmp_path / "recording.mka", + started_at="2026-09-02T12:00:00+00:00", + stopped_at="2026-09-02T12:00:20+00:00", + duration_seconds=duration_seconds, + file_size=file_size, + ) + + +def test_sqlite_recording_store_persists_completed_session(tmp_path: Path) -> None: + db_file = tmp_path / "fluxtuner.db" + media_file = tmp_path / "recording.mka" + media_file.write_bytes(b"recorded-media") + + store = SqliteRecordingStore(db_path=db_file) + session = RecordingSession( + station_name="Flux FM", + source_url="https://radio.example/stream", + output_path=media_file, + started_at="2026-09-02T12:00:00+00:00", + stopped_at="2026-09-02T12:00:20.500000+00:00", + ) + + recording_id = store.save(session) + + with db.connect(db_file) as conn: + recording = get_recording(conn, recording_id) + + assert recording is not None + assert recording["station_name"] == "Flux FM" + assert recording["duration_seconds"] == 20.5 + assert recording["file_size"] == len(b"recorded-media") + assert recording["file_path"] == str(media_file) + + +def test_sqlite_recording_store_rejects_active_session(tmp_path: Path) -> None: + store = SqliteRecordingStore(db_path=tmp_path / "fluxtuner.db") + session = RecordingSession( + station_name="Flux FM", + source_url="https://radio.example/stream", + output_path=tmp_path / "recording.mka", + started_at="2026-09-02T12:00:00+00:00", + ) + + with pytest.raises(RecordingError, match="active recording"): + store.save(session) + + +def test_sqlite_recording_store_rejects_missing_output_file(tmp_path: Path) -> None: + store = SqliteRecordingStore(db_path=tmp_path / "fluxtuner.db") + session = RecordingSession( + station_name="Flux FM", + source_url="https://radio.example/stream", + output_path=tmp_path / "missing.mka", + started_at="2026-09-02T12:00:00+00:00", + stopped_at="2026-09-02T12:00:20+00:00", + ) + + with pytest.raises(RecordingError, match="does not exist"): + store.save(session) diff --git a/tests/test_tui_recording.py b/tests/test_tui_recording.py new file mode 100644 index 0000000..c66dd89 --- /dev/null +++ b/tests/test_tui_recording.py @@ -0,0 +1,114 @@ +from pathlib import Path +from types import SimpleNamespace + +from fluxtuner.tui import FluxTunerTUI + + +class FakeRecordingManager: + def __init__(self) -> None: + self.active_session = None + self.started_request = None + self.stop_calls = 0 + + def start(self, request): + self.started_request = request + self.active_session = SimpleNamespace( + station_name=request.station_name, + output_path=request.output_path, + ) + return self.active_session + + def stop(self): + self.stop_calls += 1 + active = self.active_session + self.active_session = None + if active is None: + return None + return SimpleNamespace( + recording_id=7, + station_name=active.station_name, + output_path=active.output_path, + ) + + +class FakeTUI: + def __init__(self) -> None: + self.recording_available = True + self.recording_manager = FakeRecordingManager() + self.selected_station = { + "name": "Test Radio", + "url": "https://example.com/stream", + } + self.status = "" + self.button_updates = 0 + self.notifications = [] + + def set_status(self, message: str) -> None: + self.status = message + + def update_record_button(self) -> None: + self.button_updates += 1 + + def stop_recording(self) -> None: + FluxTunerTUI.stop_recording(self) # type: ignore[arg-type] + + def notify(self, message: str, **kwargs) -> None: + self.notifications.append((message, kwargs)) + + +def test_tui_recording_starts_selected_station(monkeypatch, tmp_path: Path) -> None: + app = FakeTUI() + output = tmp_path / "test-radio.mka" + monkeypatch.setattr("fluxtuner.tui.recording_output_path", lambda _name: output) + + FluxTunerTUI.toggle_recording(app) # type: ignore[arg-type] + + request = app.recording_manager.started_request + assert request is not None + assert request.station_name == "Test Radio" + assert request.source_url == "https://example.com/stream" + assert request.output_path == output + assert "Recording: Test Radio" in app.status + + +def test_tui_recording_toggle_stops_active_recording(monkeypatch, tmp_path: Path) -> None: + app = FakeTUI() + output = tmp_path / "test-radio.mka" + monkeypatch.setattr("fluxtuner.tui.recording_output_path", lambda _name: output) + + FluxTunerTUI.toggle_recording(app) # type: ignore[arg-type] + FluxTunerTUI.toggle_recording(app) # type: ignore[arg-type] + + assert app.recording_manager.stop_calls == 1 + assert app.recording_manager.active_session is None + assert "Saved recording #7" in app.status + + +def test_tui_recording_does_not_require_playback(monkeypatch, tmp_path: Path) -> None: + app = FakeTUI() + output = tmp_path / "independent.mka" + monkeypatch.setattr("fluxtuner.tui.recording_output_path", lambda _name: output) + + FluxTunerTUI.toggle_recording(app) # type: ignore[arg-type] + + assert app.recording_manager.started_request is not None + + +def test_tui_recording_requires_selected_station() -> None: + app = FakeTUI() + app.selected_station = None + + FluxTunerTUI.toggle_recording(app) # type: ignore[arg-type] + + assert app.recording_manager.started_request is None + assert app.status == "No station selected to record." + + +def test_tui_recording_reports_missing_ffmpeg() -> None: + app = FakeTUI() + app.recording_available = False + + FluxTunerTUI.toggle_recording(app) # type: ignore[arg-type] + + assert app.recording_manager.started_request is None + assert "ffmpeg" in app.status