{
  "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"
      ]
    }
  ]
}

# 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.

# 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` (1–100, **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.

# 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.

# 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` (1–100, 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.

# 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", "path": "/api/v1/...",
"params": {...}}` — every plan carries exactly `dry_run`, `method`, `path`, and `params`
(test-pinned in `scripts/test_peertube.py`); composite commands emit a `requests` array of
those same keyed steps, and the `login` plan is a `POST /api/v1/users/token` whose
`form_fields` lists the field NAMES only (never values). 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.

"""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_me_tolerates_scalar_role_without_attribute_error(self):
        """VAL-PT-010: a non-dict `role` (scalar id from a version-drifted
        server) must degrade to a readable line, never AttributeError."""
        client = pt.PeerTubeClient(
            server="https://inst.example", config_dir=tempfile.mkdtemp(prefix="pt-ho-")
        )
        profile = {"username": "bob", "role": 2, "videoQuota": None}
        client.save_session(TOKEN_RESPONSE)
        with patch.object(pt.requests, "get", return_value=FakeResponse(200, profile)):
            out = io.StringIO()
            err = io.StringIO()
            with contextlib.redirect_stdout(out), contextlib.redirect_stderr(err):
                pt.cmd_me(client, [])
        self.assertNotIn("Traceback", err.getvalue())
        self.assertEqual(json.loads(out.getvalue())["role"], 2)
        client.clear_token()

    def test_me_tolerates_missing_role(self):
        client = pt.PeerTubeClient(
            server="https://inst.example", config_dir=tempfile.mkdtemp(prefix="pt-ho-")
        )
        client.save_session(TOKEN_RESPONSE)
        with patch.object(pt.requests, "get", return_value=FakeResponse(200, {"username": "carol"})):
            out = io.StringIO()
            with contextlib.redirect_stdout(out):
                pt.cmd_me(client, [])
        self.assertEqual(json.loads(out.getvalue())["username"], "carol")
        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()