diff --git a/.env.example b/.env.example index 9ce5d9e..8d00aab 100644 --- a/.env.example +++ b/.env.example @@ -53,6 +53,10 @@ DEV_PAGES_USE_VITE=true DEV_PAGES_VITE_URL=http://localhost:5173 DEV_PAGES_MANIFEST_PATH=frontend/dev-pages/dist/.vite/manifest.json +# Execute enabled acquisition rules on this interval (seconds). Set a value that +# respects each indexer's API limits. +ACQUISITION_ENABLED=false + # Prefer file-based PowerSync keys for local development. # Generate them with: ./scripts/generate_dev_powersync_keys.sh POWERSYNC_JWT_PRIVATE_KEY_FILE=.local/powersync/private.pem diff --git a/alembic/versions/d4e5f6a7b8c9_add_acquisition.py b/alembic/versions/d4e5f6a7b8c9_add_acquisition.py new file mode 100644 index 0000000..88090d0 --- /dev/null +++ b/alembic/versions/d4e5f6a7b8c9_add_acquisition.py @@ -0,0 +1,87 @@ +"""add acquisition endpoints, rules and jobs + +Revision ID: d4e5f6a7b8c9 +Revises: c3f8b2a9d1e4 +""" + +from collections.abc import Sequence + +import sqlalchemy as sa +from sqlalchemy.dialects import postgresql + +from alembic import op + +revision: str = "d4e5f6a7b8c9" +down_revision: str | Sequence[str] | None = "c3f8b2a9d1e4" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + """Create private acquisition integration tables.""" + op.create_table( + "acquisition_endpoints", + sa.Column("endpoint_id", sa.Uuid(), nullable=False), + sa.Column("owner_user_id", sa.Uuid(), nullable=False), + sa.Column("name", sa.String(120), nullable=False), + sa.Column("kind", sa.String(32), nullable=False), + sa.Column("base_url", sa.String(2048), nullable=False), + sa.Column("credentials", postgresql.JSONB(), nullable=True), + sa.Column("settings", postgresql.JSONB(), nullable=True), + sa.Column("enabled", sa.Boolean(), server_default=sa.text("true"), nullable=False), + sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False), + sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False), + sa.ForeignKeyConstraint(["owner_user_id"], ["users.user_id"], ondelete="CASCADE"), + sa.PrimaryKeyConstraint("endpoint_id"), + ) + op.create_index("ix_acquisition_endpoints_owner_user_id", "acquisition_endpoints", ["owner_user_id"]) + op.create_table( + "acquisition_rules", + sa.Column("rule_id", sa.Uuid(), nullable=False), + sa.Column("owner_user_id", sa.Uuid(), nullable=False), + sa.Column("name", sa.String(120), nullable=False), + sa.Column("query", sa.String(500), nullable=False), + sa.Column("endpoint_ids", postgresql.JSONB(), nullable=True), + sa.Column("download_client_id", sa.Uuid(), nullable=True), + sa.Column("filters", postgresql.JSONB(), nullable=True), + sa.Column("enabled", sa.Boolean(), server_default=sa.text("true"), nullable=False), + sa.Column("last_run_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False), + sa.ForeignKeyConstraint( + ["download_client_id"], + ["acquisition_endpoints.endpoint_id"], + ondelete="SET NULL", + ), + sa.ForeignKeyConstraint(["owner_user_id"], ["users.user_id"], ondelete="CASCADE"), + sa.PrimaryKeyConstraint("rule_id"), + ) + op.create_index("ix_acquisition_rules_owner_user_id", "acquisition_rules", ["owner_user_id"]) + op.create_table( + "acquisition_jobs", + sa.Column("job_id", sa.Uuid(), nullable=False), + sa.Column("owner_user_id", sa.Uuid(), nullable=False), + sa.Column("endpoint_id", sa.Uuid(), nullable=True), + sa.Column("rule_id", sa.Uuid(), nullable=True), + sa.Column("title", sa.String(500), nullable=False), + sa.Column("download_url", sa.Text(), nullable=False), + sa.Column("status", sa.String(32), server_default=sa.text("'queued'"), nullable=False), + sa.Column("client_reference", sa.String(255), nullable=True), + sa.Column("error", sa.Text(), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False), + sa.ForeignKeyConstraint( + ["endpoint_id"], + ["acquisition_endpoints.endpoint_id"], + ondelete="SET NULL", + ), + sa.ForeignKeyConstraint(["owner_user_id"], ["users.user_id"], ondelete="CASCADE"), + sa.ForeignKeyConstraint(["rule_id"], ["acquisition_rules.rule_id"], ondelete="SET NULL"), + sa.PrimaryKeyConstraint("job_id"), + ) + op.create_index("ix_acquisition_jobs_owner_user_id", "acquisition_jobs", ["owner_user_id"]) + + +def downgrade() -> None: + """Drop acquisition tables.""" + op.drop_table("acquisition_jobs") + op.drop_table("acquisition_rules") + op.drop_table("acquisition_endpoints") diff --git a/docs/superpowers/plans/2026-07-17-acquisition-integration-hardening.md b/docs/superpowers/plans/2026-07-17-acquisition-integration-hardening.md new file mode 100644 index 0000000..b29ce7b --- /dev/null +++ b/docs/superpowers/plans/2026-07-17-acquisition-integration-hardening.md @@ -0,0 +1,624 @@ +# Acquisition Integration Hardening Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Make server PR #4 and client PR #18 safe to merge by adding an operator activation boundary, removing automatic execution, correcting persistence and remote-operation semantics, and completing the client management workflow. + +**Architecture:** The server remains the only component that stores credentials or contacts acquisition integrations. A disabled-by-default server capability gates every private route, protocol adapters return controlled job outcomes, and manual rules replace the unsafe worker. The Flutter client caches capability availability, consumes typed job outcomes, and owns presentation state for integration testing, search selection, and submissions. + +**Tech Stack:** FastAPI, Pydantic, async SQLAlchemy, PostgreSQL, Alembic, pytest, Flutter, Dart, Provider, go_router, package:http, flutter_test. + +--- + +## Repository map + +Server checkout: `/tmp/papyrus-server-pr4-work.aXSYmc/repo` + +- `papyrus/config.py`: operator feature setting. +- `papyrus/api/routes/acquisition.py`: HTTP dependencies and thin route wiring. +- `papyrus/services/acquisition.py`: connection testing, protocol adapters, deletion cleanup, and job orchestration. +- `papyrus/schemas/acquisition.py`: capability, connection-test, and nullable job contracts. +- `papyrus/models/acquisition.py`: nullable foreign-key model metadata. +- `alembic/versions/d4e5f6a7b8c9_add_acquisition.py`: new-table migration aligned with the models. +- `papyrus/main.py`: application lifespan without acquisition automation. +- `tests/api/routes/test_acquisition.py`: authenticated API and persistence regressions. +- `tests/services/test_acquisition.py`: protocol adapter regressions. + +Client checkout: `/tmp/papyrus-client-pr18-work.hH4BiS/repo` + +- `app/lib/acquisition/acquisition_models.dart`: capability and job response types. +- `app/lib/acquisition/acquisition_api_client.dart`: typed job and connection-test calls. +- `app/lib/providers/acquisition_availability_provider.dart`: lifecycle-owned capability cache. +- `app/lib/main.dart`: provider construction, replacement, disposal, and registration. +- `app/lib/config/app_router.dart`: synchronous preference plus capability guard. +- `app/lib/pages/profile_page.dart`: capability-aware management visibility. +- `app/lib/pages/acquisition_page.dart`: integration dialog, search selection, and per-release state. +- `app/test/acquisition/`: API, model, and availability-provider tests. +- `app/test/config/app_router_test.dart`: route guard regression. +- `app/test/pages/profile_storage_sync_test.dart`: Profile visibility regression. +- `app/test/pages/acquisition_page_test.dart`: management and action-state widget regressions. + +### Task 1: Add the server activation boundary and remove automation + +**Files:** +- Modify: `.env.example` +- Modify: `papyrus/config.py` +- Modify: `papyrus/api/routes/acquisition.py` +- Modify: `papyrus/main.py` +- Modify: `tests/api/routes/test_acquisition.py` + +- [ ] **Step 1: Write failing activation tests** + +Add tests that temporarily set the singleton application setting and restore it in `finally`: + +```python +async def test_disabled_capabilities_are_empty(client: AsyncClient, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(app_settings, "acquisition_enabled", False) + + response = await client.get("/v1/acquisition/capabilities") + + assert response.status_code == 200 + assert response.json() == { + "enabled": False, + "endpoint_kinds": [], + "indexer_kinds": [], + "download_client_kinds": [], + "arr_kinds": [], + "arr_commands": {}, + } + + +async def test_disabled_acquisition_routes_are_not_available( + client: AsyncClient, + auth_headers: dict[str, str], + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr(app_settings, "acquisition_enabled", False) + + response = await client.get("/v1/acquisition/endpoints", headers=auth_headers) + + assert response.status_code == 404 +``` + +- [ ] **Step 2: Verify the tests fail for the expected reason** + +Run: `uv run pytest tests/api/routes/test_acquisition.py -k "disabled" -q` + +Expected: capability assertion fails because `enabled` is true and endpoints remain accessible. + +- [ ] **Step 3: Implement the setting, dependency, and manual-only lifespan** + +Add the setting and route dependency: + +```python +class Settings(BaseSettings): + acquisition_enabled: bool = False + + +def require_acquisition_enabled() -> None: + if not get_settings().acquisition_enabled: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Acquisition is disabled") +``` + +Attach `Depends(require_acquisition_enabled)` to every acquisition route except `/capabilities`. Return empty capability collections when disabled. Remove `acquisition_worker`, its task lifecycle, the `run_enabled_rules` import, and `acquisition_automation_interval_seconds`. Document `ACQUISITION_ENABLED=false` in `.env.example`. + +- [ ] **Step 4: Verify activation tests pass** + +Run: `uv run pytest tests/api/routes/test_acquisition.py -k "capabilities or disabled" -q` + +Expected: all selected tests pass. + +- [ ] **Step 5: Commit the activation boundary** + +```bash +git add .env.example papyrus/config.py papyrus/api/routes/acquisition.py papyrus/main.py tests/api/routes/test_acquisition.py +git commit -m "fix: gate acquisition behind server opt-in" +``` + +### Task 2: Correct remote adapter and job outcome behavior + +**Files:** +- Modify: `papyrus/services/acquisition.py` +- Modify: `papyrus/api/routes/acquisition.py` +- Create: `tests/services/test_acquisition.py` +- Modify: `tests/api/routes/test_acquisition.py` + +- [ ] **Step 1: Write failing Transmission and Deluge tests** + +Use real adapter functions with only `_request` replaced: + +```python +async def test_transmission_rejects_rpc_failure(monkeypatch: pytest.MonkeyPatch) -> None: + async def request(*args: object, **kwargs: object) -> tuple[int, dict[str, str], bytes]: + return 200, {}, b'{"result":"invalid or corrupt torrent file","arguments":{}}' + + monkeypatch.setattr(acquisition, "_request", request) + + with pytest.raises(HTTPException) as exc_info: + await acquisition.submit_to_client(_endpoint("transmission"), "magnet:?xt=urn:btih:test", None, None) + + assert exc_info.value.status_code == 502 + + +async def test_deluge_uses_url_method_for_http_torrent(monkeypatch: pytest.MonkeyPatch) -> None: + bodies: list[dict[str, object]] = [] + + async def request(*args: object, **kwargs: object) -> tuple[int, dict[str, str], bytes]: + bodies.append(json.loads(cast(bytes, kwargs["body"]))) + if len(bodies) == 1: + return 200, {"Set-Cookie": "_session_id=test"}, b'{"result":true,"error":null,"id":1}' + return 200, {}, b'{"result":"torrent-id","error":null,"id":2}' + + monkeypatch.setattr(acquisition, "_request", request) + + await acquisition.submit_to_client(_endpoint("deluge"), "https://indexer.test/release.torrent", None, None) + + assert bodies[1]["method"] == "core.add_torrent_url" +``` + +Add a route regression asserting a rejected adapter produces a persisted `status="failed"` job with HTTP 201 and a safe `error` string. + +- [ ] **Step 2: Verify the adapter tests fail** + +Run: `uv run pytest tests/services/test_acquisition.py -q` + +Expected: Transmission does not raise and Deluge uses `core.add_torrent_magnet` for the URL. + +- [ ] **Step 3: Implement strict remote response parsing** + +Add focused helpers: + +```python +def _json_object(payload: bytes, integration: str) -> dict[str, object]: + try: + value = json.loads(payload) + except (UnicodeDecodeError, json.JSONDecodeError) as exc: + raise HTTPException(status_code=502, detail=f"{integration} returned invalid JSON") from exc + if not isinstance(value, dict): + raise HTTPException(status_code=502, detail=f"{integration} returned an invalid response") + return value + + +def _require_deluge_result(payload: bytes) -> object: + response = _json_object(payload, "Deluge") + if response.get("error") is not None or response.get("result") in {None, False}: + raise HTTPException(status_code=502, detail="Deluge rejected the request") + return response["result"] +``` + +Require Transmission `result == "success"`. Validate Deluge login and add responses. Choose the Deluge method from `download_url.startswith("magnet:")`. Use `_json_object` for Prowlarr and Arr payloads so malformed responses remain controlled failures. + +- [ ] **Step 4: Verify service and route tests pass** + +Run: `uv run pytest tests/services/test_acquisition.py tests/api/routes/test_acquisition.py -q` + +Expected: all selected tests pass. + +- [ ] **Step 5: Commit adapter corrections** + +```bash +git add papyrus/services/acquisition.py papyrus/api/routes/acquisition.py tests/services/test_acquisition.py tests/api/routes/test_acquisition.py +git commit -m "fix: validate acquisition client responses" +``` + +### Task 3: Make endpoint deletion preserve audit history + +**Files:** +- Modify: `papyrus/models/acquisition.py` +- Modify: `papyrus/schemas/acquisition.py` +- Modify: `papyrus/services/acquisition.py` +- Modify: `papyrus/api/routes/acquisition.py` +- Modify: `alembic/versions/d4e5f6a7b8c9_add_acquisition.py` +- Modify: `tests/api/routes/test_acquisition.py` + +- [ ] **Step 1: Write the failing deletion regression** + +Create an endpoint, a job referencing it, and a rule that uses it as both an indexer and download client. Delete through the API and assert: + +```python +assert response.status_code == 204 +assert persisted_job.endpoint_id is None +assert persisted_rule.download_client_id is None +assert persisted_rule.endpoint_ids == [] +assert persisted_rule.enabled is False +``` + +- [ ] **Step 2: Verify deletion fails with a foreign-key error** + +Run: `uv run pytest tests/api/routes/test_acquisition.py -k "delete_endpoint_preserves" -q` + +Expected: the request returns 500 or raises `IntegrityError` because dependent rows still reference the endpoint. + +- [ ] **Step 3: Implement nullable foreign keys and cleanup** + +Update model metadata and the unmerged migration: + +```python +download_client_id: Mapped[UUID | None] = mapped_column( + Uuid, + ForeignKey("acquisition_endpoints.endpoint_id", ondelete="SET NULL"), + nullable=True, +) +endpoint_id: Mapped[UUID | None] = mapped_column( + Uuid, + ForeignKey("acquisition_endpoints.endpoint_id", ondelete="SET NULL"), + nullable=True, +) +``` + +Make `AcquisitionJob.endpoint_id` optional in the response schema. Add a service that locks/loads the owner's rules, removes the deleted ID from JSON lists, clears matching download clients, disables unusable rules, deletes the endpoint, and commits once. Delegate the route to that service. + +- [ ] **Step 4: Verify deletion and migration metadata** + +Run: `uv run pytest tests/api/routes/test_acquisition.py -k "delete_endpoint_preserves" -q` + +Run: `uv run alembic heads` + +Expected: the regression passes and `d4e5f6a7b8c9 (head)` is the only head. + +- [ ] **Step 5: Commit persistence corrections** + +```bash +git add papyrus/models/acquisition.py papyrus/schemas/acquisition.py papyrus/services/acquisition.py papyrus/api/routes/acquisition.py alembic/versions/d4e5f6a7b8c9_add_acquisition.py tests/api/routes/test_acquisition.py +git commit -m "fix: preserve acquisition jobs on endpoint deletion" +``` + +### Task 4: Add non-persisting connection tests + +**Files:** +- Modify: `papyrus/schemas/acquisition.py` +- Modify: `papyrus/services/acquisition.py` +- Modify: `papyrus/api/routes/acquisition.py` +- Modify: `tests/api/routes/test_acquisition.py` +- Modify: `tests/services/test_acquisition.py` + +- [ ] **Step 1: Write failing API tests for unsaved and edited connections** + +Add tests that call `/v1/acquisition/endpoints/test` with an unsaved Prowlarr payload and with an owned endpoint ID plus credential overrides. Replace `test_endpoint_connection` at the route boundary and capture the transient endpoint. Assert no new `AcquisitionEndpoint` row is persisted and another user's endpoint returns 404. + +```python +assert response.status_code == 200 +assert response.json() == {"ok": True} +assert captured.kind == "prowlarr" +assert decrypt_secret_payload(captured.credentials["encrypted"])["api_key"] == "override" +assert await endpoint_count(db_session) == before_count +``` + +- [ ] **Step 2: Verify the connection-test route is missing** + +Run: `uv run pytest tests/api/routes/test_acquisition.py -k "test_connection" -q` + +Expected: requests return 404. + +- [ ] **Step 3: Implement schemas, transient configuration, and protocol checks** + +Add: + +```python +class AcquisitionEndpointTest(BaseModel): + endpoint_id: UUID | None = None + kind: EndpointKind | None = None + base_url: HttpUrl | None = None + api_key: SecretStr | None = None + username: SecretStr | None = None + password: SecretStr | None = None + + +class AcquisitionEndpointTestResult(BaseModel): + ok: bool +``` + +Validate that `kind` and `base_url` are present without `endpoint_id`. For an edit, load only an owned endpoint and merge supplied values with decrypted stored credentials in memory. Implement protocol-specific bounded checks and return `AcquisitionEndpointTestResult(ok=True)` only after the remote accepts the request. + +- [ ] **Step 4: Verify connection-test coverage** + +Run: `uv run pytest tests/api/routes/test_acquisition.py tests/services/test_acquisition.py -q` + +Expected: all selected tests pass. + +- [ ] **Step 5: Commit connection testing** + +```bash +git add papyrus/schemas/acquisition.py papyrus/services/acquisition.py papyrus/api/routes/acquisition.py tests/api/routes/test_acquisition.py tests/services/test_acquisition.py +git commit -m "feat: test acquisition connections without saving" +``` + +### Task 5: Verify the complete server branch + +**Files:** +- Review all server files changed by Tasks 1-4. + +- [ ] **Step 1: Run formatting and lint checks** + +Run: `uv run ruff format --check .` + +Run: `uv run ruff check .` + +Expected: both commands exit 0. + +- [ ] **Step 2: Run type checking** + +Run: `uv run mypy .` + +Expected: exit 0 with no type errors. + +- [ ] **Step 3: Run the full server test suite** + +Run: `uv run pytest` + +Expected: all tests pass. + +- [ ] **Step 4: Verify migration shape** + +Run: `uv run alembic heads` + +Run against a disposable test database: `uv run alembic upgrade head` + +Expected: one head and a successful upgrade creating nullable `SET NULL` acquisition foreign keys. + +### Task 6: Add typed client capability, job, and connection-test contracts + +**Files:** +- Modify: `app/lib/acquisition/acquisition_models.dart` +- Modify: `app/lib/acquisition/acquisition_api_client.dart` +- Modify: `app/test/acquisition/acquisition_models_test.dart` +- Modify: `app/test/acquisition/acquisition_api_client_test.dart` + +- [ ] **Step 1: Write failing model and API tests** + +Add assertions for `enabled`, nullable job endpoint IDs, failed jobs, and connection-test payloads: + +```dart +expect(AcquisitionCapabilities.fromJson({'enabled': false}).enabled, isFalse); + +final job = AcquisitionJob.fromJson({ + 'job_id': 'job-1', + 'endpoint_id': null, + 'rule_id': null, + 'title': 'Release', + 'download_url': 'magnet:?xt=urn:btih:test', + 'status': 'failed', + 'error': 'Transmission rejected the release', +}); +expect(job.isSubmitted, isFalse); +expect(job.error, 'Transmission rejected the release'); +``` + +Expect `submitRelease` and `runArrCommand` to return `AcquisitionJob`. Expect `testEndpoint` to POST `/v1/acquisition/endpoints/test` with either unsaved values or `endpoint_id` and overrides. + +- [ ] **Step 2: Verify the new contract tests fail** + +Run: `flutter test --no-pub test/acquisition/acquisition_models_test.dart test/acquisition/acquisition_api_client_test.dart` + +Expected: missing `enabled`, `AcquisitionJob`, and `testEndpoint` APIs fail compilation. + +- [ ] **Step 3: Implement typed contracts** + +Add immutable model fields and parsing: + +```dart +class AcquisitionJob { + final String id; + final String? endpointId; + final String status; + final String? error; + + bool get isSubmitted => status == 'submitted'; +} +``` + +Return parsed jobs from submission methods and add the connection-test call. Keep all authentication errors represented by `AuthApiException`. + +- [ ] **Step 4: Verify model and API tests pass** + +Run: `flutter test --no-pub test/acquisition/acquisition_models_test.dart test/acquisition/acquisition_api_client_test.dart` + +Expected: all selected tests pass. + +- [ ] **Step 5: Commit client contracts** + +```bash +git add app/lib/acquisition/acquisition_models.dart app/lib/acquisition/acquisition_api_client.dart app/test/acquisition/acquisition_models_test.dart app/test/acquisition/acquisition_api_client_test.dart +git commit -m "fix: consume acquisition capability and job outcomes" +``` + +### Task 7: Add lifecycle-owned capability availability and route gating + +**Files:** +- Create: `app/lib/providers/acquisition_availability_provider.dart` +- Modify: `app/lib/main.dart` +- Modify: `app/lib/config/app_router.dart` +- Modify: `app/lib/pages/profile_page.dart` +- Create: `app/test/acquisition/acquisition_availability_provider_test.dart` +- Modify: `app/test/config/app_router_test.dart` +- Modify: `app/test/pages/profile_storage_sync_test.dart` + +- [ ] **Step 1: Write failing provider, router, and Profile tests** + +Define provider tests with an injected capability loader: + +```dart +final provider = AcquisitionAvailabilityProvider( + loadCapabilities: (_) async => const AcquisitionCapabilities( + enabled: true, + endpointKinds: [], + indexerKinds: [], + downloadClientKinds: [], + arrKinds: [], + arrCommands: {}, + ), +); +await provider.refresh(serverBaseUri); +expect(provider.isAvailableFor(serverBaseUri), isTrue); +``` + +Extend router tests so `/acquisition` redirects when the preference is on but availability is false, then permits the route after availability becomes true. Extend Profile tests so the toggle remains visible while `Torrent & automation` appears only for enabled preference plus available server. + +- [ ] **Step 2: Verify availability tests fail** + +Run: `flutter test --no-pub test/acquisition/acquisition_availability_provider_test.dart test/config/app_router_test.dart test/pages/profile_storage_sync_test.dart` + +Expected: the provider type is missing and router/Profile only check local preferences. + +- [ ] **Step 3: Implement the availability provider and wiring** + +Create a `ChangeNotifier` keyed by active server URI with `unknown`, `loading`, `available`, and `unavailable` state. It owns one `AcquisitionApiClient`, closes it on server change/disposal, and loads through `AuthProvider.withFreshAccessToken`. + +Construct it in `_PapyrusState`, pass it to `AppRouter`, register it in `MultiProvider`, refresh it after auth/server changes, and dispose it. Include it in `Listenable.merge`. Guard with: + +```dart +if (location == '/acquisition' && + (!preferencesProvider.acquisitionEnabled || + !acquisitionAvailabilityProvider.isAvailableFor(activeServerUri))) { + return '/profile'; +} +``` + +Profile watches the provider and gates only the management row/button, not the opt-in toggle. + +- [ ] **Step 4: Verify provider, router, and Profile tests pass** + +Run: `flutter test --no-pub test/acquisition/acquisition_availability_provider_test.dart test/config/app_router_test.dart test/pages/profile_storage_sync_test.dart` + +Expected: all selected tests pass. + +- [ ] **Step 5: Commit availability gating** + +```bash +git add app/lib/providers/acquisition_availability_provider.dart app/lib/main.dart app/lib/config/app_router.dart app/lib/pages/profile_page.dart app/test/acquisition/acquisition_availability_provider_test.dart app/test/config/app_router_test.dart app/test/pages/profile_storage_sync_test.dart +git commit -m "fix: gate acquisition UI by server capability" +``` + +### Task 8: Complete integration testing and form state + +**Files:** +- Modify: `app/lib/pages/acquisition_page.dart` +- Create: `app/test/pages/acquisition_page_test.dart` + +- [ ] **Step 1: Write failing dialog widget tests** + +Pump `AcquisitionPage` with injected/fake API behavior and assert: + +- Prowlarr shows API key but not username/password. +- qBittorrent shows username/password but not API key. +- Deluge shows password only. +- Test connection displays progress, prevents a second request, and renders a returned error inside the dialog. +- Save is disabled while a test/save request is active. + +- [ ] **Step 2: Verify dialog tests fail** + +Run: `flutter test --no-pub test/pages/acquisition_page_test.dart --plain-name "integration dialog"` + +Expected: all credential fields are currently unconditional and there is no connection-test action. + +- [ ] **Step 3: Implement conditional fields and inline operation state** + +Extract a focused stateful `_EndpointDialog` widget with injected callbacks. Derive fields from `AcquisitionEndpointKind` and maintain distinct `testing`, `saving`, and `error` state. Send current form values to `testEndpoint`; for edits include `endpointId` so blank credentials preserve stored secrets. + +- [ ] **Step 4: Verify dialog tests pass** + +Run: `flutter test --no-pub test/pages/acquisition_page_test.dart --plain-name "integration dialog"` + +Expected: all selected tests pass. + +- [ ] **Step 5: Commit integration form completion** + +```bash +git add app/lib/pages/acquisition_page.dart app/test/pages/acquisition_page_test.dart +git commit -m "feat: test acquisition integrations from the client" +``` + +### Task 9: Complete search selection and per-release outcomes + +**Files:** +- Modify: `app/lib/pages/acquisition_page.dart` +- Modify: `app/test/pages/acquisition_page_test.dart` + +- [ ] **Step 1: Write failing search and submission widget tests** + +Assert that search remains disabled without both a selected enabled indexer and an enabled download client. Select one of two indexer chips and verify only its ID is sent. Start one submission and assert only that release/client pair is disabled. Complete with a failed job and assert the backend error is displayed instead of success. + +- [ ] **Step 2: Verify state tests fail** + +Run: `flutter test --no-pub test/pages/acquisition_page_test.dart --plain-name "search and submission"` + +Expected: no indexer controls exist, search ignores client availability, and the page uses one global submission flag. + +- [ ] **Step 3: Implement selected indexers and pair-scoped submissions** + +Maintain `Set _selectedIndexerIds` and initialize newly loaded enabled indexers only when no explicit selection exists. Render `FilterChip` controls. Compute `canSearch` from selected enabled indexers plus enabled clients. + +Maintain `Set _submittingKeys` where the key combines release URL and client ID. Add before awaiting, remove in `finally`, and pass disabled state to each release tile. Inspect the returned job and show success only for `submitted`; otherwise show `job.error ?? 'Submission failed.'`. Apply the same outcome check to Arr commands. + +- [ ] **Step 4: Verify state tests pass** + +Run: `flutter test --no-pub test/pages/acquisition_page_test.dart` + +Expected: all acquisition page tests pass. + +- [ ] **Step 5: Commit search and submission completion** + +```bash +git add app/lib/pages/acquisition_page.dart app/test/pages/acquisition_page_test.dart +git commit -m "fix: complete acquisition search and submission states" +``` + +### Task 10: Verify the complete client branch + +**Files:** +- Review all client files changed by Tasks 6-9. + +- [ ] **Step 1: Format and verify formatting** + +Run: `dart format app/lib app/test` + +Run: `dart format --output=none --set-exit-if-changed app/lib app/test` + +Expected: the second command exits 0 with zero changed files. + +- [ ] **Step 2: Run static analysis** + +Run from `app/`: `flutter analyze --no-pub` + +Expected: exit 0 with no issues. + +- [ ] **Step 3: Run the full Flutter test suite** + +Run from `app/`: `flutter test --no-pub` + +Expected: all tests pass; only explicitly skipped integration tests remain skipped. + +### Task 11: Final cross-repository verification and publication + +**Files:** +- Review both repository diffs and commit histories. + +- [ ] **Step 1: Confirm clean worktrees and intended commits** + +Run in each checkout: `git status --short --branch` + +Run in each checkout: `git log --oneline --decorate -8` + +Expected: clean worktrees with focused commits on top of the original PR heads. + +- [ ] **Step 2: Re-run the narrow cross-contract tests** + +Server: `uv run pytest tests/api/routes/test_acquisition.py tests/services/test_acquisition.py -q` + +Client from `app/`: `flutter test --no-pub test/acquisition test/config/app_router_test.dart test/pages/acquisition_page_test.dart test/pages/profile_storage_sync_test.dart` + +Expected: both commands exit 0. + +- [ ] **Step 3: Push to the existing PR source branches** + +After confirming the source branch names and remote permissions: + +```bash +git push github HEAD:feature/torrent-acquisition +``` + +Run once from each repository checkout. Do not force-push. + +- [ ] **Step 4: Verify GitHub heads and checks** + +Confirm server PR #4 and client PR #18 point at the new commits. Report CI as pending until GitHub Actions completes; do not claim remote success from local checks alone. diff --git a/docs/superpowers/specs/2026-07-17-acquisition-integration-hardening-design.md b/docs/superpowers/specs/2026-07-17-acquisition-integration-hardening-design.md new file mode 100644 index 0000000..b15f8e2 --- /dev/null +++ b/docs/superpowers/specs/2026-07-17-acquisition-integration-hardening-design.md @@ -0,0 +1,143 @@ +# Acquisition Integration Hardening Technical Specification + +- Status: Approved for implementation +- Date: 2026-07-17 +- Audience: Papyrus server and Flutter client maintainers +- Scope: `PapyrusReader/server#4` and `PapyrusReader/client#18` + +## Overview + +This specification defines the changes required to make the server-mediated torrent acquisition feature safe to enable, predictable across the HTTP boundary, and complete enough for client integration. It records the accepted decisions to disable automatic rule execution and require an explicit server-operator opt-in. + +## Problem + +The current pull requests expose useful integration primitives, but they do not yet provide a safe release boundary. Remote submission failures are returned as successful HTTP requests and ignored by the client, endpoint deletion conflicts with audit-history foreign keys, every API process starts an unsafe repeating worker, and arbitrary outbound integration requests are enabled on every server. The client also lacks capability-aware navigation and several required management states. + +## Goals + +- Require a server operator to enable acquisition explicitly. +- Keep manual endpoint, search, submission, Arr-command, and rule execution workflows. +- Remove automatic background rule execution from this release. +- Preserve audit jobs while allowing integrations to be deleted safely. +- Make failed remote operations visible to the Flutter client. +- Support non-persisting connection tests for new and edited integrations. +- Complete capability gating, integration forms, indexer selection, and per-release submission state. +- Add regression coverage for each corrected behavior. + +## Non-goals + +- Scheduling or automatically claiming acquisition rules. +- Cross-replica worker leadership or distributed job queues. +- Usenet or Newznab support. +- A general-purpose outbound proxy or administrator-managed hostname allowlist. +- Redesigning unrelated authentication, profile, or synchronization code. + +## Constraints + +- Private and loopback integration URLs remain valid after an operator enables acquisition because LAN services are the primary use case. +- Credentials stay encrypted at rest and are never returned by API schemas. +- Existing PR dependencies and repository architecture remain unchanged. +- Route handlers remain thin; protocol and transaction-aware behavior belongs in `papyrus/services`. +- The Flutter feature remains disabled by default in local preferences. + +## Proposed design + +### Server activation boundary + +Add `acquisition_enabled: bool = False` to server settings and document `ACQUISITION_ENABLED=false` in `.env.example`. `GET /v1/acquisition/capabilities` remains available and returns `enabled=false` with empty supported-kind and command collections when disabled. Every other acquisition route uses a shared dependency that returns HTTP 404 while the feature is disabled. + +Enabling this setting is the operator's authorization for authenticated users to connect the server to private/LAN HTTP services. Deployments that do not trust their users must leave the setting disabled or enforce outbound restrictions outside the application. + +### Manual rules only + +Remove the acquisition worker from the FastAPI lifespan and remove the automation interval setting. Retain rule creation, listing, and `POST /rules/{rule_id}/run` for explicit execution. No code in this release executes rules on a timer. + +### Job outcome contract + +Submission and Arr-command endpoints continue returning HTTP 201 with an `AcquisitionJob` audit record. The domain outcome is the job's `status`: + +- `submitted`: the remote system accepted the operation. +- `failed`: the remote adapter rejected or could not complete the operation; `error` contains a user-safe message. +- `queued`: reserved for a future asynchronous execution model. + +The Flutter API client parses the job response. UI success feedback is shown only for `submitted`; `failed` becomes an operation error using the returned message. + +Transmission adapters validate the RPC `result` field. Deluge adapters validate JSON-RPC errors/results and choose `core.add_torrent_magnet` for magnets or `core.add_torrent_url` for HTTP(S) torrent URLs. Malformed remote payloads become controlled HTTP 502 errors instead of uncaught exceptions. + +### Safe endpoint deletion + +Audit jobs remain after integration deletion. Change `acquisition_jobs.endpoint_id` and `acquisition_rules.download_client_id` to nullable foreign keys with `ON DELETE SET NULL`. Before deleting an endpoint, remove its identifier from the owner's rule `endpoint_ids`; disable any rule that loses its download client or all configured indexers. The endpoint credentials are deleted with the endpoint. + +The unmerged Alembic revision is updated in place so model metadata and the migration remain aligned. Downgrade continues to drop the three acquisition tables. + +### Connection testing + +Add authenticated `POST /v1/acquisition/endpoints/test`. The request accepts either: + +- complete unsaved endpoint values (`kind`, `base_url`, and relevant credentials), or +- an owned `endpoint_id` plus optional URL or credential overrides for an edit flow. + +The server builds an in-memory endpoint configuration, merges stored credentials only for an owned endpoint, performs a bounded protocol-specific health/authentication request, and returns `{ "ok": true }`. Failures use controlled 4xx/502 responses and never persist the supplied credentials. + +Protocol checks use the smallest supported request: system status/capabilities for indexers and Arr apps, authentication or session inspection for download clients. + +### Client availability and navigation + +Add a lifecycle-owned acquisition availability provider that caches capability state for the active server. It closes/replaces its HTTP client when the server changes and refreshes through the existing access-token refresh path. + +The Profile toggle remains visible and disabled by default. The management row appears only when the preference is enabled and the active server reports `enabled=true`. The router redirects `/acquisition` to Profile unless both conditions are true. Unknown/loading capability state is treated as unavailable. + +### Client management and action states + +The endpoint dialog: + +- shows API key fields for Prowlarr, Torznab, and Arr applications; +- shows username/password for qBittorrent and Transmission; +- shows password for Deluge; +- validates HTTP(S) URLs without embedded credentials; +- supports a non-persisting connection test; +- shows inline test/save progress and error text; +- prevents duplicate test or save requests. + +Search exposes enabled indexers as selectable controls. Search is enabled only when at least one indexer is selected and at least one enabled download client exists. A new search clears stale results and errors. + +Submission state is tracked by release and target client. An in-flight pair cannot be submitted twice, while unrelated releases remain actionable. Returned failed jobs display their server-provided error. Arr commands use the same job-outcome handling. + +## Interfaces and dependencies + +- Server configuration: `ACQUISITION_ENABLED`. +- Existing authenticated acquisition routes retain their paths. +- New route: `POST /v1/acquisition/endpoints/test`. +- `AcquisitionCapabilities.enabled` becomes authoritative in the Flutter model. +- `AcquisitionJob` becomes a typed Flutter response model. +- No new third-party dependencies are introduced. + +## Testing strategy + +Server tests cover disabled capabilities/routes, connection-test ownership and non-persistence, failed job responses, Transmission/Deluge protocol errors, magnet versus URL dispatch, endpoint deletion with jobs/rules, and manual-only application lifespan. Migration metadata and a single Alembic head are verified alongside Ruff, Mypy, and Pytest. + +Client tests cover capability parsing, availability refresh and server switching, route/profile gating, conditional credential fields, connection-test serialization and error state, indexer selection/search prerequisites, per-release duplicate prevention, and submitted/failed job feedback. Dart formatting, Flutter analysis, and the full Flutter test suite must pass. + +## Risks and mitigations + +- Private-network access is intentionally powerful. It is mitigated by a disabled-by-default operator setting and documented trust requirement. +- Making endpoint foreign keys nullable changes response schemas. The server schema and any client job model must accept missing endpoint IDs. +- Capability checks can race server changes. Availability state is keyed to the active server URI and reset before each refresh. +- Protocol health endpoints vary by version. Tests cover request shape, and failures are surfaced rather than treated as success. + +## Rollout and rollback + +1. Merge and deploy the server PR with the migration. +2. Leave acquisition disabled until the operator explicitly sets `ACQUISITION_ENABLED=true`. +3. Verify capabilities, connection testing, manual search, submission, and Arr commands against supported services. +4. Merge and release the client PR after server verification. + +Rollback the client independently if UI behavior regresses. Roll back the server only before users depend on stored acquisition configuration; the downgrade deletes acquisition endpoints, encrypted credentials, rules, and job history. + +## Accepted decisions + +- Automatic background rule execution is excluded from this release. +- Acquisition is disabled by default and enabled by server operators. +- Enabled servers permit private/LAN integration URLs. +- Audit jobs survive endpoint deletion through nullable foreign keys. +- HTTP 201 plus typed job status is the submission outcome contract. diff --git a/papyrus/api/routes/__init__.py b/papyrus/api/routes/__init__.py index 7835593..19772d8 100644 --- a/papyrus/api/routes/__init__.py +++ b/papyrus/api/routes/__init__.py @@ -3,6 +3,7 @@ from fastapi import APIRouter, FastAPI from papyrus.api.routes import ( + acquisition, annotations, auth, bookmarks, @@ -26,6 +27,7 @@ api_router = APIRouter() +api_router.include_router(acquisition.router, prefix="/acquisition", tags=["Acquisition"]) api_router.include_router(auth.router, prefix="/auth", tags=["Auth"]) api_router.include_router(users.router, prefix="/users", tags=["Users"]) api_router.include_router(books.router, prefix="/books", tags=["Books"]) diff --git a/papyrus/api/routes/acquisition.py b/papyrus/api/routes/acquisition.py new file mode 100644 index 0000000..fb61434 --- /dev/null +++ b/papyrus/api/routes/acquisition.py @@ -0,0 +1,291 @@ +"""Private torrent and indexer acquisition endpoints.""" + +from typing import Annotated +from uuid import UUID + +from fastapi import APIRouter, Depends, HTTPException, status +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from papyrus.api.deps import CurrentUserId +from papyrus.config import get_settings +from papyrus.core.database import get_db +from papyrus.core.security import decrypt_secret_payload, encrypt_secret_payload +from papyrus.models.acquisition import AcquisitionEndpoint as AcquisitionEndpointModel +from papyrus.models.acquisition import AcquisitionJob as AcquisitionJobModel +from papyrus.models.acquisition import AcquisitionRule as AcquisitionRuleModel +from papyrus.schemas.acquisition import ( + AcquisitionCapabilities, + AcquisitionEndpoint, + AcquisitionEndpointCreate, + AcquisitionEndpointTest, + AcquisitionEndpointTestResult, + AcquisitionEndpointUpdate, + AcquisitionJob, + AcquisitionRule, + AcquisitionRuleCreate, + ArrCommandRequest, + Release, + SearchRequest, + SubmitRequest, +) +from papyrus.services.acquisition import ( + build_test_endpoint, + delete_acquisition_endpoint, + dispatch_arr_command, + owned_endpoint, + run_rule, + search_endpoint, + submit_to_client, + test_endpoint_connection, +) + +router = APIRouter() +DbSession = Annotated[AsyncSession, Depends(get_db)] + + +def require_acquisition_enabled() -> None: + if not get_settings().acquisition_enabled: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Not found") + + +protected_router = APIRouter(dependencies=[Depends(require_acquisition_enabled)]) + + +def _credentials(api_key: str | None, username: str | None, password: str | None) -> dict[str, str] | None: + values = { + key: value for key, value in {"api_key": api_key, "username": username, "password": password}.items() if value + } + if not values: + return None + return {"encrypted": encrypt_secret_payload(values)} + + +def _merge_credentials(endpoint: AcquisitionEndpointModel, replacement: dict[str, str] | None) -> None: + if replacement is None: + return + current = endpoint.credentials or {} + if encrypted := current.get("encrypted"): + current = decrypt_secret_payload(encrypted) + endpoint.credentials = { + "encrypted": encrypt_secret_payload({**current, **decrypt_secret_payload(replacement["encrypted"])}) + } + + +@router.get("/capabilities", response_model=AcquisitionCapabilities) +async def acquisition_capabilities() -> AcquisitionCapabilities: + if not get_settings().acquisition_enabled: + return AcquisitionCapabilities( + enabled=False, + endpoint_kinds=[], + indexer_kinds=[], + download_client_kinds=[], + arr_kinds=[], + arr_commands={}, + ) + + return AcquisitionCapabilities( + endpoint_kinds=[ + "qbittorrent", + "transmission", + "deluge", + "prowlarr", + "torznab", + "readarr", + "sonarr", + "radarr", + "lidarr", + "whisparr", + ], + indexer_kinds=["prowlarr", "torznab"], + download_client_kinds=["qbittorrent", "transmission", "deluge"], + arr_kinds=["readarr", "sonarr", "radarr", "lidarr", "whisparr"], + arr_commands={ + "readarr": ["AuthorSearch", "BookSearch"], + "sonarr": ["SeriesSearch", "EpisodeSearch", "MissingEpisodeSearch"], + "radarr": ["MoviesSearch", "MissingMoviesSearch"], + "lidarr": ["ArtistSearch", "AlbumSearch", "MissingAlbumSearch"], + "whisparr": ["SeriesSearch", "EpisodeSearch", "MissingEpisodeSearch"], + }, + ) + + +@protected_router.get("/endpoints", response_model=list[AcquisitionEndpoint]) +async def list_endpoints(user_id: CurrentUserId, db: DbSession) -> list[AcquisitionEndpointModel]: + result = await db.execute( + select(AcquisitionEndpointModel) + .where(AcquisitionEndpointModel.owner_user_id == user_id) + .order_by(AcquisitionEndpointModel.name) + ) + return list(result.scalars()) + + +@protected_router.post("/endpoints", response_model=AcquisitionEndpoint, status_code=status.HTTP_201_CREATED) +async def create_endpoint( + user_id: CurrentUserId, request: AcquisitionEndpointCreate, db: DbSession +) -> AcquisitionEndpointModel: + endpoint = AcquisitionEndpointModel( + owner_user_id=user_id, + name=request.name, + kind=request.kind.value, + base_url=str(request.base_url), + credentials=_credentials( + request.api_key.get_secret_value() if request.api_key else None, + request.username.get_secret_value() if request.username else None, + request.password.get_secret_value() if request.password else None, + ), + settings=request.settings, + ) + db.add(endpoint) + await db.commit() + await db.refresh(endpoint) + return endpoint + + +@protected_router.patch("/endpoints/{endpoint_id}", response_model=AcquisitionEndpoint) +async def update_endpoint( + user_id: CurrentUserId, endpoint_id: UUID, request: AcquisitionEndpointUpdate, db: DbSession +) -> AcquisitionEndpointModel: + endpoint = await owned_endpoint(db, user_id, endpoint_id) + updates = request.model_dump(exclude_unset=True, exclude={"api_key", "username", "password"}) + for field, value in updates.items(): + setattr(endpoint, field, str(value) if field == "base_url" else value) + replacement = _credentials( + request.api_key.get_secret_value() if request.api_key else None, + request.username.get_secret_value() if request.username else None, + request.password.get_secret_value() if request.password else None, + ) + _merge_credentials(endpoint, replacement) + await db.commit() + await db.refresh(endpoint) + return endpoint + + +@protected_router.delete("/endpoints/{endpoint_id}", status_code=status.HTTP_204_NO_CONTENT) +async def delete_endpoint(user_id: CurrentUserId, endpoint_id: UUID, db: DbSession) -> None: + await delete_acquisition_endpoint(db, user_id, endpoint_id) + + +@protected_router.post("/endpoints/test", response_model=AcquisitionEndpointTestResult) +async def test_connection( + user_id: CurrentUserId, + request: AcquisitionEndpointTest, + db: DbSession, +) -> AcquisitionEndpointTestResult: + endpoint = await build_test_endpoint(db, user_id, request) + + await test_endpoint_connection(endpoint) + + return AcquisitionEndpointTestResult(ok=True) + + +@protected_router.post("/search", response_model=list[Release]) +async def search_releases(user_id: CurrentUserId, request: SearchRequest, db: DbSession) -> list[Release]: + statement = select(AcquisitionEndpointModel).where( + AcquisitionEndpointModel.owner_user_id == user_id, + AcquisitionEndpointModel.enabled.is_(True), + AcquisitionEndpointModel.kind.in_(("prowlarr", "torznab")), + ) + if request.endpoint_ids: + statement = statement.where(AcquisitionEndpointModel.endpoint_id.in_(request.endpoint_ids)) + result = await db.execute(statement) + releases: list[Release] = [] + for endpoint in result.scalars(): + releases.extend(await search_endpoint(endpoint, request.query)) + return releases + + +@protected_router.post("/submissions", response_model=AcquisitionJob, status_code=status.HTTP_201_CREATED) +async def submit_release(user_id: CurrentUserId, request: SubmitRequest, db: DbSession) -> AcquisitionJobModel: + endpoint = await owned_endpoint(db, user_id, request.endpoint_id) + job = AcquisitionJobModel( + owner_user_id=user_id, endpoint_id=endpoint.endpoint_id, title=request.title, download_url=request.download_url + ) + db.add(job) + try: + job.client_reference = await submit_to_client( + endpoint, request.download_url, request.category, request.save_path + ) + job.status = "submitted" + except HTTPException as exc: + job.status = "failed" + job.error = str(exc.detail) + await db.commit() + await db.refresh(job) + return job + + +@protected_router.post( + "/arr/{endpoint_id}/commands", response_model=AcquisitionJob, status_code=status.HTTP_201_CREATED +) +async def run_arr_command( + user_id: CurrentUserId, endpoint_id: UUID, request: ArrCommandRequest, db: DbSession +) -> AcquisitionJobModel: + """Delegate a managed search/acquisition to Readarr or another Arr app.""" + endpoint = await owned_endpoint(db, user_id, endpoint_id) + job = AcquisitionJobModel( + owner_user_id=user_id, + endpoint_id=endpoint.endpoint_id, + title=request.command, + download_url=f"arr-command:{request.command}", + ) + db.add(job) + try: + job.client_reference = await dispatch_arr_command(endpoint, request.command, request.ids) + job.status = "submitted" + except HTTPException as exc: + job.status = "failed" + job.error = str(exc.detail) + await db.commit() + await db.refresh(job) + return job + + +@protected_router.get("/jobs", response_model=list[AcquisitionJob]) +async def list_jobs(user_id: CurrentUserId, db: DbSession) -> list[AcquisitionJobModel]: + result = await db.execute( + select(AcquisitionJobModel) + .where(AcquisitionJobModel.owner_user_id == user_id) + .order_by(AcquisitionJobModel.created_at.desc()) + ) + return list(result.scalars()) + + +@protected_router.get("/rules", response_model=list[AcquisitionRule]) +async def list_rules(user_id: CurrentUserId, db: DbSession) -> list[AcquisitionRuleModel]: + result = await db.execute(select(AcquisitionRuleModel).where(AcquisitionRuleModel.owner_user_id == user_id)) + return list(result.scalars()) + + +@protected_router.post("/rules", response_model=AcquisitionRule, status_code=status.HTTP_201_CREATED) +async def create_rule(user_id: CurrentUserId, request: AcquisitionRuleCreate, db: DbSession) -> AcquisitionRuleModel: + await owned_endpoint(db, user_id, request.download_client_id) + rule = AcquisitionRuleModel( + owner_user_id=user_id, + name=request.name, + query=request.query, + endpoint_ids=[str(value) for value in request.endpoint_ids] if request.endpoint_ids else None, + download_client_id=request.download_client_id, + filters=request.filters, + enabled=request.enabled, + ) + db.add(rule) + await db.commit() + await db.refresh(rule) + return rule + + +@protected_router.post("/rules/{rule_id}/run", response_model=list[AcquisitionJob]) +async def run_acquisition_rule(user_id: CurrentUserId, rule_id: UUID, db: DbSession) -> list[AcquisitionJobModel]: + result = await db.execute( + select(AcquisitionRuleModel).where( + AcquisitionRuleModel.rule_id == rule_id, AcquisitionRuleModel.owner_user_id == user_id + ) + ) + rule = result.scalar_one_or_none() + if rule is None: + raise HTTPException(status_code=404, detail="Acquisition rule not found") + return await run_rule(db, rule) + + +router.include_router(protected_router) diff --git a/papyrus/config.py b/papyrus/config.py index 4208038..2e58f92 100644 --- a/papyrus/config.py +++ b/papyrus/config.py @@ -71,6 +71,7 @@ class Settings(BaseSettings): dev_pages_use_vite: bool = False dev_pages_vite_url: str = "http://localhost:5173" dev_pages_manifest_path: str = "frontend/dev-pages/dist/.vite/manifest.json" + acquisition_enabled: bool = False @field_validator("debug", mode="before") @classmethod diff --git a/papyrus/core/security.py b/papyrus/core/security.py index 8334630..b35cd48 100644 --- a/papyrus/core/security.py +++ b/papyrus/core/security.py @@ -1,13 +1,17 @@ """Authentication and token security helpers.""" +from base64 import urlsafe_b64encode from datetime import UTC, datetime, timedelta from functools import lru_cache from hashlib import sha256 +from json import dumps as json_dumps +from json import loads as json_loads from pathlib import Path from secrets import token_urlsafe from typing import Any import jwt +from cryptography.fernet import Fernet, InvalidToken from cryptography.hazmat.primitives import serialization from jwt.algorithms import RSAAlgorithm from pwdlib import PasswordHash @@ -125,6 +129,29 @@ def generate_opaque_token() -> str: return token_urlsafe(48) +@lru_cache +def _get_credentials_cipher() -> Fernet: + key = urlsafe_b64encode(sha256(get_settings().secret_key.encode("utf-8")).digest()) + return Fernet(key) + + +def encrypt_secret_payload(payload: dict[str, str]) -> str: + return _get_credentials_cipher().encrypt(json_dumps(payload).encode("utf-8")).decode("utf-8") + + +def decrypt_secret_payload(token: str) -> dict[str, str]: + try: + decoded = _get_credentials_cipher().decrypt(token.encode("utf-8")).decode("utf-8") + except InvalidToken as exc: + raise ValueError("Encrypted credential payload is invalid") from exc + payload = json_loads(decoded) + if not isinstance(payload, dict) or not all( + isinstance(key, str) and isinstance(value, str) for key, value in payload.items() + ): + raise ValueError("Encrypted credential payload has an invalid shape") + return payload + + def create_access_token(data: dict[str, Any], expires_delta: timedelta | None = None) -> str: settings = get_settings() ttl = expires_delta or timedelta(minutes=settings.access_token_expire_minutes) diff --git a/papyrus/main.py b/papyrus/main.py index f3db84d..1f3d84d 100644 --- a/papyrus/main.py +++ b/papyrus/main.py @@ -40,9 +40,7 @@ def configure_logging() -> None: @asynccontextmanager async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]: """Application lifespan events.""" - # Startup yield - # Shutdown def create_app() -> FastAPI: diff --git a/papyrus/models/__init__.py b/papyrus/models/__init__.py index be17f15..a94957f 100644 --- a/papyrus/models/__init__.py +++ b/papyrus/models/__init__.py @@ -1,4 +1,5 @@ from papyrus.core.database import Base +from papyrus.models.acquisition import AcquisitionEndpoint, AcquisitionJob, AcquisitionRule from papyrus.models.auth import AuthExchangeCode, AuthSession, EmailActionToken, PasswordCredential, UserIdentity from papyrus.models.media import MediaAsset from papyrus.models.powersync_demo import PowerSyncDemoItem @@ -6,6 +7,9 @@ from papyrus.models.user import User __all__ = [ + "AcquisitionEndpoint", + "AcquisitionJob", + "AcquisitionRule", "AuthExchangeCode", "AuthSession", "Base", diff --git a/papyrus/models/acquisition.py b/papyrus/models/acquisition.py new file mode 100644 index 0000000..49e397a --- /dev/null +++ b/papyrus/models/acquisition.py @@ -0,0 +1,71 @@ +"""Acquisition sources, download clients and automatic acquisition rules.""" + +from __future__ import annotations + +from datetime import datetime +from uuid import UUID, uuid4 + +from sqlalchemy import Boolean, DateTime, ForeignKey, String, Text, Uuid, func +from sqlalchemy.dialects.postgresql import JSONB +from sqlalchemy.orm import Mapped, mapped_column + +from papyrus.core.database import Base + + +class AcquisitionEndpoint(Base): + """A private indexer, Prowlarr instance, or download client connection.""" + + __tablename__ = "acquisition_endpoints" + + endpoint_id: Mapped[UUID] = mapped_column(Uuid, primary_key=True, default=uuid4) + owner_user_id: Mapped[UUID] = mapped_column(ForeignKey("users.user_id", ondelete="CASCADE"), index=True) + name: Mapped[str] = mapped_column(String(120), nullable=False) + kind: Mapped[str] = mapped_column(String(32), nullable=False) + base_url: Mapped[str] = mapped_column(String(2048), nullable=False) + credentials: Mapped[dict[str, str] | None] = mapped_column(JSONB, nullable=True) + settings: Mapped[dict[str, object] | None] = mapped_column(JSONB, nullable=True) + enabled: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True, server_default="true") + created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False, server_default=func.now()) + updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False, server_default=func.now()) + + +class AcquisitionRule(Base): + """A saved search that can automatically submit matching releases.""" + + __tablename__ = "acquisition_rules" + + rule_id: Mapped[UUID] = mapped_column(Uuid, primary_key=True, default=uuid4) + owner_user_id: Mapped[UUID] = mapped_column(ForeignKey("users.user_id", ondelete="CASCADE"), index=True) + name: Mapped[str] = mapped_column(String(120), nullable=False) + query: Mapped[str] = mapped_column(String(500), nullable=False) + endpoint_ids: Mapped[list[str] | None] = mapped_column(JSONB, nullable=True) + download_client_id: Mapped[UUID | None] = mapped_column( + Uuid, + ForeignKey("acquisition_endpoints.endpoint_id", ondelete="SET NULL"), + nullable=True, + ) + filters: Mapped[dict[str, object] | None] = mapped_column(JSONB, nullable=True) + enabled: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True, server_default="true") + last_run_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False, server_default=func.now()) + + +class AcquisitionJob(Base): + """An auditable submission to a download client.""" + + __tablename__ = "acquisition_jobs" + + job_id: Mapped[UUID] = mapped_column(Uuid, primary_key=True, default=uuid4) + owner_user_id: Mapped[UUID] = mapped_column(ForeignKey("users.user_id", ondelete="CASCADE"), index=True) + endpoint_id: Mapped[UUID | None] = mapped_column( + Uuid, + ForeignKey("acquisition_endpoints.endpoint_id", ondelete="SET NULL"), + nullable=True, + ) + rule_id: Mapped[UUID | None] = mapped_column(Uuid, ForeignKey("acquisition_rules.rule_id", ondelete="SET NULL")) + title: Mapped[str] = mapped_column(String(500), nullable=False) + download_url: Mapped[str] = mapped_column(Text, nullable=False) + status: Mapped[str] = mapped_column(String(32), nullable=False, default="queued", server_default="queued") + client_reference: Mapped[str | None] = mapped_column(String(255), nullable=True) + error: Mapped[str | None] = mapped_column(Text, nullable=True) + created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False, server_default=func.now()) diff --git a/papyrus/schemas/acquisition.py b/papyrus/schemas/acquisition.py new file mode 100644 index 0000000..3880b9e --- /dev/null +++ b/papyrus/schemas/acquisition.py @@ -0,0 +1,180 @@ +"""HTTP schemas for private acquisition integrations.""" + +from datetime import datetime +from enum import StrEnum +from typing import Self +from uuid import UUID + +from pydantic import BaseModel, ConfigDict, Field, HttpUrl, SecretStr, field_validator, model_validator + + +class EndpointKind(StrEnum): + QBITTORRENT = "qbittorrent" + TRANSMISSION = "transmission" + DELUGE = "deluge" + PROWLARR = "prowlarr" + TORZNAB = "torznab" + READARR = "readarr" + SONARR = "sonarr" + RADARR = "radarr" + LIDARR = "lidarr" + WHISPARR = "whisparr" + + +class AcquisitionCapabilities(BaseModel): + enabled: bool = True + endpoint_kinds: list[EndpointKind] + indexer_kinds: list[EndpointKind] + download_client_kinds: list[EndpointKind] + arr_kinds: list[EndpointKind] + arr_commands: dict[EndpointKind, list[str]] + + +class AcquisitionEndpointCreate(BaseModel): + name: str = Field(min_length=1, max_length=120) + kind: EndpointKind + base_url: HttpUrl + api_key: SecretStr | None = None + username: SecretStr | None = None + password: SecretStr | None = None + settings: dict[str, object] | None = None + + @field_validator("base_url") + @classmethod + def validate_endpoint_url(cls, value: HttpUrl) -> HttpUrl: + return validate_endpoint_url(value) + + +class AcquisitionEndpointUpdate(BaseModel): + name: str | None = Field(None, min_length=1, max_length=120) + base_url: HttpUrl | None = None + api_key: SecretStr | None = None + username: SecretStr | None = None + password: SecretStr | None = None + settings: dict[str, object] | None = None + enabled: bool | None = None + + @field_validator("base_url") + @classmethod + def validate_endpoint_url(cls, value: HttpUrl | None) -> HttpUrl | None: + if value is None: + return None + return validate_endpoint_url(value) + + +class AcquisitionEndpointTest(BaseModel): + endpoint_id: UUID | None = None + kind: EndpointKind | None = None + base_url: HttpUrl | None = None + api_key: SecretStr | None = None + username: SecretStr | None = None + password: SecretStr | None = None + + @field_validator("base_url") + @classmethod + def validate_endpoint_url(cls, value: HttpUrl | None) -> HttpUrl | None: + if value is None: + return None + return validate_endpoint_url(value) + + @model_validator(mode="after") + def validate_endpoint_target(self) -> Self: + if self.endpoint_id is None and (self.kind is None or self.base_url is None): + raise ValueError("kind and base_url are required for an unsaved endpoint") + return self + + +class AcquisitionEndpointTestResult(BaseModel): + ok: bool + + +class AcquisitionEndpoint(BaseModel): + model_config = ConfigDict(from_attributes=True) + endpoint_id: UUID + name: str + kind: EndpointKind + base_url: str + enabled: bool + settings: dict[str, object] | None = None + created_at: datetime | None = None + + +class Release(BaseModel): + title: str + download_url: str + protocol: str + indexer: str + size_bytes: int | None = None + seeders: int | None = None + publish_date: datetime | None = None + + +class SearchRequest(BaseModel): + query: str = Field(min_length=1, max_length=500) + endpoint_ids: list[UUID] | None = None + + +class SubmitRequest(BaseModel): + endpoint_id: UUID + title: str = Field(min_length=1, max_length=500) + download_url: str = Field(min_length=1, max_length=4096) + category: str | None = Field(None, max_length=100) + save_path: str | None = Field(None, max_length=1024) + + @field_validator("download_url") + @classmethod + def validate_torrent_download_url(cls, value: str) -> str: + if value.startswith("magnet:"): + return value + if value.startswith(("http://", "https://")): + return value + raise ValueError("download_url must be a magnet, HTTP, or HTTPS torrent URL") + + +class ArrCommandRequest(BaseModel): + """A Servarr command and the IDs scoped to that application's library.""" + + command: str = Field(min_length=1, max_length=80) + ids: list[int] = Field(default_factory=list, max_length=100) + + +class AcquisitionRuleCreate(BaseModel): + name: str = Field(min_length=1, max_length=120) + query: str = Field(min_length=1, max_length=500) + endpoint_ids: list[UUID] | None = None + download_client_id: UUID + filters: dict[str, object] | None = None + enabled: bool = True + + +class AcquisitionRule(BaseModel): + model_config = ConfigDict(from_attributes=True) + rule_id: UUID + name: str + query: str + endpoint_ids: list[UUID] | None = None + download_client_id: UUID | None + filters: dict[str, object] | None = None + enabled: bool + last_run_at: datetime | None = None + + +class AcquisitionJob(BaseModel): + model_config = ConfigDict(from_attributes=True) + job_id: UUID + endpoint_id: UUID | None + rule_id: UUID | None + title: str + download_url: str + status: str + client_reference: str | None = None + error: str | None = None + created_at: datetime | None = None + + +def validate_endpoint_url(value: HttpUrl) -> HttpUrl: + if value.username or value.password: + raise ValueError("Endpoint URL must not include credentials") + if value.scheme not in {"http", "https"}: + raise ValueError("Endpoint URL must use HTTP or HTTPS") + return value diff --git a/papyrus/services/acquisition.py b/papyrus/services/acquisition.py new file mode 100644 index 0000000..d6c950f --- /dev/null +++ b/papyrus/services/acquisition.py @@ -0,0 +1,558 @@ +"""Adapters for indexer protocols and self-hosted download clients.""" + +from __future__ import annotations + +import asyncio +import base64 +import json +from datetime import UTC, datetime +from typing import Any +from urllib.error import HTTPError, URLError +from urllib.parse import urlencode, urljoin +from urllib.request import Request, urlopen +from xml.etree import ElementTree + +from fastapi import HTTPException, status +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from papyrus.core.security import decrypt_secret_payload, encrypt_secret_payload +from papyrus.models.acquisition import AcquisitionEndpoint, AcquisitionJob, AcquisitionRule +from papyrus.schemas.acquisition import AcquisitionEndpointTest, Release + + +def _url(endpoint: AcquisitionEndpoint, path: str) -> str: + return urljoin(endpoint.base_url.rstrip("/") + "/", path.lstrip("/")) + + +async def _request( + url: str, *, method: str = "GET", headers: dict[str, str] | None = None, body: bytes | None = None +) -> tuple[int, dict[str, str], bytes]: + """Perform a bounded blocking HTTP request off the event loop.""" + + def send() -> tuple[int, dict[str, str], bytes]: + request = Request(url, data=body, headers=headers or {}, method=method) + try: + with urlopen(request, timeout=15) as response: # noqa: S310 - user-owned self-hosted integrations + return response.status, dict(response.headers.items()), response.read(5_000_000) + except HTTPError as exc: + return exc.code, dict(exc.headers.items()), exc.read(1_000_000) + except URLError as exc: + raise HTTPException(status_code=502, detail=f"Integration request failed: {exc.reason}") from exc + + return await asyncio.to_thread(send) + + +def _credentials(endpoint: AcquisitionEndpoint) -> dict[str, str]: + credentials = endpoint.credentials or {} + encrypted = credentials.get("encrypted") + if encrypted is None: + return credentials + try: + return decrypt_secret_payload(encrypted) + except ValueError as exc: + raise HTTPException(status_code=500, detail="Stored integration credentials are invalid") from exc + + +def _json_value(payload: bytes, integration: str) -> object: + try: + return json.loads(payload) + except (UnicodeDecodeError, json.JSONDecodeError) as exc: + raise HTTPException(status_code=502, detail=f"{integration} returned invalid JSON") from exc + + +def _json_object(payload: bytes, integration: str) -> dict[str, object]: + value = _json_value(payload, integration) + if not isinstance(value, dict): + raise HTTPException(status_code=502, detail=f"{integration} returned an invalid response") + return value + + +def _json_array(payload: bytes, integration: str) -> list[dict[str, object]]: + value = _json_value(payload, integration) + if not isinstance(value, list) or not all(isinstance(item, dict) for item in value): + raise HTTPException(status_code=502, detail=f"{integration} returned an invalid response") + return value + + +def _require_deluge_result(payload: bytes) -> object: + response = _json_object(payload, "Deluge") + result = response.get("result") + if response.get("error") is not None or result is None or result is False: + raise HTTPException(status_code=502, detail="Deluge rejected the request") + return result + + +async def search_endpoint(endpoint: AcquisitionEndpoint, query: str) -> list[Release]: + """Search Prowlarr or a Torznab-compatible torrent indexer.""" + credentials = _credentials(endpoint) + if endpoint.kind == "prowlarr": + request_url = _url(endpoint, f"api/v1/search?{urlencode({'query': query})}") + response_status, _, payload = await _request(request_url, headers={"X-Api-Key": credentials.get("api_key", "")}) + if response_status >= 400: + raise HTTPException(status_code=502, detail="Prowlarr search failed") + data = _json_array(payload, "Prowlarr") + return [ + Release( + title=item.get("title", "Untitled"), + download_url=item.get("downloadUrl") or item.get("magnetUrl") or item.get("guid", ""), + protocol="torrent", + indexer=item.get("indexer", "Prowlarr"), + size_bytes=item.get("size"), + seeders=item.get("seeders"), + ) + for item in data + if (item.get("downloadUrl") or item.get("magnetUrl") or item.get("guid")) + and item.get("protocol", "torrent") == "torrent" + ] + + params = urlencode({"t": "search", "q": query, "apikey": credentials.get("api_key", "")}) + response_status, _, payload = await _request(_url(endpoint, f"api?{params}")) + if response_status >= 400: + raise HTTPException(status_code=502, detail=f"{endpoint.kind.title()} search failed") + return _parse_torznab(payload, endpoint.name) + + +def _parse_torznab(payload: bytes, indexer: str) -> list[Release]: + try: + root = ElementTree.fromstring(payload) + except ElementTree.ParseError as exc: + raise HTTPException(status_code=502, detail="Indexer returned invalid XML") from exc + releases: list[Release] = [] + for item in root.findall(".//item"): + enclosure = item.find("enclosure") + link = (enclosure.get("url") if enclosure is not None else None) or item.findtext("link") + if not link: + continue + attrs = {child.attrib.get("name"): child.attrib.get("value") for child in item if child.tag.endswith("attr")} + size = attrs.get("size") + seeders = attrs.get("seeders") + releases.append( + Release( + title=item.findtext("title") or "Untitled", + download_url=link, + protocol="torrent", + indexer=indexer, + size_bytes=int(size) if size is not None and size.isdigit() else None, + seeders=int(seeders) if seeders is not None and seeders.isdigit() else None, + ) + ) + return releases + + +async def submit_to_client( + endpoint: AcquisitionEndpoint, download_url: str, category: str | None, save_path: str | None +) -> str | None: + """Submit a magnet or torrent URL to a supported BitTorrent client.""" + if endpoint.kind == "qbittorrent": + return await _submit_qbittorrent(endpoint, download_url, category, save_path) + if endpoint.kind == "transmission": + return await _submit_transmission(endpoint, download_url, save_path) + if endpoint.kind == "deluge": + return await _submit_deluge(endpoint, download_url, save_path) + raise HTTPException(status_code=422, detail="Endpoint is not a download client") + + +async def dispatch_arr_command(endpoint: AcquisitionEndpoint, command: str, ids: list[int]) -> str | None: + """Start an acquisition/search command in a Servarr application. + + The Arr apps own their library and release-grab decisions, so they must be + driven through their command API rather than handed an arbitrary magnet. + """ + if endpoint.kind not in {"readarr", "sonarr", "radarr", "lidarr", "whisparr"}: + raise HTTPException(status_code=422, detail="Endpoint is not a Servarr application") + + allowed_commands = { + "readarr": {"AuthorSearch", "BookSearch"}, + "sonarr": {"SeriesSearch", "EpisodeSearch", "MissingEpisodeSearch"}, + "radarr": {"MoviesSearch", "MissingMoviesSearch"}, + "lidarr": {"ArtistSearch", "AlbumSearch", "MissingAlbumSearch"}, + "whisparr": {"SeriesSearch", "EpisodeSearch", "MissingEpisodeSearch"}, + } + if command not in allowed_commands[endpoint.kind]: + raise HTTPException(status_code=422, detail="Command is not supported by this Servarr application") + + id_field = { + "AuthorSearch": "authorIds", + "BookSearch": "bookIds", + "SeriesSearch": "seriesId", + "EpisodeSearch": "episodeIds", + "MissingEpisodeSearch": "seriesId", + "MoviesSearch": "movieIds", + "MissingMoviesSearch": "movieIds", + "ArtistSearch": "artistIds", + "AlbumSearch": "albumIds", + "MissingAlbumSearch": "artistIds", + }[command] + payload: dict[str, object] = {"name": command} + if ids: + payload[id_field] = ids[0] if id_field in {"seriesId"} else ids + + response_status, _, response_payload = await _request( + _url(endpoint, "api/v3/command"), + method="POST", + headers={"Content-Type": "application/json", "X-Api-Key": _credentials(endpoint).get("api_key", "")}, + body=json.dumps(payload).encode(), + ) + if response_status >= 400: + raise HTTPException(status_code=502, detail=f"{endpoint.kind.title()} command failed") + response = _json_object(response_payload, endpoint.kind.title()) + return str(response.get("id")) if response.get("id") is not None else None + + +async def _submit_qbittorrent( + endpoint: AcquisitionEndpoint, download_url: str, category: str | None, save_path: str | None +) -> str | None: + credentials = _credentials(endpoint) + login = urlencode( + {"username": credentials.get("username", ""), "password": credentials.get("password", "")} + ).encode() + response_status, headers, response_payload = await _request( + _url(endpoint, "api/v2/auth/login"), + method="POST", + headers={"Content-Type": "application/x-www-form-urlencoded"}, + body=login, + ) + if response_status >= 400 or response_payload.strip() != b"Ok.": + raise HTTPException(status_code=502, detail="qBittorrent authentication failed") + payload = {"urls": download_url} + if category: + payload["category"] = category + if save_path: + payload["savepath"] = save_path + response_status, _, _ = await _request( + _url(endpoint, "api/v2/torrents/add"), + method="POST", + headers={ + "Content-Type": "application/x-www-form-urlencoded", + "Cookie": headers.get("Set-Cookie", "").split(";", 1)[0], + }, + body=urlencode(payload).encode(), + ) + if response_status >= 400: + raise HTTPException(status_code=502, detail="qBittorrent rejected the release") + return None + + +async def _submit_transmission(endpoint: AcquisitionEndpoint, download_url: str, save_path: str | None) -> str | None: + credentials = _credentials(endpoint) + arguments: dict[str, str] = {"filename": download_url} + if save_path: + arguments["download-dir"] = save_path + body = json.dumps({"method": "torrent-add", "arguments": arguments}).encode() + headers = {"Content-Type": "application/json"} + if credentials.get("username"): + token = base64.b64encode(f"{credentials['username']}:{credentials.get('password', '')}".encode()).decode() + headers["Authorization"] = f"Basic {token}" + response_status, response_headers, payload = await _request( + _url(endpoint, "transmission/rpc"), method="POST", headers=headers, body=body + ) + if response_status == 409: + headers["X-Transmission-Session-Id"] = response_headers.get("X-Transmission-Session-Id", "") + response_status, _, payload = await _request( + _url(endpoint, "transmission/rpc"), method="POST", headers=headers, body=body + ) + if response_status >= 400: + raise HTTPException(status_code=502, detail="Transmission rejected the release") + + response = _json_object(payload, "Transmission") + if response.get("result") != "success": + raise HTTPException(status_code=502, detail="Transmission rejected the release") + + response_arguments = response.get("arguments") + if not isinstance(response_arguments, dict): + raise HTTPException(status_code=502, detail="Transmission returned an invalid response") + + torrent = response_arguments.get("torrent-added") or response_arguments.get("torrent-duplicate") + if not isinstance(torrent, dict): + return None + + reference = torrent.get("hashString") + return str(reference) if reference is not None else None + + +async def _submit_deluge(endpoint: AcquisitionEndpoint, download_url: str, save_path: str | None) -> str | None: + credentials = _credentials(endpoint) + headers = {"Content-Type": "application/json"} + login = json.dumps({"method": "auth.login", "params": [credentials.get("password", "")], "id": 1}).encode() + response_status, response_headers, login_payload = await _request( + _url(endpoint, "json"), method="POST", headers=headers, body=login + ) + if response_status >= 400: + raise HTTPException(status_code=502, detail="Deluge authentication failed") + + try: + _require_deluge_result(login_payload) + except HTTPException as exc: + raise HTTPException(status_code=502, detail="Deluge authentication failed") from exc + + options = {"download_location": save_path} if save_path else {} + method = "core.add_torrent_magnet" if download_url.startswith("magnet:") else "core.add_torrent_url" + body = json.dumps({"method": method, "params": [download_url, options], "id": 2}).encode() + headers["Cookie"] = response_headers.get("Set-Cookie", "").split(";", 1)[0] + response_status, _, payload = await _request(_url(endpoint, "json"), method="POST", headers=headers, body=body) + if response_status >= 400: + raise HTTPException(status_code=502, detail="Deluge rejected the release") + + try: + result = _require_deluge_result(payload) + except HTTPException as exc: + raise HTTPException(status_code=502, detail="Deluge rejected the release") from exc + + return str(result) + + +async def owned_endpoint(session: AsyncSession, owner_user_id: Any, endpoint_id: Any) -> AcquisitionEndpoint: + result = await session.execute( + select(AcquisitionEndpoint).where( + AcquisitionEndpoint.endpoint_id == endpoint_id, AcquisitionEndpoint.owner_user_id == owner_user_id + ) + ) + endpoint = result.scalar_one_or_none() + if endpoint is None: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Acquisition endpoint not found") + return endpoint + + +async def delete_acquisition_endpoint(session: AsyncSession, owner_user_id: Any, endpoint_id: Any) -> None: + endpoint = await owned_endpoint(session, owner_user_id, endpoint_id) + result = await session.execute( + select(AcquisitionRule).where(AcquisitionRule.owner_user_id == owner_user_id).with_for_update() + ) + + for rule in result.scalars(): + endpoint_ids = rule.endpoint_ids or [] + remaining_endpoint_ids = [value for value in endpoint_ids if value != str(endpoint_id)] + indexer_deleted = remaining_endpoint_ids != endpoint_ids + download_client_deleted = rule.download_client_id == endpoint_id + + if indexer_deleted: + rule.endpoint_ids = remaining_endpoint_ids + + if download_client_deleted: + rule.download_client_id = None + + if download_client_deleted or (indexer_deleted and not remaining_endpoint_ids): + rule.enabled = False + + await session.flush() + + await session.delete(endpoint) + await session.commit() + + +async def build_test_endpoint( + session: AsyncSession, + owner_user_id: Any, + request: AcquisitionEndpointTest, +) -> AcquisitionEndpoint: + stored_endpoint = None + stored_credentials: dict[str, str] = {} + + if request.endpoint_id is not None: + stored_endpoint = await owned_endpoint(session, owner_user_id, request.endpoint_id) + stored_credentials = _credentials(stored_endpoint) + + credentials = dict(stored_credentials) + for field in ("api_key", "username", "password"): + value = getattr(request, field) + if value is not None: + credentials[field] = value.get_secret_value() + + if request.kind is not None: + kind = request.kind.value + elif stored_endpoint is not None: + kind = stored_endpoint.kind + else: + raise HTTPException(status_code=422, detail="Endpoint kind is required") + + if request.base_url is not None: + base_url = str(request.base_url) + elif stored_endpoint is not None: + base_url = stored_endpoint.base_url + else: + raise HTTPException(status_code=422, detail="Endpoint URL is required") + + encrypted_credentials = {"encrypted": encrypt_secret_payload(credentials)} if credentials else None + + return AcquisitionEndpoint( + owner_user_id=owner_user_id, + name=stored_endpoint.name if stored_endpoint is not None else "Connection test", + kind=kind, + base_url=base_url, + credentials=encrypted_credentials, + settings=stored_endpoint.settings if stored_endpoint is not None else None, + ) + + +async def test_endpoint_connection(endpoint: AcquisitionEndpoint) -> None: + credentials = _credentials(endpoint) + + if endpoint.kind == "prowlarr": + response_status, _, payload = await _request( + _url(endpoint, "api/v1/system/status"), + headers={"X-Api-Key": credentials.get("api_key", "")}, + ) + if response_status >= 400: + raise HTTPException(status_code=502, detail="Prowlarr connection test failed") + + _json_object(payload, "Prowlarr") + return + + if endpoint.kind == "torznab": + params = urlencode({"t": "caps", "apikey": credentials.get("api_key", "")}) + response_status, _, payload = await _request(_url(endpoint, f"api?{params}")) + if response_status >= 400: + raise HTTPException(status_code=502, detail="Torznab connection test failed") + + try: + ElementTree.fromstring(payload) + except ElementTree.ParseError as exc: + raise HTTPException(status_code=502, detail="Torznab returned invalid XML") from exc + return + + if endpoint.kind in {"readarr", "sonarr", "radarr", "lidarr", "whisparr"}: + response_status, _, payload = await _request( + _url(endpoint, "api/v3/system/status"), + headers={"X-Api-Key": credentials.get("api_key", "")}, + ) + if response_status >= 400: + raise HTTPException(status_code=502, detail=f"{endpoint.kind.title()} connection test failed") + + _json_object(payload, endpoint.kind.title()) + return + + if endpoint.kind == "qbittorrent": + login = urlencode( + {"username": credentials.get("username", ""), "password": credentials.get("password", "")} + ).encode() + response_status, _, payload = await _request( + _url(endpoint, "api/v2/auth/login"), + method="POST", + headers={"Content-Type": "application/x-www-form-urlencoded"}, + body=login, + ) + if response_status >= 400 or payload.strip() != b"Ok.": + raise HTTPException(status_code=502, detail="qBittorrent authentication failed") + return + + if endpoint.kind == "transmission": + body = json.dumps({"method": "session-get", "arguments": {}}).encode() + headers = {"Content-Type": "application/json"} + if credentials.get("username"): + token = base64.b64encode(f"{credentials['username']}:{credentials.get('password', '')}".encode()).decode() + headers["Authorization"] = f"Basic {token}" + + response_status, response_headers, payload = await _request( + _url(endpoint, "transmission/rpc"), + method="POST", + headers=headers, + body=body, + ) + if response_status == 409: + headers["X-Transmission-Session-Id"] = response_headers.get("X-Transmission-Session-Id", "") + response_status, _, payload = await _request( + _url(endpoint, "transmission/rpc"), + method="POST", + headers=headers, + body=body, + ) + if response_status >= 400 or _json_object(payload, "Transmission").get("result") != "success": + raise HTTPException(status_code=502, detail="Transmission connection test failed") + return + + if endpoint.kind == "deluge": + login = json.dumps({"method": "auth.login", "params": [credentials.get("password", "")], "id": 1}).encode() + response_status, _, payload = await _request( + _url(endpoint, "json"), + method="POST", + headers={"Content-Type": "application/json"}, + body=login, + ) + if response_status >= 400: + raise HTTPException(status_code=502, detail="Deluge authentication failed") + + try: + _require_deluge_result(payload) + except HTTPException as exc: + raise HTTPException(status_code=502, detail="Deluge authentication failed") from exc + return + + raise HTTPException(status_code=422, detail="Endpoint kind is not supported") + + +async def run_rule(session: AsyncSession, rule: AcquisitionRule) -> list[AcquisitionJob]: + """Run one rule once; callers may schedule this from their worker/cron service.""" + client = await owned_endpoint(session, rule.owner_user_id, rule.download_client_id) + if client.kind in {"readarr", "sonarr", "radarr", "lidarr", "whisparr"}: + filters = rule.filters or {} + command = filters.get("arr_command") + ids = filters.get("arr_ids", []) + if ( + not isinstance(command, str) + or not isinstance(ids, list) + or not all(isinstance(value, int) for value in ids) + ): + raise HTTPException( + status_code=422, + detail="Arr rules require filters.arr_command and filters.arr_ids", + ) + job = AcquisitionJob( + owner_user_id=rule.owner_user_id, + endpoint_id=client.endpoint_id, + rule_id=rule.rule_id, + title=command, + download_url=f"arr-command:{command}", + ) + session.add(job) + try: + job.client_reference = await dispatch_arr_command(client, command, ids) + job.status = "submitted" + except HTTPException as exc: + job.status = "failed" + job.error = str(exc.detail) + rule.last_run_at = datetime.now(UTC) + await session.commit() + return [job] + + endpoint_ids = rule.endpoint_ids or [] + result = await session.execute( + select(AcquisitionEndpoint).where( + AcquisitionEndpoint.owner_user_id == rule.owner_user_id, + AcquisitionEndpoint.endpoint_id.in_(endpoint_ids), + AcquisitionEndpoint.enabled.is_(True), + ) + ) + releases = [release for endpoint in result.scalars() for release in await search_endpoint(endpoint, rule.query)] + if not releases: + rule.last_run_at = datetime.now(UTC) + await session.commit() + return [] + selected = sorted(releases, key=lambda release: (release.seeders or 0, release.size_bytes or 0), reverse=True)[0] + job = AcquisitionJob( + owner_user_id=rule.owner_user_id, + endpoint_id=client.endpoint_id, + rule_id=rule.rule_id, + title=selected.title, + download_url=selected.download_url, + ) + session.add(job) + try: + job.client_reference = await submit_to_client(client, selected.download_url, None, None) + job.status = "submitted" + except HTTPException as exc: + job.status = "failed" + job.error = str(exc.detail) + rule.last_run_at = datetime.now(UTC) + await session.commit() + return [job] + + +async def run_enabled_rules(session: AsyncSession) -> None: + """Run enabled rules, isolating a failed remote integration from the others.""" + result = await session.execute(select(AcquisitionRule).where(AcquisitionRule.enabled.is_(True))) + for rule in result.scalars(): + try: + await run_rule(session, rule) + except HTTPException: + await session.rollback() diff --git a/tests/api/routes/test_acquisition.py b/tests/api/routes/test_acquisition.py new file mode 100644 index 0000000..a41a4f2 --- /dev/null +++ b/tests/api/routes/test_acquisition.py @@ -0,0 +1,338 @@ +"""Tests for private BitTorrent acquisition configuration.""" + +from uuid import UUID + +import pytest +from httpx import AsyncClient +from sqlalchemy import func, select +from sqlalchemy.ext.asyncio import AsyncSession + +from papyrus.api.routes import acquisition as acquisition_routes +from papyrus.core.security import decrypt_secret_payload +from papyrus.main import settings as app_settings +from papyrus.models.acquisition import AcquisitionEndpoint, AcquisitionJob, AcquisitionRule +from papyrus.models.user import User +from papyrus.services import acquisition as acquisition_service + + +@pytest.fixture(autouse=True) +def enable_acquisition(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(app_settings, "acquisition_enabled", True, raising=False) + + +async def test_disabled_capabilities_hide_acquisition_scope(client: AsyncClient) -> None: + app_settings.acquisition_enabled = False + + response = await client.get("/v1/acquisition/capabilities") + + assert response.status_code == 200 + assert response.json() == { + "enabled": False, + "endpoint_kinds": [], + "indexer_kinds": [], + "download_client_kinds": [], + "arr_kinds": [], + "arr_commands": {}, + } + + +async def test_disabled_acquisition_routes_are_not_found( + client: AsyncClient, + auth_headers: dict[str, str], +) -> None: + app_settings.acquisition_enabled = False + + response = await client.get("/v1/acquisition/endpoints", headers=auth_headers) + + assert response.status_code == 404 + + +async def test_capabilities_advertise_torrent_only_scope(client: AsyncClient, auth_headers: dict[str, str]) -> None: + response = await client.get("/v1/acquisition/capabilities", headers=auth_headers) + + assert response.status_code == 200 + body = response.json() + assert body["indexer_kinds"] == ["prowlarr", "torznab"] + assert body["download_client_kinds"] == ["qbittorrent", "transmission", "deluge"] + assert "newznab" not in body["endpoint_kinds"] + assert body["arr_commands"]["readarr"] == ["AuthorSearch", "BookSearch"] + + +async def test_create_and_list_endpoint_hides_and_encrypts_credentials( + client: AsyncClient, + auth_headers: dict[str, str], + db_session: AsyncSession, +) -> None: + """Credentials can be configured but must never be returned to a client.""" + response = await client.post( + "/v1/acquisition/endpoints", + headers=auth_headers, + json={ + "name": "Home qBittorrent", + "kind": "qbittorrent", + "base_url": "http://127.0.0.1:8080", + "username": "admin", + "password": "secret", + }, + ) + + assert response.status_code == 201 + assert "password" not in response.text + assert "username" not in response.text + endpoint = response.json() + + response = await client.get("/v1/acquisition/endpoints", headers=auth_headers) + assert response.status_code == 200 + assert response.json() == [endpoint] + + stored = (await db_session.execute(select(AcquisitionEndpoint))).scalar_one() + assert stored.credentials is not None + assert "password" not in stored.credentials + assert decrypt_secret_payload(stored.credentials["encrypted"]) == { + "username": "admin", + "password": "secret", + } + + +async def test_create_rule_requires_owned_download_client(client: AsyncClient, auth_headers: dict[str, str]) -> None: + """Rules cannot target a client belonging to another user or a missing client.""" + response = await client.post( + "/v1/acquisition/rules", + headers=auth_headers, + json={ + "name": "Monthly reading", + "query": "example book", + "download_client_id": "00000000-0000-0000-0000-000000000001", + }, + ) + + assert response.status_code == 404 + + +async def test_readarr_endpoint_is_a_supported_integration(client: AsyncClient, auth_headers: dict[str, str]) -> None: + """Readarr uses its own acquisition workflow and is accepted as an endpoint.""" + response = await client.post( + "/v1/acquisition/endpoints", + headers=auth_headers, + json={ + "name": "Reading automation", + "kind": "readarr", + "base_url": "http://readarr.local:8787", + "api_key": "private-key", + }, + ) + + assert response.status_code == 201 + assert response.json()["kind"] == "readarr" + + +async def test_newznab_endpoint_is_rejected(client: AsyncClient, auth_headers: dict[str, str]) -> None: + response = await client.post( + "/v1/acquisition/endpoints", + headers=auth_headers, + json={ + "name": "Usenet", + "kind": "newznab", + "base_url": "http://newznab.local", + "api_key": "private-key", + }, + ) + + assert response.status_code == 422 + + +async def test_endpoint_url_rejects_embedded_credentials(client: AsyncClient, auth_headers: dict[str, str]) -> None: + response = await client.post( + "/v1/acquisition/endpoints", + headers=auth_headers, + json={ + "name": "Bad URL", + "kind": "qbittorrent", + "base_url": "http://user:secret@127.0.0.1:8080", + }, + ) + + assert response.status_code == 422 + + +async def test_rejected_submission_persists_failed_job( + client: AsyncClient, + auth_headers: dict[str, str], + monkeypatch: pytest.MonkeyPatch, +) -> None: + async def request(*args: object, **kwargs: object) -> tuple[int, dict[str, str], bytes]: + return 200, {}, b'{"result":"invalid or corrupt torrent file","arguments":{}}' + + monkeypatch.setattr(acquisition_service, "_request", request) + endpoint_response = await client.post( + "/v1/acquisition/endpoints", + headers=auth_headers, + json={ + "name": "Transmission", + "kind": "transmission", + "base_url": "http://transmission.local:9091", + }, + ) + endpoint_id = endpoint_response.json()["endpoint_id"] + + response = await client.post( + "/v1/acquisition/submissions", + headers=auth_headers, + json={ + "endpoint_id": endpoint_id, + "title": "Rejected release", + "download_url": "magnet:?xt=urn:btih:test", + }, + ) + + assert response.status_code == 201 + assert response.json()["status"] == "failed" + assert response.json()["error"] == "Transmission rejected the release" + + +async def test_delete_endpoint_preserves_jobs_and_disables_rules( + client: AsyncClient, + auth_headers: dict[str, str], + auth_user: dict[str, str], + db_session: AsyncSession, +) -> None: + endpoint_response = await client.post( + "/v1/acquisition/endpoints", + headers=auth_headers, + json={ + "name": "Disposable client", + "kind": "qbittorrent", + "base_url": "http://qbittorrent.local:8080", + }, + ) + endpoint_id = UUID(endpoint_response.json()["endpoint_id"]) + owner_user_id = UUID(auth_user["user_id"]) + rule = AcquisitionRule( + owner_user_id=owner_user_id, + name="Affected rule", + query="book", + endpoint_ids=[str(endpoint_id)], + download_client_id=endpoint_id, + enabled=True, + ) + job = AcquisitionJob( + owner_user_id=owner_user_id, + endpoint_id=endpoint_id, + title="Audited release", + download_url="magnet:?xt=urn:btih:test", + status="submitted", + ) + + db_session.add_all([rule, job]) + await db_session.commit() + + response = await client.delete(f"/v1/acquisition/endpoints/{endpoint_id}", headers=auth_headers) + + assert response.status_code == 204 + await db_session.refresh(job) + await db_session.refresh(rule) + assert job.endpoint_id is None + assert rule.download_client_id is None + assert rule.endpoint_ids == [] + assert rule.enabled is False + + +async def test_connection_checks_unsaved_endpoint_without_persisting( + client: AsyncClient, + auth_headers: dict[str, str], + db_session: AsyncSession, + monkeypatch: pytest.MonkeyPatch, +) -> None: + captured: list[AcquisitionEndpoint] = [] + + async def test_connection(endpoint: AcquisitionEndpoint) -> None: + captured.append(endpoint) + + monkeypatch.setattr(acquisition_routes, "test_endpoint_connection", test_connection, raising=False) + before_count = await db_session.scalar(select(func.count()).select_from(AcquisitionEndpoint)) + + response = await client.post( + "/v1/acquisition/endpoints/test", + headers=auth_headers, + json={ + "kind": "prowlarr", + "base_url": "http://prowlarr.local:9696", + "api_key": "unsaved-key", + }, + ) + + assert response.status_code == 200 + assert response.json() == {"ok": True} + assert captured[0].kind == "prowlarr" + assert decrypt_secret_payload(captured[0].credentials["encrypted"])["api_key"] == "unsaved-key" + after_count = await db_session.scalar(select(func.count()).select_from(AcquisitionEndpoint)) + assert after_count == before_count + + +async def test_connection_merges_owned_endpoint_overrides_without_persisting( + client: AsyncClient, + auth_headers: dict[str, str], + db_session: AsyncSession, + monkeypatch: pytest.MonkeyPatch, +) -> None: + endpoint_response = await client.post( + "/v1/acquisition/endpoints", + headers=auth_headers, + json={ + "name": "Saved Prowlarr", + "kind": "prowlarr", + "base_url": "http://prowlarr.local:9696", + "api_key": "saved-key", + }, + ) + captured: list[AcquisitionEndpoint] = [] + + async def test_connection(endpoint: AcquisitionEndpoint) -> None: + captured.append(endpoint) + + monkeypatch.setattr(acquisition_routes, "test_endpoint_connection", test_connection, raising=False) + before_count = await db_session.scalar(select(func.count()).select_from(AcquisitionEndpoint)) + + response = await client.post( + "/v1/acquisition/endpoints/test", + headers=auth_headers, + json={ + "endpoint_id": endpoint_response.json()["endpoint_id"], + "base_url": "http://prowlarr-edited.local:9696", + "api_key": "override", + }, + ) + + assert response.status_code == 200 + assert response.json() == {"ok": True} + assert captured[0].base_url == "http://prowlarr-edited.local:9696/" + assert decrypt_secret_payload(captured[0].credentials["encrypted"])["api_key"] == "override" + after_count = await db_session.scalar(select(func.count()).select_from(AcquisitionEndpoint)) + assert after_count == before_count + + +async def test_connection_rejects_another_users_endpoint( + client: AsyncClient, + auth_headers: dict[str, str], + db_session: AsyncSession, +) -> None: + other_user = User(display_name="Other User") + db_session.add(other_user) + await db_session.flush() + + endpoint = AcquisitionEndpoint( + owner_user_id=other_user.user_id, + name="Other Prowlarr", + kind="prowlarr", + base_url="http://other-prowlarr.local:9696", + ) + db_session.add(endpoint) + await db_session.commit() + + response = await client.post( + "/v1/acquisition/endpoints/test", + headers=auth_headers, + json={"endpoint_id": str(endpoint.endpoint_id)}, + ) + + assert response.status_code == 404 diff --git a/tests/services/test_acquisition.py b/tests/services/test_acquisition.py new file mode 100644 index 0000000..0029261 --- /dev/null +++ b/tests/services/test_acquisition.py @@ -0,0 +1,129 @@ +"""Tests for acquisition integration adapters.""" + +import json +from typing import cast +from uuid import uuid4 + +import pytest +from fastapi import HTTPException + +from papyrus.models.acquisition import AcquisitionEndpoint +from papyrus.services import acquisition + + +def _endpoint(kind: str) -> AcquisitionEndpoint: + return AcquisitionEndpoint( + owner_user_id=uuid4(), + name=f"Test {kind}", + kind=kind, + base_url="http://integration.test", + ) + + +async def test_transmission_rejects_rpc_failure(monkeypatch: pytest.MonkeyPatch) -> None: + async def request(*args: object, **kwargs: object) -> tuple[int, dict[str, str], bytes]: + return 200, {}, b'{"result":"invalid or corrupt torrent file","arguments":{}}' + + monkeypatch.setattr(acquisition, "_request", request) + + with pytest.raises(HTTPException) as exc_info: + await acquisition.submit_to_client( + _endpoint("transmission"), + "magnet:?xt=urn:btih:test", + None, + None, + ) + + assert exc_info.value.status_code == 502 + + +async def test_deluge_uses_url_method_for_http_torrent(monkeypatch: pytest.MonkeyPatch) -> None: + bodies: list[dict[str, object]] = [] + + async def request(*args: object, **kwargs: object) -> tuple[int, dict[str, str], bytes]: + bodies.append(json.loads(cast(bytes, kwargs["body"]))) + + if len(bodies) == 1: + return 200, {"Set-Cookie": "_session_id=test"}, b'{"result":true,"error":null,"id":1}' + + return 200, {}, b'{"result":"torrent-id","error":null,"id":2}' + + monkeypatch.setattr(acquisition, "_request", request) + + await acquisition.submit_to_client( + _endpoint("deluge"), + "https://indexer.test/release.torrent", + None, + None, + ) + + assert bodies[1]["method"] == "core.add_torrent_url" + + +async def test_deluge_rejects_json_rpc_error(monkeypatch: pytest.MonkeyPatch) -> None: + responses = iter( + [ + (200, {"Set-Cookie": "_session_id=test"}, b'{"result":true,"error":null,"id":1}'), + (200, {}, b'{"result":null,"error":{"message":"invalid torrent"},"id":2}'), + ] + ) + + async def request(*args: object, **kwargs: object) -> tuple[int, dict[str, str], bytes]: + return next(responses) + + monkeypatch.setattr(acquisition, "_request", request) + + with pytest.raises(HTTPException) as exc_info: + await acquisition.submit_to_client( + _endpoint("deluge"), + "magnet:?xt=urn:btih:test", + None, + None, + ) + + assert exc_info.value.status_code == 502 + + +async def test_prowlarr_rejects_invalid_json(monkeypatch: pytest.MonkeyPatch) -> None: + async def request(*args: object, **kwargs: object) -> tuple[int, dict[str, str], bytes]: + return 200, {}, b"not-json" + + monkeypatch.setattr(acquisition, "_request", request) + + with pytest.raises(HTTPException) as exc_info: + await acquisition.search_endpoint(_endpoint("prowlarr"), "book") + + assert exc_info.value.status_code == 502 + assert exc_info.value.detail == "Prowlarr returned invalid JSON" + + +async def test_qbittorrent_rejects_failed_login_body(monkeypatch: pytest.MonkeyPatch) -> None: + async def request(*args: object, **kwargs: object) -> tuple[int, dict[str, str], bytes]: + return 200, {}, b"Fails." + + monkeypatch.setattr(acquisition, "_request", request) + + with pytest.raises(HTTPException) as exc_info: + await acquisition.submit_to_client( + _endpoint("qbittorrent"), + "magnet:?xt=urn:btih:test", + None, + None, + ) + + assert exc_info.value.status_code == 502 + assert exc_info.value.detail == "qBittorrent authentication failed" + + +async def test_arr_commands_use_v3_api(monkeypatch: pytest.MonkeyPatch) -> None: + urls: list[str] = [] + + async def request(url: str, **kwargs: object) -> tuple[int, dict[str, str], bytes]: + urls.append(url) + return 201, {}, b'{"id":1}' + + monkeypatch.setattr(acquisition, "_request", request) + + await acquisition.dispatch_arr_command(_endpoint("readarr"), "BookSearch", [1]) + + assert urls == ["http://integration.test/api/v3/command"]