docs(peertube): thicken federated video skill against current API research

Research-driven rebuild of the peertube skill (docs.joinpeertube.org REST
reference 8.1.0 + SepiaSearch + server source + live anonymous probes):

- SKILL.md rewritten to the lastfm model: intent-grouped commands, pipeline
  recipes, jq guidance, researched gotchas, When-to-use/When-not-to-use,
  reference routing table. New negative boundary in the description
  (YouTube/Vimeo uploads, video editing, server administration).
- scripts/peertube-cli -> scripts/peertube, rewritten and extended:
  offset (start/count) pagination replaces the nonexistent page param,
  comments fixed to the hyphenated /comment-threads route, server command
  now composes /config/about + /server/stats (canonical paths), search
  gains --search-target with searchTarget=local default and help text
  stating its instance-local scope, new video/comments/channel/account/
  my-videos/logout commands, --server hoisted before or after the
  subcommand, OAuth2 password grant hardened for 2FA (x-peertube-otp)
  and the production client_secret masking behavior, per-instance
  owner-only token file with refresh-before-expiry and revocation.
- references/: auth-and-tokens, search-and-discovery, endpoint-catalog,
  gotchas-field-guide, worked-recipes - all cited to official docs with
  Sources footers (URLs verified live at authoring time).
- scripts/test_peertube.py: 54 offline tests (help, argument errors,
  dry-run plans, mocked OAuth2 persistence/refresh/revocation, handler
  contracts, documented pipeline chains) passing pytest strict-markers,
  unittest discovery, and the proxy-trap zero-egress rerun; one
  env-guarded anonymous live probe (PEERTUBE_LIVE_TESTS=1).
- evals/evals.json: six schema-v1 cases incl. SepiaSearch-scope and
  masked-secret cases plus a should-not-trigger YouTube negative probe.
- README refreshed for humans; root README blurb and skill-triggers row
  synced; marketplace.json/llms.txt regenerated (codex artifacts
  unchanged); test-results/ gitignored (pytest runner artifact).

