diff --git a/README.md b/README.md index 1653f945..598f5d76 100644 --- a/README.md +++ b/README.md @@ -125,7 +125,8 @@ default (trusted) flow to avoid silently bypassing 2FA. The `icloud` command line interface is organized around top-level subcommands such as `auth`, `account`, `devices`, `calendar`, -`contacts`, `drive`, `photos`, `hidemyemail`, `notes`, and `reminders`, +`contacts`, `drive`, `photos`, `hidemyemail`, `notes`, `reminders`, and +`invites`, plus a standalone `doctor` command described under [Diagnostics](#diagnostics). @@ -1759,6 +1760,25 @@ enums used elsewhere in this service. `resolve()` only looks; `accept()` joins the event, after which it appears in `events()` with the shared scope. +### Command line + +```console +icloud invites list +icloud invites show 1F9D5936 +icloud invites rsvps 1F9D5936 +icloud invites resolve 008ABCDEFGHIJ +``` + +`list` shows events you host and events shared with you. `show` and `rsvps` +take a full event id or any unambiguous prefix of one, since the listing +truncates ids to fit the terminal. `resolve` previews an invite link without +joining it. + +Every command is read-only, and each accepts `--format json`. Responding to an +invitation and joining one from a link are available on the service +(`api.invites.rsvp()` and `api.invites.accept()`) but not yet on the command +line. + ### Errors Every call raises a subclass of `InvitesError`: `InvitesAuthError` when the diff --git a/pyicloud/cli/app.py b/pyicloud/cli/app.py index f2cc2a95..f739deae 100644 --- a/pyicloud/cli/app.py +++ b/pyicloud/cli/app.py @@ -12,6 +12,7 @@ from pyicloud.cli.commands.doctor import doctor from pyicloud.cli.commands.drive import app as drive_app from pyicloud.cli.commands.hidemyemail import app as hidemyemail_app +from pyicloud.cli.commands.invites import app as invites_app from pyicloud.cli.commands.notes import app as notes_app from pyicloud.cli.commands.photos import app as photos_app from pyicloud.cli.commands.reminders import app as reminders_app @@ -85,6 +86,9 @@ def root_callback( app.add_typer( notes_app, name="notes", invoke_without_command=True, callback=_group_root ) +app.add_typer( + invites_app, name="invites", invoke_without_command=True, callback=_group_root +) # A leaf rather than a group: `icloud doctor` is what someone types when a # service stopped working, and it should run rather than print a group listing. diff --git a/pyicloud/cli/commands/invites.py b/pyicloud/cli/commands/invites.py new file mode 100644 index 00000000..2ff40105 --- /dev/null +++ b/pyicloud/cli/commands/invites.py @@ -0,0 +1,338 @@ +"""Apple Invites commands. + +Read-only. Responding to an invitation and joining one from a link are writes +and are deliberately not exposed here yet; see the tracking issue. +""" + +from __future__ import annotations + +from collections.abc import Callable +from datetime import datetime +from typing import Any + +import typer + +from pyicloud.base import PyiCloudService +from pyicloud.cli.context import CLIAbort, get_state, service_call +from pyicloud.cli.normalize import ( + normalize_invite_event, + normalize_invite_event_details, + normalize_invite_rsvp, + normalize_invite_share, +) +from pyicloud.cli.options import ( + DEFAULT_LOG_LEVEL, + DEFAULT_OUTPUT_FORMAT, + HttpProxyOption, + HttpsProxyOption, + LogLevelOption, + NoVerifySslOption, + OutputFormatOption, + SessionDirOption, + UsernameOption, + store_command_options, +) +from pyicloud.cli.output import console_kv_table, console_table +from pyicloud.services.invites import Event, EventNotFound, InvitesError + +app = typer.Typer(help="Inspect Apple Invites events.") + +EVENT_ID_HELP = ( + "Event id, or any unambiguous prefix of one, as shown by `icloud invites list`." +) +SHORT_GUID_HELP = ( + "An invite link, or just its trailing part: either " + "https://www.icloud.com/invites/008ABCDEFGHIJ or 008ABCDEFGHIJ." +) + + +def _format_when(payload: dict[str, Any]) -> str: + """Render an event's start compactly enough to fit a table cell. + + The full ISO value keeps seconds and a UTC offset, which wraps onto three + lines in an eighty-column terminal and pushes the event id out of view. + """ + + starts_at = payload.get("starts_at") + if not isinstance(starts_at, datetime): + return "" if starts_at is None else str(starts_at) + if payload.get("is_all_day"): + return f"{starts_at:%Y-%m-%d} (all day)" + return f"{starts_at:%Y-%m-%d %H:%M}" + + +def _resolve_event_id(api: PyiCloudService, given: str) -> str: + """Return the full event id for a full id or an unambiguous prefix. + + Listing truncates ids to fit the terminal, so requiring the full value + would make the id column decorative. Prefixes work the way they do in git. + """ + + candidates = [ + event.event_id + for event in api.invites.events() + if event.event_id.lower().startswith(given.lower()) + ] + if len(candidates) == 1: + return candidates[0] + if not candidates: + raise CLIAbort( + f"No event id starts with {given!r}. Run `icloud invites list` to " + "see the events available to you." + ) + listed = "\n".join(f" - {candidate}" for candidate in candidates) + raise CLIAbort(f"{given!r} matches more than one event:\n{listed}") + + +def _short_guid_from(given: str) -> str: + """Accept either a short guid or the invite URL it came from. + + People copy the whole link far more often than they pick the trailing + segment out of it, and sending the URL to Apple only earns a 400. + """ + + candidate = given.strip().rstrip("/") + if "/" in candidate: + candidate = candidate.rsplit("/", 1)[-1] + if not candidate: + raise CLIAbort( + "No invite id given. Pass the trailing part of an invite link, " + "for example 008ABCDEFGHIJ from " + "https://www.icloud.com/invites/008ABCDEFGHIJ." + ) + return candidate + + +def _invites_call(api: PyiCloudService, fn: Callable[[], Any]) -> Any: + """Wrap an Invites call so its errors reach the user as advice.""" + + try: + return service_call("Invites", fn, account_name=api.account_name) + except InvitesError as err: + raise CLIAbort(f"Invites: {err}") from err + + +def _lookup_event(api: PyiCloudService, event_id: str) -> Event: + """Fetch one event, turning an unknown id into advice rather than a trace.""" + + try: + return api.invites.event(_resolve_event_id(api, event_id)) + except EventNotFound as err: + raise CLIAbort( + f"No event with id {event_id}. Run `icloud invites list` to see " + "the events available to you." + ) from err + + +@app.command("list") +def invites_list( + ctx: typer.Context, + username: UsernameOption = None, + session_dir: SessionDirOption = None, + http_proxy: HttpProxyOption = None, + https_proxy: HttpsProxyOption = None, + no_verify_ssl: NoVerifySslOption = False, + output_format: OutputFormatOption = DEFAULT_OUTPUT_FORMAT, + log_level: LogLevelOption = DEFAULT_LOG_LEVEL, +) -> None: + """List events you host and events shared with you.""" + + store_command_options( + ctx, + username=username, + session_dir=session_dir, + http_proxy=http_proxy, + https_proxy=https_proxy, + no_verify_ssl=no_verify_ssl, + output_format=output_format, + log_level=log_level, + ) + state = get_state(ctx) + api = state.get_api() + payload = [ + normalize_invite_event(event) + for event in _invites_call( + api, + # The lambda is load-bearing: `api.invites` is a property that + # builds the service and can raise PyiCloudServiceUnavailable. + # Passing `api.invites.events` would evaluate it outside the + # guard, and the failure would escape as a traceback. + lambda: api.invites.events(), # pylint: disable=unnecessary-lambda + ) + ] + if state.json_output: + state.write_json(payload) + return + state.console.print( + console_table( + "Invites", + ["Scope", "Title", "When", "Host", "ID"], + [ + ( + event["scope"], + event["title"], + _format_when(event), + event["host_display_name"], + event["event_id"], + ) + for event in payload + ], + ) + ) + + +@app.command("show") +def invites_show( + ctx: typer.Context, + event_id: str = typer.Argument(..., help=EVENT_ID_HELP), + username: UsernameOption = None, + session_dir: SessionDirOption = None, + http_proxy: HttpProxyOption = None, + https_proxy: HttpsProxyOption = None, + no_verify_ssl: NoVerifySslOption = False, + output_format: OutputFormatOption = DEFAULT_OUTPUT_FORMAT, + log_level: LogLevelOption = DEFAULT_LOG_LEVEL, +) -> None: + """Show one event in full, with its share details.""" + + store_command_options( + ctx, + username=username, + session_dir=session_dir, + http_proxy=http_proxy, + https_proxy=https_proxy, + no_verify_ssl=no_verify_ssl, + output_format=output_format, + log_level=log_level, + ) + state = get_state(ctx) + api = state.get_api() + event = _invites_call(api, lambda: _lookup_event(api, event_id)) + payload = normalize_invite_event_details(event) + if state.json_output: + state.write_json(payload) + return + state.console.print( + console_kv_table( + payload["title"] or "Event", + [ + ("Scope", payload["scope"]), + ("Host", payload["host_display_name"]), + ("When", _format_when(payload)), + ("Ends", payload["ends_at"]), + ("Location", payload["location"]), + ("City", payload["city"]), + ("Notes", payload["notes"]), + ("Cancelled", payload["is_cancelled"]), + ("Published", payload["is_published"]), + ("New RSVPs blocked", payload["block_new_rsvps"]), + ("Participants", payload["participant_count"]), + ("Invite link", payload["share_url"]), + ("ID", payload["event_id"]), + ], + ) + ) + + +@app.command("rsvps") +def invites_rsvps( + ctx: typer.Context, + event_id: str = typer.Argument(..., help=EVENT_ID_HELP), + username: UsernameOption = None, + session_dir: SessionDirOption = None, + http_proxy: HttpProxyOption = None, + https_proxy: HttpsProxyOption = None, + no_verify_ssl: NoVerifySslOption = False, + output_format: OutputFormatOption = DEFAULT_OUTPUT_FORMAT, + log_level: LogLevelOption = DEFAULT_LOG_LEVEL, +) -> None: + """List who has responded to an event.""" + + store_command_options( + ctx, + username=username, + session_dir=session_dir, + http_proxy=http_proxy, + https_proxy=https_proxy, + no_verify_ssl=no_verify_ssl, + output_format=output_format, + log_level=log_level, + ) + state = get_state(ctx) + api = state.get_api() + payload = _invites_call( + api, + lambda: [ + normalize_invite_rsvp(rsvp) + for rsvp in api.invites.rsvps(_lookup_event(api, event_id)) + ], + ) + if state.json_output: + state.write_json(payload) + return + state.console.print( + console_table( + "RSVPs", + ["Name", "Status", "Adults", "Kids", "Message"], + [ + ( + rsvp["name"], + rsvp["status"], + rsvp["additional_adults"], + rsvp["additional_kids"], + rsvp["message"], + ) + for rsvp in payload + ], + ) + ) + + +@app.command("resolve") +def invites_resolve( + ctx: typer.Context, + short_guid: str = typer.Argument(..., help=SHORT_GUID_HELP), + username: UsernameOption = None, + session_dir: SessionDirOption = None, + http_proxy: HttpProxyOption = None, + https_proxy: HttpsProxyOption = None, + no_verify_ssl: NoVerifySslOption = False, + output_format: OutputFormatOption = DEFAULT_OUTPUT_FORMAT, + log_level: LogLevelOption = DEFAULT_LOG_LEVEL, +) -> None: + """Preview an invite link without joining it.""" + + store_command_options( + ctx, + username=username, + session_dir=session_dir, + http_proxy=http_proxy, + https_proxy=https_proxy, + no_verify_ssl=no_verify_ssl, + output_format=output_format, + log_level=log_level, + ) + state = get_state(ctx) + api = state.get_api() + payload = normalize_invite_share( + _invites_call(api, lambda: api.invites.resolve(_short_guid_from(short_guid))) + ) + if state.json_output: + state.write_json(payload) + return + state.console.print( + console_kv_table( + "Invite", + [ + ("Event", payload["event_id"]), + ( + "Host", + f"{payload['owner_given_name']} {payload['owner_family_name']}", + ), + ("Your status", payload["participant_status"]), + ("Your role", payload["participant_type"]), + ("Permission", payload["participant_permission"]), + ("Short guid", payload["short_guid"]), + ], + ) + ) diff --git a/pyicloud/cli/normalize.py b/pyicloud/cli/normalize.py index cafae4d5..18587d98 100644 --- a/pyicloud/cli/normalize.py +++ b/pyicloud/cli/normalize.py @@ -357,3 +357,72 @@ def sort_key(item: Any) -> datetime: candidates.sort(key=sort_key, reverse=True) return candidates[:limit] + + +def normalize_invite_event(event: Any) -> dict[str, Any]: + """Normalize one Invites event for listing.""" + + time = getattr(event, "time", None) + place = getattr(event, "place", None) + return { + "event_id": event.event_id, + "scope": event.scope.value, + "title": event.title, + "host_display_name": event.host_display_name, + "starts_at": getattr(time, "start", None), + "ends_at": getattr(time, "end", None), + "is_all_day": getattr(time, "is_all_day", None), + "location": getattr(place, "title", None), + "is_cancelled": event.is_cancelled, + } + + +def normalize_invite_event_details(event: Any) -> dict[str, Any]: + """Normalize one Invites event with its share, for a single-event view.""" + + share = getattr(event, "share", None) + place = getattr(event, "place", None) + payload = normalize_invite_event(event) + payload.update({ + "notes": event.notes, + "is_published": event.is_published, + "is_private": event.is_private, + "block_new_rsvps": event.block_new_rsvps, + "max_attendees": event.max_attendees, + "city": getattr(place, "city", None), + "share_short_guid": getattr(share, "short_guid", None), + "share_url": getattr(share, "url", None), + "participant_count": len(getattr(share, "participants", ()) or ()), + }) + return payload + + +def normalize_invite_rsvp(rsvp: Any) -> dict[str, Any]: + """Normalize one Invites RSVP.""" + + return { + "participant_id": rsvp.participant_id, + "name": rsvp.name, + "status": rsvp.status.name, + "message": rsvp.message, + "additional_adults": rsvp.num_additional_adults, + "additional_kids": rsvp.num_additional_kids, + } + + +def normalize_invite_share(share: Any) -> dict[str, Any]: + """Normalize a resolved invite link. + + ``participant_status`` and friends are plain strings from Apple here, + unlike the enums used elsewhere in the Invites service. + """ + + return { + "short_guid": share.short_guid, + "event_id": share.event_id, + "owner_given_name": share.owner_given_name, + "owner_family_name": share.owner_family_name, + "participant_status": share.participant_status, + "participant_type": share.participant_type, + "participant_permission": share.participant_permission, + } diff --git a/pyicloud/services/invites/client.py b/pyicloud/services/invites/client.py index 50133de0..fa272797 100644 --- a/pyicloud/services/invites/client.py +++ b/pyicloud/services/invites/client.py @@ -54,6 +54,29 @@ _RATE_LIMITED_STATUS = 429 +def _retry_after_of(exc: PyiCloudAPIResponseException) -> float | None: + """Return the ``Retry-After`` Apple sent with a rate limit, if any. + + The session attaches the response to the exception, so the header survives + the trip. Dropping it would leave callers backing off on a guess when Apple + has told them exactly how long to wait. + """ + + response = getattr(exc, "response", None) + if response is None: + return None + try: + header = response.headers.get("Retry-After") + except AttributeError: + return None + if not header: + return None + try: + return float(header) + except (TypeError, ValueError): + return None + + def _status_of(exc: PyiCloudAPIResponseException) -> int | None: """Return the exception's status as an int, whichever form it arrived in. @@ -168,7 +191,9 @@ def _raise_invites_error(exc: Exception) -> NoReturn: if status in _UNAUTHORIZED_STATUSES: raise InvitesAuthError(str(exc)) from cause if status == _RATE_LIMITED_STATUS: - raise InvitesRateLimited(str(exc)) from cause + raise InvitesRateLimited( + str(exc), retry_after=_retry_after_of(exc) + ) from cause raise InvitesApiError(str(exc)) from cause raise exc @@ -340,7 +365,13 @@ def _post_public(self, path: str, payload: dict[str, Any]) -> dict[str, Any]: url = f"{url}?{urlencode(params)}" LOGGER.debug("CloudKit Invites POST %s", path) - resp = self._session.post(url, json=payload, timeout=self._timeout) + try: + resp = self._session.post(url, json=payload, timeout=self._timeout) + except PyiCloudAPIResponseException as exc: + # Same trap as the scoped wrappers above: PyiCloudSession raises on + # a non-ok JSON response, so every status check below is dead code + # for a 4xx. Map it the way those checks would have. + self._raise_invites_error(exc) code = getattr(resp, "status_code", 0) if not isinstance(code, int): code = 200 diff --git a/tests/test_cmdline.py b/tests/test_cmdline.py index b4ca7f7f..25e5c033 100644 --- a/tests/test_cmdline.py +++ b/tests/test_cmdline.py @@ -21,6 +21,18 @@ from typer.testing import CliRunner, Result from pyicloud.endpoints import WEBSERVICES +from pyicloud.exceptions import PyiCloudServiceUnavailable +from pyicloud.services.invites import ( + Event, + EventNotFound, + EventPlace, + EventScope, + EventShare, + EventTime, + ResolvedShare, + Rsvp, + RsvpStatus, +) from pyicloud.services.notes.models import Attachment as NoteAttachment from pyicloud.services.notes.models import ChangeEvent as NoteChangeEvent from pyicloud.services.notes.models import ( @@ -53,6 +65,7 @@ account_index_module = importlib.import_module("pyicloud.cli.account_index") cli_module = importlib.import_module("pyicloud.cli.app") +invites_module = importlib.import_module("pyicloud.cli.commands.invites") context_module = importlib.import_module("pyicloud.cli.context") output_module = importlib.import_module("pyicloud.cli.output") app = cli_module.app @@ -619,6 +632,75 @@ def sync_cursor(self) -> str: return self.cursor +class FakeInvites: + """Invites service fixture.""" + + def __init__(self) -> None: + self._events = [ + Event( + event_id="AAAA1111-0000-0000-0000-000000000001", + scope=EventScope.PRIVATE, + title="Standup", + time=EventTime(start=datetime(2026, 5, 1, 9, 0, tzinfo=timezone.utc)), + place=EventPlace(title="Room 2", city="Dublin"), + is_cancelled=False, + is_published=True, + ), + Event( + event_id="AAAA2222-0000-0000-0000-000000000002", + scope=EventScope.SHARED, + title="Birthday", + host_display_name="Sam Appleseed", + time=EventTime( + start=datetime(2026, 6, 2, tzinfo=timezone.utc), is_all_day=True + ), + place=EventPlace(title="The Park", city="Cork"), + share=EventShare(short_guid="008FIXTURE", public_permission="READ"), + is_cancelled=False, + is_published=True, + ), + ] + + def events(self) -> list[Event]: + """Return both fixture events.""" + return list(self._events) + + def event(self, event_id: str) -> Event: + """Return one fixture event by exact id.""" + for event in self._events: + if event.event_id == event_id: + return event + raise EventNotFound(f"Event not found: {event_id!r}") + + def rsvps(self, event: Event) -> list[Rsvp]: + """Return one response for any event.""" + assert event is not None + return [ + Rsvp( + record_name="RSVP/1", + participant_id="P1", + name="Dana Appleseed", + status=RsvpStatus.GOING, + message="See you there", + num_additional_adults=1, + num_additional_kids=0, + ) + ] + + def resolve(self, short_guid: str) -> ResolvedShare: + """Return a preview of a share.""" + return ResolvedShare( + short_guid=short_guid, + event_id="AAAA2222-0000-0000-0000-000000000002", + owner_given_name="Sam", + owner_family_name="Appleseed", + participant_status="INVITED", + participant_type="USER", + participant_permission="READ_WRITE", + share=EventShare(short_guid=short_guid, public_permission="READ"), + ) + + class FakeReminders: """Reminders service fixture.""" @@ -1213,6 +1295,7 @@ def __init__( self.hidemyemail = FakeHideMyEmail() self.notes = FakeNotes() self.reminders = FakeReminders() + self.invites = FakeInvites() def _logout( self, @@ -1241,6 +1324,23 @@ def _logout( } +class UnavailableInvitesAPI(FakeAPI): + """An account where the Invites service cannot be constructed. + + base.py raises PyiCloudServiceUnavailable from the property itself, not + from a method on the service, so the failure has to be provoked there. + """ + + @property + def invites(self) -> FakeInvites: + """Fail the way base.py does when the webservice is missing.""" + raise PyiCloudServiceUnavailable("Invites service not available") + + @invites.setter + def invites(self, value: FakeInvites) -> None: + """Absorb the assignment FakeAPI makes in its own constructor.""" + + def _runner() -> CliRunner: return CliRunner() @@ -4284,11 +4384,14 @@ def _advertised_webservices() -> dict[str, Any]: def _unwrapped(result: Any) -> str: """Return command output with rich's line wrapping collapsed. - Terminal width is not part of what these tests assert, and a sentence split - across two lines would otherwise fail on a narrower runner than this one. + Terminal width is not part of what these tests assert. A cell that wraps + puts the table's own border characters between the fragments, so those go + first and the remaining whitespace is then collapsed. """ - return " ".join(_plain_output(result).split()) + text = _plain_output(result) + stripped = "".join(" " if "\u2500" <= ch <= "\u257f" else ch for ch in text) + return " ".join(stripped.split()) def _doctor_api(webservices: dict[str, Any] | None = None) -> FakeAPI: @@ -4421,3 +4524,190 @@ def test_doctor_without_a_session_still_reports_what_it_can() -> None: assert "Authenticated" in text assert "icloud auth login" in text assert "report is incomplete" in text + + +def test_invites_list_shows_both_scopes() -> None: + """Listing covers events you host and events shared with you.""" + + result = _invoke(FakeAPI(), "invites", "list") + text = _unwrapped(result) + + assert result.exit_code == 0 + assert "private" in text + assert "shared" in text + assert "Standup" in text + assert "Sam Appleseed" in text + + +def test_invites_start_times_are_rendered_compactly() -> None: + """A full ISO timestamp wraps a row onto three lines and hides the id. + + The id column is only worth showing if enough of it survives to pass to + the next command, so the start time is rendered to the minute. Checked on + the formatter rather than the table, because a wrapped cell's fragments + land in different columns and cannot be reassembled from the output. + """ + + fmt = invites_module._format_when + + assert ( + fmt({"starts_at": datetime(2026, 5, 1, 9, 0, tzinfo=timezone.utc)}) + == "2026-05-01 09:00" + ) + assert ( + fmt({ + "starts_at": datetime(2026, 6, 2, tzinfo=timezone.utc), + "is_all_day": True, + }) + == "2026-06-02 (all day)" + ) + assert fmt({"starts_at": None}) == "" + + +def test_invites_list_shows_the_compact_time_and_the_id() -> None: + """The rendered row carries both, which is the point of the formatting.""" + + text = _unwrapped(_invoke(FakeAPI(), "invites", "list")) + + assert "2026-05-01 09:00" in text + assert "09:00:00+00:00" not in text + assert "AAAA1111" in text + + +def test_invites_show_accepts_a_short_id_prefix() -> None: + """Listing truncates ids, so a few characters have to be enough to act on. + + This is the whole reason prefix matching exists: the id column is cut to + fit the terminal, so requiring all 36 characters would make it decorative. + """ + + result = _invoke(FakeAPI(), "invites", "show", "AAAA2") + text = _unwrapped(result) + + assert result.exit_code == 0 + assert "Birthday" in text + assert "Sam Appleseed" in text + assert "The Park" in text + + +def test_invites_show_matches_an_id_case_insensitively() -> None: + """Ids render upper-case but nobody types them that way.""" + + upper = _invoke(FakeAPI(), "invites", "show", "AAAA2222") + lower = _invoke(FakeAPI(), "invites", "show", "aaaa2222") + + assert upper.exit_code == 0 + assert lower.exit_code == 0 + assert _unwrapped(upper) == _unwrapped(lower) + + +def test_invites_show_accepts_a_full_id() -> None: + """A complete id is still an unambiguous prefix of itself.""" + + result = _invoke( + FakeAPI(), "invites", "show", "AAAA2222-0000-0000-0000-000000000002" + ) + + assert result.exit_code == 0 + assert "Birthday" in _unwrapped(result) + + +def test_invites_show_rejects_an_ambiguous_prefix() -> None: + """A prefix matching two events names both rather than guessing.""" + + # Both ids start "AAAA"; only the fifth character tells them apart. + result = _invoke(FakeAPI(), "invites", "show", "AAAA") + + assert result.exit_code != 0 + message = str(result.exception) + assert "matches more than one event" in message + assert "000000000001" in message + assert "000000000002" in message + + +def test_invites_show_rejects_an_unknown_prefix() -> None: + """An id that matches nothing points at the command that lists them.""" + + result = _invoke(FakeAPI(), "invites", "show", "ZZZZ") + + assert result.exit_code != 0 + assert "icloud invites list" in str(result.exception) + + +def test_invites_rsvps_lists_responses() -> None: + """Responses render with status and guest counts.""" + + result = _invoke( + FakeAPI(), "invites", "rsvps", "AAAA2222-0000-0000-0000-000000000002" + ) + text = _unwrapped(result) + + assert result.exit_code == 0 + assert "Dana Appleseed" in text + assert "GOING" in text + assert "See you there" in text + + +def test_invites_resolve_previews_a_link() -> None: + """Resolving shows who is inviting you without joining.""" + + result = _invoke(FakeAPI(), "invites", "resolve", "008FIXTURE") + text = _unwrapped(result) + + assert result.exit_code == 0 + assert "Sam Appleseed" in text + assert "INVITED" in text + + +def test_invites_json_output() -> None: + """JSON mode carries the fields the tables show.""" + + result = _invoke(FakeAPI(), "invites", "list", output_format="json") + payload = json.loads(result.stdout) + + assert result.exit_code == 0 + assert {event["scope"] for event in payload} == {"private", "shared"} + shared = next(e for e in payload if e["scope"] == "shared") + assert shared["title"] == "Birthday" + assert shared["is_all_day"] is True + assert shared["location"] == "The Park" + + +def test_invites_reports_an_unavailable_service_without_a_traceback() -> None: + """The service is built by a property, so the call has to be deferred. + + `api.invites` can raise PyiCloudServiceUnavailable while constructing the + service. Passing `api.invites.events` to service_call instead of a lambda + would evaluate the property outside the guard and let that escape as a + traceback, which is why the lambda there is not redundant. + """ + + result = _invoke(UnavailableInvitesAPI(), "invites", "list") + + assert result.exit_code != 0 + assert "Invites service unavailable" in str(result.exception) + + +def test_invites_resolve_accepts_a_whole_invite_link() -> None: + """People copy the link, not the trailing segment out of it.""" + + from_link = _invoke( + FakeAPI(), + "invites", + "resolve", + "https://www.icloud.com/invites/008FIXTURE", + ) + from_guid = _invoke(FakeAPI(), "invites", "resolve", "008FIXTURE") + + assert from_link.exit_code == 0 + assert from_guid.exit_code == 0 + assert _unwrapped(from_link) == _unwrapped(from_guid) + + +def test_invites_resolve_rejects_an_empty_id_without_calling_apple() -> None: + """An empty id should cost a message, not a round trip and a 400.""" + + result = _invoke(FakeAPI(), "invites", "resolve", " ") + + assert result.exit_code != 0 + assert "No invite id given" in str(result.exception) diff --git a/tests/test_invites.py b/tests/test_invites.py index 1da9aea0..20ffa364 100644 --- a/tests/test_invites.py +++ b/tests/test_invites.py @@ -453,6 +453,70 @@ def test_an_unknown_shared_zone_falls_back_to_no_owner(self) -> None: self.assertEqual(zone_id.zoneName, "MISSING-ZONE") self.assertIsNone(zone_id.ownerRecordName) + def test_the_public_endpoint_translates_a_transport_error(self) -> None: + """resolve() and accept() go through a different path to the rest. + + `_post_public` inspects `resp.status_code` itself, but PyiCloudSession + raises on a non-ok JSON response before it ever returns, so those + checks are dead for a 4xx and the raw exception reached callers. The + five scoped wrappers were fixed for this; this sixth site uses a + different shape and was missed. + """ + session = MagicMock() + session.post.side_effect = PyiCloudAPIResponseException("Bad Request", 400) + self._monkeypatch.setattr(self.service.raw, "_session", session) + + with self.assertRaises(InvitesApiError): + self.service.raw.resolve(["008FIXTURE"]) + with self.assertRaises(InvitesApiError): + self.service.raw.accept(["008FIXTURE"]) + + def test_the_public_endpoint_maps_auth_and_rate_limits(self) -> None: + """The same mapping as everywhere else, not a generic error.""" + for code, expected in ( + (401, InvitesAuthError), + (429, InvitesRateLimited), + ): + session = MagicMock() + session.post.side_effect = PyiCloudAPIResponseException("nope", code) + self._monkeypatch.setattr(self.service.raw, "_session", session) + with self.subTest(code=code), self.assertRaises(expected): + self.service.raw.resolve(["008FIXTURE"]) + + def test_a_rate_limit_keeps_the_retry_after_apple_sent(self) -> None: + """Apple says how long to wait; dropping it makes callers guess. + + The header survives on the exception's response, so the mapping has to + read it rather than raise a bare InvitesRateLimited. + """ + response = MagicMock() + response.headers = {"Retry-After": "42"} + session = MagicMock() + session.post.side_effect = PyiCloudAPIResponseException( + "slow down", 429, response + ) + self._monkeypatch.setattr(self.service.raw, "_session", session) + + with self.assertRaises(InvitesRateLimited) as caught: + self.service.raw.resolve(["008FIXTURE"]) + + self.assertEqual(caught.exception.retry_after, 42.0) + + def test_a_rate_limit_without_a_retry_after_is_still_mapped(self) -> None: + """A missing or unreadable header must not break the mapping.""" + for headers in ({}, {"Retry-After": "soon"}): + session = MagicMock() + response = MagicMock() + response.headers = headers + session.post.side_effect = PyiCloudAPIResponseException( + "slow down", 429, response + ) + self._monkeypatch.setattr(self.service.raw, "_session", session) + with self.subTest(headers=headers): + with self.assertRaises(InvitesRateLimited) as caught: + self.service.raw.resolve(["008FIXTURE"]) + self.assertIsNone(caught.exception.retry_after) + def test_a_status_code_maps_the_same_as_a_string_or_an_int(self) -> None: """`PyiCloudAPIResponseException.code` is typed `int | str | None`.