diff --git a/README.md b/README.md index 598c200..7c248e7 100644 --- a/README.md +++ b/README.md @@ -59,6 +59,12 @@ Reverse API Engineer stays what it is: local, MIT, no account required, and the client it generates is yours to keep. Reach for the cloud when you'd rather not own the maintenance. +Because a capture run costs time and tokens, agent mode checks the marketplace +for the target site first and tells you when something already exists — press +enter to carry on capturing anyway. The lookup is anonymous, needs no account, +and only ever reports functions whose own domain matches your target. Turn it +off with `RAE_NO_CLOUD=1` or **Cloud Suggestions** in `/settings`. + ## Install ```bash @@ -121,6 +127,7 @@ Settings live in `~/.reverse-api/config.json` and can be edited via `/settings` "agent_browser_npx_package": "agent-browser@0", "agent_browser_notes": "", "claude_code_model": "claude-sonnet-4-6", + "cloud_suggestions": true, "collector_model": "claude-sonnet-4-6", "ollama_auto_start": true, "ollama_base_url": "http://127.0.0.1:11434", @@ -150,6 +157,7 @@ Slash commands inside the CLI: - `/settings`: configure model, SDK, agent provider, and sync settings. - `/history`: list past runs with timestamps, costs, and status. - `/messages `: view detailed message logs for a run. +- `/cloud`: show the hosted version and how to search its marketplace. - `/help` (alias: `/commands`): show the command list. - `/exit` (alias: `/quit`): leave the CLI. @@ -161,6 +169,11 @@ reverse-api-engineer agent --prompt "capture the public jobs api" \ reverse-api-engineer list --json reverse-api-engineer show --json + +# Check whether a hosted function already covers a site (public, no account). +reverse-api-engineer marketplace search --site https://www.nfl.com +reverse-api-engineer marketplace search --category Jobs +reverse-api-engineer marketplace search "nfl standings" --json | jq reverse-api-engineer run --file api_client.py \ --no-interactive --auto-install -- --org acme ``` diff --git a/src/reverse_api/cli.py b/src/reverse_api/cli.py index a756e35..afd2875 100644 --- a/src/reverse_api/cli.py +++ b/src/reverse_api/cli.py @@ -19,7 +19,7 @@ from rich.console import Console from rich.markup import escape -from . import __version__ +from . import __version__, cloud from .config import DEFAULT_OPENCODE_MODEL, DEFAULT_OPENCODE_PROVIDER, ConfigManager from .engineer import run_reverse_engineering from .messages import MessageStore @@ -768,6 +768,7 @@ def prompt_interactive_options( "/settings", "/history", "/messages", + "/cloud", "/help", "/exit", "/quit", @@ -1075,6 +1076,8 @@ def repl_loop(): handle_settings(mode_color) elif cmd == "/history": handle_history(mode_color) + elif cmd == "/cloud": + _print_cloud_overview() elif cmd == "/help" or cmd == "/commands": handle_help(mode_color) elif cmd.startswith("/messages"): @@ -1086,7 +1089,7 @@ def repl_loop(): else: # Unknown command - show error and available commands console.print(f" [red]Unknown command:[/red] {cmd}") - console.print(" [dim]Available commands: /settings, /history, /messages, /help, /commands, /exit[/dim]") + console.print(" [dim]Available commands: /settings, /history, /messages, /cloud, /help, /commands, /exit[/dim]") continue mode = options.get("mode", "agent") @@ -1186,6 +1189,7 @@ def _handle_settings_action(mode_color=THEME_PRIMARY) -> bool: choices = [ Choice(title="Agent Provider", value="agent_provider"), Choice(title="Claude Code Model", value="claude_code_model"), + Choice(title="Cloud Suggestions", value="cloud_suggestions"), Choice(title="Copilot Model", value="copilot_model"), Choice(title="Cursor Model", value="cursor_model"), Choice(title="Cursor Web Search", value="cursor_web_search"), @@ -1414,6 +1418,34 @@ def _handle_settings_action(mode_color=THEME_PRIMARY) -> bool: config_manager.set("cursor_web_search", pick) console.print(f" [dim]updated[/dim] cursor web search: {'on' if pick else 'off'}\n") + elif action == "cloud_suggestions": + console.print( + " [dim]Before a capture, check whether anything.notte.cc already hosts a\n" + " function for that site. Public lookup, no account. RAE_NO_CLOUD=1\n" + " overrides this setting.[/dim]\n" + ) + cloud_choices = [ + Choice(title="Enabled", value=True), + Choice(title="Disabled", value=False), + Choice(title="Back", value="back"), + ] + choice = questionary.select( + "", + choices=cloud_choices, + pointer=">", + qmark="", + style=questionary.Style( + [ + ("pointer", f"fg:{mode_color} bold"), + ("highlighted", f"fg:{mode_color} bold"), + ] + ), + ).ask() + if choice is not None and choice != "back": + config_manager.set("cloud_suggestions", choice) + status = "enabled" if choice else "disabled" + console.print(f" [dim]updated[/dim] cloud suggestions: {status}\n") + elif action == "real_time_sync": current = config_manager.get("real_time_sync", True) sync_choices = [ @@ -1557,6 +1589,10 @@ def handle_help(mode_color=THEME_PRIMARY): ) commands_table.add_row("", "") + commands_table.add_row( + "/cloud", + "Show the hosted version and how to search its marketplace\n[dim]Usage: /cloud[/dim]", + ) commands_table.add_row( "/help or /commands", "Show this help message\n[dim]Usage: /help[/dim]", @@ -1802,6 +1838,108 @@ def agent(prompt, url, model, output_dir, no_interactive, as_json, json_stream, sys.exit(0 if payload["status"] == "ok" else 1) +@main.group(invoke_without_command=True) +@click.pass_context +def marketplace(ctx: click.Context): + """Browse ready-made API functions on the hosted marketplace. + + The marketplace behind anything.notte.cc holds functions other people + already built. Searching it is public — no account, no API key — so it is + worth a look before reverse-engineering a site from scratch. + """ + if ctx.invoked_subcommand is None: + _print_cloud_overview() + + +@marketplace.command("search") +@click.argument("query", required=False) +@click.option( + "--site", + default=None, + help="Restrict to one site (URL or hostname). Only exact domain matches are returned.", +) +@click.option( + "--category", + default=None, + help="Restrict to one marketplace category, e.g. Jobs, Finance, E-commerce.", +) +@click.option("--limit", "-n", default=5, show_default=True, help="Maximum results.") +@click.option( + "--json", + "as_json", + is_flag=True, + help="Emit results as a single JSON document on stdout.", +) +def marketplace_search(query, site, category, limit, as_json): + """Search the marketplace for an existing function. + + \b + Examples: + reverse-api-engineer marketplace search "nfl standings" + reverse-api-engineer marketplace search --site https://www.nfl.com + reverse-api-engineer marketplace search --category Jobs + reverse-api-engineer marketplace search instagram --json | jq + """ + if not query and not site and not category: + if as_json: + click.echo(json.dumps({"error": "provide a QUERY, --site, or --category", "results": []})) + sys.exit(2) + click.echo("error: provide a QUERY, --site, or --category", err=True) + sys.exit(2) + + with console.status(" [dim]searching the marketplace...[/dim]", spinner="dots"): + if site and not query and not category: + # Domain-only lookups get the extra no-wrong-site guard. + matches = cloud.search_for_site(site, limit=limit) + else: + matches = cloud.search(query, base_url=site, category=category, limit=limit) + + if as_json: + click.echo( + json.dumps( + { + "query": query, + "site": site, + "category": category, + "count": len(matches), + "marketplace_url": cloud.MARKETPLACE_URL, + "results": [ + { + "function_id": fn.function_id, + "label": fn.label, + "description": fn.description, + "domain": fn.domain, + "run_count": fn.run_count, + "url": fn.url, + } + for fn in matches + ], + } + ) + ) + return + + if not matches: + console.print() + console.print(" [dim]no hosted function matches that yet[/dim]") + console.print(f" [dim]build one at {cloud.CLOUD_URL}, or capture it yourself with agent mode[/dim]") + console.print() + return + + console.print() + for fn in matches: + console.print( + f" [white]{escape(fn.label)}[/white] [dim]{escape(fn.domain)} · {fn.run_count} runs[/dim]" + ) + if fn.description: + desc = fn.description.replace("\n", " ").strip() + if len(desc) > 100: + desc = desc[:97] + "..." + console.print(f" [dim]{escape(desc)}[/dim]") + console.print(f" [dim]{fn.url}[/dim]") + console.print() + + @main.command( epilog="""\b Examples: @@ -2003,6 +2141,105 @@ def run_collector(prompt=None, model=None, output_dir=None): } +def _print_cloud_overview(): + """Explain the hosted version. Backs both `/cloud` and the empty search.""" + console.print() + console.print(f" [{THEME_SECONDARY}]anything[/{THEME_SECONDARY}] [dim]the hosted version of this tool[/dim]") + console.print(" [dim]describe the task, get a deployed API function instead of a local file[/dim]") + console.print() + console.print(f" [dim]home[/dim] [white]{cloud.CLOUD_URL}[/white]") + console.print(f" [dim]marketplace[/dim] [white]{cloud.MARKETPLACE_URL}[/white] [dim]ready-made functions, no account needed[/dim]") + console.print(f" [dim]mcp[/dim] [white]{cloud.MCP_URL}[/white] [dim]point an agent here to search and run them[/dim]") + console.print() + console.print(" [dim]search from here:[/dim] [white]reverse-api-engineer marketplace search [/white]") + console.print() + + +def _render_marketplace_matches(matches, *, domain: str): + """Show existing hosted functions for a site the user is about to capture.""" + plural = "function" if len(matches) == 1 else "functions" + console.print() + console.print( + f" [{THEME_SECONDARY}]anything[/{THEME_SECONDARY}] [dim]already has[/dim] " + f"[white]{len(matches)}[/white] [dim]{plural} for[/dim] [white]{domain}[/white]" + ) + for fn in matches: + console.print(f" [dim]·[/dim] [white]{escape(fn.label)}[/white] [dim]({fn.run_count} runs)[/dim]") + if fn.description: + desc = fn.description.replace("\n", " ").strip() + if len(desc) > 96: + desc = desc[:93] + "..." + console.print(f" [dim]{escape(desc)}[/dim]") + console.print(f" [dim]{fn.url}[/dim]") + console.print() + + +def _maybe_suggest_marketplace(url, *, interactive: bool) -> bool: + """Offer an existing hosted function before capturing a site from scratch. + + Returns True when the user chose the marketplace instead, meaning the + caller should abandon the capture. Never raises: a lookup problem must not + cost someone their run. + """ + if not url or not interactive: + return False + if not cloud.suggestions_enabled(config_manager): + return False + + try: + domain = cloud.registrable_domain(url) + if not domain: + return False + # The lookup is usually instant but cold-starts at several seconds, so + # show a spinner rather than an unexplained pause. Ctrl+C skips it. + try: + with console.status(f" [dim]checking if {domain} is already covered...[/dim]", spinner="dots"): + matches = cloud.search_for_site(url) + except KeyboardInterrupt: + return False + if not matches: + return False + + _render_marketplace_matches(matches, domain=domain) + + # Default is No, so a bare enter carries on capturing — the suggestion + # is an offer, not a toll gate. + open_it = questionary.confirm( + "Open the marketplace instead of capturing?", + default=False, + qmark="", + style=questionary.Style([("question", "")]), + ).ask() + except Exception: + return False + + if not open_it: + return False + + target = cloud.cloud_link(matches[0].url, "cli_precapture") + console.print(f" [dim]opening[/dim] [white]{matches[0].url}[/white]") + try: + import webbrowser + + webbrowser.open(target) + except Exception: + pass + console.print(" [dim]capture skipped[/dim]") + console.print() + return True + + +def _print_cloud_deploy_hint(): + """One quiet line after a successful capture, pointing at the hosted path.""" + if not cloud.suggestions_enabled(config_manager): + return + console.print( + f" [dim]want this hosted, scheduled, and repaired when the site changes? " + f"{cloud.CLOUD_URL}[/dim]" + ) + console.print() + + def run_auto_capture( prompt=None, url=None, @@ -2029,6 +2266,11 @@ def run_auto_capture( url = options.get("url") model = options["model"] + # Before burning a capture run, check whether the site is already covered + # by a hosted function. Skipped entirely when non-interactive or headless. + if _maybe_suggest_marketplace(url, interactive=interactive and not headless): + return None + if agent_provider == "chrome-mcp" and not headless: console.print() console.print(" [dim]chrome devtools mcp (auto-connect)[/dim]") @@ -2188,6 +2430,8 @@ def run_auto_capture( usage=result.get("usage", {}), paths={"script_path": result.get("script_path")}, ) + if interactive and not interrupted and result.get("script_path"): + _print_cloud_deploy_hint() return { "run_id": run_id, diff --git a/src/reverse_api/cloud.py b/src/reverse_api/cloud.py new file mode 100644 index 0000000..8444227 --- /dev/null +++ b/src/reverse_api/cloud.py @@ -0,0 +1,250 @@ +"""Anything marketplace lookups. + +Anything (https://anything.notte.cc) is the hosted version of this tool: you +describe a task and get back a deployed API function instead of a local file. +Its marketplace already holds several hundred ready-made functions, so before +we spend a capture run reverse-engineering a site from scratch it is worth +asking whether somebody already did it. + +Everything here is best-effort and strictly optional. The marketplace search +endpoint is public (no key, no account), every call is wrapped in a short +timeout, and any failure degrades to "no suggestions" rather than an error — +a capture must never fail because a marketing lookup did. + +Set ``RAE_NO_CLOUD=1`` (or turn off ``cloud_suggestions`` in settings) to +disable the network call entirely. +""" + +from __future__ import annotations + +import os +from dataclasses import dataclass +from urllib.parse import urlparse + +CLOUD_HOST = "anything.notte.cc" +CLOUD_URL = f"https://{CLOUD_HOST}" +MARKETPLACE_URL = f"{CLOUD_URL}/marketplace" +MCP_URL = f"{CLOUD_URL}/mcp" +SEARCH_ENDPOINT = f"{CLOUD_URL}/api/marketplace/search" + +#: The search endpoint is serverless: ~0.3s warm, but several seconds after an +#: idle period and occasionally worse. A tight timeout would drop real matches +#: exactly when the answer is most useful, so the budget is generous and +#: callers show a spinner rather than hiding the wait. A very cold call can +#: still exceed this, which is fine — the caller just shows no suggestions. +DEFAULT_TIMEOUT = 12.0 + +#: Second-level labels that are effectively part of the public suffix. Used to +#: keep "bbc.co.uk" from collapsing to the useless registrable domain "co.uk". +_COMPOUND_SLDS = frozenset({"co", "com", "org", "net", "gov", "edu", "ac"}) + + +@dataclass(frozen=True) +class MarketplaceFunction: + """One ready-made function published on the Anything marketplace.""" + + function_id: str + label: str + description: str + domain: str + run_count: int = 0 + + @property + def url(self) -> str: + """Canonical marketplace page for this function.""" + return f"{MARKETPLACE_URL}/{self.function_id}" + + @classmethod + def from_api(cls, raw: dict) -> MarketplaceFunction | None: + """Build from one search-result object, or None if it is unusable.""" + function_id = str(raw.get("function_id") or "").strip() + label = str(raw.get("label") or "").strip() + if not function_id or not label: + return None + try: + run_count = int(raw.get("run_count") or 0) + except (TypeError, ValueError): + run_count = 0 + return cls( + function_id=function_id, + label=label, + description=str(raw.get("description") or "").strip(), + domain=_normalize_host(str(raw.get("domain") or "")), + run_count=run_count, + ) + + +def suggestions_enabled(config_manager=None) -> bool: + """Whether marketplace lookups are allowed at all. + + The env var wins over config so CI and scripted wrappers can turn the + network call off without touching the user's settings file. + """ + if _env_flag("RAE_NO_CLOUD"): + return False + if config_manager is None: + return True + return bool(config_manager.get("cloud_suggestions", True)) + + +def cloud_link(url: str, campaign: str) -> str: + """Tag an outbound cloud URL with the placement it came from.""" + joiner = "&" if "?" in url else "?" + return f"{url}{joiner}utm_source=rae&utm_medium=cli&utm_campaign={campaign}" + + +def registrable_domain(url_or_host: str) -> str | None: + """Reduce a URL or hostname to the domain a marketplace entry would use. + + ``https://jobs.ashbyhq.com/openai?x=1`` -> ``ashbyhq.com`` + ``www.bbc.co.uk`` -> ``bbc.co.uk`` + + Returns None when there is no usable hostname (bare paths, IPs, garbage). + """ + host = _extract_host(url_or_host) + if not host: + return None + + labels = host.split(".") + if len(labels) < 2: + return None + # An IPv4 literal has no registrable domain worth searching for. + if all(label.isdigit() for label in labels): + return None + + if len(labels) >= 3 and labels[-2] in _COMPOUND_SLDS and len(labels[-1]) <= 3: + return ".".join(labels[-3:]) + return ".".join(labels[-2:]) + + +def search( + query: str | None = None, + *, + base_url: str | None = None, + category: str | None = None, + limit: int = 5, + timeout: float = DEFAULT_TIMEOUT, +) -> list[MarketplaceFunction]: + """Search the public marketplace. Returns [] on any failure. + + The endpoint needs no authentication, so this works for every user. All + three filters are optional and compose: `base_url` scopes to one site, + `category` to one marketplace category, and `query` ranks within whatever + is left. At least one must be supplied — an unfiltered call would just + return the most-run functions overall, which is never what a caller here + wants. + + `base_url` accepts any form the site is written in: a bare hostname, a + full URL, or a glob. Matching happens server-side and covers subdomains. + """ + params: dict[str, str | int] = {"limit": max(1, limit)} + if query and query.strip(): + params["q"] = query.strip() + if base_url and base_url.strip(): + params["base_url"] = base_url.strip() + if category and category.strip(): + params["category"] = category.strip() + + if not any(key in params for key in ("q", "base_url", "category")): + return [] + + try: + import requests + + response = requests.get( + SEARCH_ENDPOINT, + params=params, + timeout=timeout, + headers={"User-Agent": _user_agent()}, + ) + if response.status_code != 200: + return [] + payload = response.json() + except Exception: + # Offline, DNS failure, timeout, non-JSON body, requests missing — + # all mean the same thing here: no suggestions. + return [] + + if not isinstance(payload, dict): + return [] + results = payload.get("results") + if not isinstance(results, list): + return [] + + functions = [] + for raw in results: + if isinstance(raw, dict): + fn = MarketplaceFunction.from_api(raw) + if fn is not None: + functions.append(fn) + return functions + + +def search_for_site( + url_or_host: str, + *, + limit: int = 3, + timeout: float = DEFAULT_TIMEOUT, +) -> list[MarketplaceFunction]: + """Find existing functions for one site, with no false positives. + + `base_url` scopes the search server-side and understands full URLs, so the + target is passed through untouched. `registrable_domain` is still consulted + first to skip inputs that are not sites at all (localhost, IP literals, + free text), and the results are re-checked against it afterwards: server- + side matching is deliberately fuzzy, and suggesting the wrong site is worse + than suggesting nothing. + """ + domain = registrable_domain(url_or_host) + if not domain: + return [] + + matches = [fn for fn in search(base_url=url_or_host, limit=limit, timeout=timeout) if _covers(fn.domain, domain)] + matches.sort(key=lambda fn: fn.run_count, reverse=True) + return matches[:limit] + + +def _covers(candidate_host: str, domain: str) -> bool: + """True when `candidate_host` belongs to `domain`.""" + if not candidate_host: + return False + return candidate_host == domain or candidate_host.endswith(f".{domain}") + + +def _extract_host(url_or_host: str) -> str: + raw = (url_or_host or "").strip() + if not raw: + return "" + if "//" not in raw: + # urlparse needs a scheme to populate `netloc`. + raw = f"//{raw}" + try: + netloc = urlparse(raw).netloc + except ValueError: + return "" + # Drop credentials and port. + netloc = netloc.rsplit("@", 1)[-1] + if netloc.startswith("["): # IPv6 literal + return "" + netloc = netloc.split(":", 1)[0] + return _normalize_host(netloc) + + +def _normalize_host(host: str) -> str: + host = (host or "").strip().lower().rstrip(".") + if host.startswith("www."): + host = host[4:] + return host + + +def _env_flag(name: str) -> bool: + return os.environ.get(name, "").strip().lower() in {"1", "true", "yes", "on"} + + +def _user_agent() -> str: + try: + from . import __version__ + + return f"reverse-api-engineer/{__version__}" + except Exception: + return "reverse-api-engineer" diff --git a/src/reverse_api/config.py b/src/reverse_api/config.py index 9586db5..247a613 100644 --- a/src/reverse_api/config.py +++ b/src/reverse_api/config.py @@ -12,6 +12,10 @@ "agent_browser_notes": "", # extra instructions merged into agent-browser prompt / RAE_AGENT_BROWSER_NOTES env "agent_browser_npx_package": "agent-browser@0", "claude_code_model": "claude-sonnet-4-6", + # Look up the Anything marketplace before a capture, so users are told when + # the API they are about to reverse-engineer already exists. Also honours + # the RAE_NO_CLOUD env var, which wins over this setting. + "cloud_suggestions": True, "collector_model": "claude-sonnet-4-6", # Model for collector mode "cursor_model": "composer-2.5", # Model id for Cursor SDK (see Cursor.models.list()) # When True, local agents load broader Cursor setting layers (plugins/team) so WebFetch/WebSearch diff --git a/tests/test_cli_marketplace.py b/tests/test_cli_marketplace.py new file mode 100644 index 0000000..17b025b --- /dev/null +++ b/tests/test_cli_marketplace.py @@ -0,0 +1,167 @@ +"""Tests for the marketplace CLI surface and the pre-capture suggestion.""" + +import json +from unittest.mock import patch + +from click.testing import CliRunner + +from reverse_api import cloud +from reverse_api.cli import _maybe_suggest_marketplace, marketplace + + +def _fn(function_id="f1", label="get_nfl_standings", domain="nfl.com", runs=8): + return cloud.MarketplaceFunction( + function_id=function_id, + label=label, + description="Returns NFL team standings.", + domain=domain, + run_count=runs, + ) + + +class TestMarketplaceSearchCommand: + """`reverse-api-engineer marketplace search`.""" + + def test_json_output_shape(self): + with patch("reverse_api.cli.cloud.search", return_value=[_fn()]): + result = CliRunner().invoke(marketplace, ["search", "nfl", "--json"]) + assert result.exit_code == 0 + payload = json.loads(result.output) + assert payload["count"] == 1 + assert payload["query"] == "nfl" + assert payload["results"][0]["function_id"] == "f1" + assert payload["results"][0]["url"].endswith("/f1") + + def test_site_flag_uses_domain_scoped_search(self): + with patch("reverse_api.cli.cloud.search_for_site", return_value=[_fn()]) as scoped: + result = CliRunner().invoke(marketplace, ["search", "--site", "https://www.nfl.com", "--json"]) + assert result.exit_code == 0 + scoped.assert_called_once() + assert scoped.call_args.args[0] == "https://www.nfl.com" + + def test_limit_is_forwarded(self): + with patch("reverse_api.cli.cloud.search", return_value=[]) as search: + CliRunner().invoke(marketplace, ["search", "nfl", "-n", "3", "--json"]) + assert search.call_args.kwargs["limit"] == 3 + + def test_category_flag_is_forwarded(self): + with patch("reverse_api.cli.cloud.search", return_value=[]) as search: + result = CliRunner().invoke(marketplace, ["search", "--category", "Jobs", "--json"]) + assert result.exit_code == 0 + assert search.call_args.kwargs["category"] == "Jobs" + + def test_site_with_query_uses_composed_search(self): + # base_url and q compose server-side, so a combined request skips the + # domain-only helper. + with ( + patch("reverse_api.cli.cloud.search", return_value=[]) as search, + patch("reverse_api.cli.cloud.search_for_site") as scoped, + ): + CliRunner().invoke(marketplace, ["search", "standings", "--site", "nfl.com", "--json"]) + scoped.assert_not_called() + assert search.call_args.kwargs["base_url"] == "nfl.com" + + def test_requires_a_query_site_or_category(self): + result = CliRunner().invoke(marketplace, ["search"]) + assert result.exit_code == 2 + + def test_missing_query_still_emits_json_with_json_flag(self): + result = CliRunner().invoke(marketplace, ["search", "--json"]) + assert result.exit_code == 2 + assert json.loads(result.output)["results"] == [] + + def test_no_matches_is_not_an_error(self): + with patch("reverse_api.cli.cloud.search", return_value=[]): + result = CliRunner().invoke(marketplace, ["search", "nothing-here"]) + assert result.exit_code == 0 + assert "no hosted function matches" in result.output + + def test_human_output_lists_matches(self): + with patch("reverse_api.cli.cloud.search", return_value=[_fn()]): + result = CliRunner().invoke(marketplace, ["search", "nfl"]) + assert result.exit_code == 0 + assert "get_nfl_standings" in result.output + + def test_bare_group_prints_overview(self): + result = CliRunner().invoke(marketplace, []) + assert result.exit_code == 0 + assert cloud.MARKETPLACE_URL in result.output + + def test_network_failure_degrades_to_no_matches(self): + # cloud.search already swallows errors, so the command sees []. + with patch("reverse_api.cli.cloud.search", return_value=[]): + result = CliRunner().invoke(marketplace, ["search", "nfl", "--json"]) + assert result.exit_code == 0 + assert json.loads(result.output)["count"] == 0 + + +class TestPreCaptureSuggestion: + """The offer shown before a capture starts.""" + + def test_skipped_without_a_url(self): + with patch("reverse_api.cli.cloud.search_for_site") as search: + assert _maybe_suggest_marketplace(None, interactive=True) is False + search.assert_not_called() + + def test_skipped_when_not_interactive(self): + with patch("reverse_api.cli.cloud.search_for_site") as search: + assert _maybe_suggest_marketplace("https://nfl.com", interactive=False) is False + search.assert_not_called() + + def test_skipped_when_disabled(self, monkeypatch): + monkeypatch.setenv("RAE_NO_CLOUD", "1") + with patch("reverse_api.cli.cloud.search_for_site") as search: + assert _maybe_suggest_marketplace("https://nfl.com", interactive=True) is False + search.assert_not_called() + + def test_no_matches_continues_the_capture(self, monkeypatch): + monkeypatch.delenv("RAE_NO_CLOUD", raising=False) + with patch("reverse_api.cli.cloud.search_for_site", return_value=[]): + assert _maybe_suggest_marketplace("https://nfl.com", interactive=True) is False + + def test_declining_continues_the_capture(self, monkeypatch): + monkeypatch.delenv("RAE_NO_CLOUD", raising=False) + with ( + patch("reverse_api.cli.cloud.search_for_site", return_value=[_fn()]), + patch("reverse_api.cli.questionary.confirm") as confirm, + ): + confirm.return_value.ask.return_value = False + assert _maybe_suggest_marketplace("https://nfl.com", interactive=True) is False + + def test_ctrl_c_at_the_prompt_continues_the_capture(self, monkeypatch): + # questionary returns None when the user interrupts the prompt. + monkeypatch.delenv("RAE_NO_CLOUD", raising=False) + with ( + patch("reverse_api.cli.cloud.search_for_site", return_value=[_fn()]), + patch("reverse_api.cli.questionary.confirm") as confirm, + ): + confirm.return_value.ask.return_value = None + assert _maybe_suggest_marketplace("https://nfl.com", interactive=True) is False + + def test_accepting_opens_the_page_and_aborts(self, monkeypatch): + monkeypatch.delenv("RAE_NO_CLOUD", raising=False) + with ( + patch("reverse_api.cli.cloud.search_for_site", return_value=[_fn()]), + patch("reverse_api.cli.questionary.confirm") as confirm, + patch("webbrowser.open") as opener, + ): + confirm.return_value.ask.return_value = True + assert _maybe_suggest_marketplace("https://nfl.com", interactive=True) is True + opened = opener.call_args.args[0] + assert "utm_campaign=cli_precapture" in opened + assert "/f1" in opened + + def test_browser_failure_still_aborts_cleanly(self, monkeypatch): + monkeypatch.delenv("RAE_NO_CLOUD", raising=False) + with ( + patch("reverse_api.cli.cloud.search_for_site", return_value=[_fn()]), + patch("reverse_api.cli.questionary.confirm") as confirm, + patch("webbrowser.open", side_effect=OSError("no browser")), + ): + confirm.return_value.ask.return_value = True + assert _maybe_suggest_marketplace("https://nfl.com", interactive=True) is True + + def test_lookup_crash_never_blocks_a_capture(self, monkeypatch): + monkeypatch.delenv("RAE_NO_CLOUD", raising=False) + with patch("reverse_api.cli.cloud.search_for_site", side_effect=RuntimeError("boom")): + assert _maybe_suggest_marketplace("https://nfl.com", interactive=True) is False diff --git a/tests/test_cloud.py b/tests/test_cloud.py new file mode 100644 index 0000000..f8a9c78 --- /dev/null +++ b/tests/test_cloud.py @@ -0,0 +1,245 @@ +"""Tests for cloud.py - Anything marketplace lookups. + +Nothing here touches the network: `requests.get` is stubbed everywhere so the +suite stays offline-safe and deterministic. +""" + +import json +from types import SimpleNamespace + +import pytest + +from reverse_api import cloud + + +def _result(function_id="f1", label="get_thing", domain="example.com", runs=3, description="Does a thing."): + return { + "function_id": function_id, + "label": label, + "description": description, + "domain": domain, + "run_count": runs, + } + + +class _FakeResponse: + def __init__(self, payload, status_code=200): + self._payload = payload + self.status_code = status_code + + def json(self): + if isinstance(self._payload, Exception): + raise self._payload + return self._payload + + +@pytest.fixture +def fake_get(monkeypatch): + """Stub requests.get and record the calls it received.""" + calls = [] + + def _install(payload, status_code=200, raises=None): + def _get(url, params=None, timeout=None, headers=None): + calls.append(SimpleNamespace(url=url, params=params or {}, timeout=timeout, headers=headers or {})) + if raises is not None: + raise raises + return _FakeResponse(payload, status_code) + + import requests + + monkeypatch.setattr(requests, "get", _get) + return calls + + return _install + + +class TestRegistrableDomain: + """Reducing a URL to the domain a marketplace entry would carry.""" + + @pytest.mark.parametrize( + "value,expected", + [ + ("https://jobs.ashbyhq.com/openai?x=1", "ashbyhq.com"), + ("https://www.nfl.com/scores", "nfl.com"), + ("instagram.com", "instagram.com"), + ("www.bbc.co.uk", "bbc.co.uk"), + ("https://shop.example.com.au/cart", "example.com.au"), + ("HTTPS://WWW.Example.COM/", "example.com"), + ("https://user:pw@example.com:8443/x", "example.com"), + ("example.com.", "example.com"), + ], + ) + def test_extracts_domain(self, value, expected): + assert cloud.registrable_domain(value) == expected + + @pytest.mark.parametrize( + "value", + ["", " ", "not a url", "localhost", "http://192.168.1.1/x", "https://[::1]:8080/x", "/just/a/path"], + ) + def test_rejects_unusable_input(self, value): + assert cloud.registrable_domain(value) is None + + +class TestSuggestionsEnabled: + """The kill switches.""" + + def test_enabled_by_default(self, monkeypatch): + monkeypatch.delenv("RAE_NO_CLOUD", raising=False) + assert cloud.suggestions_enabled() is True + + @pytest.mark.parametrize("value", ["1", "true", "TRUE", "yes", "on"]) + def test_env_var_disables(self, monkeypatch, value): + monkeypatch.setenv("RAE_NO_CLOUD", value) + assert cloud.suggestions_enabled() is False + + def test_env_var_beats_config(self, monkeypatch): + monkeypatch.setenv("RAE_NO_CLOUD", "1") + cm = SimpleNamespace(get=lambda key, default=None: True) + assert cloud.suggestions_enabled(cm) is False + + def test_config_disables(self, monkeypatch): + monkeypatch.delenv("RAE_NO_CLOUD", raising=False) + cm = SimpleNamespace(get=lambda key, default=None: False) + assert cloud.suggestions_enabled(cm) is False + + def test_unset_env_is_ignored(self, monkeypatch): + monkeypatch.setenv("RAE_NO_CLOUD", "0") + assert cloud.suggestions_enabled() is True + + +class TestSearch: + """The public marketplace search call.""" + + def test_parses_results(self, fake_get): + fake_get({"results": [_result(), _result(function_id="f2", label="other")], "total": 2}) + found = cloud.search("thing") + assert [f.function_id for f in found] == ["f1", "f2"] + assert found[0].label == "get_thing" + assert found[0].run_count == 3 + + def test_sends_query_and_limit(self, fake_get): + calls = fake_get({"results": []}) + cloud.search("thing", limit=7) + assert calls[0].params == {"q": "thing", "limit": 7} + assert calls[0].url == cloud.SEARCH_ENDPOINT + assert "reverse-api-engineer" in calls[0].headers["User-Agent"] + + def test_sends_base_url_and_category(self, fake_get): + calls = fake_get({"results": []}) + cloud.search(base_url="nfl.com", category="Sports", limit=3) + assert calls[0].params == {"base_url": "nfl.com", "category": "Sports", "limit": 3} + + def test_filters_compose(self, fake_get): + calls = fake_get({"results": []}) + cloud.search("standings", base_url="nfl.com") + assert calls[0].params["q"] == "standings" + assert calls[0].params["base_url"] == "nfl.com" + + def test_blank_query_skips_the_call(self, fake_get): + calls = fake_get({"results": [_result()]}) + assert cloud.search(" ") == [] + assert calls == [] + + def test_no_filters_skips_the_call(self, fake_get): + # An unfiltered call returns the most-run functions overall, which is + # never what a caller here means. + calls = fake_get({"results": [_result()]}) + assert cloud.search() == [] + assert calls == [] + + def test_blank_filters_are_dropped(self, fake_get): + calls = fake_get({"results": []}) + cloud.search(" thing ", base_url=" ", category="") + assert calls[0].params == {"q": "thing", "limit": 5} + + def test_non_200_returns_empty(self, fake_get): + fake_get({"results": [_result()]}, status_code=500) + assert cloud.search("thing") == [] + + def test_network_error_returns_empty(self, fake_get): + fake_get(None, raises=OSError("no route to host")) + assert cloud.search("thing") == [] + + def test_invalid_json_returns_empty(self, fake_get): + fake_get(json.JSONDecodeError("bad", "", 0)) + assert cloud.search("thing") == [] + + @pytest.mark.parametrize("payload", [[], "nope", None, {"results": "nope"}, {}]) + def test_unexpected_shapes_return_empty(self, fake_get, payload): + fake_get(payload) + assert cloud.search("thing") == [] + + def test_skips_malformed_entries(self, fake_get): + fake_get( + { + "results": [ + {"label": "no id"}, + {"function_id": "f2"}, + "not a dict", + _result(function_id="f3", runs="not-a-number"), + ] + } + ) + found = cloud.search("thing") + assert [f.function_id for f in found] == ["f3"] + assert found[0].run_count == 0 + + def test_url_points_at_the_function_page(self, fake_get): + fake_get({"results": [_result(function_id="abc")]}) + assert cloud.search("thing")[0].url == f"{cloud.MARKETPLACE_URL}/abc" + + +class TestSearchForSite: + """Domain-scoped lookups must never surface an unrelated site.""" + + def test_scopes_server_side_by_base_url(self, fake_get): + calls = fake_get({"results": []}) + cloud.search_for_site("https://jobs.ashbyhq.com/openai?x=1") + # The endpoint understands full URLs, so the target is passed through + # untouched rather than reduced first. + assert calls[0].params["base_url"] == "https://jobs.ashbyhq.com/openai?x=1" + assert "q" not in calls[0].params + + def test_keeps_matching_domains_only(self, fake_get): + fake_get( + { + "results": [ + _result(function_id="a", domain="nfl.com", runs=2), + _result(function_id="b", domain="fantasy.nfl.com", runs=9), + _result(function_id="c", domain="paisabazaar.com", runs=99), + _result(function_id="d", domain="notnfl.com", runs=50), + ] + } + ) + found = cloud.search_for_site("https://www.nfl.com/standings") + assert [f.function_id for f in found] == ["b", "a"] + + def test_unrelated_results_yield_nothing(self, fake_get): + # Server-side matching is deliberately fuzzy, so the domain guard stays + # as a backstop: a suggestion for the wrong site is worse than none. + fake_get({"results": [_result(domain="paisabazaar.com"), _result(domain="ratings.fide.com")]}) + assert cloud.search_for_site("https://jobs.ashbyhq.com/openai") == [] + + def test_respects_limit(self, fake_get): + fake_get({"results": [_result(function_id=f"f{i}", domain="nfl.com", runs=i) for i in range(10)]}) + assert len(cloud.search_for_site("nfl.com", limit=2)) == 2 + + def test_unusable_url_skips_the_call(self, fake_get): + calls = fake_get({"results": [_result()]}) + assert cloud.search_for_site("localhost") == [] + assert calls == [] + + +class TestCloudLink: + """UTM tagging for attribution.""" + + def test_adds_tags(self): + tagged = cloud.cloud_link(cloud.CLOUD_URL, "cli_precapture") + assert tagged.startswith(f"{cloud.CLOUD_URL}?") + assert "utm_source=rae" in tagged + assert "utm_medium=cli" in tagged + assert "utm_campaign=cli_precapture" in tagged + + def test_appends_to_existing_query(self): + tagged = cloud.cloud_link("https://anything.notte.cc/x?a=1", "somewhere") + assert "?a=1&utm_source=rae" in tagged diff --git a/tests/test_config.py b/tests/test_config.py index bc27654..062bbd1 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -146,6 +146,7 @@ def test_has_required_keys(self): "agent_browser_notes", "agent_browser_npx_package", "claude_code_model", + "cloud_suggestions", "collector_model", "copilot_model", "cursor_model",