Co-authored-by: factory-droid[bot] <138933559+factory-droid[bot]@users.noreply.github.com>
This commit is contained in:
Magnus Hedemark
2026-08-29 18:43:00 -04:00
co-authored by factory-droid[bot] <138933559+factory-droid[bot]@users.noreply.github.com>
parent bf1bba6fc6
commit 83e07b9ac2
16 changed files with 2827 additions and 316 deletions
+1 -1
View File
@@ -822,7 +822,7 @@
"./peertube" "./peertube"
], ],
"strict": false, "strict": false,
"description": "Browse PeerTube federated video from the terminal: view videos and channels, search across instances, check server stats, and manage your account. Uses OAuth2 authentication with token persistence. Use when the user mentions PeerTube, federated video, decentralized video platforms, or browsing/uploading to a PeerTube instance." "description": "Browse PeerTube federated video from the terminal — instance stats, latest videos, video detail, comment threads, channels, accounts, instance-local search, and OAuth2 login with per-instance token persistence. Set PEERTUBE_SERVER to any instance; point it at sepiasearch.org for fediverse-wide search. Use when the user mentions PeerTube, federated video, SepiaSearch, or browsing a specific PeerTube instance. Do not use this skill for YouTube/Vimeo uploads, video editing, or installing and administering a PeerTube server."
}, },
{ {
"name": "platform-engineering", "name": "platform-engineering",
+1
View File
@@ -31,6 +31,7 @@ cashew-brain/
# ─── Python ─────────────────────────────────────────────────── # ─── Python ───────────────────────────────────────────────────
__pycache__/ __pycache__/
test-results/
*.py[cod] *.py[cod]
*.egg-info/ *.egg-info/
*.egg *.egg
+1 -1
View File
@@ -377,7 +377,7 @@ Build and maintain owner-approved Primary, Alternate, Contingency, and Emergency
### [peertube](peertube/SKILL.md) ### [peertube](peertube/SKILL.md)
PeerTube federated video platform from the terminal. Browse videos and channels, search across instances, view server info. OAuth2 login with token persistence. Set PEERTUBE_SERVER to point at any instance. PeerTube federated video from the terminal. Browse videos, channels, and comment threads on any instance, search instance-local or the whole fediverse via SepiaSearch, check server stats, and log in with OAuth2 with per-instance token persistence and pagination/search-scope quirks documented.
### [platform-engineering](platform-engineering/SKILL.md) ### [platform-engineering](platform-engineering/SKILL.md)
+1 -1
View File
@@ -92,7 +92,7 @@
- [operational-design](operational-design/SKILL.md): Design and improve operational processes and organizational scaling — process design, operational metrics, compliance and audit, vendor management, and team topology. Covers value stream mapping, BPMN, bottleneck analysis, scaling from 10 to 100 to 1000 people, KPI design, balanced scorecard, SOC 2, ISO 27001, GDPR readiness, RFP processes, SLA design, vendor scorecards, team topologies, Conway's Law, and Dunbar's Number. Do not use for engineering delivery, financial modeling, or technology evaluation. - [operational-design](operational-design/SKILL.md): Design and improve operational processes and organizational scaling — process design, operational metrics, compliance and audit, vendor management, and team topology. Covers value stream mapping, BPMN, bottleneck analysis, scaling from 10 to 100 to 1000 people, KPI design, balanced scorecard, SOC 2, ISO 27001, GDPR readiness, RFP processes, SLA design, vendor scorecards, team topologies, Conway's Law, and Dunbar's Number. Do not use for engineering delivery, financial modeling, or technology evaluation.
- [org-design](org-design/SKILL.md): CHRO methodology — organizational design (team topologies, span of control, reporting structures), talent strategy (make-vs-buy, skill taxonomies, succession planning), compensation frameworks (market benchmarking, equity design, leveling), culture architecture (values codification, rituals, psychological safety), organizational health metrics (eNPS, retention risk, engagement surveys), DEI strategy (inclusive design, equitable systems, belonging). - [org-design](org-design/SKILL.md): CHRO methodology — organizational design (team topologies, span of control, reporting structures), talent strategy (make-vs-buy, skill taxonomies, succession planning), compensation frameworks (market benchmarking, equity design, leveling), culture architecture (values codification, rituals, psychological safety), organizational health metrics (eNPS, retention risk, engagement surveys), DEI strategy (inclusive design, equitable systems, belonging).
- [pace-plan](pace-plan/SKILL.md): Build, coordinate, operate, troubleshoot, exercise, and improve an authorized Primary, Alternate, Contingency, and Emergency communications plan. Use for resilient emergency-communications paths and their ownership, triggers, check-ins, tests, and corrective actions. Do not use for generic incident status messaging, frequency or channel planning, radio programming, or unauthorized transmission and activation. - [pace-plan](pace-plan/SKILL.md): Build, coordinate, operate, troubleshoot, exercise, and improve an authorized Primary, Alternate, Contingency, and Emergency communications plan. Use for resilient emergency-communications paths and their ownership, triggers, check-ins, tests, and corrective actions. Do not use for generic incident status messaging, frequency or channel planning, radio programming, or unauthorized transmission and activation.
- [peertube](peertube/SKILL.md): Browse PeerTube federated video from the terminal: view videos and channels, search across instances, check server stats, and manage your account. Uses OAuth2 authentication with token persistence. Use when the user mentions PeerTube, federated video, decentralized video platforms, or browsing/uploading to a PeerTube instance. - [peertube](peertube/SKILL.md): Browse PeerTube federated video from the terminal — instance stats, latest videos, video detail, comment threads, channels, accounts, instance-local search, and OAuth2 login with per-instance token persistence. Set PEERTUBE_SERVER to any instance; point it at sepiasearch.org for fediverse-wide search. Use when the user mentions PeerTube, federated video, SepiaSearch, or browsing a specific PeerTube instance. Do not use this skill for YouTube/Vimeo uploads, video editing, or installing and administering a PeerTube server.
- [platform-engineering](platform-engineering/SKILL.md): Use this skill when building or operating internal developer platforms: infrastructure as code, CI/CD, container orchestration, service networking, secrets, and observability. Do not use it to define release process, promotion, rollout, or rollback policy; use release-engineering for that delivery model. - [platform-engineering](platform-engineering/SKILL.md): Use this skill when building or operating internal developer platforms: infrastructure as code, CI/CD, container orchestration, service networking, secrets, and observability. Do not use it to define release process, promotion, rollout, or rollback policy; use release-engineering for that delivery model.
- [playwright](playwright/SKILL.md): Operate Playwright for browser automation end to end: author and debug E2E test suites (robust locators, network interception and mocking, parallel workers, accessibility snapshot checks), wire them into CI, and drive headless browsing and scraping with an extract -> validate -> save loop. Use when writing, running, fixing, or scraping with Playwright, when a Playwright CI failure or JSON report needs triage, or when the bundled pwrun script should analyze a run. Do not use for QA strategy or test framework selection (route to qa-methodology), for frontend component or architecture design (route to frontend-engineering), or for Cloudflare/DDoS-GUARD challenge bypass (use flaresolverr). - [playwright](playwright/SKILL.md): Operate Playwright for browser automation end to end: author and debug E2E test suites (robust locators, network interception and mocking, parallel workers, accessibility snapshot checks), wire them into CI, and drive headless browsing and scraping with an extract -> validate -> save loop. Use when writing, running, fixing, or scraping with Playwright, when a Playwright CI failure or JSON report needs triage, or when the bundled pwrun script should analyze a run. Do not use for QA strategy or test framework selection (route to qa-methodology), for frontend component or architecture design (route to frontend-engineering), or for Cloudflare/DDoS-GUARD challenge bypass (use flaresolverr).
- [postgres](postgres/SKILL.md): Operate PostgreSQL instances safely: configuration review, index and query-plan analysis, vacuum and bloat management, WAL archiving and point-in-time recovery, replication and failover, extensions, major-version upgrades, and evidence-based diagnostics with the bundled read-only pgdiag script. Use when running or inspecting a PostgreSQL server, diagnosing performance or backup health, or planning an upgrade or failover. Do not use for application-level data access patterns (that's backend-engineering) or schema design (that's data-architect/data-engineering). - [postgres](postgres/SKILL.md): Operate PostgreSQL instances safely: configuration review, index and query-plan analysis, vacuum and bloat management, WAL archiving and point-in-time recovery, replication and failover, extensions, major-version upgrades, and evidence-based diagnostics with the bundled read-only pgdiag script. Use when running or inspecting a PostgreSQL server, diagnosing performance or backup health, or planning an upgrade or failover. Do not use for application-level data access patterns (that's backend-engineering) or schema design (that's data-architect/data-engineering).
+58 -15
View File
@@ -1,35 +1,78 @@
# PeerTube — Federated Video from the Terminal # PeerTube — Federated Video from the Terminal
Browse videos, channels, and server info on any PeerTube instance. Search across the fediverse, list channels, check your account stats, and manage authentication. Browse any PeerTube instance from the command line: latest videos, video detail, comment
threads, channels and accounts, instance stats, and OAuth2 login for your own account —
plus fediverse-wide search through SepiaSearch.
## Why Install This Skill ## Why Install This Skill
When your agent loads this skill, it can **navigate the federated video universe** without a browser. That means: When your agent loads this skill, it can **navigate the federated video universe** without
a browser. That means:
- **Browse videos** — recent uploads from any instance - **Browse any instance** — latest videos with real offset pagination (the API has no
- **Search across instances** — find content in the fediverse `page` parameter, and most naive wrappers get this wrong)
- **Explore channels** — list channels and their videos - **Search the right scope** — instance-local search or the whole fediverse via
- **Check server info** — instance name, description, user/video/view stats SepiaSearch, with the `searchTarget` semantics documented instead of guessed
- **Authenticate** — OAuth2 login with token persistence - **Inspect videos deeply** — full metadata, comment threads (the hyphenated
`/comment-threads` route), channels, and accounts by handle (`name@host`)
- **Check instance health** — name, description, and user/video/view counters composed
from `/config/about` + `/server/stats` anonymously
- **Authenticate safely** — OAuth2 password grant with per-instance, owner-only token
persistence, automatic refresh, and proper server-side revocation on logout
- **Avoid the traps** — masked `client_secret` responses, token lifetimes that vary per
instance, 2FA `x-peertube-otp`, rate-limit headers, RFC7807 error bodies
Every command is read-only except `login`/`logout`, and `--dry-run` previews any request
without touching the network.
## What You Get ## What You Get
| Directory | Purpose | | Path | Purpose |
|-----------|---------| |------|---------|
| `SKILL.md` | Complete command reference with auth setup | | `SKILL.md` | Complete command reference with setup, gotchas, and recipes |
| `scripts/peertube-cli` | CLI tool for PeerTube API | | `scripts/peertube` | CLI for PeerTube API operations (`--json`, `--dry-run`, `--verbose`) |
| `scripts/test_peertube.py` | Offline test suite (all HTTP mocked, zero egress) |
| `references/auth-and-tokens.md` | The full OAuth2 flow, secret masking, token hygiene |
| `references/search-and-discovery.md` | Instance-local vs SepiaSearch search scopes |
| `references/endpoint-catalog.md` | Endpoint-by-endpoint parameters and response shapes |
| `references/gotchas-field-guide.md` | Failure signatures and version drift |
| `references/worked-recipes.md` | Multi-step CLI/jq and curl workflows |
| `evals/evals.json` | Behavioral eval cases including negative triggers |
## Quick Start ## Quick Start
```bash ```bash
export PEERTUBE_SERVER="https://your-instance.example.com" export PEERTUBE_SERVER="https://<INSTANCE_HOST>" # any PeerTube instance
peertube-cli auth login --username "myuser" --password "mypassword" scripts/peertube server # instance stats, anonymous
scripts/peertube videos --limit 5 --json
scripts/peertube search --query "linux" # searches THIS instance
```
Fediverse-wide search through SepiaSearch (same API shape, wider index):
```bash
PEERTUBE_SERVER="https://sepiasearch.org" scripts/peertube search --query "linux"
```
Optional login for your own account commands:
```bash
scripts/peertube login --username "<USERNAME>" --prompt
scripts/peertube me --json | jq '.role.label'
scripts/peertube logout # revokes server-side + deletes token file
``` ```
## Triggers ## Triggers
Load this for PeerTube, federated video, decentralized video platforms, or browsing PeerTube content. Load this when asking about PeerTube, federated video, decentralized video platforms,
SepiaSearch, browsing a specific PeerTube instance's videos or channels, or PeerTube API
authentication.
## Requirements ## Requirements
Python 3.8+ with `requests` library. Python 3.8+ with `requests`. One thing this skill always needs from you: **an instance
host** — export `PEERTUBE_SERVER` (e.g. `https://<INSTANCE_HOST>`) or pass
`--server https://...` per command, since PeerTube is federated and every command targets
one instance. Reads are anonymous; `me`/`my-videos` need a token from `scripts/peertube
login`. Tokens persist to `~/.config/peertube/token.json` (override the directory with
`PEERTUBE_CONFIG_DIR`). Find public instances at [joinpeertube.org](https://joinpeertube.org).
+205 -59
View File
@@ -1,116 +1,262 @@
--- ---
name: peertube name: peertube
description: 'Browse PeerTube federated video from the terminal: view videos and channels, description: Browse PeerTube federated video from the terminal — instance stats, latest
search across instances, check server stats, and manage your account. Uses OAuth2 videos, video detail, comment threads, channels, accounts, instance-local search, and
authentication with token persistence. Use when the user mentions PeerTube, federated OAuth2 login with per-instance token persistence. Set PEERTUBE_SERVER to any instance;
video, decentralized video platforms, or browsing/uploading to a PeerTube instance.' point it at sepiasearch.org for fediverse-wide search. Use when the user mentions
PeerTube, federated video, SepiaSearch, or browsing a specific PeerTube instance.
Do not use this skill for YouTube/Vimeo uploads, video editing, or installing and
administering a PeerTube server.
license: MIT license: MIT
compatibility: Requires PEERTUBE_SERVER env var (set to your instance URL, e.g. https://watch.nousresearch.com), compatibility: Requires Python 3.8+ and `requests`. Reads are anonymous; authenticated
Python 3.8+, and the `requests` library. OAuth2 tokens persisted to ~/.config/peertube-cli/token.json. commands (`me`, `my-videos`) need a token from `scripts/peertube login`. Tokens persist
per-instance to ~/.config/peertube/token.json (owner-only).
metadata: metadata:
tags: peertube, federated-video, video-platform, activitypub, api-client tags: peertube, federated-video, activitypub, video-platform, sepiasearch, api-client
sources: https://joinpeertube.org/, https://docs.joinpeertube.org/api/reference sources: https://docs.joinpeertube.org/api-rest-reference.html, https://sepiasearch.org/
--- ---
# peertube-cli — PeerTube Federated Video # peertube — PeerTube federated video from the terminal
Browse videos, channels, and server info on any PeerTube instance. Search across the fediverse, list channels, check your account stats, and manage authentication — all from the terminal. Browse any PeerTube instance — a federated deployment, not a single API — from the
terminal: instance stats, latest videos, full video detail, comment threads, channels,
accounts, and instance-local search. Authenticate with OAuth2 only for your own account
commands. Every command is read-only except `login`/`logout`.
## Setup ## Setup
1. Set the PeerTube instance URL: 1. Choose the instance to talk to. Every command is per-instance; the API shape is
identical everywhere, but accounts, tokens, rules, and catalogs are not:
```bash ```bash
export PEERTUBE_SERVER="https://your-instance.example.com" export PEERTUBE_SERVER="https://<INSTANCE_HOST>" # e.g. https://tilvids.com
``` ```
2. (Optional) Log in for authenticated operations: To search the whole fediverse instead of one instance, point the same variable at the
public search index: `PEERTUBE_SERVER=https://sepiasearch.org` (same API shape — see
[references/search-and-discovery.md](references/search-and-discovery.md)).
2. Nothing else is required to browse: videos, search, channels, comments, and instance
info are anonymous reads.
3. (Optional) Log in only for your own account commands (`me`, `my-videos`):
```bash ```bash
peertube-cli auth login --username "your-username" --password "your-password" scripts/peertube login --username <NAME> --prompt
``` ```
The OAuth2 token is persisted to `~/.config/peertube-cli/token.json` and automatically reused. `--dry-run` works without authentication. ### How authentication works
PeerTube uses plain OAuth2 with per-instance client credentials: the CLI anonymously
fetches the client pair from `GET /api/v1/oauth-clients/local` (singular `local`), then
exchanges your username/password for a bearer token at `POST /api/v1/users/token`
(`grant_type=password`, form-encoded). The token rides `Authorization: Bearer <token>`,
lives for the instance's configured lifetime (read `expires_in` from the response — do not
assume a fixed number), and is refreshed automatically when it expires. The token file is
written owner-only to `~/.config/peertube/token.json` keyed by server URL.
**Do not commit tokens** — they are account credentials; revoke with `scripts/peertube
logout` (`POST /users/revoke-token`) when done. Current production instances mask
`client_secret` in the API response; the CLI detects this and explains the workaround.
Details and wire-level error signatures:
[references/auth-and-tokens.md](references/auth-and-tokens.md).
## Essential Commands ## Essential Commands
### auth login — Authenticate to a PeerTube instance ### server — instance stats and identity (anonymous)
```bash ```bash
peertube-cli auth login --username "myuser" --password "mypassword" # login scripts/peertube server # name, description, user/video/view counters
scripts/peertube server --json
``` ```
Token is saved to `~/.config/peertube-cli/token.json` and reused on subsequent calls. Tokens expire after the server-configured lifetime (typically 24h). Composes `GET /config/about` + `GET /server/stats` (canonical paths — there is no
`/instance/stats`).
### server — Instance information ### videos — browse the instance's uploads (anonymous)
```bash ```bash
peertube-cli server # instance name, description, stats scripts/peertube videos # latest 15, offset pagination
peertube-cli server --json # machine-readable scripts/peertube videos --limit 50 --offset 50
scripts/peertube videos --sort -views --json # popular first
``` ```
Shows: instance name, short description, total users, total videos, total views. Pagination is `start`/`count` offsets (max count 100) — the API has **no `page`
parameter**.
### videos — Browse recent videos ### search — find videos on THIS instance (anonymous)
```bash ```bash
peertube-cli videos # last 12 videos scripts/peertube search --query "linux" # instance-local (searchTarget=local)
peertube-cli videos --limit 24 # more results scripts/peertube search -q "docker" --limit 20 --json
peertube-cli videos --json # machine-readable PEERTUBE_SERVER="https://sepiasearch.org" scripts/peertube search -q "linux" # fediverse-wide
``` ```
Shows: title, duration, views, author/channel, publish date. The bundled CLI performs **instance-local** search only (`searchTarget=local`). For
fediverse-wide search, point `PEERTUBE_SERVER` at SepiaSearch — same commands, wider
index. Search results carry `channel.host`/`url`, the origin instance of federated hits.
### search — Search videos across the fediverse ### video — full detail for one video (anonymous)
```bash ```bash
peertube-cli search --query "linux tutorial" # search videos scripts/peertube video --id <UUID> # numeric id, UUID, or shortUUID all work
peertube-cli search -q "peer" --limit 24 # more results scripts/peertube video --id <UUID> --json | jq '{name, description, views, url}'
peertube-cli search -q "docker" --json # machine-readable
``` ```
### channels — List video channels ### comments — top-level comment threads (anonymous)
```bash ```bash
peertube-cli channels # all channels on the instance scripts/peertube comments --id <UUID> # GET /videos/{id}/comment-threads
peertube-cli channels --json # machine-readable with subscriber counts scripts/peertube comments --id <UUID> --limit 30 --json
``` ```
Shows: display name, channel handle (@name), video count, subscriber count. ### channels / channel / account — creators (anonymous)
### me — Your profile
```bash ```bash
peertube-cli me # your account stats scripts/peertube channels --limit 20 --json # instance channel list
peertube-cli me --json # machine-readable scripts/peertube channel --handle framasoft@framatube.org # name or name@host
scripts/peertube account --name chocobozzz@framatube.org
``` ```
Shows: username, role, video count, view count. Requires authentication. `channel` shows metadata plus the channel's uploads (offset-paginated).
## Global Flags ### me / my-videos — your account (requires login)
All flags work in any position:
```bash ```bash
peertube-cli --json videos # flag before subcommand scripts/peertube me --json | jq '.role.label'
peertube-cli videos --json # flag after subcommand scripts/peertube my-videos --limit 50 --json
peertube-cli --dry-run search --query "test" # preview (no API call)
peertube-cli --quiet videos # suppress non-essential output
peertube-cli --verbose channels # detailed logging
``` ```
### login / logout — OAuth2 session management
```bash
scripts/peertube login --username <NAME> --prompt # hidden prompt
echo "<PASSWORD>" | scripts/peertube login --username <NAME> --password-stdin
scripts/peertube login --username <NAME> --otp <CODE> # 2FA-enabled accounts
scripts/peertube logout # revoke server-side + delete file
```
## Global flags
```bash
scripts/peertube --json videos # flag before or after the subcommand
scripts/peertube videos --json
scripts/peertube --dry-run search --query test # request plan, zero network
scripts/peertube --verbose videos --limit 2 # trace requests on stderr
scripts/peertube --server https://tilvids.com server # per-invocation instance override
```
`--dry-run` emits `{"dry_run": true, "method", "path", "params"}` (login adds
`form_fields` names only, never values) — use it to verify a jq chain before running it
live. `--help` and `--dry-run` never require credentials.
## Pipeline recipes
### Search, then inspect the top hit
```bash
scripts/peertube search --query "linux" --limit 5 --json | jq -r '.videos[0].uuid'
scripts/peertube video --id "$(scripts/peertube search -q linux --limit 1 --json | jq -r '.videos[0].uuid')" --json
```
### Page through a channel's uploads
```bash
scripts/peertube channel --handle framasoft@framatube.org --limit 100 --offset 0 --json | jq -r '.videos[].name'
# loop: advance --offset by the returned count until you reach .total (no page param exists)
```
### Instance report card
```bash
scripts/peertube server --json | jq '{name: .instance.name, videos: .stats.totalLocalVideos, users: .stats.totalUsers, views: .stats.totalLocalVideoViews}'
```
### Log in, check quota, log out
```bash
scripts/peertube login --username <NAME> --prompt
scripts/peertube me --json | jq '{username, role: .role.label, quota_bytes: .videoQuota}'
scripts/peertube logout
```
## JSON and jq
`--json` output keys are stable snake_case wrappers around raw API objects: `videos`
(the API's `{total, data}` list objects), `channels`, `threads` (+ `total_not_deleted`),
`instance` + `stats`, `channel`, `dry_run`/`method`/`path`/`params` for plans. Video
objects keep PeerTube's own field names — `uuid`, `shortUUID`, `name`, `duration`
(seconds), `views`, `publishedAt`, `account{name,displayName,host}`,
`channel{name,displayName,host}` — so jq selectors transfer directly to raw `curl`
against `/api/v1`. Example: `jq -r '.videos[] | [.name, .views, .channel.displayName] | @tsv'`.
## Known Gotchas ## Known Gotchas
- **Set PEERTUBE_SERVER first** — Without this env var, the CLI defaults to `https://your-instance.example.com` (which won't resolve). Always export the correct instance URL. - **Instances are independent (federated, not one API)** — accounts, tokens, rules,
- **Authentication is required for most commands** — `server` and public video browsing work without auth. `me`, `channels`, and personal video lists require a valid OAuth token. Use `--dry-run` to preview without auth. enabled features, and catalogs differ per instance. A token from instance A 401s on
- **Token is persisted automatically** — After `auth login`, the token is saved to `~/.config/peertube-cli/token.json`. No need to login again unless the token expires. Delete this file to force re-login. instance B; the CLI keys the token file by server URL. Content federated *onto* an
- **Token expiry** — PeerTube OAuth2 tokens have a configurable expiry (default ~24h). Expired tokens cause 401 errors. Re-run `auth login` to refresh. instance still belongs to its origin (`channel.host`, video `url`).
- **Cross-instance search** — `search --query` searches across the fediverse, not just the local instance. Results may include videos from remote instances. - **Search scope is two different things** — `searchTarget=local` searches the
- **API pagination** — PeerTube uses offset-based pagination. The `--limit` flag controls the page size (default: 12 for videos, 15 for channels). instance's own catalog; `search-index` (or SepiaSearch's base URL) searches the
- **Rate limits** — PeerTube instances have configurable rate limits. The CLI does not auto-retry on 429 responses. fediverse via an external index. Omitting `searchTarget` gives the instance's own
scope on current servers, not the fediverse. The bundled CLI is instance-local unless
you point it at sepiasearch.org.
- **`page` does not exist** — collections paginate with `start`/`count` (max 100).
Clients sending `page=` silently re-read the first page forever.
- **The comments route is `/comment-threads`** (hyphenated) — `/comments` and
`/commentthreads` are not routes (they 400 on current servers).
- **Instance metadata paths are mixed** — stats at `/server/stats` (operation titled
"instance stats"), about at `/config/about`, config at `/config`. No
`/instance/*` metadata paths exist.
- **Production masks `client_secret`** — `oauth-clients/local` answers
`"********************************"` on current production instances; a token request
with the masked value 400s. The CLI detects it and explains the front-end-asset
workaround. `response_type=code` appears in old quick-start curls but is not part of
the current token schema — the CLI omits it.
- **Token lifetimes are instance-configurable** — read `expires_in` per response; the
CLI persists the absolute `expires_at` and refreshes automatically. Store tokens
owner-only, never commit them, revoke on logout (deleting the file alone leaves the
session live).
- **2FA needs an OTP header** — `x-peertube-otp` on the token request; the CLI maps a
bare 401 to "pass --otp".
- **Rate limits** — default 50 calls/10 s per IP (token endpoint tighter); on 429 read
`Retry-After` and back off. Errors use RFC7807 `application/problem+json` bodies, and
unknown routes answer 400 (not 404) — read the body.
- **`duration` is seconds**; ids are triple (`id`, `uuid`, `shortUUID` — all accepted by
detail endpoints); `role` is an object `{id, label}`; `videoQuota` is bytes.
- **Anonymous vs authed** — browsing/search/comments/instance-info need no token;
`/users/me*` and mutations always do.
## References ## When to use
- [scripts/peertube-cli](scripts/peertube-cli) — The CLI binary. Built following the cli-builder patterns: `--json`, `--dry-run`, `--quiet`, `--verbose`, dual-output via `emit()`, lazy auth, config file persistence. Use this skill for read-only interaction with PeerTube instances: browsing and filtering
- [PeerTube API Reference](https://docs.joinpeertube.org/api/reference) — Official API reference. videos, instance-local or fediverse-wide search (via SepiaSearch), video detail and
- [JoinPeerTube.org](https://joinpeertube.org/) — Find instances and learn about the federated video platform. comments, channel/account exploration, instance stats, and managing your own account
session with OAuth2 (login, profile, my videos, logout).
## When not to use
Do not use this skill for YouTube, Vimeo, or other platform uploads or any video
editing/transcoding (route to those platforms' own tooling and ffmpeg); for installing,
hosting, or administering a PeerTube server (instance administration is out of scope —
the bundled CLI is read-only plus login/logout); or for generic ActivityPub/Mastodon
federation questions (use a Mastodon or ActivityPub skill).
## Reference Files
| File | Use it for |
| ---- | ---------- |
| [references/auth-and-tokens.md](references/auth-and-tokens.md) | OAuth2 flow (oauth-clients/local, password grant), secret masking, refresh/revocation, token-file hygiene, wire error signatures |
| [references/search-and-discovery.md](references/search-and-discovery.md) | searchTarget local vs search-index, SepiaSearch semantics, search parameters and sorts |
| [references/endpoint-catalog.md](references/endpoint-catalog.md) | Every read endpoint's parameters, response shapes, pagination, rate limits |
| [references/gotchas-field-guide.md](references/gotchas-field-guide.md) | Symptom → cause → fix table for every failure signature and version drift |
| [references/worked-recipes.md](references/worked-recipes.md) | Multi-step CLI/jq workflows, raw curl auth chain, jq processing patterns |
## Available Scripts and Prerequisites
- `scripts/peertube` — the bundled Python CLI (`--json`, `--dry-run`, `--verbose`,
`--server` override). Imports only the standard library and `requests`.
- `scripts/test_peertube.py` — offline test suite (pytest + unittest compatible); all
HTTP is mocked, zero network egress.
- Requires Python 3.8+ and `requests`. Any reachable PeerTube instance (or SepiaSearch)
works; no credentials exist or are required by default. No service is started by this
skill.
+72
View File
@@ -0,0 +1,72 @@
{
"schema_version": 1,
"skill_name": "peertube",
"evals": [
{
"id": "browse-latest-videos-json",
"prompt": "Show me the ten most recent videos on my PeerTube instance (framatube.org) as JSON.",
"expected_output": "Set PEERTUBE_SERVER=https://framatube.org (or pass --server) and run scripts/peertube videos --limit 10 --json. No login is needed: /api/v1/videos is anonymous and returns {total, data} with offset pagination (start/count, no page parameter).",
"assertions": [
"exports PEERTUBE_SERVER or passes --server with the instance host",
"runs scripts/peertube videos with --limit 10 and --json",
"does not require login because public video listing is anonymous",
"pages with start/count offsets, never a page parameter"
]
},
{
"id": "search-then-detail-pipeline",
"prompt": "Search my PeerTube instance for videos about linux, then show me the full details of the best result.",
"expected_output": "Chain scripts/peertube search --query linux --json (instance-local search, searchTarget=local) to get results, extract .videos[0].uuid with jq, then run scripts/peertube video --id <UUID> --json for full metadata. The video detail endpoint accepts the numeric id, UUID, or shortUUID.",
"assertions": [
"starts with scripts/peertube search using --query",
"extracts the uuid field from the search output before the next stage",
"feeds the extracted id into scripts/peertube video --id",
"does not claim the search covered the whole fediverse since the CLI performs instance-local search"
]
},
{
"id": "fediverse-search-via-sepiasearch",
"prompt": "I searched my PeerTube instance for a popular video and got no results, but I know it exists on another instance. How do I search across the whole fediverse?",
"expected_output": "Instance search (searchTarget=local) only finds objects the instance knows. For fediverse-wide search, point the same CLI at SepiaSearch - the public search index that indexes public PeerTube instances and speaks the identical API shape: PEERTUBE_SERVER=https://sepiasearch.org scripts/peertube search --query '<query>'. Follow results back to their origin instance using the account/channel host or the video url field; an instance may also support searchTarget=search-index if its admin enabled an external index.",
"assertions": [
"explains the instance-local versus search-index/fediverse search scopes",
"uses SepiaSearch as the fediverse-wide search base host with the same API shape",
"directs following results to their origin instance via host/url fields",
"mentions searchTarget=local as the explicit instance-scope value"
]
},
{
"id": "oauth-client-secret-masked-login",
"prompt": "I'm writing a script that logs in to a PeerTube instance. GET /api/v1/oauth-clients/local returns client_secret as '********************************' and my subsequent POST to /users/token fails with 400 invalid client. What is going on?",
"expected_output": "Current production instances mask the client_secret in the oauth-clients/local response (the real pair reaches the web front end via its served assets). The correct flow is still: fetch /api/v1/oauth-clients/local (singular 'local'), obtain the unmasked client pair the way the instance's own front end does, then POST /api/v1/users/token with x-www-form-urlencoded fields client_id, client_secret, grant_type=password, username, password - no response_type needed. A 400 can also mean wrong credentials; the bundled CLI detects the masked secret and stops with guidance before sending a doomed token request.",
"assertions": [
"identifies production client_secret masking as the cause of the invalid-client 400",
"names the oauth-clients/local (singular) endpoint as step one",
"lists the exact password-grant form fields including grant_type=password",
"does not treat the masked asterisk value as a usable secret"
]
},
{
"id": "token-persistence-and-logout-hygiene",
"prompt": "How should a PeerTube CLI store the OAuth token after login, and how do I log out properly?",
"expected_output": "Persist the token per-instance in an owner-only file (the bundled CLI uses ~/.config/peertube/token.json, override with PEERTUBE_CONFIG_DIR) recording the server URL, access token, refresh token, and absolute expires_at from the token response - lifetimes are instance-configurable, so never hard-code one. Log out with scripts/peertube logout, which calls POST /api/v1/users/revoke-token (revoking the access and refresh tokens server-side) and then deletes the local file; deleting the file alone leaves a live session. Never commit or log tokens.",
"assertions": [
"stores the token outside the repository in an owner-only location",
"records expiry from expires_in instead of assuming a fixed lifetime",
"revokes server-side via POST /users/revoke-token before deleting the local file",
"keeps tokens per-instance because tokens are not valid across instances"
]
},
{
"id": "youtube-upload-not-peertube",
"prompt": "Help me upload my video to YouTube and trim the intro with ffmpeg.",
"expected_output": "This must not trigger the peertube skill: YouTube uploading and video editing are outside its read-only PeerTube API scope, and PeerTube is a different federated platform from YouTube. Use YouTube's own upload tooling (e.g. youtube-cli or YouTube Studio) and ffmpeg directly for trimming. The PeerTube skill supports no upload operation at all - it is read-only plus login/logout.",
"assertions": [
"must not trigger peertube for YouTube uploads or video editing",
"routes the upload to YouTube's own tooling instead",
"routes the trim to ffmpeg directly",
"does not invent a peertube upload command since the bundled CLI is read-only"
]
}
]
}
+152
View File
@@ -0,0 +1,152 @@
# PeerTube authentication and tokens
How PeerTube's OAuth2 flow actually behaves on the wire, what each failure looks like, and
how to store tokens without leaking them. Every behavioral claim traces to the official
REST reference, the official quick start, or the PeerTube server source (Sources footer).
PeerTube has exactly one authenticated posture: an OAuth2 **bearer access token** minted
from per-instance client credentials. There is no API-key alternative (unlike Jellyfin) and
no header scheme beyond standard `Authorization: Bearer <token>`.
## Step 1 — fetch the instance's OAuth client credentials
`GET /api/v1/oauth-clients/local` (singular `local`, not `locals`) returns the
per-instance client pair. It is anonymous — no authorization block on the operation — and
PeerTube's own web UI calls it before every login:
```json
{ "client_id": "<CLIENT_ID>", "client_secret": "<CLIENT_SECRET>" }
```
**Production servers mask the client_secret.** The current server code returns the real
secret from this endpoint (it is the same secret the web client uses), but production
instances in recent versions respond with the secret replaced by
`"********************************"` — observed live on multiple public instances in
2026 and consistent with the reference page's own masked response example
(`client_secret: "********************************"`). Practical consequences:
- Never persist the response of `oauth-clients/local` as if it were a working secret.
- A login attempt using the masked value fails with HTTP 400 (invalid client). This is
what you are seeing if your script fetched the client pair and the very next token
request 400s on a public instance.
- The legacy workaround mirrors what the web client does: the client pair is embedded in
the instance's front-end JavaScript, and official PeerTube tooling reads it from the
served assets when the API masks it. Treat the masked behavior as version-dependent —
always attempt the API first, then fall back to scraping the served JS bundle if the
secret comes back masked.
The endpoint is also Host-header-guarded: the server compares the request's `Host` header
against its configured webserver hostname and answers **HTTP 403
"Getting client tokens for host ... is forbidden"** when they disagree (proxies that
rewrite the header break clients here; the guard is skipped on test/dev instances).
## Step 2 — the password grant
`POST /api/v1/users/token`, `Content-Type: application/x-www-form-urlencoded`, with form
fields (names exactly as in the reference):
| Field | Required? | Notes |
| --- | --- | --- |
| `client_id` | yes | from step 1 |
| `client_secret` | yes | from step 1 (unmasked) |
| `grant_type` | yes | `password` for login |
| `username` | yes | |
| `password` | yes | |
| `response_type` | no | the official quick-start curl sends `response_type=code`; it is absent from the current OpenAPI request schema. Sending it is harmless; omitting it works with `requests` |
| `x-peertube-otp` | conditional | request header, only when the account has 2FA enabled (server answers 401 without it) |
```bash
curl -X POST "$BASE/users/token" \
-H 'Content-Type: application/x-www-form-urlencoded' \
--data-urlencode 'client_id=<CLIENT_ID>' \
--data-urlencode 'client_secret=<CLIENT_SECRET>' \
--data-urlencode 'grant_type=password' \
--data-urlencode 'username=<USERNAME>' \
--data-urlencode 'password=<PASSWORD>'
```
Success response fields: `access_token`, `token_type` (`"Bearer"`), `expires_in` (seconds),
`refresh_token`, and `refresh_token_expires_in` (seconds; present in the current reference
sample, `1209600` there — a sample value, not a guaranteed default). The quick-start
example shows `expires_in: 14399` (~4 hours). Sample values are not contract: instances can
configure token lifetimes server-side, so read `expires_in` from each response and schedule
refresh from it rather than hard-coding "24 hours" or any other number.
## Refresh, revocation, and lifetime
- **Refresh grant**: `grant_type=refresh_token` is a documented allowed value on the token
endpoint. The rendered reference does not display a `refresh_token` form-field row, so
the exact refresh request body is not fully specified in official docs; standard OAuth2
practice (send `refresh_token` alongside the client pair) is the community-established
shape, but verify against your instance's version before relying on it.
- **Revocation**: `POST /api/v1/users/revoke-token` with `Authorization: Bearer <token>`,
no body, returns HTTP 200 and revokes the access token **and** its associated refresh
token, destroying the session. This is the correct "logout" operation: revoke before
discarding a stored token.
- **Lifetimes**: no official page documents default lifetime values or the server config
keys that change them; the only official evidence is the sample `expires_in: 14399` /
`refresh_token_expires_in: 1209600` (~4 h / ~14 d). Server operators can adjust token
lifetimes via their production config (all options documented as living in
`config/default.yaml` overridable by `production.yaml`), so treat expiry as per-instance
and honor `expires_in`.
## Error signatures on the wire
| Symptom | Status | Meaning / response |
| --- | --- | --- |
| Bad client_id/client_secret, or the masked secret, or wrong username/password | `400` on `POST /users/token` | Reference documents 400 for invalid client or credentials; bodies are RFC7807-style (`application/problem+json` with `type`, `title`, `status`, `detail`, sometimes `code`) |
| 2FA enabled, no `x-peertube-otp` header | `401` on `POST /users/token` | header must be supplied on the token request |
| Expired/revoked token on any authenticated call | `401` | re-run the password grant (or refresh) |
| Wrong `Host` header reaching `oauth-clients/local` | `403` | proxy/header rewriting problem, not auth |
| Rate limit exceeded | `429` | all endpoints are rate-limited; the token endpoint is tighter than most (documented sample: 15 calls per 5 minutes). Inspect `Retry-After` (seconds) and `X-RateLimit-Limit` / `X-RateLimit-Remaining` / `X-RateLimit-Reset` (Unix timestamp) and back off |
| Connection refused / DNS failure | no HTTP response | transport failure — classify separately from API errors; usually `PEERTUBE_SERVER` is wrong or unreachable |
Anonymous-read endpoints (videos, search, channels, `/config`, `/config/about`,
`/server/stats`) need no token at all. Authenticated-only calls include `/users/me`,
`/users/me/videos`, and any state-changing operation. The API answers `401` when a call
needs a token you did not send.
## Token persistence hygiene
PeerTube's docs do not prescribe storage mechanics, so a CLI should follow these
sanctioned-by-logout-support practices:
1. **Store under the user's own profile, not in the repo.** The bundled CLI defaults to
`~/.config/peertube/token.json` (override with `PEERTUBE_CONFIG_DIR` for tests). Never
write tokens into a working tree, a shell history, or an eval manifest.
2. **Restrict file permissions.** Create the directory and file so only the owner can read
the token file (e.g. `os.makedirs(..., mode=0o700)` and `0o600` on the file).
3. **Persist the refresh token alongside the access token and the server base URL**, plus
the absolute `expires_at` computed from `expires_in`. A token file is only valid for
the instance it was minted by — re-authenticate when `PEERTUBE_SERVER` changes.
4. **Refresh before expiry; fall back to password re-grant.** Because refresh-request
semantics are underspecified in official docs, treat refresh as an optimization: try
`grant_type=refresh_token`, and on any failure re-run the password grant.
5. **Revoke on logout.** `POST /users/revoke-token` invalidates both tokens server-side,
then delete the local file. Deleting the file alone leaves a live session behind.
6. **Never commit or log tokens.** Examples everywhere in this skill use
`<ACCESS_TOKEN>`-style placeholders. If a token file ever lands in a diff, revoke it —
deleting the file does not invalidate the session.
7. **Multi-instance note**: one token file per server URL (or include the server in the
file) avoids "works on instance A, 401 on instance B" confusion when switching
`PEERTUBE_SERVER`.
## Detection and headers for API clients
- API responses carry `x-powered-by: PeerTube` and `/api/*` is CORS-enabled; HTML pages
include `<meta property="og:platform" content="PeerTube">`; NodeInfo is exposed at
`/nodeinfo/2.0.json`. Any of these distinguishes a PeerTube instance from other servers.
- No special `User-Agent` is required. Use `Accept: application/json`.
## Sources
- https://docs.joinpeertube.org/api-rest-reference.html (Session: getOAuthClient,
getOAuthToken, revokeOAuthToken; Errors; Rate-limits; CORS; Config; Stats)
- https://docs.joinpeertube.org/api/rest-getting-started (client fetch, password grant
curl, token response example, instance detection)
- https://docs.joinpeertube.org/maintain/configuration (config file layering)
- https://github.com/Chocobozzz/PeerTube/blob/develop/support/doc/api/openapi.yaml
(generated OpenAPI spec; openapi-generator clients)
- https://raw.githubusercontent.com/Chocobozzz/PeerTube/develop/server/core/controllers/api/oauth-clients.ts
(Host-header guard; response construction)
- Live anonymous probes of public instances (`oauth-clients/local` masking, 400 on token
misuse), 2026-08-29.
+131
View File
@@ -0,0 +1,131 @@
# PeerTube endpoint catalog for CLI clients
The read surface of the PeerTube REST API with exact parameter names, response shapes, and
pagination semantics — everything a CLI needs to list, filter, and page through videos,
channels, accounts, and instance metadata. Base path: `/api/v1` on any instance
(`https://<INSTANCE_HOST>/api/v1`). Sources footer cites the official reference; a few
shapes were additionally confirmed by live anonymous probes (noted inline).
## The one pagination model: start/count offsets
Every collection endpoint uses **offset pagination**: query params `start` (integer >= 0)
and `count` (1100, **default 15**). There is no `page` parameter anywhere in the current
API — a client sending `page=` silently gets default paging while believing it paginated
(this bit the original bundled CLI). Responses wrap as:
```json
{ "total": 23792, "data": [ /* resource objects */ ] }
```
Loop by advancing `start` by the number of rows received until `start >= total` (or an
empty page). `skipCount=true` on video collections/search omits the `total` computation —
faster, but then you must stop on the first short/empty page. Max `count` per request is
100; a `count` above the allowed range is rejected.
## Videos
| Endpoint | Auth | Notes |
| --- | --- | --- |
| `GET /videos` | anonymous | instance-wide video list; filters below |
| `GET /videos/{id}` | anonymous | full detail; `{id}` accepts **numeric id, UUIDv4, or shortUUID** |
| `GET /videos/{id}/comment-threads` | anonymous | top-level comment threads; `start`, `count`, `sort` in {-createdAt, -totalReplies}; response `{total, totalNotDeletedComments, data}` |
- The comments route is **`/comment-threads`** (hyphenated). `/comments` and
`/commentthreads` are not the route (probes: `/comments` 400s on current servers; the
OpenAPI shows `/comment-threads`). A newer `/videos/{id}/comments/{commentId}/replies`
route (v8.3 changelog) fetches replies, not top-level threads.
- Listing filters (current exact names): `start`, `count`, `sort`, `categoryOneOf`,
`tagsOneOf`, `tagsAllOf`, `languageOneOf`, `licenceOneOf`, `nsfw`, `nsfwFlagsIncluded`,
`nsfwFlagsExcluded`, `isLive`, `isLocal`, `host`, `skipCount`, `search`, plus
admin-only `include`/`privacyOneOf`/`stateOneOf` (>=8.2)/`autoTagOneOf` (>=6.2) and
file-format filters `hasHLSFiles`/`hasWebVideoFiles`.
- Sort values: `name`, `-duration`, `-createdAt`, `-publishedAt`, `-views`, `-likes`,
`-comments`, `-trending`, `-hot`, `-best`.
- List-item shape (probe-confirmed field names): `id`, `uuid`, `shortUUID`, `url`, `name`,
`category{id,label}`, `licence{id,label}`, `language{id,label}`, `privacy{id,label}`,
`nsfw`, `truncatedDescription`, `duration` (**seconds** — sample `1419` is ~23.6 min),
`views`, `likes`, `dislikes`, `comments`, `publishedAt`/`originallyPublishedAt`/`createdAt`
(ISO-8601), `isLocal`, `isLive`, thumbnail/preview `path`s, and actor summaries:
`account{id,name,displayName,host,url,avatars[]}`,
`channel{id,name,displayName,host,url,avatars[]}`.
- `account`/`channel` `host` tells you the **origin instance** of a federated video — on a
search-index result this is how you find where the video actually lives.
- Detail adds full `description`, `files[]`/`streamingPlaylists[]` (resolutions,
`fileUrl`/`fileDownloadUrl`, `metadataUrl`s), `commentsEnabled`, `downloadEnabled`,
`trackerUrls`, `support`, `tags`, `scheduledUpdate` for scheduled/live videos.
## Channels and accounts
| Endpoint | Auth | Notes |
| --- | --- | --- |
| `GET /video-channels` | anonymous | **does exist** (current reference): lists the instance's channels, `start`/`count`/`sort`, `{total,data}` |
| `GET /video-channels/{channelHandle}` | anonymous | handle format `my_username` or `my_username@example.com` (`name@host` for remote channels) |
| `GET /video-channels/{channelHandle}/videos` | anonymous | channel's videos, standard video filters + offset pagination |
| `GET /accounts/{name}` | anonymous | account actor; 404 for unknown; `name` accepts `chocobozzz` or `chocobozzz@example.org` |
| `GET /accounts/{name}/videos` | anonymous | account's videos, offset pagination |
| `GET /accounts/{name}/video-channels` | anonymous | an account's channels |
| `GET /search/video-channels` | anonymous | see search-and-discovery.md |
Channel object fields include `name`, `displayName`, `host`, `url`, `avatars`,
`followersCount` (subscribers), `videosCount` — but note the **global** `/video-channels`
list rows additionally observed carrying `videosCount`/`followersCount` per channel in
list responses (probe 2026-08-29). Historical route drift: pre-1.0 `/videos/channels/*`
routes became `/video-channels/*` and `/videos/accounts/{id}/channels` became
`/accounts/{id}/video-channels` (changelog, v1.0.0-beta.4) — ancient wrappers still using
the old shapes will 404.
## Instance metadata (all anonymous, all public)
| Endpoint | Returns |
| --- | --- |
| `GET /config` | public runtime configuration: `client{}`, `defaults{}`, `webadmin{}`, and an `instance{}` block with `name`, `shortDescription`, classifications, customization, avatars/banners |
| `GET /config/about` | `{instance:{name, shortDescription, description, terms, codeOfConduct, hardwareInformation, administrationInformation, maintenanceInformation, businessInformation, languages, categories, banners}}` |
| `GET /server/stats` | instance counters: `totalUsers`, `totalLocalVideos`, `totalLocalVideoViews`, `totalLocalVideoDownloads`, `totalLocalVideoComments`, `totalVideos`, `totalVideoComments`, `totalLocalVideoChannels`, `totalLocalDailyActiveVideoChannels`, `totalLocalVideoChannels`, `totalLocalVideoPlaylists`, moderation/registration counters, activity-processing stats. Public and cached by the server. |
| `GET /nodeinfo/2.0.json` | standard NodeInfo document (software name/version, usage counts) — handy for instance detection |
**Naming trap:** the stats operation is titled "Get instance stats" but the canonical
current path is **`/server/stats`** (there is no `/instance/stats`), while the config
endpoints are **`/config`** and **`/config/about`** (there is no `/instance/config` or
`/instance/about`). Mixed naming is current reality, not a docs bug. A CLI's `server` /
`info` command should compose `/config/about` + `/server/stats` to give name, description,
and user/video/view counts in one screenful.
## My user (OAuth2 required)
| Endpoint | Notes |
| --- | --- |
| `GET /users/me` | identity + preferences: `id`, `username`, `email`, `role{id,label}`, `videoQuota`, `videoQuotaDaily`, `account{}`, `videoChannels[]`, `twoFactorEnabled`, theme/NSFW/p2p preferences, `createdAt`. The current reference sample is rendered as an array; every live server returns a **single user object** — clients should tolerate both. |
| `GET /users/me/videos` | `{total, data}` of your uploads with the standard video-list fields and filters (`start`, `count`, `sort`, privacy/scope filters) |
The `role` block is `{id, label}` (e.g. `{id: 1, label: "User"}`); `videoQuota` is bytes.
Channel rows inside `videoChannels` carry the same `name`/`displayName`/`host` actor shape
used everywhere else.
## Rate limits (all endpoints)
Default server-side limiter: **50 calls per 10 seconds** per IP across `/*` (the token
endpoint is documented at a tighter 15 per 5 minutes in its operation docs; administrators
can customize all values). On exhaustion you get **HTTP 429** with
`X-RateLimit-Limit`, `X-RateLimit-Remaining`, `X-RateLimit-Reset` (Unix timestamp) and
`Retry-After` (seconds). A CLI should read `Retry-After` and back off; aggressive parallel
listing (count=100 × many pages) on a small instance will trip the limiter.
## Error bodies
Errors use RFC7807-style `application/problem+json` documents with `type`, `title`,
`status`, `detail`, and sometimes a `code`. Unknown routes on current servers typically
answer 400 (not the classic 404) with an `error` body — check the body, not just the
status, when a route mysteriously "doesn't exist".
## Sources
- https://docs.joinpeertube.org/api-rest-reference.html (getVideos, getVideo,
getVideoChannels, getVideoChannel, getVideoChannelVideos, getAccount, getAccountVideos,
searchChannels, getConfig, getAbout, getInstanceStats, getUserInfo, comment-threads
operations; Errors and Rate-limits sections)
- https://docs.joinpeertube.org/api/rest-getting-started (pagination/filter basics,
instance detection via NodeInfo / x-powered-by / og:platform)
- https://docs.joinpeertube.org/CHANGELOG (route renames v1.0.0-beta.4; v8.2 stateOneOf;
v8.3 comment routes)
- Live anonymous probes on a public instance (list shapes, channel list fields,
`/config/about`, `/server/stats`, `/comment-threads` vs `/comments` status), 2026-08-29.
+125
View File
@@ -0,0 +1,125 @@
# PeerTube gotchas field guide
Failure signatures and behavioral traps, distilled from the official docs, the server
source, and live probes. Each entry: symptom → cause → what to do.
## Instance plurality (the big one)
- **Symptom**: same CLI command works on one instance and 400s/401s/empty-results on
another; or a token that worked on instance A 401s on instance B.
- **Cause**: PeerTube is federated software, not a single API. Every instance is an
independent deployment with its own rules, allowances, moderation policy, enabled
features (NSFW policy, search-index support, registration, transcoding) and its own user
accounts and OAuth tokens. A token minted by instance A is meaningless to instance B;
instance B may have closed registrations, disabled uploads, or set its own NSFW default.
- **Do**: always configure the instance host per operation (`PEERTUBE_SERVER` or
`--server`); keep per-instance token files; never assume an account or video exists on a
different instance. Federated content viewed on instance X still **belongs** to the
origin instance (`channel.host` / `account.host` / video `url` tell you which).
## Search scope confusion
- **Symptom**: "search across the fediverse" expectations return only a handful of local
results; or results reference videos the instance doesn't host.
- **Cause**: `searchTarget` has two scopes: `local` (instance-known objects only) and
`search-index` (external fediverse index, admin-enabled). Omitting the parameter gives
the instance's own scope on current servers (observed), not the fediverse.
- **Do**: pass `searchTarget=local` explicitly for instance scope; use SepiaSearch
(`https://sepiasearch.org/api/v1/search/videos`) for fediverse-wide scope. Index results
point at origin instances — follow `channel.host`/`url` rather than expecting the
queried instance to serve them.
## Pagination: `start`/`count`, never `page`
- **Symptom**: client pages with `page=1&count=15` and gets identical results forever.
- **Cause**: the API has no `page` parameter; unknown params are ignored, so `page=1`
requests silently return the first `count` rows every time.
- **Do**: advance `start` by the page size until `start >= total` or an empty page. Max
`count` is 100 (higher values are rejected). `skipCount=true` trades the `total` field
for speed — then you must stop on the first short page.
## Comment route spelling
- **Symptom**: fetching comments with `/videos/{id}/comments` or
`/videos/{id}/commentthreads` returns 400 (current servers answer 400, not 404, for bad
routes — see below) while other endpoints work.
- **Cause**: the route is `GET /videos/{id}/comment-threads` (hyphenated). The v8.3
`/comments/{commentId}/replies` route is for replies, not top-level threads.
- **Do**: use `/comment-threads` with `start`/`count`/`sort=-createdAt|-totalReplies`.
## Instance metadata endpoint names
- **Symptom**: `/instance/stats`, `/instance/about`, `/instance/config` all 400/404.
- **Cause**: mixed current naming: stats live at **`/server/stats`** (operation *titled*
"Get instance stats"), about at **`/config/about`**, config at **`/config`**.
- **Do**: compose `/config/about` + `/server/stats` for a full instance picture.
## oauth-clients/local secret masking
- **Symptom**: `GET /oauth-clients/local` returns
`"client_secret": "********************************"`; the following token request 400s
with invalid_client.
- **Cause**: current production servers mask the secret in this response (the value is
still delivered to the web client via served front-end assets; the API response masks
it). Older instances/versions return the real secret.
- **Do**: detect the masked value; if masked, obtain the client pair from the instance's
served front-end JS (the same source its own web UI uses) before the token request. Never
persist the masked string as a secret. The endpoint is also Host-header-guarded (403 if
the `Host` header disagrees with the configured webserver hostname — mind reverse
proxies).
## Auth error signatures
| Status | Where | Meaning |
| --- | --- | --- |
| 400 on `POST /users/token` | invalid client pair (including the masked-secret case) or wrong credentials | RFC7807-style `application/problem+json` body; check `detail` |
| 401 on `POST /users/token` | account has 2FA and no `x-peertube-otp` header supplied | supply OTP header |
| 401 on authenticated GETs | token expired/revoked/malformed, or missing | re-run password grant |
| 403 on `oauth-clients/local` | Host-header mismatch (proxy misconfiguration) | fix the proxy/Host |
| 429 anywhere | rate limit (default 50 req/10 s; token endpoint tighter) | read `Retry-After` + `X-RateLimit-*` headers, back off |
| 400 on unknown routes | current servers answer 400 with an error body for unrecognized API routes | read the body; the classic "404 means missing route" assumption misleads here |
| connection errors | wrong/unreachable `PEERTUBE_SERVER` | no HTTP response at all; classify as transport failure |
## Shape and value traps
- **duration is seconds** (integer). Sample list value `1419` = 23:39, not milliseconds.
- **ids are triple**: numeric `id`, `uuid` (UUIDv4), and `shortUUID` — all three are
accepted by `/videos/{id}` and family; `uuid` is the safest portable choice in scripts.
- **`users/me` sample is an array in the docs**; live servers return a single object.
Tolerate both when writing generic parsers.
- **`role` is an object** `{id, label}` on `/users/me` — don't stringify the dict.
- **`videoQuota` is bytes** (large integer).
- **`{total, data}` everywhere**: collections never wrap in `{"videos": []}` at the API
layer (the bundled CLI adds that key in its JSON output; know which layer you're
reading).
- **`nsfw` filter is a string** (`"true"`/`"false"`) in query params.
- **filter names end in `OneOf`/`AllOf`** (`categoryOneOf`, `tagsAllOf`, ...); bare
`category=` from old wrappers is ignored silently.
- **federated results**: a video listed on instance X may be hosted on instance Y
(`account.host`/`channel.host`). Views/likes counters are local-ish and eventually
consistent across the federation — don't expect exact global numbers.
## Version drift
- Docs reference page currently identifies PeerTube **8.1.0** while the changelog already
carries 8.3.0 material — instance versions vary; validate optional parameters
(`stateOneOf` >= 8.2, `autoTagOneOf` >= 6.2) before relying on them.
- Historical renames worth knowing when reading old code: `/videos/channels/*`
`/video-channels/*`, `/videos/accounts/{id}/channels``/accounts/{id}/video-channels`
(v1.0.0-beta.4).
- Refresh-token request fields are underspecified in official docs; don't build
refresh-critical logic without testing against your target instance.
## Sources
- https://docs.joinpeertube.org/api-rest-reference.html (operation pages: searchVideos,
getVideos, comment-threads, getOAuthToken, revokeOAuthToken, getInstanceStats; Errors,
Rate-limits sections)
- https://docs.joinpeertube.org/api/rest-getting-started
- https://docs.joinpeertube.org/use/search (scope semantics)
- https://docs.joinpeertube.org/admin/configuration (global-search admin enablement)
- https://docs.joinpeertube.org/CHANGELOG (route renames, version additions)
- https://raw.githubusercontent.com/Chocobozzz/PeerTube/develop/server/core/controllers/api/oauth-clients.ts
(Host guard)
- Live anonymous probes (secret masking, search default scope, route status codes,
response shapes), 2026-08-29.
+146
View File
@@ -0,0 +1,146 @@
# PeerTube search: instance-local vs the fediverse-wide index
PeerTube search has two distinct scopes, and confusing them is the single most common
mistake clients make. This file pins down exactly what each scope does, what SepiaSearch
is, and which one the bundled CLI performs.
## The two scopes
`GET /api/v1/search/videos` accepts `searchTarget` with exactly two documented values:
| `searchTarget` | Scope | What you get |
| --- | --- | --- |
| `local` | platform/instance search | Results known to the platform you are querying: its own videos plus objects it has discovered/federated from instances it follows. Same behavior as the instance's web UI search box. |
| `search-index` | global/fediverse search | Results served through an **external search index** configured by the instance administrator. The result set is not scoped to objects your instance knows. The reference warns these results come from a third-party service, and the instance may not yet know (have copies of) the returned objects. |
Facts that matter operationally:
- `remote` is **not** a current `searchTarget` value (it appears in old blog posts and
older wrappers); the current enum is `local` | `search-index`.
- The current reference does not state what happens when `searchTarget` is omitted.
Observed behavior on a public instance (2026-08-29): omitting it returned local results
identical to `searchTarget=local`, i.e. **the default scope is the instance's own
index**, not the fediverse. Do not assume otherwise; if you need the instance's results,
pass `searchTarget=local` explicitly, and if you want the fediverse, use a search-index
host (below) rather than an undocumented default.
- `searchTarget=search-index` only works when the administrator has enabled and configured
an external search index (admin config section "Global search"); instances without one
cannot serve index results. Errors when the index is unavailable surface as HTTP 500 on
search endpoints.
- Index results may reference videos your instance has never federated. The official
recommendation for consuming them: if URI search is enabled, fetch the result's URL into
your instance first, then use the classic REST endpoint; otherwise fetch from or redirect
to the **origin instance** (every result carries its origin in `account`/`channel.host`
and the video `url`).
## SepiaSearch: the fediverse-wide index
[SepiaSearch](https://sepiasearch.org) is Framasoft's public search index for PeerTube: a
separately hosted service that crawls and indexes public PeerTube instances (its front
page advertises ~1,700 sites indexed) and exposes **the same REST API shape** under its own
base URL:
```
GET https://sepiasearch.org/api/v1/search/videos?search=<query>&start=0&count=15
```
Verified live (2026-08-29): the response is the standard `{total, data: [...]}` collection
of PeerTube-shaped video objects (`uuid`, `shortUUID`, `name`, `category`, `language`,
`privacy`, `publishedAt`, `account`, `channel`, `views`, `duration`, plus a `score` field
the instance endpoints do not return). Consequences:
- A client only needs to swap the base host from an instance to `https://sepiasearch.org`
to get fediverse-wide search — same parameters, same pagination, same parsing.
- There is no documented indexing-latency guarantee; freshly published videos may take an
unspecified time to appear. Treat indexing lag as variable.
- SepiaSearch is a search service, not a video host: play/upload URLs in results point at
the origin instances.
- PeerTube administrators may instead configure their own index URL (Framasoft also
publishes one at `https://search.joinpeertube.org/` built on the same idea); that is what
`searchTarget=search-index` talks to on such instances. SepiaSearch is simply the
well-known public instance of this concept.
- SepiaSearch results are not moderated by anyone you are talking to; the official
documentation explicitly warns the index content is not moderated.
## Search endpoint catalog
| Endpoint | Notes |
| --- | --- |
| `GET /api/v1/search/videos` | required `search`; `searchTarget`, `start`, `count` (1100, default 15), `sort`, plus video filters below |
| `GET /api/v1/search/video-channels` | required `search`; optional `handles`, `host`, `searchTarget`, `start`, `count`, `sort`; returns 500 if the search index is unavailable |
### Sort values (search + video listing)
`name`, `-duration`, `-createdAt`, `-publishedAt`, `-views`, `-likes`, `-comments`,
`-trending`, `-hot`, `-best`. The last three are relevance/popularity orders computed by
the instance (hot/trending window definitions are instance-side).
### Filter parameters (exact names)
`categoryOneOf`, `licenceOneOf`, `languageOneOf`, `tagsOneOf`, `tagsAllOf`, `nsfw`
(`"true"`/`"false"` string), `nsfwFlagsIncluded`/`nsfwFlagsExcluded`, `isLive`,
`durationMin`/`durationMax` (seconds), `startDate`/`endDate` and
`originallyPublishedStartDate`/`originallyPublishedEndDate` (ISO dates), `host`,
`uuids`, `skipCount` (`true` avoids computing `total`), plus admin-only
`autoTagOneOf` (>=6.2), `include` (bitmask), `privacyOneOf`, `stateOneOf` (>=8.2).
`category` (without `OneOf`) is not the current parameter name — older wrappers using it
silently drop the filter.
## Which scope does the bundled CLI use?
The bundled `scripts/peertube` performs **instance-local search only**: it issues
`GET /search/videos` with `searchTarget=local` against `PEERTUBE_SERVER` and never claims
fediverse-wide coverage. For fediverse-wide search, point the same commands at SepiaSearch
(`PEERTUBE_SERVER=https://sepiasearch.org scripts/peertube search --query ...`) — the CLI
is instance-agnostic by design, and SepiaSearch speaks the same API. The CLI's `search
--help` text states its scope so nobody mistakes local results for the whole fediverse.
## Worked recipes
### Instance-local search, then full video detail
```bash
BASE="https://<INSTANCE_HOST>"
curl -G "$BASE/api/v1/search/videos" \
--data-urlencode 'search=<QUERY>' \
--data-urlencode 'searchTarget=local' \
--data-urlencode 'start=0' --data-urlencode 'count=10'
# data[].uuid / shortUUID / id all work as the {id} path parameter below
curl "$BASE/api/v1/videos/<UUID_OR_SHORTUUID>"
```
### Fediverse-wide search via SepiaSearch
```bash
curl -G 'https://sepiasearch.org/api/v1/search/videos' \
--data-urlencode 'search=<QUERY>' \
--data-urlencode 'start=0' --data-urlencode 'count=10'
# follow a result to its origin instance:
# data[0].url / data[0].channel.host tell you where the video lives
```
### Local search with filters and relevance sort
```bash
curl -G "$BASE/api/v1/search/videos" \
--data-urlencode 'search=<QUERY>' \
--data-urlencode 'searchTarget=local' \
--data-urlencode 'sort=-views' \
--data-urlencode 'durationMin=300' \
--data-urlencode 'languageOneOf=en' \
--data-urlencode 'count=20'
```
## Sources
- https://docs.joinpeertube.org/api-rest-reference.html (searchVideos, searchChannels
operations: searchTarget enum, parameter tables, third-party-index warning)
- https://docs.joinpeertube.org/use/search (platform search vs global search semantics)
- https://docs.joinpeertube.org/admin/configuration (Global search: external index
configuration, search.joinpeertube.org, non-moderation warning)
- https://sepiasearch.org/ (what SepiaSearch is; indexed-site count)
- https://sepiasearch.org/api/v1/search/videos?search=peertube&start=0&count=1
(live response shape, 2026-08-29)
- https://docs.joinpeertube.org/CHANGELOG (version-drift notes)
- Live anonymous probe of a public instance's `/search/videos` with and without
`searchTarget` (default-scope observation), 2026-08-29.
+151
View File
@@ -0,0 +1,151 @@
# Worked recipes and CLI workflows
Multi-step workflows for the bundled `scripts/peertube` CLI, plus raw curl/jq equivalents.
Every stage's output field names and JSON types are what the next stage consumes — the
pipelines are proven by the CLI's offline test suite. `PEERTUBE_SERVER` must be exported
for all commands (any instance host works; SepiaSearch works too — see below).
```bash
export PEERTUBE_SERVER="https://<INSTANCE_HOST>" # e.g. https://tilvids.com
```
## CLI command map
| Command | Does | Auth needed |
| --- | --- | --- |
| `server` | instance name + description (`/config/about`) + stats (`/server/stats`) | no |
| `videos` | latest instance videos (`/videos`, offset paging) | no |
| `search --query Q` | **instance-local** search (`/search/videos`, `searchTarget=local`) | no |
| `video --id ID` | full video detail (id, UUID, or shortUUID) | no |
| `comments --id ID` | top-level comment threads (`/comment-threads`) | no |
| `channels` | instance channel list (`/video-channels`) | no |
| `channel --handle H` | one channel's metadata + recent uploads | no |
| `account --name N` | account metadata (`/accounts/{name}`) | no |
| `me` | your profile (`/users/me`) | yes |
| `my-videos` | your uploads (`/users/me/videos`) | yes |
| `login` | OAuth2 password grant → persists token file | yes (credentials) |
| `logout` | revoke token server-side + delete token file | yes (token) |
Global flags: `--json` (machine output), `--dry-run` (print the request plan, zero
network), `--limit N` (page size, max 100), `--offset N` (start offset). All flags work
before or after the subcommand. `--help` and `--dry-run` never require credentials.
## Recipe 1 — browse what's new, then inspect one video
```bash
scripts/peertube videos --limit 5 --json | jq -r '.videos[] | [.name, .uuid, .duration] | @tsv'
UUID=$(scripts/peertube videos --limit 1 --json | jq -r '.videos[0].uuid')
scripts/peertube video --id "$UUID" --json | jq '{name, description, views, likes, url}'
```
`videos` emits `{"total": <number>, "videos": [...each raw video object with uuid/name/
duration/views/publishedAt/channel/account...]}`; `video` emits the raw detail object
(fields include `description`, `files[]`, `commentsEnabled`).
## Recipe 2 — instance-local search, then pull the description
```bash
scripts/peertube search --query "linux" --limit 10 --json | jq -r '.videos[0].uuid'
scripts/peertube search --query "linux" --limit 5 --json \
| jq -r '.videos[] | select(.language.label == "English") | .name'
# detail for the top hit:
scripts/peertube video --id "$(scripts/peertube search --query linux --limit 1 --json | jq -r '.videos[0].uuid')" --json
```
Search results are the same video-object shape as `videos` (plus nothing missing that the
detail call needs — `uuid` is always present). To search the **whole fediverse** instead of
one instance, point the same CLI at SepiaSearch:
```bash
PEERTUBE_SERVER="https://sepiasearch.org" scripts/peertube search --query "linux" --limit 10
```
## Recipe 3 — channels: find the busy ones, then page their uploads
```bash
scripts/peertube channels --json | jq -r '.channels[] | [.displayName, .name, .host, .videosCount, .followersCount] | @tsv' \
| sort -t$'\t' -k4,4nr | head
# page through a channel's uploads with offsets (no page param exists):
scripts/peertube channel --handle "framasoft@framatube.org" --limit 100 --offset 0 --json | jq -r '.videos[].name'
scripts/peertube channel --handle "framasoft@framatube.org" --limit 100 --offset 100 --json | jq -c '{returned: (.videos | length), total}'
```
Handles accept `name` (local) or `name@host` (remote). The offset loop is the only
pagination mechanism — stop when `returned` is 0 or `offset >= total`.
## Recipe 4 — log in, check your quota, upload-aware housekeeping, log out
```bash
scripts/peertube login --username "<USERNAME>" # prompts for password (hidden)
scripts/peertube me --json | jq '{username, role: .role.label, quota_bytes: .videoQuota}'
scripts/peertube my-videos --limit 100 --json | jq -r '.videos[] | [.name, .privacy.label, .duration] | @tsv'
scripts/peertube logout # revokes server-side + deletes local file
```
The token file lands in `~/.config/peertube/token.json` (owner-only permissions;
`PEERTUBE_CONFIG_DIR` overrides the directory for tests). It records the server URL,
access token, refresh token, and absolute `expires_at`; the CLI re-authenticates if the
server changes or the token is expired. `login --dry-run --json` previews the token
request (fields only — no secret values) without network.
## Recipe 5 — instance report card (compose three anonymous endpoints)
```bash
scripts/peertube server --json \
| jq '{name: .instance.name, description: .instance.shortDescription,
local_videos: .stats.totalLocalVideos, total_videos: .stats.totalVideos,
users: .stats.totalUsers, views: .stats.totalLocalVideoViews}'
```
Equivalent raw curl: `/api/v1/config/about` for identity, `/api/v1/server/stats` for the
counters (note: the stats path is `/server/stats`, not `/instance/stats`).
## Recipe 6 — jq processing patterns
```bash
# TSV table of the five most-viewed local videos
scripts/peertube videos --limit 100 --json \
| jq -r '.videos | sort_by(-.views)[:5][] | [.name, .views, .channel.displayName] | @tsv'
# Count videos per origin host on a search-index-style result set
PEERTUBE_SERVER="https://sepiasearch.org" scripts/peertube search --query "peertube" --limit 100 --json \
| jq -r '.videos | group_by(.channel.host) | map({host: .[0].channel.host, n: length}) | sort_by(-.n)[] | "\(.n)\t\(.host)"'
# Comments of a video, flattening thread counts
scripts/peertube comments --id "<UUID>" --json | jq '{total, total_not_deleted: .totalNotDeletedComments, threads: (.threads | length)}'
# Verify the request plan before running it live (zero network)
scripts/peertube --dry-run --json search --query "test" | jq '{path, params: (.params | keys)}'
```
`--dry-run` output shape: `{"dry_run": true, "method": "GET", "url": "...", "path":
"/api/v1/...", "params": {...}}` (token requests additionally carry `"form"` keys without
values). Every handler's dry-run plan uses the same keys, so one jq pattern audits any
command.
## Raw curl equivalents (auth chain end-to-end)
```bash
BASE="$PEERTUBE_SERVER/api/v1"
CLIENT_ID=$(curl -sS "$BASE/oauth-clients/local" | jq -r .client_id)
# NOTE: production instances mask client_secret ("****...") in this response; if masked,
# obtain the secret as the web client does (served front-end assets) before proceeding.
curl -sS -X POST "$BASE/users/token" \
-H 'Content-Type: application/x-www-form-urlencoded' \
--data-urlencode "client_id=$CLIENT_ID" \
--data-urlencode 'client_secret=<CLIENT_SECRET>' \
--data-urlencode 'grant_type=password' \
--data-urlencode 'username=<USERNAME>' \
--data-urlencode 'password=<PASSWORD>'
ACCESS_TOKEN="<ACCESS_TOKEN>"
curl -sS -H "Authorization: Bearer $ACCESS_TOKEN" "$BASE/users/me"
```
## Sources
- https://docs.joinpeertube.org/api/rest-getting-started (auth chain, pagination basics)
- https://docs.joinpeertube.org/api-rest-reference.html (endpoint parameter tables and
response shapes referenced per recipe)
- https://sepiasearch.org/api/v1/search/videos (fediverse-wide search base URL)
- Field names/types corroborated by live anonymous probes and the CLI's offline mocked
tests, 2026-08-29.
+821
View File
@@ -0,0 +1,821 @@
#!/usr/bin/env python3
"""peertube — browse PeerTube federated video from the terminal.
Read-only client for any PeerTube instance's REST API, plus OAuth2 login.
Commands: server, videos, video, search, comments, channels, channel,
account, me, my-videos, login, logout. Set PEERTUBE_SERVER (or pass
--server) to choose the instance; point it at https://sepiasearch.org
for fediverse-wide search — the API shape is identical. `--json` emits
machine-readable output, `--dry-run` prints the exact request plan with
zero network activity, and `--help` works without credentials.
"""
import argparse
import getpass
import json
import os
import sys
import time
import warnings
from typing import Any, Dict, List, Optional, Tuple
warnings.simplefilter("ignore")
import requests
ENV_SERVER = os.getenv("PEERTUBE_SERVER", "")
ENV_CONFIG_DIR = os.getenv("PEERTUBE_CONFIG_DIR", "")
API_BASE = "/api/v1"
DEFAULT_CONFIG_DIR = "~/.config/peertube"
TOKEN_FILE_NAME = "token.json"
MAX_COUNT = 100
DEFAULT_COUNT = 15
GLOBAL_FLAGS: Dict[str, Any] = {"json": False, "dry_run": False, "quiet": False, "verbose": False}
def die(message, exit_code=1):
print(f"Error: {message}", file=sys.stderr)
sys.exit(exit_code)
def warn(message):
print(f"Warning: {message}", file=sys.stderr)
def emit(human, data):
if GLOBAL_FLAGS.get("json"):
print(json.dumps(data, default=str))
else:
print(human)
def log(message):
if GLOBAL_FLAGS.get("verbose") and not GLOBAL_FLAGS.get("json"):
print(message, file=sys.stderr)
def _preparse_global_flags(argv):
"""Pull global flags out so they work before or after the subcommand.
Boolean flags are recorded as True; --server consumes its value. Anything
else passes through to the subparsers untouched."""
bools = {"--json", "--dry-run", "--quiet", "--verbose"}
flags, filtered = {}, [argv[0]]
i = 1
while i < len(argv):
arg = argv[i]
if arg in bools:
flags[arg.lstrip("-").replace("-", "_")] = True
i += 1
elif arg == "--server":
if i + 1 >= len(argv):
die("--server requires a value (the instance URL).")
flags["server"] = argv[i + 1]
i += 2
elif arg in ("--help", "-h"):
return flags, argv
elif arg == "--":
filtered.extend(argv[i:])
break
else:
filtered.append(arg)
i += 1
return flags, filtered
def default_config_dir():
if ENV_CONFIG_DIR:
return os.path.expanduser(ENV_CONFIG_DIR)
return os.path.expanduser(DEFAULT_CONFIG_DIR)
def fmt_duration(seconds):
try:
total = int(seconds or 0)
except (TypeError, ValueError):
return "?"
return f"{total // 60}:{total % 60:02d}"
def fmt_date(value):
return str(value or "")[:10]
def is_masked_secret(value):
"""Production instances reply to oauth-clients/local with the secret
replaced by a run of '*' characters."""
return bool(value) and set(value) == {"*"}
def request_plan(method, path, params=None, **extra):
plan: Dict[str, Any] = {"dry_run": True, "method": method, "path": path, "params": params or {}}
plan.update(extra)
return plan
def channel_label(video):
channel = video.get("channel") or {}
account = video.get("account") or {}
display = channel.get("displayName") or account.get("displayName") or "?"
host = channel.get("host") or account.get("host")
return f"{display}@{host}" if host else display
def video_line(video, index=None):
name = str(video.get("name") or "?")[:52]
views = video.get("views") or 0
when = fmt_date(video.get("publishedAt"))
prefix = f"{index:>3}. " if index else " - "
return f" {prefix}{name:<52} {fmt_duration(video.get('duration')):>6} {views:>8} views {channel_label(video)} {when}"
class PeerTubeClient:
"""REST client for one PeerTube instance (federated: tokens are per-instance)."""
def __init__(self, server="", dry_run=False, config_dir=None):
self.server = (server or ENV_SERVER).rstrip("/")
self.dry_run = dry_run
self.config_dir = config_dir or default_config_dir()
self._token: Optional[str] = None
self._refresh_token: Optional[str] = None
self._expires_at: Optional[float] = None
self._load_token()
# ----- instance / URL helpers -------------------------------------
def base_url(self):
if not self.server:
die("No instance configured. Export PEERTUBE_SERVER (e.g. https://<INSTANCE_HOST>) "
"or pass --server.")
return self.server
def display_server(self):
return self.server or "https://<INSTANCE_HOST>"
def _url(self, path):
return f"{self.base_url()}{API_BASE}{path}"
def _headers(self, with_token=True):
headers = {"Accept": "application/json"}
if with_token and self._token:
headers["Authorization"] = f"Bearer {self._token}"
return headers
# ----- token file persistence (per-instance, owner-only) ----------
def token_path(self):
return os.path.join(self.config_dir, TOKEN_FILE_NAME)
def _load_token(self):
try:
with open(self.token_path()) as handle:
data = json.load(handle)
except (OSError, ValueError):
return
stored_server = str(data.get("server") or "").rstrip("/")
if self.server and stored_server and stored_server != self.server:
return # token was minted by a different instance
self._token = data.get("access_token") or None
self._refresh_token = data.get("refresh_token") or None
expires_at = data.get("expires_at")
self._expires_at = float(expires_at) if expires_at else None
def save_session(self, token_response):
access = token_response.get("access_token")
if not access:
die("Login response carried no access_token; nothing to persist.")
expires_in = token_response.get("expires_in")
expires_at = time.time() + float(expires_in) if expires_in else None
record = {
"server": self.server,
"access_token": access,
"refresh_token": token_response.get("refresh_token"),
"token_type": token_response.get("token_type", "Bearer"),
"expires_at": expires_at,
"expires_in": expires_in,
}
os.makedirs(self.config_dir, mode=0o700, exist_ok=True)
fd = os.open(self.token_path(), os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
with os.fdopen(fd, "w") as handle:
json.dump(record, handle, indent=2)
self._token = access
self._refresh_token = record["refresh_token"]
self._expires_at = expires_at
def clear_token(self):
try:
os.remove(self.token_path())
except OSError:
pass
self._token = None
self._refresh_token = None
self._expires_at = None
def token_is_valid(self):
if not self._token:
return False
return self._expires_at is None or time.time() < self._expires_at
def ensure_token(self):
"""Authed commands need a live token; try refresh before giving up."""
if self.token_is_valid():
return
if self._refresh_token and not self.dry_run and self.refresh():
return
die(f"Not authenticated for {self.display_server()}. Run "
f"'peertube login --username <name>' first, or use --dry-run to preview.")
# ----- OAuth2 flows ------------------------------------------------
def fetch_oauth_client(self):
"""Anonymous GET /oauth-clients/local (singular 'local')."""
url = self._url("/oauth-clients/local")
log(f"--> GET {url}")
try:
response = requests.get(url, headers={"Accept": "application/json"}, timeout=15)
except requests.exceptions.RequestException as exc:
die(f"Cannot reach {self.display_server()} ({exc.__class__.__name__}). "
f"Check PEERTUBE_SERVER and connectivity.")
if response.status_code >= 400:
die(f"oauth-clients/local returned {response.status_code}: {response.text[:200]}")
try:
payload = response.json()
except ValueError:
die(f"Non-JSON response from oauth-clients/local; is {self.display_server()} a PeerTube instance?")
client_id = payload.get("client_id")
if not client_id:
die("oauth-clients/local response missing client_id.")
return client_id, payload.get("client_secret") or ""
def request_token(self, client_id, client_secret, username, password, otp=None):
"""POST /users/token with the password grant (x-www-form-urlencoded)."""
form = {
"client_id": client_id,
"client_secret": client_secret,
"grant_type": "password",
"username": username,
"password": password,
}
headers = {"Accept": "application/json"}
if otp:
headers["x-peertube-otp"] = otp
url = self._url("/users/token")
log("--> POST /users/token (password grant, values not logged)")
try:
response = requests.post(url, data=form, headers=headers, timeout=15)
except requests.exceptions.RequestException as exc:
die(f"Cannot reach {self.display_server()} ({exc.__class__.__name__}) during token request.")
if response.status_code == 401:
die("Login failed (401): token request rejected. If the account uses two-factor "
"authentication, pass --otp.")
if response.status_code >= 400:
detail = ""
try:
body = response.json()
if isinstance(body, dict):
detail = body.get("detail") or body.get("title") or body.get("error") or ""
except ValueError:
detail = (response.text or "")[:200]
die(f"Login failed ({response.status_code}): invalid client credentials or wrong "
f"username/password. {detail}".rstrip())
try:
return response.json()
except ValueError:
die("Token endpoint returned non-JSON.")
def refresh(self):
"""grant_type=refresh_token. The official reference does not render the
exact refresh form fields, so this uses the community-established shape
and treats any failure as a signal to fall back to a password re-grant."""
if not self._refresh_token:
return False
try:
client_id, client_secret = self.fetch_oauth_client()
except SystemExit:
return False
if is_masked_secret(client_secret) or not client_secret:
return False
form = {
"client_id": client_id,
"client_secret": client_secret,
"grant_type": "refresh_token",
"refresh_token": self._refresh_token,
}
url = self._url("/users/token")
log("--> POST /users/token (refresh grant, values not logged)")
try:
response = requests.post(url, data=form, headers={"Accept": "application/json"}, timeout=15)
except requests.exceptions.RequestException:
return False
if response.status_code >= 400:
return False
try:
data = response.json()
except ValueError:
return False
if not data.get("access_token"):
return False
self.save_session(data)
return True
def revoke_token(self):
"""POST /users/revoke-token invalidates the access and refresh tokens."""
url = self._url("/users/revoke-token")
log(f"--> POST {url}")
try:
return requests.post(url, headers=self._headers(), timeout=15)
except requests.exceptions.RequestException as exc:
die(f"Cannot reach {self.display_server()} ({exc.__class__.__name__}) during revocation.")
# ----- read endpoints ----------------------------------------------
def _get(self, path, params=None):
url = self._url(path)
if self.dry_run:
return request_plan("GET", f"{API_BASE}{path}", params)
log(f"--> GET {url}")
try:
response = requests.get(url, params=params, headers=self._headers(), timeout=30)
except requests.exceptions.RequestException as exc:
die(f"Cannot reach {self.display_server()} ({exc.__class__.__name__}). "
f"Check PEERTUBE_SERVER and connectivity.")
return self._handle(response, f"{API_BASE}{path}")
def _handle(self, response, path):
status = response.status_code
if status >= 400:
detail = ""
try:
body = response.json()
if isinstance(body, dict):
detail = body.get("detail") or body.get("title") or body.get("error") or ""
except ValueError:
detail = (response.text or "")[:200]
if status == 401:
die(f"401 Unauthorized on {path}: {detail or 'token missing, expired, or revoked'}. "
f"Run 'peertube login' or check the instance.")
if status == 429:
retry = response.headers.get("Retry-After", "?")
die(f"429 rate limited on {path}; Retry-After: {retry}s "
f"(default server limit: 50 calls per 10 seconds).")
die(f"API error {status} on {path}: {detail or 'no detail in body'}")
try:
return response.json()
except ValueError:
die(f"Non-JSON response from {path} (status {status}); "
f"is {self.display_server()} a PeerTube instance?")
def get_about(self):
return self._get("/config/about")
def get_stats(self):
return self._get("/server/stats")
def list_videos(self, start=0, count=DEFAULT_COUNT, sort="-publishedAt"):
return self._get("/videos", {"start": start, "count": count, "sort": sort})
def get_video(self, video_id):
return self._get(f"/videos/{video_id}")
def search_videos(self, query, start=0, count=DEFAULT_COUNT, search_target="local", sort=None):
params: Dict[str, Any] = {"search": query, "searchTarget": search_target,
"start": start, "count": count}
if sort:
params["sort"] = sort
return self._get("/search/videos", params)
def list_comment_threads(self, video_id, start=0, count=DEFAULT_COUNT):
return self._get(f"/videos/{video_id}/comment-threads",
{"start": start, "count": count, "sort": "-createdAt"})
def list_channels(self, start=0, count=DEFAULT_COUNT):
return self._get("/video-channels", {"start": start, "count": count})
def get_channel(self, handle):
return self._get(f"/video-channels/{handle}")
def list_channel_videos(self, handle, start=0, count=DEFAULT_COUNT):
return self._get(f"/video-channels/{handle}/videos", {"start": start, "count": count})
def get_account(self, name):
return self._get(f"/accounts/{name}")
def my_profile(self):
return self._get("/users/me")
def my_videos(self, start=0, count=DEFAULT_COUNT):
return self._get("/users/me/videos", {"start": start, "count": count})
# ----- command handlers ------------------------------------------------
def validate_page_args(count, start):
if count < 1 or count > MAX_COUNT:
die(f"--limit must be between 1 and {MAX_COUNT} (server maximum).")
if start < 0:
die("--offset must be >= 0.")
def cmd_server(client, args):
if client.dry_run:
plan = {"dry_run": True, "requests": [
{"method": "GET", "path": f"{API_BASE}/config/about", "params": {}},
{"method": "GET", "path": f"{API_BASE}/server/stats", "params": {}},
]}
return emit("[dry-run] GET /config/about + /server/stats", plan)
about = client.get_about() or {}
stats = client.get_stats() or {}
instance = about.get("instance") or {}
name = instance.get("name", "?")
description = instance.get("shortDescription") or ""
keys = ("totalUsers", "totalLocalVideos", "totalVideos",
"totalLocalVideoViews", "totalLocalVideoDownloads", "totalLocalVideoChannels")
payload = {
"instance": {
"name": name,
"shortDescription": instance.get("shortDescription"),
"description": instance.get("description"),
},
"stats": {key: stats.get(key) for key in keys},
}
stats_line = " ".join(f"{key}={payload['stats'][key]}" for key in keys if payload['stats'][key] is not None)
emit(f"Instance {name}\n {description[:100]}\n {stats_line}", payload)
def cmd_videos(client, args):
parser = argparse.ArgumentParser(prog="peertube videos")
parser.add_argument("--limit", type=int, default=DEFAULT_COUNT)
parser.add_argument("--offset", type=int, default=0)
parser.add_argument("--sort", default="-publishedAt")
parsed, _ = parser.parse_known_args(args)
validate_page_args(parsed.limit, parsed.offset)
params = {"start": parsed.offset, "count": parsed.limit, "sort": parsed.sort}
if client.dry_run:
return emit("[dry-run] GET /api/v1/videos " + json.dumps(params),
request_plan("GET", f"{API_BASE}/videos", params))
data = client._get("/videos", params) or {}
videos = data.get("data", [])
total = data.get("total", len(videos))
if not videos:
return emit("No videos.", {"total": total, "start": parsed.offset, "count": 0, "videos": []})
lines = [video_line(v, i + 1) for i, v in enumerate(videos)]
emit(f"{total} video(s) on {client.display_server()}:\n" + "\n".join(lines),
{"total": total, "start": parsed.offset, "count": len(videos), "videos": videos})
def cmd_video(client, args):
parser = argparse.ArgumentParser(prog="peertube video")
parser.add_argument("--id", required=True, help="Numeric id, UUID, or shortUUID")
parsed, _ = parser.parse_known_args(args)
if client.dry_run:
return emit(f"[dry-run] GET {API_BASE}/videos/{parsed.id}",
request_plan("GET", f"{API_BASE}/videos/{parsed.id}", {}))
detail = client._get(f"/videos/{parsed.id}") or {}
emit(f"{detail.get('name', '?')} [{fmt_duration(detail.get('duration'))}] "
f"{detail.get('views', 0)} views by {channel_label(detail)}", detail)
def cmd_search(client, args):
parser = argparse.ArgumentParser(
prog="peertube search",
description="Search the configured instance's OWN catalog (searchTarget=local). "
"For fediverse-wide search set PEERTUBE_SERVER=https://sepiasearch.org — "
"same commands, wider index.")
parser.add_argument("--query", "-q", required=True)
parser.add_argument("--limit", type=int, default=DEFAULT_COUNT)
parser.add_argument("--offset", type=int, default=0)
parser.add_argument("--sort", default=None)
parser.add_argument("--search-target", choices=["local", "search-index"], default="local")
parsed, _ = parser.parse_known_args(args)
validate_page_args(parsed.limit, parsed.offset)
params: Dict[str, Any] = {"search": parsed.query, "searchTarget": parsed.search_target,
"start": parsed.offset, "count": parsed.limit}
if parsed.sort:
params["sort"] = parsed.sort
if client.dry_run:
return emit(f"[dry-run] GET {API_BASE}/search/videos " + json.dumps(params),
request_plan("GET", f"{API_BASE}/search/videos", params))
data = client._get("/search/videos", params) or {}
videos = data.get("data", [])
total = data.get("total", len(videos))
if not videos:
return emit("No results.", {"total": total, "start": parsed.offset, "count": 0, "videos": []})
lines = [video_line(v, i + 1) for i, v in enumerate(videos)]
scope = ("instance-local" if parsed.search_target == "local" else "search-index")
emit(f"{total} result(s) [{scope} search on {client.display_server()}]:\n" + "\n".join(lines),
{"total": total, "start": parsed.offset, "count": len(videos), "videos": videos})
def cmd_comments(client, args):
parser = argparse.ArgumentParser(prog="peertube comments")
parser.add_argument("--id", required=True, help="Video id, UUID, or shortUUID")
parser.add_argument("--limit", type=int, default=DEFAULT_COUNT)
parser.add_argument("--offset", type=int, default=0)
parsed, _ = parser.parse_known_args(args)
validate_page_args(parsed.limit, parsed.offset)
params = {"start": parsed.offset, "count": parsed.limit, "sort": "-createdAt"}
if client.dry_run:
return emit(f"[dry-run] GET {API_BASE}/videos/{parsed.id}/comment-threads " + json.dumps(params),
request_plan("GET", f"{API_BASE}/videos/{parsed.id}/comment-threads", params))
data = client._get(f"/videos/{parsed.id}/comment-threads", params) or {}
threads = data.get("data", [])
total = data.get("total", len(threads))
lines = []
for thread in threads:
comment = thread.get("comment") or {}
account = comment.get("account") or {}
text = str(comment.get("text") or "")[:70].replace("\n", " ")
lines.append(f" {thread.get('totalReplies', 0):>3} replies @{account.get('name', '?')} {text}")
emit(f"{total} comment thread(s):\n" + ("\n".join(lines) if lines else " (none)"),
{"total": total, "total_not_deleted": data.get("totalNotDeletedComments"),
"start": parsed.offset, "threads": threads})
def cmd_channels(client, args):
parser = argparse.ArgumentParser(prog="peertube channels")
parser.add_argument("--limit", type=int, default=DEFAULT_COUNT)
parser.add_argument("--offset", type=int, default=0)
parsed, _ = parser.parse_known_args(args)
validate_page_args(parsed.limit, parsed.offset)
params = {"start": parsed.offset, "count": parsed.limit}
if client.dry_run:
return emit(f"[dry-run] GET {API_BASE}/video-channels " + json.dumps(params),
request_plan("GET", f"{API_BASE}/video-channels", params))
data = client._get("/video-channels", params) or {}
channels = data.get("data", [])
total = data.get("total", len(channels))
if not channels:
return emit("No channels.", {"total": total, "start": parsed.offset, "count": 0, "channels": []})
lines, _ = [], []
for channel in channels:
display = channel.get("displayName") or channel.get("name") or "?"
handle = f"{channel.get('name', '?')}@{channel.get('host', '?')}"
lines.append(f" {display:<30} {handle:<40} "
f"{channel.get('videosCount', '?')} videos {channel.get('followersCount', '?')} followers")
emit(f"{total} channel(s) on {client.display_server()}:\n" + "\n".join(lines),
{"total": total, "start": parsed.offset, "count": len(channels), "channels": channels})
def cmd_channel(client, args):
parser = argparse.ArgumentParser(prog="peertube channel")
parser.add_argument("--handle", required=True, help="Channel name or name@host")
parser.add_argument("--limit", type=int, default=DEFAULT_COUNT)
parser.add_argument("--offset", type=int, default=0)
parsed, _ = parser.parse_known_args(args)
validate_page_args(parsed.limit, parsed.offset)
params = {"start": parsed.offset, "count": parsed.limit}
if client.dry_run:
plan = {"dry_run": True, "requests": [
{"method": "GET", "path": f"{API_BASE}/video-channels/{parsed.handle}", "params": {}},
{"method": "GET", "path": f"{API_BASE}/video-channels/{parsed.handle}/videos", "params": params},
]}
return emit(f"[dry-run] GET /video-channels/{parsed.handle} (+ /videos)", plan)
meta = client._get(f"/video-channels/{parsed.handle}") or {}
videos_data = client._get(f"/video-channels/{parsed.handle}/videos", params) or {}
videos = videos_data.get("data", [])
display = meta.get("displayName") or meta.get("name") or parsed.handle
lines = [video_line(v, i + 1) for i, v in enumerate(videos)]
emit(f"{display} ({meta.get('followersCount', '?')} followers, {meta.get('videosCount', '?')} videos)\n"
+ ("\n".join(lines) if lines else " (no videos listed)"),
{"channel": meta, "total": videos_data.get("total", len(videos)),
"start": parsed.offset, "videos": videos})
def cmd_account(client, args):
parser = argparse.ArgumentParser(prog="peertube account")
parser.add_argument("--name", required=True, help="Account name or name@host")
parsed, _ = parser.parse_known_args(args)
if client.dry_run:
return emit(f"[dry-run] GET {API_BASE}/accounts/{parsed.name}",
request_plan("GET", f"{API_BASE}/accounts/{parsed.name}", {}))
account = client._get(f"/accounts/{parsed.name}") or {}
display = account.get("displayName") or account.get("name") or parsed.name
handle = f"{account.get('name', parsed.name)}@{account.get('host', '?')}"
emit(f"{display} ({handle}) {account.get('followersCount', '?')} followers", account)
def cmd_me(client, args):
if client.dry_run:
return emit(f"[dry-run] GET {API_BASE}/users/me (Authorization: Bearer <ACCESS_TOKEN>)",
request_plan("GET", f"{API_BASE}/users/me", {}))
client.ensure_token()
data = client._get("/users/me") or {}
if isinstance(data, list): # docs render an array sample; live servers return one object
data = data[0] if data else {}
role = data.get("role") or {}
quota = data.get("videoQuota")
quota_text = f"{quota} bytes" if quota is not None else "?"
emit(f"@{data.get('username', '?')} Role: {role.get('label', '?')} Quota: {quota_text}", data)
def cmd_my_videos(client, args):
parser = argparse.ArgumentParser(prog="peertube my-videos")
parser.add_argument("--limit", type=int, default=DEFAULT_COUNT)
parser.add_argument("--offset", type=int, default=0)
parsed, _ = parser.parse_known_args(args)
validate_page_args(parsed.limit, parsed.offset)
params = {"start": parsed.offset, "count": parsed.limit}
if client.dry_run:
return emit(f"[dry-run] GET {API_BASE}/users/me/videos " + json.dumps(params),
request_plan("GET", f"{API_BASE}/users/me/videos", params))
client.ensure_token()
data = client._get("/users/me/videos", params) or {}
videos = data.get("data", [])
total = data.get("total", len(videos))
lines = [video_line(v, i + 1) for i, v in enumerate(videos)]
emit(f"{total} of your video(s):\n" + ("\n".join(lines) if lines else " (none)"),
{"total": total, "start": parsed.offset, "count": len(videos), "videos": videos})
def cmd_login(client, args):
parser = argparse.ArgumentParser(
prog="peertube login",
description="OAuth2 password grant: fetch the instance's client pair, exchange your "
"credentials for a bearer token, and persist it (owner-only file, "
"per-instance). Password is never echoed.")
parser.add_argument("--username", "-u", required=True)
parser.add_argument("--server", help="Instance URL override, e.g. https://<INSTANCE_HOST>")
password_group = parser.add_mutually_exclusive_group()
password_group.add_argument("--password", help="Account password (prefer --password-stdin or --prompt)")
password_group.add_argument("--password-stdin", action="store_true", help="Read password from stdin")
password_group.add_argument("--prompt", action="store_true", help="Prompt for the password (hidden)")
parser.add_argument("--otp", help="One-time password when the account has 2FA enabled")
parsed, _ = parser.parse_known_args(args)
if client.dry_run:
return emit("[dry-run] POST /api/v1/users/token (password grant)",
request_plan("POST", f"{API_BASE}/users/token", {},
form_fields=["client_id", "client_secret", "grant_type",
"username", "password"],
note="field values suppressed; run without --dry-run to authenticate"))
server = (parsed.server or client.server).rstrip("/")
if not server:
die("No instance configured. Export PEERTUBE_SERVER or pass --server https://<INSTANCE_HOST>.")
login_client = PeerTubeClient(server=server, config_dir=client.config_dir)
if not parsed.password_stdin and not parsed.prompt and parsed.password is None:
die("No password supplied. Use --password, --password-stdin, or --prompt.")
if parsed.password_stdin:
password = sys.stdin.readline().rstrip("\n")
elif parsed.prompt:
password = getpass.getpass(f"Password for {parsed.username}@{server}: ")
else:
password = parsed.password
client_id, client_secret = login_client.fetch_oauth_client()
if is_masked_secret(client_secret):
die("This instance masks client_secret in oauth-clients/local ('*' characters). "
"Production PeerTube front ends obtain the real pair from their own served assets; "
"see references/auth-and-tokens.md for the workaround before retrying login.")
token_response = login_client.request_token(client_id, client_secret, parsed.username,
password, otp=parsed.otp)
login_client.save_session(token_response)
expires_in = token_response.get("expires_in")
expires_text = f", expires in {expires_in}s" if expires_in else ""
emit(f"Logged in to {server} as {parsed.username}{expires_text}\n"
f"Token file: {login_client.token_path()}",
{"status": "logged_in", "server": server, "username": parsed.username,
"token_file": login_client.token_path(), "expires_in": expires_in})
def cmd_logout(client, args):
if client.dry_run:
return emit(f"[dry-run] POST {API_BASE}/users/revoke-token",
request_plan("POST", f"{API_BASE}/users/revoke-token", {}))
if not client.server:
die("No instance configured. Export PEERTUBE_SERVER or pass --server.")
if not client._token and not client._refresh_token:
die(f"No stored token for {client.server}; nothing to revoke.")
response = client.revoke_token()
status = response.status_code
if status == 200:
client.clear_token()
return emit("Token revoked and local token file removed.",
{"status": "logged_out", "revoked": True, "server": client.server})
if status == 401:
client.clear_token()
warn("Server rejected revocation (401); the token was likely already invalid. "
"Local token file removed.")
return emit("Logged out (token was already invalid server-side).",
{"status": "logged_out", "revoked": False, "server": client.server})
die(f"Revocation failed ({status}). Local token file kept; retry or remove "
f"{client.token_path()} manually.")
def build_parser():
parser = argparse.ArgumentParser(
prog="peertube",
description="PeerTube federated video from the terminal.",
epilog="Set PEERTUBE_SERVER (any instance host, e.g. https://<INSTANCE_HOST>) or pass "
"--server. Fediverse-wide search: PEERTUBE_SERVER=https://sepiasearch.org. "
"Login: peertube login --username NAME --prompt")
parser.add_argument("--server",
help="Instance URL (default: PEERTUBE_SERVER env var)")
parser.add_argument("--json", action="store_true", help="Machine-readable JSON output")
parser.add_argument("--dry-run", action="store_true",
help="Print the request plan without any network call")
parser.add_argument("--quiet", action="store_true",
help="Accepted for compatibility; no effect")
parser.add_argument("--verbose", action="store_true", help="Trace requests on stderr")
sub = parser.add_subparsers(dest="command")
sub.add_parser("server", help="Instance name, description, and stats (anonymous)",
description="Compose /config/about and /server/stats for an instance picture.",
epilog="Example: peertube server --json")
vp = sub.add_parser("videos", help="List the instance's videos (offset pagination)",
description="GET /videos with start/count pagination (there is no page parameter).",
epilog="Example: peertube videos --limit 5 --json")
vp.add_argument("--limit", type=int, default=DEFAULT_COUNT, help="Page size, 1-100")
vp.add_argument("--offset", type=int, default=0, help="Zero-based result offset")
vp.add_argument("--sort", default="-publishedAt", help="e.g. -publishedAt, -views, -trending")
vi = sub.add_parser("video", help="Full detail for one video",
description="GET /videos/{id}; id accepts numeric id, UUID, or shortUUID.",
epilog="Example: peertube video --id <UUID>")
vi.add_argument("--id", required=True)
sp = sub.add_parser("search", help="Search this instance's own catalog",
description="Instance-local search (searchTarget=local). For fediverse-wide "
"search point PEERTUBE_SERVER at https://sepiasearch.org.",
epilog="Example: peertube search --query linux --limit 10")
sp.add_argument("--query", "-q", required=True)
sp.add_argument("--limit", type=int, default=DEFAULT_COUNT, help="Page size, 1-100")
sp.add_argument("--offset", type=int, default=0)
sp.add_argument("--sort", default=None, help="e.g. -views, -match, -publishedAt")
sp.add_argument("--search-target", choices=["local", "search-index"], default="local",
help="local = this instance's catalog; search-index = external index if the "
"instance has one")
cp = sub.add_parser("comments", help="Top-level comment threads of a video",
description="GET /videos/{id}/comment-threads (hyphenated route).",
epilog="Example: peertube comments --id <UUID>")
cp.add_argument("--id", required=True)
cp.add_argument("--limit", type=int, default=DEFAULT_COUNT, help="Page size, 1-100")
cp.add_argument("--offset", type=int, default=0)
hp = sub.add_parser("channels", help="List the instance's channels",
description="GET /video-channels with start/count pagination.",
epilog="Example: peertube channels --limit 20 --json")
hp.add_argument("--limit", type=int, default=DEFAULT_COUNT, help="Page size, 1-100")
hp.add_argument("--offset", type=int, default=0)
cv = sub.add_parser("channel", help="One channel's metadata and uploads",
description="GET /video-channels/{handle} and its videos; handle is name or "
"name@host for remote channels.",
epilog="Example: peertube channel --handle framasoft@framatube.org")
cv.add_argument("--handle", required=True)
cv.add_argument("--limit", type=int, default=DEFAULT_COUNT, help="Page size, 1-100")
cv.add_argument("--offset", type=int, default=0)
ap = sub.add_parser("account", help="One account's metadata",
description="GET /accounts/{name}; name accepts name or name@host.",
epilog="Example: peertube account --name chocobozzz@framatube.org")
ap.add_argument("--name", required=True)
sub.add_parser("me", help="Your profile (requires login)",
description="GET /users/me with the persisted bearer token.",
epilog="Example: peertube me --json | jq .role.label")
mv = sub.add_parser("my-videos", help="Your uploads (requires login)",
description="GET /users/me/videos with start/count pagination.",
epilog="Example: peertube my-videos --limit 50 --json")
mv.add_argument("--limit", type=int, default=DEFAULT_COUNT, help="Page size, 1-100")
mv.add_argument("--offset", type=int, default=0)
lg = sub.add_parser("login", help="Authenticate (OAuth2 password grant) and persist a token",
description="Fetch the instance's OAuth client pair, run the password grant, "
"and persist the token per-instance in an owner-only file.",
epilog="Example: peertube login --username NAME --prompt")
lg.add_argument("--username", "-u", help="PeerTube username")
lg_password_group = lg.add_mutually_exclusive_group()
lg_password_group.add_argument("--password", help="Account password (prefer --password-stdin or --prompt)")
lg_password_group.add_argument("--password-stdin", action="store_true", help="Read the password from stdin")
lg_password_group.add_argument("--prompt", action="store_true", help="Prompt for the password (hidden)")
lg.add_argument("--otp", help="One-time password when the account has 2FA enabled")
lg.add_argument("--server", help="Instance URL override, e.g. https://<INSTANCE_HOST>")
lo = sub.add_parser("logout", help="Revoke the token server-side and delete the local token file",
description="POST /users/revoke-token, then remove the local token file.",
epilog="Example: peertube logout")
return parser
def main():
global GLOBAL_FLAGS
GLOBAL_FLAGS, filtered_argv = _preparse_global_flags(sys.argv)
if GLOBAL_FLAGS.get("json"):
warnings.simplefilter("ignore")
parser = build_parser()
args = parser.parse_args(filtered_argv[1:])
if not args.command:
parser.print_help()
sys.exit(1)
client = PeerTubeClient(server=args.server or GLOBAL_FLAGS.get("server") or "",
dry_run=GLOBAL_FLAGS.get("dry_run", False))
handlers = {
"server": cmd_server, "videos": cmd_videos, "video": cmd_video,
"search": cmd_search, "comments": cmd_comments, "channels": cmd_channels,
"channel": cmd_channel, "account": cmd_account, "me": cmd_me,
"my-videos": cmd_my_videos, "login": cmd_login, "logout": cmd_logout,
}
handler = handlers.get(args.command)
if not handler:
parser.print_help()
sys.exit(1)
remaining = filtered_argv[filtered_argv.index(args.command) + 1:]
handler(client, remaining)
if __name__ == "__main__":
main()
-239
View File
@@ -1,239 +0,0 @@
#!/usr/bin/env python3
"""peertube-cli — PeerTube federated video from the terminal.
Browse videos, channels, and playlists on any PeerTube instance.
Login with OAuth2 for authenticated operations.
"""
import argparse, json, os, sys, time, warnings
from typing import Any, Dict, List, Optional, Tuple
warnings.simplefilter("ignore")
import requests
ENV_SERVER = os.getenv("PEERTUBE_SERVER", "")
CONFIG_DIR = os.path.expanduser(os.getenv("PEERTUBE_CONFIG_DIR", "~/.config/peertube-cli"))
API_BASE = "/api/v1"
QUIET = False
GLOBAL_FLAGS: Dict[str, Any] = {"json": False, "dry_run": False, "quiet": False, "verbose": False}
def log(m): global QUIET; (not QUIET and not GLOBAL_FLAGS.get("json")) and print(m)
def warn(m): print(f"Warning: {m}", file=sys.stderr)
def die(m, c=1): print(f"Error: {m}", file=sys.stderr); sys.exit(c)
def emit(h, d):
if GLOBAL_FLAGS.get("json"): print(json.dumps(d, default=str))
else: print(h)
def _preparse(argv):
BOOLS = {"--json","--dry-run","--quiet","--verbose"}
f, fl = {}, [argv[0]]
i = 1
while i < len(argv):
a = argv[i]
if a in BOOLS: f[a.lstrip("-").replace("-","_")] = True; i += 1
elif a in ("--help","-h"): return f, argv
elif a == "--": fl.extend(argv[i:]); break
else: fl.append(a); i += 1
return f, fl
class PeerTubeClient:
def __init__(self, server="", dry_run=False):
self.server = (server or ENV_SERVER or "https://your-instance.example.com").rstrip("/")
self.dry_run = dry_run
self._token = None
# Try to load saved token
token_path = os.path.join(CONFIG_DIR, "token.json")
if os.path.isfile(token_path):
try:
with open(token_path) as f: data = json.load(f)
if data.get("expires_at", 0) > time.time():
self._token = data.get("access_token")
except: pass
def _oauth_login(self):
"""Fetch OAuth client creds and exchange for token."""
try:
r = requests.get(f"{self.server}{API_BASE}/oauth-clients/local", timeout=15)
clients = r.json()
except: die(f"Cannot reach {self.server}. Check PEERTUBE_SERVER.")
return clients.get("client_id", ""), clients.get("client_secret", "")
def login(self, username, password):
"""Login and persist OAuth token."""
cid, csec = self._oauth_login()
try:
r = requests.post(f"{self.server}{API_BASE}/users/token", data={
"client_id": cid, "client_secret": csec,
"grant_type": "password", "username": username,
"password": password,
"response_type": "code"}, timeout=15)
except ConnectionError as e: die(f"Cannot connect: {e}")
if r.status_code >= 400: die(f"Login failed: {r.text[:200]}")
data = r.json()
self._token = data.get("access_token")
# Persist token
os.makedirs(CONFIG_DIR, exist_ok=True)
with open(os.path.join(CONFIG_DIR, "token.json"), "w") as f:
json.dump({"access_token": self._token, "expires_at": time.time() + data.get("expires_in", 86400)}, f)
return data
def _headers(self):
h = {"Accept": "application/json"}
if self._token: h["Authorization"] = f"Bearer {self._token}"
return h
def _get(self, path, params=None):
url = f"{self.server}{API_BASE}{path}"
if self.dry_run: return {"dry_run":True, "url":url, "params":params, "total":0, "data":[]}
try:
r = requests.get(url, params=params, headers=self._headers(), timeout=30)
except ConnectionError as e: die(f"Cannot connect: {e}")
if r.status_code == 401:
die("Not authenticated. Run 'peertube auth login' first.")
if r.status_code >= 400:
try: d = r.json()
except: d = r.text[:200]
die(f"API error ({r.status_code}): {d}")
return r.json()
def get_server_info(self): return self._get("/server/")
def list_videos(self, page=1, limit=12, sort="-publishedAt"):
return self._get("/videos", {"page":page, "count":limit, "sort":sort})
def search_videos(self, query, page=1, limit=12):
return self._get("/search/videos", {"search":query, "page":page, "count":limit})
def get_video(self, vid): return self._get(f"/videos/{vid}")
def list_video_comments(self, vid, page=1, limit=10):
return self._get(f"/videos/{vid}/comments", {"page":page, "count":limit})
def list_channels(self, page=1, limit=15):
return self._get("/video-channels", {"page":page, "count":limit})
def get_channel(self, name): return self._get(f"/video-channels/{name}")
def list_channel_videos(self, name, page=1, limit=12):
return self._get(f"/video-channels/{name}/videos", {"page":page, "count":limit})
def list_playlists(self, page=1, limit=10):
return self._get("/video-playlists/user", {"page":page, "count":limit})
def get_playlist(self, pid): return self._get(f"/video-playlists/{pid}")
def my_profile(self): return self._get("/users/me")
def my_videos(self, page=1, limit=12): return self._get("/users/me/videos", {"page":page, "count":limit})
def fmt_video(v, idx=None):
prefix = f"{idx}. " if idx else ""
name = v.get("name", "?")
author = v.get("channel",{}).get("displayName", v.get("account",{}).get("displayName","?"))
dur = v.get("duration", 0)
dur_str = f"{int(dur//60)}:{int(dur%60):02d}"
views = v.get("views", 0)
pub = (v.get("publishedAt") or "")[:10]
return f" {prefix}{name:55} {dur_str} {views} views {author} {pub}"
def cmd_server(client, args):
if client.dry_run: return emit("[dry-run] Get server info", {"dry_run":True})
d = client.get_server_info() or {}
name = d.get("instance",{}).get("name","?")
desc = (d.get("instance",{}).get("shortDescription","") or "")[:80]
users = d.get("users","?"); videos = d.get("videos","?"); views = d.get("views","?")
emit(f"🖥️ {name}\n {desc}\n Users: {users} Videos: {videos} Views: {views}",
{"instance":{"name":name,"shortDescription":desc},"users":users,"videos":videos})
def cmd_videos(client, args):
p = argparse.ArgumentParser(prog="peertube videos")
p.add_argument("--limit", type=int, default=12)
parsed, _ = p.parse_known_args(args)
if client.dry_run: return emit("[dry-run] List videos", {"dry_run":True})
data = client.list_videos(limit=parsed.limit) or {}
videos = data.get("data",[])
if not videos: return emit("No videos.", {"videos":[]})
lines = [fmt_video(v, i+1) for i, v in enumerate(videos)]
total = data.get("total", len(videos))
emit(f"{total} video(s):\n"+"\n".join(lines), {"total":total,"videos":videos})
def cmd_search(client, args):
p = argparse.ArgumentParser(prog="peertube search")
p.add_argument("--query", "-q", required=True)
p.add_argument("--limit", type=int, default=12)
parsed, _ = p.parse_known_args(args)
if client.dry_run: return emit(f"[dry-run] Search: {parsed.query}", {"dry_run":True})
data = client.search_videos(parsed.query, limit=parsed.limit) or {}
videos = data.get("data",[])
if not videos: return emit("No results.", {"videos":[]})
lines = [fmt_video(v, i+1) for i, v in enumerate(videos)]
total = data.get("total", len(videos))
emit(f"{total} result(s):\n"+"\n".join(lines), {"total":total,"videos":videos})
def cmd_channels(client, args):
if client.dry_run: return emit("[dry-run] List channels", {"dry_run":True})
data = client.list_channels() or {}
channels = data.get("data",[])
if not channels: return emit("No channels.", {"channels":[]})
lines, out = [], []
for c in channels:
dn = c.get("displayName","?"); name = c.get("name","?"); vid = c.get("videosCount",0); subs = c.get("subscribersCount",0)
lines.append(f" {dn:30} @{name} {vid} videos {subs} subscribers")
out.append({"displayName":dn,"name":name,"videosCount":vid,"subscribersCount":subs})
emit(f"{len(channels)} channel(s):\n"+"\n".join(lines), {"channels":out})
def cmd_me(client, args):
if client.dry_run: return emit("[dry-run] Get profile", {"dry_run":True})
d = client.my_profile() or {}
uname = d.get("username","?"); role = d.get("role","?")
vid = d.get("videosCount",0); views = d.get("viewsCount",0)
emit(f"👤 @{uname} Role: {role} Videos: {vid} Views: {views}",
{"username":uname,"role":role,"videosCount":vid,"viewsCount":views})
def main():
global GLOBAL_FLAGS, QUIET
GLOBAL_FLAGS, filtered_argv = _preparse(sys.argv)
if GLOBAL_FLAGS.get("quiet"): QUIET = True
if GLOBAL_FLAGS.get("json"): warnings.simplefilter("ignore")
parser = argparse.ArgumentParser(prog="peertube", description="PeerTube federated video.",
epilog="Set PEERTUBE_SERVER. Login: peertube auth login --username ... --password ...")
sub = parser.add_subparsers(dest="command")
# auth
ap = sub.add_parser("auth", help="Authentication")
asub = ap.add_subparsers(dest="auth_action")
lp = asub.add_parser("login", help="Login to PeerTube instance")
lp.add_argument("--username", required=True); lp.add_argument("--password", required=True)
# other commands
sub.add_parser("server", help="Server info")
sub.add_parser("videos", help="List videos").add_argument("--limit",type=int,default=12)
sub.add_parser("channels", help="List channels")
sp = sub.add_parser("search", help="Search videos")
sp.add_argument("--query","-q",required=True); sp.add_argument("--limit",type=int,default=12)
sub.add_parser("me", help="My profile")
args = parser.parse_args(filtered_argv[1:])
if not args.command: parser.print_help(); sys.exit(1)
client = PeerTubeClient(dry_run=GLOBAL_FLAGS.get("dry_run",False))
remaining = filtered_argv[filtered_argv.index(args.command)+1:]
if args.command == "auth":
if args.auth_action == "login":
result = client.login(args.username, args.password)
emit(f"✅ Logged in to {client.server}", {"status":"logged_in","server":client.server})
else: ap.print_help()
return
# Commands that need auth
if not client._token and not GLOBAL_FLAGS.get("dry_run"):
die("Not logged in. Run 'peertube auth login --username ... --password ...' first, or use --dry-run.")
handlers = {"server":cmd_server,"videos":cmd_videos,"search":cmd_search,"channels":cmd_channels,"me":cmd_me}
h = handlers.get(args.command)
if not h: parser.print_help(); sys.exit(1)
h(client, remaining)
if __name__ == "__main__": main()
+961
View File
@@ -0,0 +1,961 @@
"""Offline test suite for the bundled peertube CLI.
All HTTP is mocked at the requests seam; the only live call in the file is the
single anonymous instance probe behind the PEERTUBE_LIVE_TESTS=1 guard (skipped
by default, so the suite is fully offline and passes the proxy-trap rerun).
Covers: help output, argument-error paths, dry-run plans, mocked OAuth2 token
persistence/refresh/revocation, handler output contracts, and the documented
multi-step pipeline stages (each stage's output fields/types feed the next).
"""
import contextlib
import importlib.machinery
import importlib.util
import io
import json
import os
import pathlib
import stat
import subprocess
import sys
import tempfile
import time
import unittest
from unittest.mock import patch
SCRIPT = pathlib.Path(__file__).resolve().parent / "peertube"
LOADER = importlib.machinery.SourceFileLoader("peertube_cli", str(SCRIPT))
SPEC = importlib.util.spec_from_loader(LOADER.name, LOADER)
pt = importlib.util.module_from_spec(SPEC)
LOADER.exec_module(pt)
def clean_env():
env = os.environ.copy()
for var in ("PEERTUBE_SERVER", "PEERTUBE_CONFIG_DIR", "PEERTUBE_LIVE_TESTS"):
env.pop(var, None)
return env
class FakeResponse:
def __init__(self, status_code=200, json_body=None, text="", headers=None):
self.status_code = status_code
self._json = json_body
self.text = text if text else (json.dumps(json_body) if json_body is not None else "")
self.headers = headers or {}
def json(self):
if self._json is None:
raise ValueError("no json body")
return self._json
VIDEO_ONE = {
"id": 1,
"uuid": "uuid-one",
"shortUUID": "sOne1",
"url": "https://inst.example/w/uuid-one",
"name": "First video",
"duration": 125,
"views": 42,
"likes": 7,
"publishedAt": "2026-08-01T10:00:00.000Z",
"privacy": {"id": 1, "label": "Public"},
"account": {"name": "alice", "displayName": "Alice", "host": "inst.example"},
"channel": {"name": "alice-channel", "displayName": "Alice Channel", "host": "inst.example"},
}
VIDEO_TWO = dict(
VIDEO_ONE,
id=2,
uuid="uuid-two",
shortUUID="sTwo2",
name="Second video",
duration=3661,
views=5,
channel={"name": "bob-channel", "displayName": "Bob Channel", "host": "other.example"},
)
TOKEN_RESPONSE = {
"access_token": "tok-1",
"refresh_token": "ref-1",
"token_type": "Bearer",
"expires_in": 3600,
"refresh_token_expires_in": 7200,
}
OAUTH_CLIENT = {"client_id": "cid-1", "client_secret": "client-secret-1"}
MASKED_OAUTH_CLIENT = {"client_id": "cid-1", "client_secret": "*" * 32}
def run_cli(*args, env=None):
return subprocess.run(
[sys.executable, str(SCRIPT), *args],
capture_output=True,
text=True,
env=env if env is not None else clean_env(),
)
class ModuleStateTestCase(unittest.TestCase):
"""Base that restores module globals mutated by in-process tests."""
def setUp(self):
self._flags = dict(pt.GLOBAL_FLAGS)
self._env_server = pt.ENV_SERVER
self._env_config = pt.ENV_CONFIG_DIR
pt.GLOBAL_FLAGS = {"json": True, "dry_run": False, "quiet": False, "verbose": False}
def tearDown(self):
pt.GLOBAL_FLAGS = self._flags
pt.ENV_SERVER = self._env_server
pt.ENV_CONFIG_DIR = self._env_config
class HelpOutputTests(unittest.TestCase):
"""Class 1: --help output."""
def test_help_lists_every_subcommand(self):
result = run_cli("--help")
self.assertEqual(result.returncode, 0, result.stderr)
for noun in (
"server",
"videos",
"video",
"search",
"comments",
"channels",
"channel",
"account",
"me",
"my-videos",
"login",
"logout",
):
self.assertIn(noun, result.stdout)
def test_help_names_the_instance_env_var(self):
result = run_cli("--help")
self.assertIn("PEERTUBE_SERVER", result.stdout)
self.assertIn("sepiasearch.org", result.stdout)
def test_search_help_states_its_scope(self):
result = run_cli("search", "--help")
self.assertEqual(result.returncode, 0, result.stderr)
self.assertIn("searchTarget", result.stdout)
self.assertIn("sepiasearch", result.stdout.lower())
def test_leaf_help_carries_examples(self):
for leaf in ("videos", "comments", "login", "logout"):
result = run_cli(leaf, "--help")
with self.subTest(leaf=leaf):
self.assertEqual(result.returncode, 0, result.stderr)
self.assertIn("Example:", result.stdout)
class ArgumentErrorTests(unittest.TestCase):
"""Class 2: argument-error paths fail cleanly before any network call."""
def test_search_requires_query(self):
result = run_cli("search")
self.assertNotEqual(result.returncode, 0)
self.assertIn("--query", result.stderr)
self.assertNotIn("Traceback", result.stderr)
def test_video_requires_id(self):
result = run_cli("video")
self.assertNotEqual(result.returncode, 0)
self.assertIn("--id", result.stderr)
def test_channel_requires_handle(self):
result = run_cli("channel")
self.assertNotEqual(result.returncode, 0)
self.assertIn("--handle", result.stderr)
def test_no_subcommand_prints_help_and_exits(self):
result = run_cli()
self.assertNotEqual(result.returncode, 0)
self.assertIn("usage", result.stdout)
def test_limit_above_server_maximum_rejected(self):
result = run_cli("--dry-run", "--json", "videos", "--limit", "101")
self.assertNotEqual(result.returncode, 0)
self.assertIn("100", result.stderr)
def test_limit_zero_rejected(self):
result = run_cli("--dry-run", "--json", "videos", "--limit", "0")
self.assertNotEqual(result.returncode, 0)
self.assertIn("--limit", result.stderr)
def test_negative_offset_rejected(self):
result = run_cli("--dry-run", "--json", "videos", "--offset", "-1")
self.assertNotEqual(result.returncode, 0)
self.assertIn("--offset", result.stderr)
def test_login_without_password_errors(self):
result = run_cli("login", "--username", "alice", "--server", "https://inst.example")
self.assertNotEqual(result.returncode, 0)
self.assertIn("password", result.stderr.lower())
self.assertNotIn("Traceback", result.stderr)
def test_missing_server_dies_before_network(self):
result = run_cli("videos", "--limit", "1")
self.assertNotEqual(result.returncode, 0)
self.assertIn("PEERTUBE_SERVER", result.stderr)
self.assertNotIn("Traceback", result.stderr)
class DryRunPlanTests(ModuleStateTestCase):
"""Class 3: --dry-run emits valid JSON plans with zero network activity."""
def run_json(self, *args):
pt.GLOBAL_FLAGS = {"json": True, "dry_run": True, "quiet": False, "verbose": False}
return io.StringIO()
def test_single_endpoint_plans_emit_method_path_params(self):
cases = (
(
pt.cmd_videos,
["--limit", "3"],
{
"method": "GET",
"path": "/api/v1/videos",
"params.start": 0,
"params.count": 3,
"params.sort": "-publishedAt",
},
),
(
pt.cmd_search,
["--query", "linux", "--limit", "5"],
{
"method": "GET",
"path": "/api/v1/search/videos",
"params.searchTarget": "local",
"params.search": "linux",
},
),
(
pt.cmd_video,
["--id", "uuid-one"],
{"method": "GET", "path": "/api/v1/videos/uuid-one"},
),
(
pt.cmd_comments,
["--id", "uuid-one"],
{"method": "GET", "path": "/api/v1/videos/uuid-one/comment-threads"},
),
(pt.cmd_channels, [], {"method": "GET", "path": "/api/v1/video-channels"}),
(pt.cmd_me, [], {"method": "GET", "path": "/api/v1/users/me"}),
(pt.cmd_my_videos, [], {"method": "GET", "path": "/api/v1/users/me/videos"}),
(pt.cmd_logout, [], {"method": "POST", "path": "/api/v1/users/revoke-token"}),
)
for handler, args, expectations in cases:
with self.subTest(handler=handler.__name__):
client = pt.PeerTubeClient(
server="https://inst.example",
dry_run=True,
config_dir=tempfile.mkdtemp(prefix="pt-dry-"),
)
out = io.StringIO()
with contextlib.redirect_stdout(out):
handler(client, args)
plan = json.loads(out.getvalue())
self.assertTrue(plan["dry_run"])
self.assertEqual(plan["method"], expectations["method"])
self.assertEqual(plan["path"], expectations["path"])
def test_dry_run_videos_plan_never_sends_page_param(self):
client = pt.PeerTubeClient(
server="https://inst.example",
dry_run=True,
config_dir=tempfile.mkdtemp(prefix="pt-dry-"),
)
out = io.StringIO()
with contextlib.redirect_stdout(out):
pt.cmd_videos(client, ["--limit", "9", "--offset", "18"])
plan = json.loads(out.getvalue())
self.assertNotIn("page", plan["params"])
self.assertEqual(plan["params"]["start"], 18)
self.assertEqual(plan["params"]["count"], 9)
def test_search_plan_defaults_to_instance_local_scope(self):
client = pt.PeerTubeClient(
server="https://inst.example",
dry_run=True,
config_dir=tempfile.mkdtemp(prefix="pt-dry-"),
)
out = io.StringIO()
with contextlib.redirect_stdout(out):
pt.cmd_search(client, ["--query", "peertube"])
plan = json.loads(out.getvalue())
self.assertEqual(plan["params"]["searchTarget"], "local")
def test_composite_commands_plan_every_request(self):
client = pt.PeerTubeClient(
server="https://inst.example",
dry_run=True,
config_dir=tempfile.mkdtemp(prefix="pt-dry-"),
)
out = io.StringIO()
with contextlib.redirect_stdout(out):
pt.cmd_server(client, [])
plan = json.loads(out.getvalue())
paths = [req["path"] for req in plan["requests"]]
self.assertEqual(paths, ["/api/v1/config/about", "/api/v1/server/stats"])
out = io.StringIO()
with contextlib.redirect_stdout(out):
pt.cmd_channel(client, ["--handle", "alice-channel"])
plan = json.loads(out.getvalue())
paths = [req["path"] for req in plan["requests"]]
self.assertEqual(
paths,
["/api/v1/video-channels/alice-channel", "/api/v1/video-channels/alice-channel/videos"],
)
def test_login_dry_run_lists_form_fields_without_values(self):
out = io.StringIO()
with contextlib.redirect_stdout(out):
pt.cmd_login(
pt.PeerTubeClient(
server="https://inst.example",
dry_run=True,
config_dir=tempfile.mkdtemp(prefix="pt-dry-"),
),
["--username", "alice"],
)
plan = json.loads(out.getvalue())
self.assertTrue(plan["dry_run"])
self.assertEqual(plan["path"], "/api/v1/users/token")
self.assertIn("grant_type", plan["form_fields"])
self.assertIn("client_secret", plan["form_fields"])
self.assertNotIn("form", plan) # no values leak in the plan
def test_dry_run_works_without_any_server_configured(self):
pt.ENV_SERVER = ""
client = pt.PeerTubeClient(
server="", dry_run=True, config_dir=tempfile.mkdtemp(prefix="pt-dry-")
)
out = io.StringIO()
with contextlib.redirect_stdout(out):
pt.cmd_videos(client, ["--limit", "2"])
self.assertTrue(json.loads(out.getvalue())["dry_run"])
def test_dry_run_never_touches_network(self):
client = pt.PeerTubeClient(
server="https://inst.example",
dry_run=True,
config_dir=tempfile.mkdtemp(prefix="pt-dry-"),
)
with (
patch.object(pt.requests, "get") as getter,
patch.object(pt.requests, "post") as poster,
):
for handler, args in (
(pt.cmd_videos, ["--limit", "2"]),
(pt.cmd_search, ["--query", "x"]),
(pt.cmd_server, []),
(pt.cmd_me, []),
(pt.cmd_login, ["--username", "a"]),
(pt.cmd_logout, []),
):
buf = io.StringIO()
with contextlib.redirect_stdout(buf):
handler(client, args)
getter.assert_not_called()
poster.assert_not_called()
def test_flags_work_before_and_after_subcommand(self):
result = run_cli("--json", "--dry-run", "search", "--query", "x", "--limit", "2")
self.assertEqual(result.returncode, 0, result.stderr)
self.assertTrue(json.loads(result.stdout)["dry_run"])
result = run_cli("search", "--query", "x", "--limit", "2", "--json", "--dry-run")
self.assertEqual(result.returncode, 0, result.stderr)
self.assertTrue(json.loads(result.stdout)["dry_run"])
class ClientContractTests(ModuleStateTestCase):
"""Class 4a: mocked requests — paths, params, and error handling."""
def mocked_get(self, responses, server="https://inst.example"):
client = pt.PeerTubeClient(server=server, config_dir=tempfile.mkdtemp(prefix="pt-cc-"))
return client, patch.object(pt.requests, "get", side_effect=responses)
def test_video_listing_sends_start_count_sort(self):
client, patcher = self.mocked_get(
[FakeResponse(200, {"total": 2, "data": [VIDEO_ONE, VIDEO_TWO]})]
)
with patcher as getter:
pt.cmd_videos(client, ["--limit", "2", "--offset", "10"])
args, kwargs = getter.call_args
self.assertEqual(args[0], "https://inst.example/api/v1/videos")
self.assertEqual(kwargs["params"], {"start": 10, "count": 2, "sort": "-publishedAt"})
self.assertNotIn("page", kwargs["params"])
def test_search_defaults_to_local_target_and_omits_empty_sort(self):
client, patcher = self.mocked_get([FakeResponse(200, {"total": 0, "data": []})])
with patcher as getter:
pt.cmd_search(client, ["--query", "linux"])
params = getter.call_args[1]["params"]
self.assertEqual(params["searchTarget"], "local")
self.assertEqual(params["search"], "linux")
self.assertNotIn("sort", params)
def test_comment_threads_route_is_hyphenated(self):
client, patcher = self.mocked_get(
[FakeResponse(200, {"total": 0, "totalNotDeletedComments": 0, "data": []})]
)
with patcher as getter:
pt.cmd_comments(client, ["--id", "uuid-one"])
self.assertEqual(
getter.call_args[0][0], "https://inst.example/api/v1/videos/uuid-one/comment-threads"
)
def test_server_composes_about_and_stats(self):
about = FakeResponse(
200, {"instance": {"name": "Inst", "shortDescription": "Desc", "description": "Long"}}
)
stats = FakeResponse(
200,
{
"totalUsers": 9,
"totalLocalVideos": 933,
"totalVideos": 23890,
"totalLocalVideoViews": 1001751,
"totalLocalVideoDownloads": 32569,
"totalLocalVideoChannels": 28,
},
)
client, patcher = self.mocked_get([about, stats])
with patcher:
out = io.StringIO()
with contextlib.redirect_stdout(out):
pt.cmd_server(client, [])
payload = json.loads(out.getvalue())
self.assertEqual(payload["instance"]["name"], "Inst")
self.assertEqual(payload["stats"]["totalLocalVideos"], 933)
self.assertIsInstance(payload["stats"]["totalUsers"], int)
def test_401_names_login_remedy(self):
client, patcher = self.mocked_get([FakeResponse(401, {"detail": "token expired"})])
client._token = "stale-token" # authed command proceeds, then server rejects
with patcher:
err = io.StringIO()
with contextlib.redirect_stderr(err), self.assertRaises(SystemExit):
pt.cmd_me(client, [])
self.assertIn("401", err.getvalue())
self.assertIn("login", err.getvalue().lower())
def test_429_surfaces_retry_after(self):
client, patcher = self.mocked_get(
[FakeResponse(429, {"detail": "rate limit"}, headers={"Retry-After": "7"})]
)
with patcher:
err = io.StringIO()
with contextlib.redirect_stderr(err), self.assertRaises(SystemExit):
pt.cmd_videos(client, ["--limit", "2"])
self.assertIn("429", err.getvalue())
self.assertIn("Retry-After", err.getvalue())
def test_rfc7807_detail_extracted_on_generic_error(self):
client, patcher = self.mocked_get(
[
FakeResponse(
400,
{
"type": "about:blank",
"title": "Bad Request",
"status": 400,
"detail": "unknown route shape",
},
)
]
)
with patcher:
err = io.StringIO()
with contextlib.redirect_stderr(err), self.assertRaises(SystemExit):
pt.cmd_videos(client, ["--limit", "2"])
self.assertIn("unknown route shape", err.getvalue())
def test_non_json_instance_response_is_diagnosed(self):
client, patcher = self.mocked_get([FakeResponse(200, text="<html>not peertube</html>")])
with patcher:
err = io.StringIO()
with contextlib.redirect_stderr(err), self.assertRaises(SystemExit):
pt.cmd_videos(client, ["--limit", "2"])
self.assertIn("Non-JSON", err.getvalue())
class OAuthFlowTests(ModuleStateTestCase):
"""Class 4b: mocked OAuth2 — client fetch, password grant, persistence,
refresh, revocation. Token files live in TemporaryDirectories only."""
def setUp(self):
super().setUp()
self.tmp = tempfile.TemporaryDirectory(prefix="pt-oauth-")
self.config_dir = self.tmp.name
pt.ENV_SERVER = "https://inst.example"
def tearDown(self):
self.tmp.cleanup()
super().tearDown()
def client(self, **kwargs):
return pt.PeerTubeClient(
server="https://inst.example", config_dir=self.config_dir, **kwargs
)
def test_fetch_oauth_client_hits_singular_local_route(self):
client = self.client()
with patch.object(
pt.requests, "get", return_value=FakeResponse(200, OAUTH_CLIENT)
) as getter:
client_id, client_secret = client.fetch_oauth_client()
self.assertEqual(getter.call_args[0][0], "https://inst.example/api/v1/oauth-clients/local")
self.assertEqual((client_id, client_secret), ("cid-1", "client-secret-1"))
def test_password_grant_sends_form_encoded_fields(self):
client = self.client()
with (
patch.object(pt.requests, "get", return_value=FakeResponse(200, OAUTH_CLIENT)),
patch.object(
pt.requests, "post", return_value=FakeResponse(200, TOKEN_RESPONSE)
) as poster,
):
out = io.StringIO()
with contextlib.redirect_stdout(out):
pt.cmd_login(client, ["--username", "alice", "--password", "pw"])
args, kwargs = poster.call_args
self.assertEqual(args[0], "https://inst.example/api/v1/users/token")
form = kwargs["data"]
self.assertEqual(form["grant_type"], "password")
self.assertEqual(form["username"], "alice")
self.assertEqual(form["client_id"], "cid-1")
self.assertNotIn("response_type", form) # not part of the documented schema
payload = json.loads(out.getvalue())
self.assertEqual(payload["status"], "logged_in")
def test_bad_password_400_exits_with_guidance(self):
client = self.client()
with (
patch.object(pt.requests, "get", return_value=FakeResponse(200, OAUTH_CLIENT)),
patch.object(
pt.requests, "post", return_value=FakeResponse(400, {"detail": "invalid_grant"})
),
):
err = io.StringIO()
with contextlib.redirect_stderr(err), self.assertRaises(SystemExit):
pt.cmd_login(client, ["--username", "alice", "--password", "wrong"])
self.assertIn("400", err.getvalue())
self.assertIn("invalid_grant", err.getvalue())
def test_two_factor_401_suggests_otp_flag(self):
client = self.client()
with (
patch.object(pt.requests, "get", return_value=FakeResponse(200, OAUTH_CLIENT)),
patch.object(pt.requests, "post", return_value=FakeResponse(401, {})),
):
err = io.StringIO()
with contextlib.redirect_stderr(err), self.assertRaises(SystemExit):
pt.cmd_login(client, ["--username", "alice", "--password", "pw"])
self.assertIn("--otp", err.getvalue())
def test_otp_header_attached_when_provided(self):
client = self.client()
with (
patch.object(pt.requests, "get", return_value=FakeResponse(200, OAUTH_CLIENT)),
patch.object(
pt.requests, "post", return_value=FakeResponse(200, TOKEN_RESPONSE)
) as poster,
):
out = io.StringIO()
with contextlib.redirect_stdout(out):
pt.cmd_login(client, ["--username", "alice", "--password", "pw", "--otp", "123456"])
self.assertEqual(poster.call_args[1]["headers"]["x-peertube-otp"], "123456")
def test_masked_client_secret_stops_login_with_guidance(self):
client = self.client()
with (
patch.object(pt.requests, "get", return_value=FakeResponse(200, MASKED_OAUTH_CLIENT)),
patch.object(pt.requests, "post") as poster,
):
err = io.StringIO()
with contextlib.redirect_stderr(err), self.assertRaises(SystemExit):
pt.cmd_login(client, ["--username", "alice", "--password", "pw"])
self.assertIn("masks client_secret", err.getvalue())
poster.assert_not_called()
def test_is_masked_secret_detection(self):
self.assertTrue(pt.is_masked_secret("*" * 32))
self.assertFalse(pt.is_masked_secret("client-secret-1"))
self.assertFalse(pt.is_masked_secret(""))
self.assertFalse(pt.is_masked_secret("*-mixed-*"))
def test_token_file_persisted_owner_only_with_expiry(self):
client = self.client()
with (
patch.object(pt.requests, "get", return_value=FakeResponse(200, OAUTH_CLIENT)),
patch.object(pt.requests, "post", return_value=FakeResponse(200, TOKEN_RESPONSE)),
):
buf = io.StringIO()
with contextlib.redirect_stdout(buf):
pt.cmd_login(client, ["--username", "alice", "--password", "pw"])
token_path = client.token_path()
self.assertTrue(os.path.isfile(token_path))
mode = stat.S_IMODE(os.stat(token_path).st_mode)
self.assertEqual(mode & 0o077, 0, "token file must be owner-only")
with open(token_path) as handle:
record = json.load(handle)
self.assertEqual(record["server"], "https://inst.example")
self.assertEqual(record["access_token"], "tok-1")
self.assertEqual(record["refresh_token"], "ref-1")
self.assertIsNotNone(record["expires_at"])
self.assertGreater(record["expires_at"], time.time())
self.assertLess(record["expires_at"], time.time() + 7200)
def test_token_from_another_instance_is_ignored(self):
client = self.client()
with (
patch.object(pt.requests, "get", return_value=FakeResponse(200, OAUTH_CLIENT)),
patch.object(pt.requests, "post", return_value=FakeResponse(200, TOKEN_RESPONSE)),
):
buf = io.StringIO()
with contextlib.redirect_stdout(buf):
pt.cmd_login(client, ["--username", "alice", "--password", "pw"])
other = pt.PeerTubeClient(server="https://other.example", config_dir=self.config_dir)
self.assertIsNone(other._token)
def test_expired_token_triggers_refresh_then_success(self):
client = self.client()
client.save_session(dict(TOKEN_RESPONSE, expires_in=-10)) # already expired
refreshed = dict(TOKEN_RESPONSE, access_token="tok-2", expires_in=3600)
with (
patch.object(pt.requests, "get", return_value=FakeResponse(200, OAUTH_CLIENT)),
patch.object(pt.requests, "post", return_value=FakeResponse(200, refreshed)) as poster,
):
buf = io.StringIO()
with contextlib.redirect_stdout(buf):
pt.cmd_me(client, [])
form = poster.call_args[1]["data"]
self.assertEqual(form["grant_type"], "refresh_token")
self.assertEqual(form["refresh_token"], "ref-1")
self.assertEqual(client._token, "tok-2")
with open(client.token_path()) as handle:
self.assertEqual(json.load(handle)["access_token"], "tok-2")
def test_failed_refresh_falls_back_to_login_guidance(self):
client = self.client()
client.save_session(dict(TOKEN_RESPONSE, expires_in=-10))
with (
patch.object(pt.requests, "get", return_value=FakeResponse(200, OAUTH_CLIENT)),
patch.object(pt.requests, "post", return_value=FakeResponse(400, {})),
):
err = io.StringIO()
with contextlib.redirect_stderr(err), self.assertRaises(SystemExit):
pt.cmd_me(client, [])
self.assertIn("Not authenticated", err.getvalue())
def test_logout_revokes_and_deletes_token_file(self):
client = self.client()
client.save_session(TOKEN_RESPONSE)
self.assertTrue(os.path.isfile(client.token_path()))
with patch.object(pt.requests, "post", return_value=FakeResponse(200, {})) as poster:
out = io.StringIO()
with contextlib.redirect_stdout(out):
pt.cmd_logout(client, [])
args, kwargs = poster.call_args
self.assertEqual(args[0], "https://inst.example/api/v1/users/revoke-token")
self.assertEqual(kwargs["headers"]["Authorization"], "Bearer tok-1")
self.assertFalse(os.path.exists(client.token_path()))
def test_logout_keeps_file_when_revocation_fails(self):
client = self.client()
client.save_session(TOKEN_RESPONSE)
with patch.object(pt.requests, "post", return_value=FakeResponse(500, {})):
err = io.StringIO()
with contextlib.redirect_stderr(err), self.assertRaises(SystemExit):
pt.cmd_logout(client, [])
self.assertTrue(os.path.isfile(client.token_path()))
def test_logout_without_token_errors_cleanly(self):
client = self.client()
err = io.StringIO()
with contextlib.redirect_stderr(err), self.assertRaises(SystemExit):
pt.cmd_logout(client, [])
self.assertIn("No stored token", err.getvalue())
class HandlerOutputTests(ModuleStateTestCase):
"""Class 4c: handler output contracts consumed by jq pipelines."""
def test_videos_output_carries_raw_video_objects(self):
client = pt.PeerTubeClient(
server="https://inst.example", config_dir=tempfile.mkdtemp(prefix="pt-ho-")
)
with patch.object(
pt.requests,
"get",
return_value=FakeResponse(200, {"total": 2, "data": [VIDEO_ONE, VIDEO_TWO]}),
):
out = io.StringIO()
with contextlib.redirect_stdout(out):
pt.cmd_videos(client, ["--limit", "2"])
payload = json.loads(out.getvalue())
self.assertEqual(payload["total"], 2)
self.assertEqual(payload["start"], 0)
self.assertEqual(payload["count"], 2)
self.assertIsInstance(payload["videos"], list)
first = payload["videos"][0]
self.assertEqual(first["uuid"], "uuid-one")
self.assertIsInstance(first["duration"], int) # seconds
self.assertEqual(first["channel"]["host"], "inst.example")
def test_search_output_marks_scope_and_carries_uuids(self):
client = pt.PeerTubeClient(
server="https://inst.example", config_dir=tempfile.mkdtemp(prefix="pt-ho-")
)
with patch.object(
pt.requests, "get", return_value=FakeResponse(200, {"total": 1, "data": [VIDEO_TWO]})
):
out = io.StringIO()
with contextlib.redirect_stdout(out):
pt.cmd_search(client, ["--query", "x"])
payload = json.loads(out.getvalue())
self.assertEqual(payload["videos"][0]["uuid"], "uuid-two")
def test_me_tolerates_docs_array_sample_and_object(self):
client = pt.PeerTubeClient(
server="https://inst.example", config_dir=tempfile.mkdtemp(prefix="pt-ho-")
)
profile = {
"username": "alice",
"role": {"id": 1, "label": "User"},
"videoQuota": 1073741824,
"videoChannels": [],
}
for body in (profile, [profile]):
client.save_session(TOKEN_RESPONSE)
with patch.object(pt.requests, "get", return_value=FakeResponse(200, body)):
out = io.StringIO()
with contextlib.redirect_stdout(out):
pt.cmd_me(client, [])
payload = json.loads(out.getvalue())
self.assertEqual(payload["username"], "alice")
self.assertEqual(payload["role"]["label"], "User")
client.clear_token()
def test_comments_output_exposes_thread_counts(self):
client = pt.PeerTubeClient(
server="https://inst.example", config_dir=tempfile.mkdtemp(prefix="pt-ho-")
)
body = {
"total": 1,
"totalNotDeletedComments": 3,
"data": [
{"totalReplies": 3, "comment": {"text": "nice video", "account": {"name": "bob"}}}
],
}
with patch.object(pt.requests, "get", return_value=FakeResponse(200, body)):
out = io.StringIO()
with contextlib.redirect_stdout(out):
pt.cmd_comments(client, ["--id", "uuid-one"])
payload = json.loads(out.getvalue())
self.assertEqual(payload["total"], 1)
self.assertEqual(payload["total_not_deleted"], 3)
self.assertEqual(payload["threads"][0]["comment"]["text"], "nice video")
def test_channels_output_carries_handles_and_counts(self):
client = pt.PeerTubeClient(
server="https://inst.example", config_dir=tempfile.mkdtemp(prefix="pt-ho-")
)
body = {
"total": 1,
"data": [
{
"name": "alice-channel",
"displayName": "Alice Channel",
"host": "inst.example",
"videosCount": 12,
"followersCount": 34,
}
],
}
with patch.object(pt.requests, "get", return_value=FakeResponse(200, body)):
out = io.StringIO()
with contextlib.redirect_stdout(out):
pt.cmd_channels(client, [])
payload = json.loads(out.getvalue())
channel = payload["channels"][0]
self.assertEqual(channel["name"], "alice-channel")
self.assertIsInstance(channel["videosCount"], int)
class PipelineChainTests(ModuleStateTestCase):
"""Documented multi-step recipes must execute stage by stage, each stage's
output field names AND JSON types consumable by the next."""
@classmethod
def setUpClass(cls):
cls.tmpdir = tempfile.TemporaryDirectory(prefix="pt-pipeline-")
@classmethod
def tearDownClass(cls):
cls.tmpdir.cleanup()
def run_cli(self, *args):
env = clean_env()
env["PEERTUBE_SERVER"] = "https://inst.example"
env["PEERTUBE_CONFIG_DIR"] = self.tmpdir.name
return subprocess.run(
[sys.executable, str(SCRIPT), "--json", "--dry-run", *args],
capture_output=True,
text=True,
env=env,
)
def run_jq(self, *jq_args, stdin_text=""):
return subprocess.run(
["jq", *jq_args], input=stdin_text, capture_output=True, text=True, env=clean_env()
)
def stage_file(self, name, document):
path = pathlib.Path(self.tmpdir.name) / name
path.write_text(json.dumps(document))
return str(path)
def test_browse_then_detail_chain_consumability(self):
# Stage 1: videos plan; jq proves the path and the start-offset type
# (number) that stage two consumes when picking an id from the listing.
r1 = self.run_cli("videos", "--limit", "2")
self.assertEqual(r1.returncode, 0, r1.stderr)
self.stage_file("s1.json", json.loads(r1.stdout))
self.assertEqual(
self.run_jq("-r", ".path", stdin_text=r1.stdout).stdout.strip(), "/api/v1/videos"
)
self.assertEqual(
self.run_jq("-r", ".params.start | type", stdin_text=r1.stdout).stdout.strip(), "number"
)
# Stage 2: detail plan consumes an id into the URL path.
r2 = self.run_cli("video", "--id", "uuid-one")
self.assertEqual(r2.returncode, 0, r2.stderr)
self.stage_file("s2.json", json.loads(r2.stdout))
self.assertEqual(
self.run_jq("-r", ".path", stdin_text=r2.stdout).stdout.strip(),
"/api/v1/videos/uuid-one",
)
def test_search_then_video_chain_consumability(self):
r1 = self.run_cli("search", "--query", "linux", "--limit", "3")
self.assertEqual(r1.returncode, 0, r1.stderr)
self.assertEqual(
self.run_jq("-r", ".params.searchTarget", stdin_text=r1.stdout).stdout.strip(), "local"
)
self.assertEqual(
self.run_jq("-r", ".params.count | type", stdin_text=r1.stdout).stdout.strip(), "number"
)
# The documented jq selector .videos[0].uuid maps to detail --id.
r2 = self.run_cli("video", "--id", "uuid-from-search")
self.assertEqual(r2.returncode, 0, r2.stderr)
self.assertEqual(
self.run_jq("-r", ".path", stdin_text=r2.stdout).stdout.strip(),
"/api/v1/videos/uuid-from-search",
)
def test_channel_offset_paging_chain_consumability(self):
r1 = self.run_cli("channels", "--limit", "100", "--offset", "0")
self.assertEqual(r1.returncode, 0, r1.stderr)
self.assertEqual(
self.run_jq("-r", ".path", stdin_text=r1.stdout).stdout.strip(),
"/api/v1/video-channels",
)
r2 = self.run_cli(
"channel", "--handle", "alice-channel", "--limit", "100", "--offset", "100"
)
self.assertEqual(r2.returncode, 0, r2.stderr)
plan = json.loads(r2.stdout)
video_req = plan["requests"][1]
self.assertEqual(video_req["params"]["start"], 100)
self.assertEqual(video_req["params"]["count"], 100)
self.assertNotIn("page", video_req["params"])
def test_login_to_me_chain_handoff(self):
# Stage 1: login plan lists the form fields (no values).
r1 = self.run_cli("login", "--username", "alice")
self.assertEqual(r1.returncode, 0, r1.stderr)
self.assertEqual(
self.run_jq("-r", ".path", stdin_text=r1.stdout).stdout.strip(), "/api/v1/users/token"
)
fields = json.loads(self.run_jq("-c", ".form_fields", stdin_text=r1.stdout).stdout)
self.assertIn("grant_type", fields)
# Stage 2: me plan rides the Authorization header the login persisted.
r2 = self.run_cli("me")
self.assertEqual(r2.returncode, 0, r2.stderr)
self.assertEqual(
self.run_jq("-r", ".path", stdin_text=r2.stdout).stdout.strip(), "/api/v1/users/me"
)
# Stage 3: logout plan revokes on the same instance.
r3 = self.run_cli("logout")
self.assertEqual(r3.returncode, 0, r3.stderr)
self.assertEqual(
self.run_jq("-r", ".path", stdin_text=r3.stdout).stdout.strip(),
"/api/v1/users/revoke-token",
)
def test_server_composition_plan_targets_both_endpoints(self):
r1 = self.run_cli("server")
self.assertEqual(r1.returncode, 0, r1.stderr)
paths = json.loads(self.run_jq("-c", "[.requests[].path]", stdin_text=r1.stdout).stdout)
self.assertEqual(paths, ["/api/v1/config/about", "/api/v1/server/stats"])
def test_mocked_browse_to_detail_stage_types(self):
"""Live-shape variant of recipe 1: the videos output's uuid (string)
feeds video --id, and the detail object carries description/url."""
pt.GLOBAL_FLAGS = {"json": True, "dry_run": False, "quiet": False, "verbose": False}
client = pt.PeerTubeClient(server="https://inst.example", config_dir=self.tmpdir.name)
detail = dict(VIDEO_ONE, description="full text", commentsEnabled=True)
with patch.object(
pt.requests,
"get",
side_effect=[
FakeResponse(200, {"total": 1, "data": [VIDEO_ONE]}),
FakeResponse(200, detail),
],
):
first = io.StringIO()
with contextlib.redirect_stdout(first):
pt.cmd_videos(client, ["--limit", "1"])
listing = json.loads(first.getvalue())
consumed_id = listing["videos"][0]["uuid"]
self.assertIsInstance(consumed_id, str)
second = io.StringIO()
with contextlib.redirect_stdout(second):
pt.cmd_video(client, ["--id", consumed_id])
video_detail = json.loads(second.getvalue())
self.assertEqual(video_detail["uuid"], consumed_id)
self.assertIsInstance(video_detail["description"], str)
self.assertIsInstance(video_detail["commentsEnabled"], bool)
class EnvGuardedLiveProbeTests(unittest.TestCase):
"""Optional anonymous instance probe (keyless public endpoint). Runs only
with PEERTUBE_LIVE_TESTS=1; skipped cleanly otherwise so the suite stays
fully offline under the proxy-trap."""
def test_public_instance_oauth_client_probe(self):
if os.getenv("PEERTUBE_LIVE_TESTS") != "1":
self.skipTest("live probe disabled (set PEERTUBE_LIVE_TESTS=1)")
result = subprocess.run(
[sys.executable, str(SCRIPT), "--json", "server", "--server", "https://framatube.org"],
capture_output=True,
text=True,
env=clean_env(),
timeout=60,
)
self.assertEqual(result.returncode, 0, result.stderr)
payload = json.loads(result.stdout)
self.assertEqual(payload["instance"]["name"], "Framatube")
self.assertIsInstance(payload["stats"]["totalLocalVideos"], int)
if __name__ == "__main__":
unittest.main()
+1
View File
@@ -22,6 +22,7 @@ Each skill's `description` field is the canonical routing contract. This conveni
| "Jellyfin", "Jellyfin media server", "recently added movies", "recently added episodes", "media library", "JELLYFIN_API_KEY", "Jellyfin API authentication" | [jellyfin](../jellyfin/SKILL.md) | | "Jellyfin", "Jellyfin media server", "recently added movies", "recently added episodes", "media library", "JELLYFIN_API_KEY", "Jellyfin API authentication" | [jellyfin](../jellyfin/SKILL.md) |
| "Jira", "Atlassian Jira", "JQL", "ticket PROJ-123", "sprint work", "JIRA_API_TOKEN" | [jira](../jira/SKILL.md) | | "Jira", "Atlassian Jira", "JQL", "ticket PROJ-123", "sprint work", "JIRA_API_TOKEN" | [jira](../jira/SKILL.md) |
| "Open Library", "openlibrary", "book search", "ISBN lookup", "author records", "work details", "book editions", "book ratings", "cover image" | [openlibrary](../openlibrary/SKILL.md) | | "Open Library", "openlibrary", "book search", "ISBN lookup", "author records", "work details", "book editions", "book ratings", "cover image" | [openlibrary](../openlibrary/SKILL.md) |
| "PeerTube", "peertube", "federated video", "SepiaSearch", "decentralized video platform", "PEERTUBE_SERVER", "my PeerTube instance" | [peertube](../peertube/SKILL.md) |
| "weather", "forecast", "temperature", "is it raining", "Tempest" | [tempest](../tempest/SKILL.md) | | "weather", "forecast", "temperature", "is it raining", "Tempest" | [tempest](../tempest/SKILL.md) |
| "TMDb", "The Movie Database", "movie search", "trending movies", "upcoming TV releases", "TMDB_ACCESS_TOKEN" | [tmdb](../tmdb/SKILL.md) | | "TMDb", "The Movie Database", "movie search", "trending movies", "upcoming TV releases", "TMDB_ACCESS_TOKEN" | [tmdb](../tmdb/SKILL.md) |
| "traefik", "reverse proxy", "load balancer", "API gateway", "Let's Encrypt", "ACME", "Docker routing", "traefik.yml", "entry point", "middleware", "TLS termination", "forward auth", "rate limit" | [traefik](../traefik/SKILL.md) | | "traefik", "reverse proxy", "load balancer", "API gateway", "Let's Encrypt", "ACME", "Docker routing", "traefik.yml", "entry point", "middleware", "TLS termination", "forward auth", "rate limit" | [traefik](../traefik/SKILL.md) |