diff --git a/.gitignore b/.gitignore index cfeb20e..d9b84a0 100644 --- a/.gitignore +++ b/.gitignore @@ -129,4 +129,8 @@ dmypy.json .pyre/ .idea/ -.vscode/ \ No newline at end of file +.vscode/ +.env + +# Generated test fixtures (recreated by conftest on demand). +tests/fixtures/ diff --git a/README.md b/README.md index c4f57fa..ad49742 100644 --- a/README.md +++ b/README.md @@ -253,20 +253,101 @@ body: src: "local_image.jpg" # Local images will be uploaded automatically ``` +## Publishing a Podcast Episode + +The `PodcastPost` class is the podcast counterpart to `Post`. A podcast draft +carries two distinct ProseMirror documents -- `draft_body` (the post-page +body, same field a newsletter Post uses) and `podcast_description` (the show +notes shown alongside the audio player) -- and adds podcast-specific fields +like `draft_podcast_upload_id` and `draft_podcast_duration`. Both ProseMirror +docs are independently optional. + +The audio upload uses Substack's three-step S3 protocol (initiate -> PUT +bytes -> trigger transcode) plus polling until the media object reaches +`state == "transcoded"`. `Api.upload_podcast_audio` wraps the whole +sequence; duration is computed locally with mutagen. + +```python +import os +from dotenv import load_dotenv + +from substack import Api, PodcastPost + +load_dotenv() + +api = Api( + cookies_string=os.getenv("COOKIES_STRING"), + publication_url=os.getenv("PUBLICATION_URL"), +) +user_id = api.get_user_id() + +# 1. Build the podcast draft in memory. +pod = ( + PodcastPost( + title="Episode 7: Async I/O", + subtitle="What asyncio buys you, and what it costs", + user_id=user_id, + audience="everyone", + ) + .set_body_from_markdown("Full episode text mirror for the post page.") + .set_show_notes_from_markdown( + "## Show notes\n\n- Event loops\n- Backpressure\n" + ) +) + +# 2. Create the draft. We need a draft id before we can upload audio +# (Substack scopes the upload to a post_id at init time). +draft = api.post_draft(pod.get_draft()) +draft_id = draft["id"] + +# 3. Upload the audio. Returns once the media object is transcoded. +media = api.upload_podcast_audio(file_path="episode-07.mp3", draft_id=draft_id) + +# 4. Attach the audio. Thread last_updated_at from the create response so +# subsequent PUTs match the server's optimistic-concurrency token. +pod.set_audio(media["id"], duration_seconds=media["duration"]) +pod.last_updated_at = draft.get("draft_updated_at") +api.put_draft(draft_id, **pod.get_draft()) + +# 5. Prepublish (server-side validation) and publish. +api.prepublish_draft(draft_id) +api.publish_draft(draft_id, send=True, share_automatically=False) +``` + +A runnable end-to-end example lives at +[`examples/publish_podcast.py`](examples/publish_podcast.py) (`--help` for +flags including `--audio`, `--body-file`, `--show-notes-file`, `--publish`, +and `--no-send` for sandbox testing). + ## MCP FastMCP server -This package now includes a FastMCP server in `substack/mcp_fastmcp.py` with the following tools: +This package includes a FastMCP server in `substack_mcp/mcp_server.py` with +the following tools: -- `post_draft_from_markdown(...)`: create draft from markdown, optional tag/add/prepublish/publish, and control send/share_automatically. +Newsletter: + +- `post_draft_from_markdown(...)`: create a newsletter draft from markdown, with optional tag/prepublish/publish. - `put_draft(draft_id, update_payload)`: update draft fields. - `add_tags(draft_id, tags)`: add tags to a draft/post. - `prepublish_draft(draft_id)`: prepublish a draft. - `publish_draft(draft_id, send=True, share_automatically=False)`: publish a draft. +Podcast: + +- `upload_podcast_audio(file_path, draft_id)`: upload an MP3 and wait for + transcoding. Returns the final media object with `state == "transcoded"`. + The audio UUID (`id` field) is what attaches the upload to a draft. +- `post_podcast_draft_from_markdown(...)`: create (and optionally publish) + a podcast episode end-to-end. Composes draft creation + optional audio + upload + attach + optional tag + optional prepublish + optional publish + in a single call. Show notes (`podcast_description`) and post-page body + (`draft_body`) are independently optional. Returns a dict surfacing each + composed step's result. + Use via stdio transport: ```bash -python -c "from substack.mcp_fastmcp import main; main()" +python -c "from substack_mcp.mcp_server import main; main()" ``` # Contributing diff --git a/examples/publish_podcast.py b/examples/publish_podcast.py new file mode 100644 index 0000000..30f976f --- /dev/null +++ b/examples/publish_podcast.py @@ -0,0 +1,195 @@ +""" +Example: Publishing a podcast episode end-to-end + +Demonstrates the full podcast flow: + 1. Build a PodcastPost with title, subtitle, body, show notes + 2. Create the podcast draft on Substack + 3. Upload the MP3 (init -> S3 PUT -> transcode -> poll-until-ready) + 4. Attach the uploaded audio to the draft + 5. Prepublish (server-side validation) + 6. (Optional) Publish the episode + +Run without --publish to leave the result as a draft; add --publish to +actually go live. Use --no-send with --publish to publish without firing +the subscriber-email notification. +""" + +import argparse +import os +from pathlib import Path + +from dotenv import load_dotenv + +from substack import Api, PodcastPost + +load_dotenv() + +DEFAULT_BODY_MD = ( + "This episode was created with the python-substack `publish_podcast.py` example.\n" +) +DEFAULT_SHOW_NOTES_MD = ( + "## Show notes\n\n" + "- Generated by `examples/publish_podcast.py`\n" + "- Edit this script to set your own title, body, and show notes.\n" +) + + +def _read_or_default(path: str | None, default: str) -> str: + """Return file contents at `path`, or `default` when `path` is None.""" + if path is None: + return default + return Path(path).read_text(encoding="utf-8") + + +if __name__ == "__main__": + parser = argparse.ArgumentParser( + description="Create, optionally publish, a Substack podcast episode.", + ) + parser.add_argument( + "-a", + "--audio", + required=True, + type=str, + help="Path to the episode MP3 file.", + ) + parser.add_argument( + "-t", + "--title", + default="Test episode", + type=str, + help="Episode title (default: 'Test episode').", + ) + parser.add_argument( + "--subtitle", + default="Created with python-substack", + type=str, + help="Episode subtitle.", + ) + parser.add_argument( + "--body-file", + default=None, + type=str, + help="Path to a Markdown file for the post-page body. " + "If omitted, a short default is used.", + ) + parser.add_argument( + "--show-notes-file", + default=None, + type=str, + help="Path to a Markdown file for the show notes (podcast_description). " + "If omitted, a short default is used.", + ) + parser.add_argument( + "--audience", + default="everyone", + choices=["everyone", "only_paid", "founding", "only_free"], + help="Audience for the episode (default: everyone).", + ) + parser.add_argument( + "--publish", + help="Publish the episode. Without this flag the script stops at the draft stage.", + action="store_true", + default=False, + ) + parser.add_argument( + "--no-send", + help="When combined with --publish, do not email subscribers. " + "Recommended for sandbox testing.", + action="store_true", + default=False, + ) + parser.add_argument( + "--cookies", + help="Path to cookies JSON file for authentication " + "(optional, can also be set via COOKIES_PATH or COOKIES_STRING env vars).", + type=str, + default=None, + ) + args = parser.parse_args() + + cookies_path = args.cookies or os.getenv("COOKIES_PATH") + cookies_string = os.getenv("COOKIES_STRING") + + api = Api( + email=os.getenv("EMAIL") if not cookies_path and not cookies_string else None, + password=os.getenv("PASSWORD") + if not cookies_path and not cookies_string + else None, + cookies_path=cookies_path, + cookies_string=cookies_string, + publication_url=os.getenv("PUBLICATION_URL"), + ) + + user_id = api.get_user_id() + + body_markdown = _read_or_default(args.body_file, DEFAULT_BODY_MD) + show_notes_markdown = _read_or_default(args.show_notes_file, DEFAULT_SHOW_NOTES_MD) + + # 1. Build the podcast draft in memory. + pod = ( + PodcastPost( + title=args.title, + subtitle=args.subtitle, + user_id=user_id, + audience=args.audience, + ) + .set_body_from_markdown(body_markdown, api=api) + .set_show_notes_from_markdown(show_notes_markdown, api=api) + ) + + # 2. Create the podcast draft. We need a draft id before we can upload + # audio (Substack scopes the upload to a post_id at init time). + print("Creating draft...") + draft = api.post_draft(pod.get_draft()) + draft_id = draft["id"] + print(f"Draft created: id={draft_id}") + + # 3. Upload the audio. This handles init + S3 PUT + transcode + polling + # in one call; returns once the media object reports state="transcoded". + print(f"Uploading audio from {args.audio} ...") + media = api.upload_podcast_audio(file_path=args.audio, draft_id=draft_id) + upload_id = media["id"] + duration_seconds = media["duration"] + print( + f"Audio uploaded and transcoded: upload_id={upload_id} " + f"duration={duration_seconds:.2f}s" + ) + + # 4. Attach the audio to the draft. Thread last_updated_at from the + # create response so subsequent PUTs match the server's optimistic + # concurrency token (see docs/PODCAST_ARCHITECTURE.md section 7). + pod.set_audio(upload_id, duration_seconds=duration_seconds) + pod.last_updated_at = draft.get("draft_updated_at") + + print("Attaching audio to draft...") + updated = api.put_draft(draft_id, **pod.get_draft()) + print(f"Audio attached: draft_podcast_upload_id={updated['draft_podcast_upload_id']}") + + if not args.publish: + print( + f"\nDraft ready (not published). View / edit in the Substack editor, " + f"or re-run with --publish to publish." + ) + print(f"Draft id: {draft_id}") + raise SystemExit(0) + + # 5. Prepublish: server-side validation. Surfaces missing required fields + # before we commit to publishing. + print("\nRunning prepublish validation...") + prepublish = api.prepublish_draft(draft_id) + if prepublish.get("errors"): + print(f"Prepublish FAILED: {prepublish['errors']}") + raise SystemExit(1) + if prepublish.get("suggestions"): + print(f"Prepublish suggestions: {prepublish['suggestions']}") + + # 6. Publish. + send = not args.no_send + print(f"Publishing episode (send={send})...") + published = api.publish_draft( + draft_id, send=send, share_automatically=False + ) + print( + f"Published: id={published['id']} is_published={published['is_published']} " + f"slug={published.get('slug')!r}" + ) diff --git a/pyproject.toml b/pyproject.toml index 9a96686..37618ab 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -21,8 +21,13 @@ python = "<4.0,>=3.10" requests = "^2.32.0" python-dotenv = "^1.2.1" PyYAML = "^6.0" +mutagen = "^1.47" [tool.poetry.group.dev.dependencies] +pytest = "^9.0" +# lameenc generates the tiny silent MP3 used by the integration test fixture. +# Pure-Python wheels with LAME bundled -- no ffmpeg / system dependency. +lameenc = "^1.8" [tool.poetry.group.mcp] @@ -34,3 +39,8 @@ fastmcp = "^3.1.1" [build-system] requires = ["poetry-core>=1.0.0"] build-backend = "poetry.core.masonry.api" + +[tool.pytest.ini_options] +markers = [ + "integration: tests that hit the real Substack API. Skipped unless RUN_INTEGRATION_TESTS=1 is set in the environment.", +] diff --git a/scripts/extract_har.py b/scripts/extract_har.py new file mode 100644 index 0000000..562c2b5 --- /dev/null +++ b/scripts/extract_har.py @@ -0,0 +1,107 @@ +"""Extract a digestible summary of a HAR capture. + +Filters to substack.com / S3 / amazonaws hosts and prints, for each request: +URL, method, status, content-type, body sizes, and a truncated body preview. + +Usage: + python extract_har.py [--filter SUBSTRING] +""" + +from __future__ import annotations + +import argparse +import io +import json +import sys +from pathlib import Path + +# Force UTF-8 stdout on Windows. +sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8", errors="replace") + + +def short(text: str | None, limit: int = 600) -> str: + if text is None: + return "" + if len(text) <= limit: + return text + return text[:limit] + f"... [+{len(text) - limit} bytes]" + + +def header(headers, name): + name_l = name.lower() + for h in headers or []: + if h.get("name", "").lower() == name_l: + return h.get("value", "") + return "" + + +def main() -> int: + p = argparse.ArgumentParser() + p.add_argument("har") + p.add_argument("--filter", default="", help="substring to require in URL") + p.add_argument("--method", default="", help="only this HTTP method") + p.add_argument("--limit", type=int, default=0, help="max entries to print (0 = all)") + p.add_argument("--full-body", action="store_true", help="print full bodies (no truncation)") + args = p.parse_args() + + har = json.loads(Path(args.har).read_text(encoding="utf-8", errors="replace")) + entries = har["log"]["entries"] + matched = 0 + for i, e in enumerate(entries): + req = e["request"] + res = e["response"] + url = req["url"] + if args.filter and args.filter not in url: + continue + if args.method and req["method"].upper() != args.method.upper(): + continue + matched += 1 + print("=" * 100) + print(f"#{i} {req['method']} {url}") + print(f" status: {res['status']} {res.get('statusText','')}") + ct_req = header(req.get("headers"), "content-type") + ct_res = header(res.get("headers"), "content-type") + print(f" req content-type: {ct_req}") + print(f" res content-type: {ct_res}") + # Show interesting auth/csrf headers + for hname in ("x-csrf-token", "x-substack-publication-id", "x-amz-meta-uuid", + "x-amz-server-side-encryption", "x-amz-acl"): + v = header(req.get("headers"), hname) + if v: + print(f" req {hname}: {v}") + # Body sizes + pd = req.get("postData") or {} + body_text = pd.get("text", "") + body_size = req.get("bodySize", -1) + pd_mime = pd.get("mimeType", "") + print(f" req bodySize: {body_size}, postData.text len: {len(body_text)}, mime: {pd_mime}") + # Don't print bodies for binary mime types + is_binary = pd_mime.startswith(("audio/", "image/", "video/", "application/octet")) or "binary" in pd_mime + if body_text and not is_binary: + print(f" req body: {body_text if args.full_body else short(body_text)}") + elif body_text and is_binary: + print(f" req body: ") + else: + if pd.get("params"): + print(f" req postData.params: {pd['params']}") + # Useful response headers + for hname in ("etag", "x-amz-version-id", "location"): + v = header(res.get("headers"), hname) + if v: + print(f" res {hname}: {v}") + # Response body + content = res.get("content") or {} + rtext = content.get("text", "") + rsize = content.get("size", -1) + print(f" res size: {rsize}, content.text len: {len(rtext)}") + if rtext: + print(f" res body: {rtext if args.full_body else short(rtext)}") + if args.limit and matched >= args.limit: + break + + print(f"\n=== matched {matched} entries ===") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/verify_baseline.py b/scripts/verify_baseline.py new file mode 100644 index 0000000..dc7fdc0 --- /dev/null +++ b/scripts/verify_baseline.py @@ -0,0 +1,152 @@ +"""Baseline verification for python-substack against a sandbox publication. + +Exercises the FastMCP tool surface end-to-end: + 1. Authenticate + list publications + 2. Create a newsletter draft from markdown via the MCP tool + 3. Publish it (send=False so no email goes out) + 4. Delete the resulting post + +Reads .env from python-substack/.env. Prints PASS/FAIL per step and exits 0 only +if every step passed. +""" + +from __future__ import annotations + +import asyncio +import os +import sys +import time +import traceback +from pathlib import Path + +REPO = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(REPO)) + +from dotenv import load_dotenv + +load_dotenv(REPO / ".env") + +from substack_mcp.mcp_server import ( # noqa: E402 + get_api, + post_draft_from_markdown, + publish_draft, +) + + +def banner(msg: str) -> None: + print(f"\n=== {msg} ===", flush=True) + + +def step(label: str, ok: bool, detail: str = "") -> None: + tag = "PASS" if ok else "FAIL" + line = f"[{tag}] {label}" + if detail: + line += f" — {detail}" + print(line, flush=True) + + +async def main() -> int: + failures: list[str] = [] + created_post_id: int | None = None + api = None + + # --- Step 1: auth + list publications --- + banner("Step 1: authenticate and list publications") + try: + api = get_api() + pubs = api.get_user_publications() + names = [p.get("subdomain") for p in pubs] + step("authenticate via get_api()", True, f"current publication_url={api.publication_url}") + step("get_user_publications()", True, f"{len(pubs)} publications: {names}") + primary = api.get_user_primary_publication() + step("get_user_primary_publication()", True, f"subdomain={primary.get('subdomain')}") + except Exception as exc: + step("authenticate / list publications", False, repr(exc)) + traceback.print_exc() + failures.append("auth") + return 1 + + # --- Step 2: create draft from markdown via MCP tool --- + banner("Step 2: post_draft_from_markdown (MCP tool)") + ts = time.strftime("%Y-%m-%d %H:%M:%S") + title = f"[baseline test {ts}] python-substack verification" + markdown = ( + "# Baseline verification\n\n" + "This draft was created by `scripts/verify_baseline.py` to confirm the " + "FastMCP server functions correctly against the sandbox publication.\n\n" + "It should be deleted immediately by the same script. If you are reading " + "this in a published post, the cleanup step failed.\n" + ) + try: + result = await post_draft_from_markdown( + title=title, + markdown=markdown, + subtitle="Automated baseline verification — safe to delete", + audience="everyone", + write_comment_permissions="everyone", + prepublish=False, + publish=False, + ) + draft = result.get("draft") or {} + created_post_id = draft.get("id") + if not created_post_id: + raise RuntimeError(f"no draft id in response: {result!r}") + step( + "post_draft_from_markdown", + True, + f"draft_id={created_post_id} type={draft.get('type')}", + ) + except Exception as exc: + step("post_draft_from_markdown", False, repr(exc)) + traceback.print_exc() + failures.append("create_draft") + + # --- Step 3: publish (send=False so no email goes out) --- + if created_post_id is not None: + banner("Step 3: publish_draft (send=False, share_automatically=False)") + try: + pub_result = await publish_draft( + draft_id=created_post_id, + send=False, + share_automatically=False, + ) + published_id = (pub_result or {}).get("id") or created_post_id + step( + "publish_draft", + True, + f"published id={published_id} is_published={pub_result.get('is_published')}", + ) + # publish typically returns the post with a new id distinct from the draft id; + # keep the original draft id for cleanup, but note both. + created_post_id = published_id + except Exception as exc: + step("publish_draft", False, repr(exc)) + traceback.print_exc() + failures.append("publish") + + # --- Step 4: cleanup --- + if created_post_id is not None: + banner("Step 4: delete published draft") + try: + del_result = api.delete_draft(created_post_id) + step("delete_draft", True, f"response={del_result!r}") + except Exception as exc: + step("delete_draft", False, repr(exc)) + traceback.print_exc() + failures.append("delete") + print( + f"\n!! NOTE: Post id={created_post_id} may still exist on the publication. " + "Delete it manually from the Substack dashboard.", + flush=True, + ) + + banner("Summary") + if failures: + print(f"FAILED steps: {failures}", flush=True) + return 1 + print("All steps passed.", flush=True) + return 0 + + +if __name__ == "__main__": + sys.exit(asyncio.run(main())) diff --git a/substack/__init__.py b/substack/__init__.py index 4963dbe..e079ff6 100644 --- a/substack/__init__.py +++ b/substack/__init__.py @@ -9,3 +9,4 @@ __description__ = "A Python wrapper around the Substack API" from .api import Api +from .podcast import PodcastPost diff --git a/substack/api.py b/substack/api.py index 676d86e..a1f7d65 100644 --- a/substack/api.py +++ b/substack/api.py @@ -8,10 +8,12 @@ import json import logging import os +import time from datetime import datetime from urllib.parse import urljoin, unquote import requests +from mutagen.mp3 import MP3 from substack.exceptions import SubstackAPIException, SubstackRequestException @@ -514,6 +516,110 @@ def get_image(self, image: str): data={"image": image}, ) return Api._handle_response(response=response) + + def upload_podcast_audio( + self, + file_path: str, + draft_id: int, + poll_timeout_seconds: float = 120, + poll_interval_seconds: float = 3, + ) -> dict: + """ + + Upload an audio file for a podcast draft and wait until transcoding completes. + + Performs the full sequence: initiate upload, PUT raw bytes to the + pre-signed S3 URL(s) returned by init, trigger transcode (which + finalizes the multipart upload and starts the async transcode job), + then poll the media object until its state is ``"transcoded"``. + Duration is computed locally via mutagen and sent with the transcode + call. + + Args: + file_path: Local path to the audio file (typically an MP3). + draft_id: ID of the podcast draft this upload is scoped to. + Substack requires a ``post_id`` on the init call. + poll_timeout_seconds: Maximum time to wait for transcoding to + complete before raising. + poll_interval_seconds: Delay between successive state polls. + + Returns: + The final media object dict (``state == "transcoded"``). Its + ``id`` field is the audio UUID to attach to a draft as + ``draft_podcast_upload_id``. + + Raises: + SubstackAPIException: on non-2xx from any Substack endpoint + (init, transcode, or poll). + SubstackRequestException: on S3 PUT failure or transcode timeout. + """ + path = os.fspath(file_path) + file_size = os.path.getsize(path) + file_name = os.path.basename(path) + duration_seconds = MP3(path).info.length + + # 1. Initiate upload -- returns the media UUID, multipart upload id, + # and one or more pre-signed S3 PUT URLs (today always 1 URL). + init_response = self._session.post( + f"{self.publication_url}/audio/upload", + params={ + "filetype": "audio/mpeg", + "fileSize": file_size, + "fileName": file_name, + "post_id": draft_id, + }, + ) + init_data = Api._handle_response(response=init_response) + upload_id = init_data["mediaUpload"]["id"] + multipart_upload_id = init_data["multipartUploadId"] + upload_urls = init_data["multipartUploadUrls"] + + # 2. PUT bytes to each part URL. S3 returns no body; the ETag header + # (including its surrounding double-quotes) is what we need. + # This is the only HTTP call in the codebase that does not route + # through _handle_response -- see docs/STYLE.md section 6. + with open(path, "rb") as f: + audio_bytes = f.read() + etags = [] + for url in upload_urls: + s3_response = self._session.put(url, data=audio_bytes) + if not s3_response.ok: + body_preview = (s3_response.text or "")[:500] + raise SubstackRequestException( + f"S3 upload failed: {s3_response.status_code} {body_preview}" + ) + etags.append(s3_response.headers["ETag"]) + + # 3. Trigger transcode. Returns immediately with state "uploaded"; + # actual transcoding happens asynchronously. Trust 2xx and proceed + # to polling; the poll loop is the source of truth on transcode + # completion. + transcode_response = self._session.post( + f"{self.publication_url}/audio/upload/{upload_id}/transcode", + json={ + "duration": duration_seconds, + "multipart_upload_id": multipart_upload_id, + "multipart_upload_etags": etags, + }, + ) + Api._handle_response(response=transcode_response) + + # 4. Poll until state == "transcoded" (or raise on timeout). + deadline = time.monotonic() + poll_timeout_seconds + while True: + poll_response = self._session.get( + f"{self.publication_url}/audio/upload/{upload_id}" + ) + media = Api._handle_response(response=poll_response) + if media.get("state") == "transcoded": + return media + if time.monotonic() >= deadline: + raise SubstackRequestException( + f"Audio transcode did not complete within " + f"{poll_timeout_seconds}s (uuid={upload_id}, " + f"last state={media.get('state')!r})" + ) + time.sleep(poll_interval_seconds) def add_tags_to_post(self, post_id: int, tag_names: list) -> dict: """ diff --git a/substack/podcast.py b/substack/podcast.py new file mode 100644 index 0000000..c8e2c83 --- /dev/null +++ b/substack/podcast.py @@ -0,0 +1,193 @@ +""" + +Podcast Post Utilities + +""" + +import json + +__all__ = ["PodcastPost"] + +from substack.exceptions import SectionNotExistsException +from substack.post import Post + + +class PodcastPost: + """ + + In-memory state for a Substack podcast draft. + + A podcast draft carries two distinct ProseMirror documents: + + * ``draft_body`` -- the post-page body (same field a newsletter Post uses). + * ``podcast_description`` -- the show notes shown alongside the audio + player. + + Both are kept as Python dicts internally and JSON-encoded by + :meth:`get_draft`. Both are independently optional. + + Like :class:`substack.post.Post`, ``vars(self)`` is the serialization + layer: every public attribute name matches the Substack API field name + exactly. Unlike ``Post``, :meth:`get_draft` is non-destructive (it + shallow-copies the instance dict before encoding) so the same instance + can be reused across the create / PUT-updates / publish lifecycle. + + """ + + def __init__( + self, + title: str, + subtitle: str, + user_id, + audience: str = None, + write_comment_permissions: str = None, + ): + """ + + Args: + title: + subtitle: + user_id: + audience: possible values: everyone, only_paid, founding, only_free + write_comment_permissions: none, only_paid, everyone + """ + # Fields shared with a newsletter draft. + self.draft_title = title + self.draft_subtitle = subtitle + self.draft_body = {"type": "doc", "content": []} + self.draft_bylines = [{"id": int(user_id), "is_guest": False}] + self.audience = audience if audience is not None else "everyone" + self.draft_section_id = None + self.section_chosen = True + + if write_comment_permissions is not None: + self.write_comment_permissions = write_comment_permissions + else: + self.write_comment_permissions = self.audience + + # Podcast-only fields. Names match the Substack API exactly. + self.type = "podcast" + self.draft_podcast_upload_id = None + self.draft_podcast_url = None + self.draft_podcast_duration = None + self.podcast_description = {"type": "doc", "content": []} + + # Optimistic-concurrency token. None on first create; callers should + # set it from the previous draft response's ``draft_updated_at`` before + # each subsequent PUT. See docs/PODCAST_ARCHITECTURE.md section 7. + self.last_updated_at = None + + def set_section(self, name: str, sections: list): + """ + + Look up a section by name and store its id as ``draft_section_id``. + + Args: + name: Section name as configured on the publication. + sections: List of section dicts as returned by + :meth:`substack.api.Api.get_sections`. + + Returns: + Self for method chaining. + """ + section = [s for s in sections if s.get("name") == name] + if len(section) != 1: + raise SectionNotExistsException(name) + self.draft_section_id = section[0].get("id") + return self + + def set_audio(self, upload_id: str, duration_seconds: float = None): + """ + + Attach a previously-uploaded audio file to the draft. + + Args: + upload_id: The audio UUID returned by + :meth:`substack.api.Api.upload_podcast_audio` (the + ``id`` field of the transcoded media object). + duration_seconds: Optional. When provided, stored as + ``draft_podcast_duration``. Substack's player can derive + duration from the transcoded media, so this is informational. + + Returns: + Self for method chaining. + """ + self.draft_podcast_upload_id = upload_id + if duration_seconds is not None: + self.draft_podcast_duration = duration_seconds + return self + + def set_body_from_markdown(self, markdown: str, api=None): + """ + + Replace the post-page body with Markdown parsed into ProseMirror. + + Internally constructs a throwaway :class:`substack.post.Post`, runs + its :meth:`~substack.post.Post.from_markdown`, then steals the + resulting ``draft_body`` dict. Any prior body content is replaced. + + Args: + markdown: Markdown source for the post-page body. + api: Optional :class:`substack.api.Api` instance. Forwarded to + ``Post.from_markdown`` so any local image paths referenced + in the Markdown are uploaded via ``api.get_image`` and + rewritten to their CDN URLs. + + Returns: + Self for method chaining. + """ + tmp = Post(title="", subtitle="", user_id=self.draft_bylines[0]["id"]) + tmp.from_markdown(markdown, api=api) + self.draft_body = tmp.draft_body + return self + + def set_show_notes_from_markdown(self, markdown: str, api=None): + """ + + Replace the show notes (``podcast_description``) with Markdown + parsed into ProseMirror. + + Internally constructs a throwaway :class:`substack.post.Post`, runs + its :meth:`~substack.post.Post.from_markdown`, then steals the + resulting ``draft_body`` dict into ``podcast_description``. Any + prior show notes are replaced. + + Args: + markdown: Markdown source for the show notes. + api: Optional :class:`substack.api.Api` instance. Forwarded to + ``Post.from_markdown`` so any local image paths referenced + in the Markdown are uploaded via ``api.get_image`` and + rewritten to their CDN URLs. + + Returns: + Self for method chaining. + """ + tmp = Post(title="", subtitle="", user_id=self.draft_bylines[0]["id"]) + tmp.from_markdown(markdown, api=api) + self.podcast_description = tmp.draft_body + return self + + def get_draft(self) -> dict: + """ + + Return the draft payload as a JSON-ready dict. + + Both ProseMirror docs (``draft_body`` and ``podcast_description``) + are JSON-encoded to strings, matching the wire format Substack + expects. ``last_updated_at`` is omitted when None (the create call + must not carry the field; PUTs after the first response should). + + Unlike :meth:`substack.post.Post.get_draft`, this method does not + mutate ``self``. It returns a shallow copy of ``vars(self)`` so the + same instance can be reused across create / PUT / publish. + + Returns: + dict ready to pass to :meth:`substack.api.Api.post_draft` or to + splat into :meth:`substack.api.Api.put_draft` as kwargs. + """ + out = dict(vars(self)) + out["draft_body"] = json.dumps(out["draft_body"]) + out["podcast_description"] = json.dumps(out["podcast_description"]) + if out.get("last_updated_at") is None: + out.pop("last_updated_at", None) + return out diff --git a/substack_mcp/mcp_server.py b/substack_mcp/mcp_server.py index df0ea8c..521802c 100644 --- a/substack_mcp/mcp_server.py +++ b/substack_mcp/mcp_server.py @@ -11,6 +11,7 @@ from mcp.server.fastmcp import FastMCP from substack.api import Api +from substack.podcast import PodcastPost from substack.post import Post if load_dotenv is not None: @@ -285,6 +286,202 @@ async def publish_draft( ) +@mcp.tool() +async def upload_podcast_audio( + file_path: str, + draft_id: int, +) -> Dict[str, Any]: + """Upload a podcast audio file (typically an MP3) and wait for transcoding. + + Performs the full Substack upload sequence: initiate, PUT raw bytes to + the pre-signed S3 URL(s), trigger transcode, and poll the media object + until its state is ``"transcoded"``. Duration is computed locally via + mutagen and sent with the transcode call. + + Args: + file_path: Local path to the audio file on the machine running this + MCP server. + draft_id: ID of the podcast draft this audio is being attached to. + Substack scopes the upload to a ``post_id`` at init time, so the + draft must already exist. + + Returns: + The final media object dict with ``state == "transcoded"``. The + ``id`` field is the audio UUID; pass it as ``audio_upload_id`` / + ``draft_podcast_upload_id`` to attach the audio to a draft. + + Examples: + ```python + from substack_mcp.mcp_server import ( + post_podcast_draft_from_markdown, + upload_podcast_audio, + put_draft, + ) + + # Create a podcast shell first so we have a draft_id to scope the + # upload to. + d = await post_podcast_draft_from_markdown(title='Episode 1') + draft_id = d['draft']['id'] + + media = await upload_podcast_audio( + file_path='/abs/path/episode-01.mp3', + draft_id=draft_id, + ) + # media['id'] is the upload UUID; attach it to the draft: + await put_draft(draft_id, {'draft_podcast_upload_id': media['id']}) + ``` + """ + client = get_api() + return client.upload_podcast_audio(file_path=file_path, draft_id=draft_id) + + +@mcp.tool() +async def post_podcast_draft_from_markdown( + title: str, + subtitle: Optional[str] = "", + show_notes_markdown: Optional[str] = None, + post_body_markdown: Optional[str] = None, + audio_file_path: Optional[str] = None, + audience: str = "everyone", + write_comment_permissions: Optional[str] = None, + draft_section_id: Optional[int] = None, + tags: Optional[Any] = None, + prepublish: bool = False, + publish: bool = False, + send: bool = True, + share_automatically: bool = False, +) -> Dict[str, Any]: + """Create (and optionally publish) a Substack podcast draft end-to-end. + + Composes the full podcast publishing flow: build a ``PodcastPost`` from + the supplied fields, create the draft, optionally upload + attach audio, + optionally tag, optionally prepublish, optionally publish. Each step is + independently opt-in and surfaced separately in the return value. + + Show notes (``podcast_description``) and the post-page body + (``draft_body``) are two distinct ProseMirror documents on a podcast + draft; both are independently optional. + + Args: + title: Episode title. + subtitle: Optional subtitle. + show_notes_markdown: Markdown for the show notes + (``podcast_description``). Renders alongside the audio player. + post_body_markdown: Markdown for the post-page body (``draft_body``). + Same field a newsletter post uses. + audio_file_path: Local path to the episode audio file (MP3). When + provided, the audio is uploaded via :func:`upload_podcast_audio` + and attached to the draft. + audience: One of ``everyone``, ``only_paid``, ``founding``, ``only_free``. + write_comment_permissions: One of ``none``, ``only_paid``, ``everyone``. + Defaults to ``audience`` when omitted. + draft_section_id: Optional section ID. + tags: Tag or list of tags to attach to the post. + prepublish: If true, calls ``prepublish_draft`` (server-side validation). + publish: If true, calls ``publish_draft``. + send: Passed to ``publish_draft`` -- newsletter email delivery on + publish. Ignored when ``publish`` is false. + share_automatically: Passed to ``publish_draft``. Substack's podcast + publish endpoint ignores this key, but it is included for + symmetry with :func:`post_draft_from_markdown`. + + Returns: + Dict with one key per composed step: + + * ``draft``: latest draft / post object (post-PUT if audio attached, + otherwise the create response; not replaced by the publish response). + * ``upload``: transcoded media object, or ``None`` if no audio uploaded. + * ``tags``: result of ``add_tags_to_post``, or ``None`` if no tags. + * ``prepublish``: result of ``prepublish_draft``, or ``None`` if skipped. + * ``publish``: result of ``publish_draft``, or ``None`` if skipped. + + Examples: + Minimal draft, no audio, stop at draft stage: + + ```python + from substack_mcp.mcp_server import post_podcast_draft_from_markdown + + result = await post_podcast_draft_from_markdown( + title='Pilot episode', + subtitle='Where it all begins', + show_notes_markdown='## Topics\\n\\n- Topic one\\n- Topic two', + ) + draft_id = result['draft']['id'] + ``` + + Full end-to-end publish (no email, safe for sandbox): + + ```python + result = await post_podcast_draft_from_markdown( + title='Episode 7: Async I/O', + subtitle='What asyncio buys you, and what it costs', + show_notes_markdown='## Show notes\\n\\n- Event loops\\n- Backpressure', + post_body_markdown='Full episode text mirror for the post page.', + audio_file_path='/abs/path/episode-07.mp3', + tags=['python', 'async'], + prepublish=True, + publish=True, + send=False, + ) + assert result['publish']['is_published'] + ``` + """ + client = get_api() + user_id = client.get_user_id() + + pod = PodcastPost( + title=title, + subtitle=subtitle or "", + user_id=user_id, + audience=audience, + write_comment_permissions=write_comment_permissions, + ) + + if show_notes_markdown: + pod.set_show_notes_from_markdown(show_notes_markdown, api=client) + if post_body_markdown: + pod.set_body_from_markdown(post_body_markdown, api=client) + if draft_section_id is not None: + pod.draft_section_id = draft_section_id + + draft = client.post_draft(pod.get_draft()) + draft_id = draft["id"] + + upload_result: Optional[Dict[str, Any]] = None + if audio_file_path: + upload_result = client.upload_podcast_audio( + file_path=audio_file_path, draft_id=draft_id + ) + pod.set_audio( + upload_result["id"], duration_seconds=upload_result.get("duration") + ) + pod.last_updated_at = draft.get("draft_updated_at") + draft = client.put_draft(draft_id, **pod.get_draft()) + + tags_list = _normalize_tags(tags) + tags_result = None + if tags_list: + tags_result = client.add_tags_to_post(draft_id, tags_list) + + prepublish_result = None + if prepublish: + prepublish_result = client.prepublish_draft(draft_id) + + publish_result = None + if publish: + publish_result = client.publish_draft( + draft_id, send=send, share_automatically=share_automatically + ) + + return { + "draft": draft, + "upload": upload_result, + "tags": tags_result, + "prepublish": prepublish_result, + "publish": publish_result, + } + + def main() -> None: mcp.run(transport="stdio") diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..5a140ad --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,44 @@ +"""Shared pytest fixtures. + +The `tiny_mp3_path` fixture provides a small, deterministic, valid MP3 file +for integration tests. It is generated once via lameenc (pure-Python LAME +bindings, no ffmpeg/system dependency) and cached to tests/fixtures/. +""" + +from __future__ import annotations + +import struct +from pathlib import Path + +import pytest + +FIXTURES_DIR = Path(__file__).resolve().parent / "fixtures" +TINY_MP3 = FIXTURES_DIR / "tiny_silent_2s.mp3" + + +def _generate_tiny_mp3(out_path: Path) -> None: + """Encode 2 seconds of silence as a 64 kbps mono MP3 (~16 KB).""" + import lameenc + + sample_rate = 44100 + n_samples = sample_rate * 2 # 2 seconds + # 16-bit signed little-endian PCM, all zeros. + silent_pcm = struct.pack("<" + "h" * n_samples, *([0] * n_samples)) + + encoder = lameenc.Encoder() + encoder.set_bit_rate(64) + encoder.set_in_sample_rate(sample_rate) + encoder.set_channels(1) + encoder.set_quality(2) + mp3 = encoder.encode(silent_pcm) + encoder.flush() + + out_path.parent.mkdir(parents=True, exist_ok=True) + out_path.write_bytes(mp3) + + +@pytest.fixture(scope="session") +def tiny_mp3_path() -> Path: + """Return the path to a small (~16 KB) valid silent MP3, generating it once.""" + if not TINY_MP3.exists(): + _generate_tiny_mp3(TINY_MP3) + return TINY_MP3 diff --git a/tests/substack/test_audio_upload.py b/tests/substack/test_audio_upload.py new file mode 100644 index 0000000..2dd5cc8 --- /dev/null +++ b/tests/substack/test_audio_upload.py @@ -0,0 +1,484 @@ +"""Tests for Api.upload_podcast_audio. + +Deliberate departure from the no-mock convention in STYLE.md §5: HTTP integration +logic for the multi-step audio upload (init -> S3 PUT -> transcode -> poll) cannot +be exercised without mocking the HTTP layer. Mocks target the underlying +`requests.Session` method calls (`post`, `put`, `get`) since `_handle_response` +inspects `response.json()` / `response.status_code` / `response.headers`. + +All fixture payload shapes are derived from real HAR captures; see +docs/PODCAST_API_CONTRACT.md for the canonical wire-level reference. +""" + +import json +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest + +from substack.api import Api +from substack.exceptions import SubstackAPIException, SubstackRequestException + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +SAMPLES_DIR = Path(__file__).resolve().parents[2].parent / "samples" +SMALL_SAMPLE = SAMPLES_DIR / "sample-speech-10m.mp3" + + +def make_api(): + """Construct an Api instance without touching the network. + + Api.__init__ authenticates and fetches the publication; tests can't run + through that. We bypass it via __new__ and set the minimum attributes the + upload code path actually touches. + """ + api = Api.__new__(Api) + api._session = MagicMock() + api.publication_url = "https://test.substack.com/api/v1" + api.base_url = "https://substack.com/api/v1" + return api + + +def fake_response(status=200, json_data=None, headers=None, text=""): + """Build a MagicMock that quacks like requests.Response.""" + r = MagicMock() + r.status_code = status + r.ok = 200 <= status < 300 + r.headers = headers or {} + r.text = text + if json_data is None: + r.json.side_effect = ValueError("no json") + else: + r.json.return_value = json_data + return r + + +# Fixture payloads modeled after captures/substack_capture_small.har entries +# #12, #14, #15, #16, #19. UUIDs and IDs are made-up but shape-faithful. + +DRAFT_ID = 199281706 +AUDIO_UUID = "198f70da-d9b5-42d7-8241-8ea327b09101" +MULTIPART_UPLOAD_ID = "56fy3HuHONW7_f6sUFFYGq5A0Ez76Bw_test_multipart_upload_id" +S3_PUT_URL = ( + "https://substack-video.s3-accelerate.amazonaws.com/video_upload/post/" + f"{DRAFT_ID}/{AUDIO_UUID}/original" + "?X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Signature=fake&partNumber=1" + f"&uploadId={MULTIPART_UPLOAD_ID}" +) +S3_ETAG = '"bf25ca5cb165f948f3c26faef0fbedba"' + +INIT_RESPONSE_BODY = { + "mediaUpload": { + "user_id": 253249334, + "name": "sample-speech-10m.mp3", + "publication_id": 9222054, + "post_id": DRAFT_ID, + "state": "created", + "media_type": "audio", + "is_mux": False, + "primary_file_size": 9601402, + "id": AUDIO_UUID, + "parts": [], + "multipart_upload_id": MULTIPART_UPLOAD_ID, + }, + "multipartUploadId": MULTIPART_UPLOAD_ID, + "multipartUploadUrls": [S3_PUT_URL], +} + +TRANSCODE_RESPONSE_BODY = { + "id": AUDIO_UUID, + "name": "sample-speech-10m.mp3", + "state": "uploaded", + "post_id": DRAFT_ID, + "duration": 600.058776, + "media_type": "audio", + "primary_file_size": 9601402, + "is_mux": False, + "explicit": False, +} + +POLL_UPLOADED_BODY = { + "id": AUDIO_UUID, + "state": "uploaded", + "duration": 600.0588, + "primary_file_size": 9601402, + "media_type": "audio", +} + +POLL_TRANSCODED_BODY = { + "id": AUDIO_UUID, + "state": "transcoded", + "duration": 600.0588, + "primary_file_size": 7201063, # smaller after transcode + "media_type": "audio", +} + +FAKE_DURATION_SECONDS = 600.032653 + + +def configure_happy_path(api, *, poll_uploaded_count=1): + """Wire api._session to return the canonical success sequence. + + Order of responses: + POST /audio/upload -> INIT_RESPONSE_BODY + PUT {S3 URL} -> 200 with ETag header + POST /audio/upload/{uuid}/transcode -> TRANSCODE_RESPONSE_BODY + GET /audio/upload/{uuid} -> "uploaded" (x poll_uploaded_count) then "transcoded" + """ + api._session.post.side_effect = [ + fake_response(200, json_data=INIT_RESPONSE_BODY), + fake_response(200, json_data=TRANSCODE_RESPONSE_BODY), + ] + api._session.put.side_effect = [ + fake_response(200, headers={"ETag": S3_ETAG}), + ] + api._session.get.side_effect = [ + fake_response(200, json_data=POLL_UPLOADED_BODY) + for _ in range(poll_uploaded_count) + ] + [ + fake_response(200, json_data=POLL_TRANSCODED_BODY), + ] + + +def call_upload(api, **overrides): + """Invoke the unit under test with sleep + mutagen patched out. + + `time.sleep` is patched so polling tests don't actually wait. + `MP3` from mutagen is patched to return a deterministic duration so tests + don't depend on a real audio file (one separate test exercises mutagen). + """ + kwargs = {"file_path": str(SMALL_SAMPLE), "draft_id": DRAFT_ID} + kwargs.update(overrides) + with patch("substack.api.time.sleep"), patch("substack.api.MP3") as mp3: + mp3.return_value.info.length = FAKE_DURATION_SECONDS + return api.upload_podcast_audio(**kwargs) + + +# --------------------------------------------------------------------------- +# TestUploadPodcastAudioInit +# --------------------------------------------------------------------------- + + +class TestUploadPodcastAudioInit: + """The init call: POST /audio/upload with query params.""" + + def test_init_hits_correct_endpoint(self): + api = make_api() + configure_happy_path(api) + call_upload(api) + init_url = api._session.post.call_args_list[0][0][0] + assert init_url == "https://test.substack.com/api/v1/audio/upload" + + def test_init_sends_filename_filesize_filetype_postid_params(self): + api = make_api() + configure_happy_path(api) + call_upload(api) + init_call = api._session.post.call_args_list[0] + params = init_call.kwargs.get("params") or init_call[1].get("params") + assert params["filetype"] == "audio/mpeg" + assert params["fileSize"] == SMALL_SAMPLE.stat().st_size + assert params["fileName"] == "sample-speech-10m.mp3" + assert params["post_id"] == DRAFT_ID + + def test_init_sends_no_body(self): + api = make_api() + configure_happy_path(api) + call_upload(api) + init_call = api._session.post.call_args_list[0] + assert init_call.kwargs.get("json") is None + assert init_call.kwargs.get("data") is None + + +# --------------------------------------------------------------------------- +# TestUploadPodcastAudioS3Put +# --------------------------------------------------------------------------- + + +class TestUploadPodcastAudioS3Put: + """The S3 PUT: uses the pre-signed URL from init, sends raw bytes.""" + + def test_s3_put_uses_url_from_init_response(self): + api = make_api() + configure_happy_path(api) + call_upload(api) + put_call = api._session.put.call_args_list[0] + assert put_call[0][0] == S3_PUT_URL + + def test_s3_put_sends_raw_audio_bytes(self): + api = make_api() + configure_happy_path(api) + call_upload(api) + put_call = api._session.put.call_args_list[0] + # Either positional `data=` or kwarg — both acceptable; tolerate both. + body = put_call.kwargs.get("data") + if body is None: + # Try positional + body = put_call[0][1] if len(put_call[0]) > 1 else None + assert body is not None + # Body should be the bytes of the source file (or a streaming handle). + expected_bytes = SMALL_SAMPLE.read_bytes() + if hasattr(body, "read"): + actual = body.read() + else: + actual = body + assert actual == expected_bytes + + def test_s3_put_iterates_all_multipart_upload_urls(self): + """Defensive: today multipartUploadUrls is always length 1, but the + implementation must loop so a future multi-part upload doesn't silently + drop bytes.""" + api = make_api() + # Hand-craft an init response with two URLs to prove the loop is real. + url2 = S3_PUT_URL.replace("partNumber=1", "partNumber=2") + etag2 = '"deadbeefdeadbeefdeadbeefdeadbeef"' + multi_init = { + **INIT_RESPONSE_BODY, + "multipartUploadUrls": [S3_PUT_URL, url2], + } + multi_transcode_request_etags = [S3_ETAG, etag2] + api._session.post.side_effect = [ + fake_response(200, json_data=multi_init), + fake_response(200, json_data=TRANSCODE_RESPONSE_BODY), + ] + api._session.put.side_effect = [ + fake_response(200, headers={"ETag": S3_ETAG}), + fake_response(200, headers={"ETag": etag2}), + ] + api._session.get.side_effect = [ + fake_response(200, json_data=POLL_TRANSCODED_BODY), + ] + call_upload(api) + assert api._session.put.call_count == 2 + # Both etags should appear in the transcode payload, in order. + transcode_call = api._session.post.call_args_list[1] + transcode_body = transcode_call.kwargs.get("json") + assert transcode_body["multipart_upload_etags"] == multi_transcode_request_etags + + +# --------------------------------------------------------------------------- +# TestUploadPodcastAudioTranscode +# --------------------------------------------------------------------------- + + +class TestUploadPodcastAudioTranscode: + """The transcode call: POST with duration + multipart upload id + etags.""" + + def test_transcode_hits_correct_endpoint(self): + api = make_api() + configure_happy_path(api) + call_upload(api) + transcode_url = api._session.post.call_args_list[1][0][0] + assert transcode_url == ( + f"https://test.substack.com/api/v1/audio/upload/{AUDIO_UUID}/transcode" + ) + + def test_transcode_payload_carries_duration_from_mutagen(self): + api = make_api() + configure_happy_path(api) + call_upload(api) + transcode_body = api._session.post.call_args_list[1].kwargs.get("json") + assert transcode_body["duration"] == FAKE_DURATION_SECONDS + + def test_transcode_payload_carries_multipart_upload_id(self): + api = make_api() + configure_happy_path(api) + call_upload(api) + transcode_body = api._session.post.call_args_list[1].kwargs.get("json") + assert transcode_body["multipart_upload_id"] == MULTIPART_UPLOAD_ID + + def test_transcode_payload_carries_etag_with_quotes(self): + """The S3 ETag header includes surrounding double-quotes that the + transcode endpoint expects to be preserved verbatim.""" + api = make_api() + configure_happy_path(api) + call_upload(api) + transcode_body = api._session.post.call_args_list[1].kwargs.get("json") + assert transcode_body["multipart_upload_etags"] == [S3_ETAG] + # Defensive: the quotes are still there, not stripped. + assert transcode_body["multipart_upload_etags"][0].startswith('"') + assert transcode_body["multipart_upload_etags"][0].endswith('"') + + +# --------------------------------------------------------------------------- +# TestUploadPodcastAudioPolling +# --------------------------------------------------------------------------- + + +class TestUploadPodcastAudioPolling: + """Polling for state transition from 'uploaded' to 'transcoded'.""" + + def test_polling_endpoint(self): + api = make_api() + configure_happy_path(api) + call_upload(api) + poll_url = api._session.get.call_args_list[0][0][0] + assert poll_url == ( + f"https://test.substack.com/api/v1/audio/upload/{AUDIO_UUID}" + ) + + def test_polling_stops_immediately_when_already_transcoded(self): + """If the first poll returns 'transcoded', no further polls happen.""" + api = make_api() + configure_happy_path(api, poll_uploaded_count=0) + call_upload(api) + assert api._session.get.call_count == 1 + + def test_polling_continues_until_transcoded(self): + """When polls return 'uploaded' first, polling continues.""" + api = make_api() + configure_happy_path(api, poll_uploaded_count=3) + call_upload(api) + assert api._session.get.call_count == 4 # 3 'uploaded' + 1 'transcoded' + + def test_polling_returns_final_transcoded_media_object(self): + api = make_api() + configure_happy_path(api, poll_uploaded_count=2) + result = call_upload(api) + assert result["state"] == "transcoded" + assert result["id"] == AUDIO_UUID + # The smaller transcoded file size is what comes back. + assert result["primary_file_size"] == 7201063 + + +# --------------------------------------------------------------------------- +# TestUploadPodcastAudioErrors +# --------------------------------------------------------------------------- + + +class TestUploadPodcastAudioErrors: + """Error paths — each step's failure raises a meaningful exception.""" + + def test_init_non_2xx_raises_substack_api_exception(self): + api = make_api() + api._session.post.side_effect = [ + fake_response(500, text='{"errors": [{"msg": "internal"}]}'), + ] + with pytest.raises(SubstackAPIException) as exc: + call_upload(api) + assert exc.value.status_code == 500 + + def test_s3_put_non_2xx_raises_substack_request_exception(self): + api = make_api() + api._session.post.side_effect = [ + fake_response(200, json_data=INIT_RESPONSE_BODY), + ] + api._session.put.side_effect = [ + fake_response(403, text="AccessDenied"), + ] + with pytest.raises(SubstackRequestException) as exc: + call_upload(api) + # The error message should mention S3 / the status code so the user + # can tell upload failures apart from Substack API failures. + msg = str(exc.value) + assert "S3" in msg or "403" in msg + + def test_transcode_non_2xx_raises_substack_api_exception(self): + api = make_api() + api._session.post.side_effect = [ + fake_response(200, json_data=INIT_RESPONSE_BODY), + fake_response(400, text='{"errors": [{"msg": "bad etag"}]}'), + ] + api._session.put.side_effect = [ + fake_response(200, headers={"ETag": S3_ETAG}), + ] + with pytest.raises(SubstackAPIException) as exc: + call_upload(api) + assert exc.value.status_code == 400 + + def test_polling_times_out_raises_substack_request_exception(self): + """If state never reaches 'transcoded' within the timeout window, + raise rather than spin forever. + + Note: does NOT patch time.sleep — the test relies on real wall clock + to advance past the (tiny) timeout. A callable side_effect avoids + StopIteration when the loop polls many times. + """ + api = make_api() + api._session.post.side_effect = [ + fake_response(200, json_data=INIT_RESPONSE_BODY), + fake_response(200, json_data=TRANSCODE_RESPONSE_BODY), + ] + api._session.put.side_effect = [ + fake_response(200, headers={"ETag": S3_ETAG}), + ] + api._session.get.side_effect = ( + lambda *a, **kw: fake_response(200, json_data=POLL_UPLOADED_BODY) + ) + with patch("substack.api.MP3") as mp3: + mp3.return_value.info.length = FAKE_DURATION_SECONDS + with pytest.raises(SubstackRequestException) as exc: + api.upload_podcast_audio( + file_path=str(SMALL_SAMPLE), + draft_id=DRAFT_ID, + poll_timeout_seconds=0.3, + poll_interval_seconds=0.05, + ) + assert "transcode" in str(exc.value).lower() or "timeout" in str(exc.value).lower() + + +# --------------------------------------------------------------------------- +# TestMutagenDuration +# --------------------------------------------------------------------------- + + +class TestMutagenDuration: + """One real-fixture test confirming mutagen reads MP3 duration. + + All other tests mock MP3 out — this test guards the integration so a + mutagen version bump or API change is caught locally. + """ + + @pytest.mark.skipif( + not SMALL_SAMPLE.exists(), + reason=f"sample MP3 not present at {SMALL_SAMPLE}", + ) + def test_mutagen_reads_sample_mp3_duration(self): + from mutagen.mp3 import MP3 + info = MP3(str(SMALL_SAMPLE)).info + # Sample is ~10 minutes; sanity bounds, not exact equality. + assert 590 < info.length < 610 + + @pytest.mark.skipif( + not SMALL_SAMPLE.exists(), + reason=f"sample MP3 not present at {SMALL_SAMPLE}", + ) + def test_upload_uses_real_mutagen_when_not_patched(self): + """End-to-end with no MP3 patch: confirms api.py imports mutagen + correctly and feeds the real duration into the transcode payload.""" + api = make_api() + configure_happy_path(api) + # No MP3 patch this time; only sleep. + with patch("substack.api.time.sleep"): + api.upload_podcast_audio(file_path=str(SMALL_SAMPLE), draft_id=DRAFT_ID) + transcode_body = api._session.post.call_args_list[1].kwargs.get("json") + # Real duration of the sample, with a small tolerance. + assert 590 < transcode_body["duration"] < 610 + + +# --------------------------------------------------------------------------- +# TestUploadPodcastAudioReturnShape +# --------------------------------------------------------------------------- + + +class TestUploadPodcastAudioReturnShape: + """The public method returns the final media object verbatim.""" + + def test_returns_dict_with_uuid_state_duration(self): + api = make_api() + configure_happy_path(api) + result = call_upload(api) + assert isinstance(result, dict) + assert result["id"] == AUDIO_UUID + assert result["state"] == "transcoded" + assert "duration" in result + + def test_does_not_mutate_response_body(self): + """Caller should receive the parsed JSON without wrapper keys added.""" + api = make_api() + configure_happy_path(api) + result = call_upload(api) + # No invented fields — should match what the API returned. + assert set(result.keys()) == set(POLL_TRANSCODED_BODY.keys()) diff --git a/tests/substack/test_audio_upload_integration.py b/tests/substack/test_audio_upload_integration.py new file mode 100644 index 0000000..7988f1e --- /dev/null +++ b/tests/substack/test_audio_upload_integration.py @@ -0,0 +1,99 @@ +"""Integration test for Api.upload_podcast_audio against a real Substack sandbox. + +Gated on the environment variable RUN_INTEGRATION_TESTS=1. Also marked with +the `integration` pytest marker so the default `pytest` run (or any run with +`-m "not integration"`) skips it. + +Requires .env in the project root with COOKIES_STRING and PUBLICATION_URL +(or EMAIL/PASSWORD). Creates a temporary podcast draft, uploads a tiny +fixture MP3, asserts the media object reaches `state == "transcoded"` with a +non-zero duration, and deletes the draft in a finally block so cleanup +happens even if assertions fail. +""" + +from __future__ import annotations + +import os +import time +from pathlib import Path + +import pytest +from dotenv import load_dotenv + +from substack.api import Api + +REPO = Path(__file__).resolve().parents[2] +load_dotenv(REPO / ".env") + + +pytestmark = [ + pytest.mark.integration, + pytest.mark.skipif( + os.getenv("RUN_INTEGRATION_TESTS") != "1", + reason="set RUN_INTEGRATION_TESTS=1 to run integration tests", + ), +] + + +def _build_api() -> Api: + """Build an Api client from .env credentials, preferring cookies.""" + cookies_path = os.getenv("COOKIES_PATH") + cookies_string = os.getenv("COOKIES_STRING") + publication_url = os.getenv("PUBLICATION_URL") + if cookies_path or cookies_string: + return Api( + cookies_path=cookies_path, + cookies_string=cookies_string, + publication_url=publication_url, + ) + return Api( + email=os.getenv("EMAIL"), + password=os.getenv("PASSWORD"), + publication_url=publication_url, + ) + + +def test_upload_real_mp3_to_sandbox(tiny_mp3_path, capsys): + """Create a podcast draft, upload a tiny MP3, verify transcoded, cleanup.""" + api = _build_api() + user_id = api.get_user_id() + + # Create a minimal podcast draft to give the upload a post_id to attach to. + draft = api.post_draft({ + "draft_title": "", + "draft_subtitle": "", + "draft_podcast_url": None, + "draft_podcast_duration": None, + "draft_body": '{"type":"doc","content":[{"type":"paragraph"}]}', + "section_chosen": True, + "draft_bylines": [{"id": int(user_id), "is_guest": False}], + "audience": "everyone", + "type": "podcast", + }) + draft_id = draft["id"] + + started_at = time.monotonic() + try: + media = api.upload_podcast_audio( + file_path=str(tiny_mp3_path), + draft_id=draft_id, + poll_timeout_seconds=120, + poll_interval_seconds=2, + ) + elapsed = time.monotonic() - started_at + + # Surface a useful one-line report regardless of pass/fail. + print( + f"\n[integration] draft_id={draft_id} upload_id={media.get('id')} " + f"state={media.get('state')!r} duration={media.get('duration')} " + f"transcoded_size={media.get('primary_file_size')} " + f"elapsed={elapsed:.2f}s" + ) + + assert media["state"] == "transcoded" + assert media["duration"] is not None and media["duration"] > 0 + # The fixture is 2s of silence; allow a wide tolerance for encoder padding + # and any server-side re-encode that nudges the duration slightly. + assert 1.5 < media["duration"] < 3.5 + finally: + api.delete_draft(draft_id) diff --git a/tests/substack/test_podcast.py b/tests/substack/test_podcast.py new file mode 100644 index 0000000..5c8ce87 --- /dev/null +++ b/tests/substack/test_podcast.py @@ -0,0 +1,400 @@ +"""Tests for PodcastPost. + +Pure unit tests -- no HTTP, no mocks. Mirrors the style of test_post.py: +pytest classes, scenario-named methods, bare assert, section banners. +Helpers at the top. +""" + +import json + +import pytest + +from substack.exceptions import SectionNotExistsException +from substack.podcast import PodcastPost + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def make_podcast(**overrides): + """Construct a PodcastPost with sensible defaults; overrides win.""" + kwargs = {"title": "Test Episode", "subtitle": "Test Sub", "user_id": 1} + kwargs.update(overrides) + return PodcastPost(**kwargs) + + +def get_draft_dict(podcast): + """get_draft() with both ProseMirror fields decoded back to dicts.""" + out = podcast.get_draft() + if isinstance(out.get("draft_body"), str): + out["draft_body"] = json.loads(out["draft_body"]) + if isinstance(out.get("podcast_description"), str): + out["podcast_description"] = json.loads(out["podcast_description"]) + return out + + +# --------------------------------------------------------------------------- +# TestPodcastPostConstruction +# --------------------------------------------------------------------------- + + +class TestPodcastPostConstruction: + def test_stores_title_and_subtitle(self): + pod = make_podcast(title="Episode 1", subtitle="The pilot") + assert pod.draft_title == "Episode 1" + assert pod.draft_subtitle == "The pilot" + + def test_coerces_user_id_to_int_in_bylines(self): + pod = make_podcast(user_id="42") + assert pod.draft_bylines == [{"id": 42, "is_guest": False}] + + def test_defaults_audience_to_everyone(self): + pod = make_podcast() + assert pod.audience == "everyone" + + def test_respects_explicit_audience(self): + pod = make_podcast(audience="only_paid") + assert pod.audience == "only_paid" + + def test_write_comment_permissions_defaults_to_audience(self): + pod = make_podcast(audience="only_paid") + assert pod.write_comment_permissions == "only_paid" + + def test_write_comment_permissions_respects_override(self): + pod = make_podcast(audience="only_paid", write_comment_permissions="everyone") + assert pod.write_comment_permissions == "everyone" + + def test_type_is_podcast(self): + pod = make_podcast() + assert pod.type == "podcast" + + def test_section_chosen_defaults_true(self): + pod = make_podcast() + assert pod.section_chosen is True + + def test_draft_section_id_starts_none(self): + pod = make_podcast() + assert pod.draft_section_id is None + + def test_draft_body_starts_empty_prosemirror_doc(self): + pod = make_podcast() + assert pod.draft_body == {"type": "doc", "content": []} + + def test_podcast_description_starts_empty_prosemirror_doc(self): + pod = make_podcast() + assert pod.podcast_description == {"type": "doc", "content": []} + + def test_draft_body_and_podcast_description_are_distinct_objects(self): + """Mutating one must not affect the other.""" + pod = make_podcast() + pod.draft_body["content"].append({"type": "paragraph"}) + assert pod.podcast_description == {"type": "doc", "content": []} + + def test_audio_fields_start_none(self): + pod = make_podcast() + assert pod.draft_podcast_upload_id is None + assert pod.draft_podcast_url is None + assert pod.draft_podcast_duration is None + + def test_last_updated_at_starts_none(self): + pod = make_podcast() + assert pod.last_updated_at is None + + +# --------------------------------------------------------------------------- +# TestGetDraftSerialization +# --------------------------------------------------------------------------- + + +class TestGetDraftSerialization: + def test_draft_body_serialized_as_json_string(self): + pod = make_podcast() + out = pod.get_draft() + assert isinstance(out["draft_body"], str) + # Parses back to the empty doc shape. + assert json.loads(out["draft_body"]) == {"type": "doc", "content": []} + + def test_podcast_description_serialized_as_json_string(self): + pod = make_podcast() + out = pod.get_draft() + assert isinstance(out["podcast_description"], str) + assert json.loads(out["podcast_description"]) == {"type": "doc", "content": []} + + def test_includes_type_podcast(self): + out = make_podcast().get_draft() + assert out["type"] == "podcast" + + def test_includes_title_subtitle_audience(self): + pod = make_podcast(title="X", subtitle="Y", audience="founding") + out = pod.get_draft() + assert out["draft_title"] == "X" + assert out["draft_subtitle"] == "Y" + assert out["audience"] == "founding" + + def test_includes_bylines(self): + out = make_podcast(user_id=99).get_draft() + assert out["draft_bylines"] == [{"id": 99, "is_guest": False}] + + def test_includes_section_chosen_and_section_id(self): + pod = make_podcast() + pod.draft_section_id = 12345 + out = pod.get_draft() + assert out["section_chosen"] is True + assert out["draft_section_id"] == 12345 + + def test_includes_podcast_audio_fields(self): + pod = make_podcast() + pod.draft_podcast_upload_id = "uuid-1" + pod.draft_podcast_duration = 123.45 + out = pod.get_draft() + assert out["draft_podcast_upload_id"] == "uuid-1" + assert out["draft_podcast_duration"] == 123.45 + # draft_podcast_url starts None and the field should still be present. + assert "draft_podcast_url" in out + assert out["draft_podcast_url"] is None + + def test_omits_last_updated_at_when_none(self): + pod = make_podcast() + out = pod.get_draft() + assert "last_updated_at" not in out + + def test_includes_last_updated_at_when_set(self): + pod = make_podcast() + pod.last_updated_at = "2026-05-26T04:52:01.886Z" + out = pod.get_draft() + assert out["last_updated_at"] == "2026-05-26T04:52:01.886Z" + + def test_get_draft_is_non_destructive(self): + """Calling get_draft must not mutate self. + + Post has a known footgun where get_draft replaces self.draft_body with + a JSON string in place. PodcastPost is designed for multi-step flows + (create -> PUT updates) and must support repeated get_draft calls. + """ + pod = make_podcast() + first = pod.get_draft() + second = pod.get_draft() + # Both calls yield the same shape. + assert isinstance(first["draft_body"], str) + assert isinstance(second["draft_body"], str) + # And self.draft_body remains a dict (not a string). + assert isinstance(pod.draft_body, dict) + assert pod.draft_body == {"type": "doc", "content": []} + assert isinstance(pod.podcast_description, dict) + + +# --------------------------------------------------------------------------- +# TestSetBodyFromMarkdown +# --------------------------------------------------------------------------- + + +class TestSetBodyFromMarkdown: + def test_returns_self_for_chaining(self): + pod = make_podcast() + assert pod.set_body_from_markdown("hello") is pod + + def test_populates_draft_body_with_prosemirror_paragraph(self): + pod = make_podcast() + pod.set_body_from_markdown("hello world") + # First content node should be a paragraph carrying the text. + body = pod.draft_body + assert body["type"] == "doc" + assert body["content"][0]["type"] == "paragraph" + # Text node carries the literal. + assert body["content"][0]["content"][0]["text"] == "hello world" + + def test_supports_inline_formatting_via_post_parser(self): + pod = make_podcast() + pod.set_body_from_markdown("This is **bold**.") + body = pod.draft_body + marks = [ + mark + for node in body["content"][0].get("content", []) + for mark in node.get("marks", []) + ] + assert {"type": "strong"} in marks + + def test_does_not_modify_podcast_description(self): + pod = make_podcast() + pod.set_body_from_markdown("body text") + assert pod.podcast_description == {"type": "doc", "content": []} + + def test_replaces_prior_body(self): + """Calling twice replaces, not appends.""" + pod = make_podcast() + pod.set_body_from_markdown("first") + pod.set_body_from_markdown("second") + # Only one paragraph remains; it contains 'second'. + paragraphs = [ + n for n in pod.draft_body["content"] if n["type"] == "paragraph" + ] + assert len(paragraphs) == 1 + assert paragraphs[0]["content"][0]["text"] == "second" + + +# --------------------------------------------------------------------------- +# TestSetShowNotesFromMarkdown +# --------------------------------------------------------------------------- + + +class TestSetShowNotesFromMarkdown: + def test_returns_self_for_chaining(self): + pod = make_podcast() + assert pod.set_show_notes_from_markdown("notes") is pod + + def test_populates_podcast_description_with_prosemirror(self): + pod = make_podcast() + pod.set_show_notes_from_markdown("show notes paragraph") + desc = pod.podcast_description + assert desc["type"] == "doc" + assert desc["content"][0]["type"] == "paragraph" + assert desc["content"][0]["content"][0]["text"] == "show notes paragraph" + + def test_does_not_modify_draft_body(self): + pod = make_podcast() + pod.set_show_notes_from_markdown("notes only") + assert pod.draft_body == {"type": "doc", "content": []} + + def test_show_notes_supports_links_and_headings(self): + pod = make_podcast() + pod.set_show_notes_from_markdown( + "# Topics\n\nSee [example](https://example.com)" + ) + desc = pod.podcast_description + types = [n["type"] for n in desc["content"]] + assert "heading" in types + # The link mark surfaces somewhere in the doc. + link_marks = [ + m + for n in desc["content"] + for child in n.get("content", []) or [] + for m in (child.get("marks") or [] if isinstance(child, dict) else []) + if m.get("type") == "link" + ] + assert link_marks, "expected at least one link mark in show notes" + + def test_both_docs_serialized_independently(self): + pod = make_podcast() + pod.set_body_from_markdown("body text") + pod.set_show_notes_from_markdown("show notes") + out = pod.get_draft() + body = json.loads(out["draft_body"]) + desc = json.loads(out["podcast_description"]) + assert body["content"][0]["content"][0]["text"] == "body text" + assert desc["content"][0]["content"][0]["text"] == "show notes" + + +# --------------------------------------------------------------------------- +# TestSetAudio +# --------------------------------------------------------------------------- + + +class TestSetAudio: + def test_returns_self_for_chaining(self): + pod = make_podcast() + assert pod.set_audio("uuid-x") is pod + + def test_sets_upload_id(self): + pod = make_podcast() + pod.set_audio("uuid-x") + assert pod.draft_podcast_upload_id == "uuid-x" + + def test_sets_duration_when_provided(self): + pod = make_podcast() + pod.set_audio("uuid-x", duration_seconds=42.5) + assert pod.draft_podcast_duration == 42.5 + + def test_omitting_duration_leaves_it_none(self): + pod = make_podcast() + pod.set_audio("uuid-x") + assert pod.draft_podcast_duration is None + + def test_upload_id_round_trips_through_get_draft(self): + pod = make_podcast() + pod.set_audio("uuid-x", duration_seconds=60) + out = pod.get_draft() + assert out["draft_podcast_upload_id"] == "uuid-x" + assert out["draft_podcast_duration"] == 60 + + +# --------------------------------------------------------------------------- +# TestSetSection +# --------------------------------------------------------------------------- + + +class TestSetSection: + SECTIONS = [ + {"id": 100, "name": "Interviews"}, + {"id": 200, "name": "Solo"}, + ] + + def test_sets_draft_section_id(self): + pod = make_podcast() + pod.set_section("Solo", self.SECTIONS) + assert pod.draft_section_id == 200 + + def test_returns_self_for_chaining(self): + pod = make_podcast() + assert pod.set_section("Solo", self.SECTIONS) is pod + + def test_raises_section_not_exists_on_unknown(self): + pod = make_podcast() + with pytest.raises(SectionNotExistsException): + pod.set_section("Nonexistent", self.SECTIONS) + + +# --------------------------------------------------------------------------- +# TestLastUpdatedAtThreading +# --------------------------------------------------------------------------- + + +class TestLastUpdatedAtThreading: + """The field exists so callers can echo the server's draft_updated_at on + subsequent PUTs. See docs/PODCAST_ARCHITECTURE.md section 7.""" + + def test_initially_omitted_from_get_draft(self): + out = make_podcast().get_draft() + assert "last_updated_at" not in out + + def test_appears_in_get_draft_once_set(self): + pod = make_podcast() + pod.last_updated_at = "2026-05-26T04:52:01.886Z" + out = pod.get_draft() + assert out["last_updated_at"] == "2026-05-26T04:52:01.886Z" + + def test_can_be_cleared_back_to_none(self): + pod = make_podcast() + pod.last_updated_at = "2026-05-26T04:52:01.886Z" + pod.last_updated_at = None + out = pod.get_draft() + assert "last_updated_at" not in out + + def test_repeated_get_drafts_remain_consistent(self): + pod = make_podcast() + pod.last_updated_at = "2026-05-26T04:52:01.886Z" + first = pod.get_draft() + second = pod.get_draft() + assert first["last_updated_at"] == second["last_updated_at"] + assert first["draft_body"] == second["draft_body"] + + +# --------------------------------------------------------------------------- +# TestFluentChaining +# --------------------------------------------------------------------------- + + +class TestFluentChaining: + def test_chain_body_show_notes_audio(self): + pod = ( + make_podcast() + .set_body_from_markdown("body") + .set_show_notes_from_markdown("notes") + .set_audio("uuid-x", duration_seconds=10) + ) + out = get_draft_dict(pod) + assert out["draft_body"]["content"][0]["content"][0]["text"] == "body" + assert out["podcast_description"]["content"][0]["content"][0]["text"] == "notes" + assert out["draft_podcast_upload_id"] == "uuid-x" + assert out["draft_podcast_duration"] == 10 diff --git a/tests/substack/test_podcast_draft_integration.py b/tests/substack/test_podcast_draft_integration.py new file mode 100644 index 0000000..7621bcd --- /dev/null +++ b/tests/substack/test_podcast_draft_integration.py @@ -0,0 +1,171 @@ +"""Integration test for PodcastPost draft creation + audio attachment. + +Exercises the Phase D flow end-to-end against a real Substack sandbox: + + 1. Build PodcastPost with body + show notes from Markdown + 2. POST /drafts to create the podcast shell, capture draft_updated_at + 3. Upload a tiny fixture MP3 (reuses Phase C: Api.upload_podcast_audio) + 4. set_audio + thread last_updated_at from the create response + 5. PUT /drafts with the updated payload + 6. Verify the API echoes back title / subtitle / audio UUID / both + ProseMirror docs, AND that draft_updated_at advanced from the + create-response value (proof last_updated_at threading is round + tripping, not just satisfying mock assertions) + 7. Delete the draft in a finally block + +No publish call -- that's Phase E. + +Gated on RUN_INTEGRATION_TESTS=1 and the `integration` marker. +""" + +from __future__ import annotations + +import json +import os +import time +from pathlib import Path + +import pytest +from dotenv import load_dotenv + +from substack import Api, PodcastPost + +REPO = Path(__file__).resolve().parents[2] +load_dotenv(REPO / ".env") + + +pytestmark = [ + pytest.mark.integration, + pytest.mark.skipif( + os.getenv("RUN_INTEGRATION_TESTS") != "1", + reason="set RUN_INTEGRATION_TESTS=1 to run integration tests", + ), +] + + +def _build_api() -> Api: + """Construct an Api client from .env credentials, preferring cookies.""" + cookies_path = os.getenv("COOKIES_PATH") + cookies_string = os.getenv("COOKIES_STRING") + publication_url = os.getenv("PUBLICATION_URL") + if cookies_path or cookies_string: + return Api( + cookies_path=cookies_path, + cookies_string=cookies_string, + publication_url=publication_url, + ) + return Api( + email=os.getenv("EMAIL"), + password=os.getenv("PASSWORD"), + publication_url=publication_url, + ) + + +def test_podcast_draft_create_and_attach_audio(tiny_mp3_path, capsys): + """Round-trip PodcastPost -> create -> upload audio -> PUT -> verify echo.""" + api = _build_api() + user_id = api.get_user_id() + + ts = time.strftime("%Y-%m-%d %H:%M:%S") + title = f"[integration test {ts}] PodcastPost round trip" + subtitle = "Auto-generated by tests/substack/test_podcast_draft_integration.py" + body_markdown = ( + "This is the **post-page body** of the episode. " + "It should round-trip through `draft_body`.\n" + ) + show_notes_markdown = ( + "## Show notes\n\n" + "- Topic one\n" + "- Topic two\n\n" + "See [example](https://example.com) for more.\n" + ) + + pod = ( + PodcastPost( + title=title, + subtitle=subtitle, + user_id=user_id, + audience="everyone", + write_comment_permissions="everyone", + ) + .set_body_from_markdown(body_markdown) + .set_show_notes_from_markdown(show_notes_markdown) + ) + + # Step 1: create the podcast draft shell. + created = api.post_draft(pod.get_draft()) + draft_id = created["id"] + create_updated_at = created.get("draft_updated_at") + + try: + assert created["type"] == "podcast", ( + f"expected type=podcast, got {created.get('type')!r}" + ) + assert created["draft_title"] == title + assert create_updated_at is not None, ( + "create response missing draft_updated_at -- " + "cannot exercise last_updated_at threading" + ) + + # Step 2: upload audio (reuses Phase C). + upload_started = time.monotonic() + media = api.upload_podcast_audio( + file_path=str(tiny_mp3_path), + draft_id=draft_id, + poll_timeout_seconds=120, + poll_interval_seconds=2, + ) + upload_elapsed = time.monotonic() - upload_started + upload_id = media["id"] + duration_seconds = media["duration"] + assert media["state"] == "transcoded" + + # Step 3: attach audio and thread last_updated_at into the PUT payload. + pod.set_audio(upload_id, duration_seconds=duration_seconds) + pod.last_updated_at = create_updated_at + + # Step 4: PUT the updated draft. + updated = api.put_draft(draft_id, **pod.get_draft()) + put_updated_at = updated.get("draft_updated_at") + + # Surface a one-line report for debugging regardless of pass/fail. + print( + f"\n[integration] draft_id={draft_id} upload_id={upload_id} " + f"audio_duration={duration_seconds} " + f"upload_elapsed={upload_elapsed:.2f}s " + f"create_updated_at={create_updated_at!r} " + f"put_updated_at={put_updated_at!r}" + ) + + # Step 5: verify echoes. + assert updated["type"] == "podcast" + assert updated["draft_title"] == title + assert updated["draft_subtitle"] == subtitle + assert updated["draft_podcast_upload_id"] == upload_id, ( + "Substack did not store the audio UUID -- attachment failed" + ) + + # Body and show notes come back as JSON-stringified ProseMirror docs. + body_doc = json.loads(updated["draft_body"]) + assert "post-page body" in json.dumps(body_doc), ( + f"body text missing from echoed draft_body: {body_doc!r}" + ) + + desc = updated.get("podcast_description") + assert desc, "podcast_description was empty/missing in PUT response" + desc_doc = json.loads(desc) + assert "Show notes" in json.dumps(desc_doc), ( + f"show notes heading missing from echoed podcast_description: {desc_doc!r}" + ) + assert "Topic one" in json.dumps(desc_doc) + + # Proof that last_updated_at threading actually round-trips: the + # server's timestamp advanced after our PUT. + assert put_updated_at is not None + assert put_updated_at != create_updated_at, ( + f"draft_updated_at did not advance after PUT " + f"({create_updated_at!r} -> {put_updated_at!r}); " + "last_updated_at threading may be a no-op" + ) + finally: + api.delete_draft(draft_id) diff --git a/tests/substack/test_podcast_publish_integration.py b/tests/substack/test_podcast_publish_integration.py new file mode 100644 index 0000000..213bcfb --- /dev/null +++ b/tests/substack/test_podcast_publish_integration.py @@ -0,0 +1,150 @@ +"""Integration test for the full podcast publish flow. + +End-to-end exercise against a real Substack sandbox: + + 1. Build a complete PodcastPost (title, subtitle, body markdown, + show notes markdown) + 2. POST /drafts to create the podcast shell + 3. Upload a tiny fixture MP3 via Api.upload_podcast_audio + 4. PUT /drafts to attach the audio with last_updated_at threading + 5. GET /drafts/{id}/prepublish, assert errors == [] + 6. POST /drafts/{id}/publish with send=False (no email goes out) + 7. Assert is_published == True, type == "podcast", + draft_podcast_upload_id preserved + 8. Delete the published post in a finally block + +Gated on RUN_INTEGRATION_TESTS=1 + the `integration` marker. + +`send=False` is critical: this test publishes against the real sandbox, +and email delivery is permanent. Even a zero-subscriber sandbox shouldn't +fire delivery side-effects from CI. +""" + +from __future__ import annotations + +import os +import time +from pathlib import Path + +import pytest +from dotenv import load_dotenv + +from substack import Api, PodcastPost + +REPO = Path(__file__).resolve().parents[2] +load_dotenv(REPO / ".env") + + +pytestmark = [ + pytest.mark.integration, + pytest.mark.skipif( + os.getenv("RUN_INTEGRATION_TESTS") != "1", + reason="set RUN_INTEGRATION_TESTS=1 to run integration tests", + ), +] + + +def _build_api() -> Api: + """Construct an Api client from .env credentials, preferring cookies.""" + cookies_path = os.getenv("COOKIES_PATH") + cookies_string = os.getenv("COOKIES_STRING") + publication_url = os.getenv("PUBLICATION_URL") + if cookies_path or cookies_string: + return Api( + cookies_path=cookies_path, + cookies_string=cookies_string, + publication_url=publication_url, + ) + return Api( + email=os.getenv("EMAIL"), + password=os.getenv("PASSWORD"), + publication_url=publication_url, + ) + + +def test_full_podcast_publish_flow(tiny_mp3_path, capsys): + """Create -> upload -> attach -> prepublish -> publish(send=False) -> delete.""" + api = _build_api() + user_id = api.get_user_id() + + ts = time.strftime("%Y-%m-%d %H:%M:%S") + title = f"[integration test {ts}] full podcast publish flow" + subtitle = "Auto-generated by tests/substack/test_podcast_publish_integration.py" + body_markdown = ( + "This is the post-page body of an automatically-published episode. " + "Safe to delete.\n" + ) + show_notes_markdown = ( + "## Show notes\n\n" + "- Generated by the python-substack integration suite\n" + "- send=False so no email goes out\n" + ) + + pod = ( + PodcastPost( + title=title, + subtitle=subtitle, + user_id=user_id, + audience="everyone", + write_comment_permissions="everyone", + ) + .set_body_from_markdown(body_markdown) + .set_show_notes_from_markdown(show_notes_markdown) + ) + + flow_started = time.monotonic() + created = api.post_draft(pod.get_draft()) + draft_id = created["id"] + create_updated_at = created.get("draft_updated_at") + + try: + assert created["type"] == "podcast" + + # Upload audio + attach. + media = api.upload_podcast_audio( + file_path=str(tiny_mp3_path), + draft_id=draft_id, + poll_timeout_seconds=120, + poll_interval_seconds=2, + ) + upload_id = media["id"] + duration_seconds = media["duration"] + assert media["state"] == "transcoded" + + pod.set_audio(upload_id, duration_seconds=duration_seconds) + pod.last_updated_at = create_updated_at + updated = api.put_draft(draft_id, **pod.get_draft()) + assert updated["draft_podcast_upload_id"] == upload_id + + # Prepublish validation. + prepublish = api.prepublish_draft(draft_id) + assert prepublish.get("errors") == [], ( + f"prepublish reported errors that should have been caught earlier: " + f"{prepublish.get('errors')!r}" + ) + + # Publish with send=False so no email goes out from the sandbox. + published = api.publish_draft(draft_id, send=False, share_automatically=False) + flow_elapsed = time.monotonic() - flow_started + + print( + f"\n[integration] draft_id={draft_id} upload_id={upload_id} " + f"published_id={published.get('id')} " + f"is_published={published.get('is_published')} " + f"type={published.get('type')!r} " + f"audio_attached={published.get('draft_podcast_upload_id') == upload_id} " + f"prepublish_errors={prepublish.get('errors')!r} " + f"prepublish_suggestions={prepublish.get('suggestions')!r} " + f"flow_elapsed={flow_elapsed:.2f}s" + ) + + assert published["is_published"] is True + assert published["type"] == "podcast" + assert published["draft_podcast_upload_id"] == upload_id, ( + "audio reference lost during publish" + ) + assert published["draft_title"] == title + # publish response uses the same draft id; publish does not re-id. + assert published["id"] == draft_id + finally: + api.delete_draft(draft_id) diff --git a/tests/substack_mcp/__init__.py b/tests/substack_mcp/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/substack_mcp/test_mcp_podcast_integration.py b/tests/substack_mcp/test_mcp_podcast_integration.py new file mode 100644 index 0000000..0f60c72 --- /dev/null +++ b/tests/substack_mcp/test_mcp_podcast_integration.py @@ -0,0 +1,134 @@ +"""End-to-end acceptance test for the podcast MCP tool surface. + +This is the v1 acceptance test: one call to `post_podcast_draft_from_markdown` +takes a fixture MP3 + Markdown for body and show notes, creates a podcast +draft, uploads the audio, attaches it via PUT (with last_updated_at +threading), runs prepublish, publishes with send=False (no email goes out +to sandbox subscribers), and deletes the resulting post in a finally +block. + +If this passes, the whole MCP surface works end-to-end -- no separate +Api or PodcastPost integration step is needed for the MCP path. + +Gated on RUN_INTEGRATION_TESTS=1 and the `integration` marker. +""" + +from __future__ import annotations + +import asyncio +import os +import time +from pathlib import Path + +import pytest +from dotenv import load_dotenv + +from substack import Api +from substack_mcp.mcp_server import post_podcast_draft_from_markdown + +REPO = Path(__file__).resolve().parents[2] +load_dotenv(REPO / ".env") + + +pytestmark = [ + pytest.mark.integration, + pytest.mark.skipif( + os.getenv("RUN_INTEGRATION_TESTS") != "1", + reason="set RUN_INTEGRATION_TESTS=1 to run integration tests", + ), +] + + +def _build_cleanup_api() -> Api: + """Build a plain Api for the post-test cleanup call (delete_draft).""" + cookies_path = os.getenv("COOKIES_PATH") + cookies_string = os.getenv("COOKIES_STRING") + publication_url = os.getenv("PUBLICATION_URL") + if cookies_path or cookies_string: + return Api( + cookies_path=cookies_path, + cookies_string=cookies_string, + publication_url=publication_url, + ) + return Api( + email=os.getenv("EMAIL"), + password=os.getenv("PASSWORD"), + publication_url=publication_url, + ) + + +def test_mcp_post_podcast_draft_from_markdown_end_to_end(tiny_mp3_path, capsys): + """One MCP call: MP3 + Markdown -> published podcast on sandbox -> deleted.""" + ts = time.strftime("%Y-%m-%d %H:%M:%S") + title = f"[mcp integration {ts}] post_podcast_draft_from_markdown" + subtitle = "Auto-generated v1 acceptance test" + show_notes = ( + "## Show notes\n\n" + "- Generated by `tests/substack_mcp/test_mcp_podcast_integration.py`\n" + "- Audio attached via `Api.upload_podcast_audio`\n" + "- Published with `send=False` so no email goes out\n\n" + "See [the contract](https://example.com/contract) for the wire details.\n" + ) + body = ( + "This is the post-page body that ships alongside the episode.\n\n" + "It should round-trip through `draft_body` and appear on the post page.\n" + ) + + started = time.monotonic() + result = asyncio.run( + post_podcast_draft_from_markdown( + title=title, + subtitle=subtitle, + show_notes_markdown=show_notes, + post_body_markdown=body, + audio_file_path=str(tiny_mp3_path), + audience="everyone", + prepublish=True, + publish=True, + send=False, + share_automatically=False, + ) + ) + elapsed = time.monotonic() - started + + published_id = result["publish"]["id"] + try: + # One-line report regardless of pass/fail. + print( + f"\n[mcp-integration] published_id={published_id} " + f"upload_id={result['upload']['id']} " + f"upload_state={result['upload']['state']!r} " + f"is_published={result['publish']['is_published']} " + f"type={result['publish']['type']!r} " + f"audio_attached={result['publish']['draft_podcast_upload_id'] == result['upload']['id']} " + f"prepublish_errors={result['prepublish']['errors']!r} " + f"flow_elapsed={elapsed:.2f}s" + ) + + # 1. Audio uploaded and transcoded. + assert result["upload"] is not None + assert result["upload"]["state"] == "transcoded" + assert result["upload"]["duration"] > 0 + + # 2. Prepublish clean (no client-side guards; server is the validator). + assert result["prepublish"] == {"errors": [], "suggestions": []} + + # 3. Publish succeeded, podcast type preserved, audio attached. + assert result["publish"]["is_published"] is True + assert result["publish"]["type"] == "podcast" + assert ( + result["publish"]["draft_podcast_upload_id"] == result["upload"]["id"] + ), "audio reference lost during publish" + assert result["publish"]["draft_title"] == title + + # 4. The draft key reflects the post-PUT state (audio attached) and + # is NOT replaced by the publish response. + assert result["draft"]["draft_podcast_upload_id"] == result["upload"]["id"] + + # 5. No tags were requested. + assert result["tags"] is None + finally: + # Cleanup outside the MCP surface -- delete_draft is not (currently) + # exposed as an MCP tool. + api = _build_cleanup_api() + api.delete_draft(published_id) diff --git a/tests/substack_mcp/test_mcp_server_podcast.py b/tests/substack_mcp/test_mcp_server_podcast.py new file mode 100644 index 0000000..f9e2a70 --- /dev/null +++ b/tests/substack_mcp/test_mcp_server_podcast.py @@ -0,0 +1,503 @@ +"""Tests for the podcast MCP tools. + +Like tests/substack/test_audio_upload.py, these tests deliberately depart +from the codebase's no-mock convention (STYLE.md section 5): MCP tool +composition is HTTP integration logic that can only be exercised by +mocking the Api layer. + +Mocks target `substack_mcp.mcp_server.get_api` (replacing the entire Api +factory) and configure a MagicMock client whose methods return canned +responses. We let the real PodcastPost class run so assertions can read +the actual payload built into `pod.get_draft()`. +""" + +import asyncio +import json +from unittest.mock import MagicMock, patch + +import pytest + +from substack_mcp.mcp_server import ( + post_podcast_draft_from_markdown, + upload_podcast_audio, +) + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def make_mock_api(): + """Construct a MagicMock Api whose draft / upload / publish methods + return shape-faithful canned responses.""" + api = MagicMock() + api.get_user_id.return_value = 253249334 + api.post_draft.return_value = { + "id": 199290000, + "type": "podcast", + "draft_updated_at": "2026-05-26T10:00:00.000Z", + "draft_title": "", + } + api.put_draft.return_value = { + "id": 199290000, + "type": "podcast", + "draft_updated_at": "2026-05-26T10:00:05.000Z", + "draft_podcast_upload_id": "uuid-attached", + } + api.upload_podcast_audio.return_value = { + "id": "uuid-attached", + "state": "transcoded", + "duration": 600.0, + "primary_file_size": 7000000, + } + api.add_tags_to_post.return_value = {"tags_added": [{"id": 1, "name": "x"}]} + api.prepublish_draft.return_value = {"errors": [], "suggestions": []} + api.publish_draft.return_value = { + "id": 199290000, + "is_published": True, + "type": "podcast", + "draft_podcast_upload_id": "uuid-attached", + } + return api + + +def run(coro): + """Synchronously run an async tool function.""" + return asyncio.run(coro) + + +# =========================================================================== +# upload_podcast_audio (MCP wrapper) +# =========================================================================== + + +class TestUploadPodcastAudioTool: + """Thin wrapper around Api.upload_podcast_audio.""" + + def test_calls_api_with_passed_args(self): + api = make_mock_api() + with patch("substack_mcp.mcp_server.get_api", return_value=api): + run(upload_podcast_audio(file_path="/tmp/test.mp3", draft_id=12345)) + api.upload_podcast_audio.assert_called_once_with( + file_path="/tmp/test.mp3", draft_id=12345 + ) + + def test_returns_api_result_verbatim(self): + api = make_mock_api() + with patch("substack_mcp.mcp_server.get_api", return_value=api): + result = run(upload_podcast_audio(file_path="/tmp/test.mp3", draft_id=12345)) + assert result == api.upload_podcast_audio.return_value + assert result["state"] == "transcoded" + assert result["id"] == "uuid-attached" + + +# =========================================================================== +# post_podcast_draft_from_markdown -- PodcastPost construction +# =========================================================================== + + +class TestPodcastPostConstruction: + """The tool builds a PodcastPost from kwargs and calls api.post_draft.""" + + def test_post_draft_called_with_podcast_type(self): + api = make_mock_api() + with patch("substack_mcp.mcp_server.get_api", return_value=api): + run(post_podcast_draft_from_markdown(title="My Episode")) + body = api.post_draft.call_args[0][0] + assert body["type"] == "podcast" + assert body["draft_title"] == "My Episode" + + def test_passes_subtitle_and_audience(self): + api = make_mock_api() + with patch("substack_mcp.mcp_server.get_api", return_value=api): + run( + post_podcast_draft_from_markdown( + title="Ep", + subtitle="A subtitle", + audience="only_paid", + ) + ) + body = api.post_draft.call_args[0][0] + assert body["draft_subtitle"] == "A subtitle" + assert body["audience"] == "only_paid" + + def test_write_comment_permissions_defaults_to_audience(self): + api = make_mock_api() + with patch("substack_mcp.mcp_server.get_api", return_value=api): + run( + post_podcast_draft_from_markdown( + title="Ep", + audience="founding", + ) + ) + body = api.post_draft.call_args[0][0] + assert body["write_comment_permissions"] == "founding" + + def test_write_comment_permissions_respects_explicit_value(self): + api = make_mock_api() + with patch("substack_mcp.mcp_server.get_api", return_value=api): + run( + post_podcast_draft_from_markdown( + title="Ep", + audience="founding", + write_comment_permissions="everyone", + ) + ) + body = api.post_draft.call_args[0][0] + assert body["write_comment_permissions"] == "everyone" + + def test_user_id_from_api(self): + api = make_mock_api() + with patch("substack_mcp.mcp_server.get_api", return_value=api): + run(post_podcast_draft_from_markdown(title="Ep")) + body = api.post_draft.call_args[0][0] + assert body["draft_bylines"] == [{"id": 253249334, "is_guest": False}] + + def test_draft_section_id_passed_through(self): + api = make_mock_api() + with patch("substack_mcp.mcp_server.get_api", return_value=api): + run(post_podcast_draft_from_markdown(title="Ep", draft_section_id=42)) + body = api.post_draft.call_args[0][0] + assert body["draft_section_id"] == 42 + + +# =========================================================================== +# post_podcast_draft_from_markdown -- two ProseMirror docs +# =========================================================================== + + +class TestTwoProseMirrorDocs: + """Both show_notes_markdown and post_body_markdown are independently optional.""" + + def test_both_omitted_yields_empty_docs(self): + api = make_mock_api() + with patch("substack_mcp.mcp_server.get_api", return_value=api): + run(post_podcast_draft_from_markdown(title="Ep")) + body = api.post_draft.call_args[0][0] + assert json.loads(body["draft_body"]) == {"type": "doc", "content": []} + assert json.loads(body["podcast_description"]) == {"type": "doc", "content": []} + + def test_show_notes_populates_podcast_description(self): + api = make_mock_api() + with patch("substack_mcp.mcp_server.get_api", return_value=api): + run( + post_podcast_draft_from_markdown( + title="Ep", + show_notes_markdown="# Show notes\n\nA paragraph.", + ) + ) + body = api.post_draft.call_args[0][0] + desc = json.loads(body["podcast_description"]) + assert "Show notes" in json.dumps(desc) + # And the body stays empty. + assert json.loads(body["draft_body"]) == {"type": "doc", "content": []} + + def test_body_populates_draft_body(self): + api = make_mock_api() + with patch("substack_mcp.mcp_server.get_api", return_value=api): + run( + post_podcast_draft_from_markdown( + title="Ep", + post_body_markdown="Body text here.", + ) + ) + body = api.post_draft.call_args[0][0] + draft_body = json.loads(body["draft_body"]) + assert "Body text here." in json.dumps(draft_body) + # And the show notes stay empty. + assert json.loads(body["podcast_description"]) == {"type": "doc", "content": []} + + def test_both_populated_independently(self): + api = make_mock_api() + with patch("substack_mcp.mcp_server.get_api", return_value=api): + run( + post_podcast_draft_from_markdown( + title="Ep", + show_notes_markdown="show notes paragraph", + post_body_markdown="body paragraph", + ) + ) + body = api.post_draft.call_args[0][0] + desc = json.dumps(json.loads(body["podcast_description"])) + body_doc = json.dumps(json.loads(body["draft_body"])) + assert "show notes paragraph" in desc + assert "body paragraph" in body_doc + + +# =========================================================================== +# post_podcast_draft_from_markdown -- audio attach flow +# =========================================================================== + + +class TestAudioAttach: + """When audio_file_path is provided, the tool uploads + attaches via PUT.""" + + def test_skips_audio_when_no_path(self): + api = make_mock_api() + with patch("substack_mcp.mcp_server.get_api", return_value=api): + run(post_podcast_draft_from_markdown(title="Ep")) + api.upload_podcast_audio.assert_not_called() + api.put_draft.assert_not_called() + + def test_uploads_audio_with_created_draft_id(self): + api = make_mock_api() + with patch("substack_mcp.mcp_server.get_api", return_value=api): + run( + post_podcast_draft_from_markdown( + title="Ep", + audio_file_path="/tmp/sample.mp3", + ) + ) + api.upload_podcast_audio.assert_called_once_with( + file_path="/tmp/sample.mp3", draft_id=199290000 + ) + + def test_put_draft_carries_audio_upload_id(self): + api = make_mock_api() + with patch("substack_mcp.mcp_server.get_api", return_value=api): + run( + post_podcast_draft_from_markdown( + title="Ep", + audio_file_path="/tmp/sample.mp3", + ) + ) + put_kwargs = api.put_draft.call_args.kwargs + assert put_kwargs["draft_podcast_upload_id"] == "uuid-attached" + + def test_put_draft_threads_last_updated_at_from_create(self): + """Proof the optimistic-concurrency token from POST flows into PUT.""" + api = make_mock_api() + with patch("substack_mcp.mcp_server.get_api", return_value=api): + run( + post_podcast_draft_from_markdown( + title="Ep", + audio_file_path="/tmp/sample.mp3", + ) + ) + put_kwargs = api.put_draft.call_args.kwargs + assert put_kwargs["last_updated_at"] == "2026-05-26T10:00:00.000Z" + + def test_put_draft_targets_created_draft_id(self): + api = make_mock_api() + with patch("substack_mcp.mcp_server.get_api", return_value=api): + run( + post_podcast_draft_from_markdown( + title="Ep", + audio_file_path="/tmp/sample.mp3", + ) + ) + # First positional arg is the draft id. + assert api.put_draft.call_args.args[0] == 199290000 + + +# =========================================================================== +# post_podcast_draft_from_markdown -- tags +# =========================================================================== + + +class TestTagsHandling: + """Tags reuse the existing _normalize_tags helper + Api.add_tags_to_post.""" + + def test_skips_tags_when_none(self): + api = make_mock_api() + with patch("substack_mcp.mcp_server.get_api", return_value=api): + run(post_podcast_draft_from_markdown(title="Ep")) + api.add_tags_to_post.assert_not_called() + + def test_string_tag_normalized_to_list(self): + api = make_mock_api() + with patch("substack_mcp.mcp_server.get_api", return_value=api): + run(post_podcast_draft_from_markdown(title="Ep", tags="tech")) + api.add_tags_to_post.assert_called_once_with(199290000, ["tech"]) + + def test_list_tags_passed_through(self): + api = make_mock_api() + with patch("substack_mcp.mcp_server.get_api", return_value=api): + run(post_podcast_draft_from_markdown(title="Ep", tags=["a", "b"])) + api.add_tags_to_post.assert_called_once_with(199290000, ["a", "b"]) + + +# =========================================================================== +# post_podcast_draft_from_markdown -- prepublish / publish gating +# =========================================================================== + + +class TestPrepublishAndPublishGating: + """Both prepublish and publish are off by default; opt-in via flags.""" + + def test_skips_prepublish_when_false(self): + api = make_mock_api() + with patch("substack_mcp.mcp_server.get_api", return_value=api): + run(post_podcast_draft_from_markdown(title="Ep")) + api.prepublish_draft.assert_not_called() + + def test_calls_prepublish_when_true(self): + api = make_mock_api() + with patch("substack_mcp.mcp_server.get_api", return_value=api): + run(post_podcast_draft_from_markdown(title="Ep", prepublish=True)) + api.prepublish_draft.assert_called_once_with(199290000) + + def test_skips_publish_when_false(self): + api = make_mock_api() + with patch("substack_mcp.mcp_server.get_api", return_value=api): + run(post_podcast_draft_from_markdown(title="Ep")) + api.publish_draft.assert_not_called() + + def test_calls_publish_with_send_and_share_automatically(self): + api = make_mock_api() + with patch("substack_mcp.mcp_server.get_api", return_value=api): + run( + post_podcast_draft_from_markdown( + title="Ep", + publish=True, + send=False, + share_automatically=True, + ) + ) + api.publish_draft.assert_called_once_with( + 199290000, send=False, share_automatically=True + ) + + def test_publish_defaults_send_true_share_false(self): + api = make_mock_api() + with patch("substack_mcp.mcp_server.get_api", return_value=api): + run(post_podcast_draft_from_markdown(title="Ep", publish=True)) + api.publish_draft.assert_called_once_with( + 199290000, send=True, share_automatically=False + ) + + +# =========================================================================== +# post_podcast_draft_from_markdown -- return shape +# =========================================================================== + + +class TestReturnShape: + """Return dict surfaces each composed step so callers can inspect results.""" + + def test_returns_dict_with_expected_keys(self): + api = make_mock_api() + with patch("substack_mcp.mcp_server.get_api", return_value=api): + result = run(post_podcast_draft_from_markdown(title="Ep")) + for key in ("draft", "upload", "tags", "prepublish", "publish"): + assert key in result, f"missing key {key!r} in {result!r}" + + def test_no_audio_yields_upload_none(self): + api = make_mock_api() + with patch("substack_mcp.mcp_server.get_api", return_value=api): + result = run(post_podcast_draft_from_markdown(title="Ep")) + assert result["upload"] is None + + def test_with_audio_returns_transcoded_media(self): + api = make_mock_api() + with patch("substack_mcp.mcp_server.get_api", return_value=api): + result = run( + post_podcast_draft_from_markdown( + title="Ep", + audio_file_path="/tmp/sample.mp3", + ) + ) + assert result["upload"]["state"] == "transcoded" + assert result["upload"]["id"] == "uuid-attached" + + def test_no_tags_yields_tags_none(self): + api = make_mock_api() + with patch("substack_mcp.mcp_server.get_api", return_value=api): + result = run(post_podcast_draft_from_markdown(title="Ep")) + assert result["tags"] is None + + def test_with_tags_returns_tag_result(self): + api = make_mock_api() + with patch("substack_mcp.mcp_server.get_api", return_value=api): + result = run(post_podcast_draft_from_markdown(title="Ep", tags=["x"])) + assert result["tags"] == api.add_tags_to_post.return_value + + def test_draft_field_reflects_latest_state_after_audio_attach(self): + """After PUT-attaching audio, the 'draft' key should hold the PUT result, + not the original POST result.""" + api = make_mock_api() + with patch("substack_mcp.mcp_server.get_api", return_value=api): + result = run( + post_podcast_draft_from_markdown( + title="Ep", + audio_file_path="/tmp/sample.mp3", + ) + ) + # The PUT response has draft_podcast_upload_id; the POST response did not. + assert result["draft"]["draft_podcast_upload_id"] == "uuid-attached" + + def test_prepublish_and_publish_results_in_return(self): + api = make_mock_api() + with patch("substack_mcp.mcp_server.get_api", return_value=api): + result = run( + post_podcast_draft_from_markdown( + title="Ep", + prepublish=True, + publish=True, + ) + ) + assert result["prepublish"] == {"errors": [], "suggestions": []} + assert result["publish"]["is_published"] is True + + +# =========================================================================== +# post_podcast_draft_from_markdown -- call order +# =========================================================================== + + +class TestCallOrder: + """The composition must follow create -> upload -> attach -> tags -> + prepublish -> publish. Order matters because each step needs the previous + step's id / token.""" + + def test_full_flow_order(self): + api = make_mock_api() + # Record the sequence of method names called on api. + order = [] + + def record(name): + orig = getattr(api, name) + def wrapper(*a, **kw): + order.append(name) + return orig.return_value + return wrapper + + api.post_draft.side_effect = lambda *a, **kw: ( + order.append("post_draft"), api.post_draft.return_value + )[1] + api.upload_podcast_audio.side_effect = lambda *a, **kw: ( + order.append("upload_podcast_audio"), api.upload_podcast_audio.return_value + )[1] + api.put_draft.side_effect = lambda *a, **kw: ( + order.append("put_draft"), api.put_draft.return_value + )[1] + api.add_tags_to_post.side_effect = lambda *a, **kw: ( + order.append("add_tags_to_post"), api.add_tags_to_post.return_value + )[1] + api.prepublish_draft.side_effect = lambda *a, **kw: ( + order.append("prepublish_draft"), api.prepublish_draft.return_value + )[1] + api.publish_draft.side_effect = lambda *a, **kw: ( + order.append("publish_draft"), api.publish_draft.return_value + )[1] + + with patch("substack_mcp.mcp_server.get_api", return_value=api): + run( + post_podcast_draft_from_markdown( + title="Ep", + audio_file_path="/tmp/x.mp3", + tags=["t1"], + prepublish=True, + publish=True, + ) + ) + + assert order == [ + "post_draft", + "upload_podcast_audio", + "put_draft", + "add_tags_to_post", + "prepublish_draft", + "publish_draft", + ]