diff --git a/CLI-COMMANDS.md b/CLI-COMMANDS.md index 50dbc2ce..2a7667af 100644 --- a/CLI-COMMANDS.md +++ b/CLI-COMMANDS.md @@ -12,6 +12,54 @@ export ROBOFLOW_API_KEY=rf_xxxxx # recommended for scripts and agents roboflow auth login # or interactive login ``` +### Select a Roboflow region + +Roboflow uses the US platform by default. To authenticate with the EU +data-residency platform, select the region during login: + +```bash +roboflow auth login --region eu +# The backwards-compatible alias accepts the same option: +roboflow login --region eu +``` + +The selection is saved in the Roboflow config file. You can change it later +or inspect the effective endpoints with: + +```bash +roboflow auth set-region eu +roboflow auth status +``` + +For CI and other non-interactive environments, set `ROBOFLOW_REGION=eu`. +`ROBOFLOW_REGION` accepts `us` or `eu` (case-insensitive); an environment +value takes precedence over the saved region. Explicit per-URL environment or +config values such as `API_URL` continue to take precedence over the region. + +For Roboflow staging, set `ROBOFLOW_ENVIRONMENT=staging` alongside the +region; it accepts `prod` (default) or `staging`, and anything else warns and +falls back to `prod`. It selects the `roboflow.one` (US) or `roboflow-eu.one` +(EU) hosts, matching `inference`. `roboflow auth status` reports both switches. + +| Endpoint | `us` (default) | `eu` | +|----------|----------------|------| +| API | `https://api.roboflow.com` | `https://api.roboflow.eu` | +| App / CLI authentication | `https://app.roboflow.com` | `https://app.roboflow.eu` | +| Object detection | `https://serverless.roboflow.com` | `https://serverless.roboflow.eu` | +| Instance segmentation | `https://serverless.roboflow.com` | `https://serverless.roboflow.eu` | +| Classification, keypoint, VLM (`SERVERLESS_URL`) | `https://serverless.roboflow.com` | `https://serverless.roboflow.eu` | +| Dedicated deployment | `https://roboflow.cloud` | `https://eu.roboflow.cloud` | +| Universe | `https://universe.roboflow.com` | `https://universe.roboflow.com` | +| Semantic segmentation | `https://segment.roboflow.com` | not available | + +Roboflow Universe remains a single global product, so its URL stays on +`.com` in the EU region. Hosted semantic segmentation has no EU deployment, so +in the EU region it raises an error instead of sending images to the US +endpoint; set `SEMANTIC_SEGMENTATION_URL` explicitly to override. EU and US use +separate authentication backends; obtain EU +API keys from `https://app.roboflow.eu` and log in again after switching if +your existing credentials were issued by the other region. + ## Global flags | Flag | Short | Description | @@ -498,7 +546,7 @@ Version numbers are always numeric — that's how `x/y` is disambiguated between | Command | Description | |---------|-------------| -| `auth` | Login, logout, status, set default workspace | +| `auth` | Login, logout, status, set region or default workspace | | `api-key` | List, create, update, protect, disable, revoke workspace API keys | | `workspace` | List and inspect workspaces | | `project` | List, get, create projects | diff --git a/README.md b/README.md index 32294918..1be0577f 100644 --- a/README.md +++ b/README.md @@ -36,7 +36,7 @@ The Python package is documented on the [official Roboflow documentation site](h ## 💻 Installation -You will need to have `Python 3.8` or higher set up to use the Roboflow Python package. +You will need to have `Python 3.10` or higher set up to use the Roboflow Python package. Run the following command to install the Roboflow Python package: @@ -104,6 +104,20 @@ import roboflow roboflow.login() ``` +### Using Roboflow EU + +The same package supports Roboflow's EU data-residency platform. Select it +when logging in with the CLI: + +```bash +roboflow auth login --region eu +``` + +For environment-based configuration and CI, set `ROBOFLOW_REGION=eu` before +running Python or CLI commands. EU and US use separate authentication +backends, so use an EU API key obtained from +[`app.roboflow.eu`](https://app.roboflow.eu). +
Authenticate with an API key diff --git a/roboflow/__init__.py b/roboflow/__init__.py index 6c556500..eb00e37f 100644 --- a/roboflow/__init__.py +++ b/roboflow/__init__.py @@ -66,7 +66,31 @@ def check_key(api_key, model, notebook, num_retries=0): return "onboarding" -def login(workspace=None, force=False): +def login(workspace=None, force=False, region=None): + from roboflow.config import ( + CREDENTIALS_REGION_KEY, + SUPPORTED_REGIONS, + credentials_region, + get_effective_region, + has_credentials, + refresh_region_urls, + region_conflict, + resolve_url, + ) + + normalized_region = None + if region is not None: + if not isinstance(region, str) or region.lower() not in SUPPORTED_REGIONS: + raise ValueError(f"Invalid region '{region}'. Expected one of: {', '.join(SUPPORTED_REGIONS)}.") + normalized_region = region.lower() + conflicting = region_conflict(normalized_region) + if conflicting is not None: + # Requests would follow the environment, not the credentials issued here. + raise ValueError( + f"ROBOFLOW_REGION={conflicting} in the environment overrides region='{normalized_region}'. " + "Unset it or make it match before logging in." + ) + os_name = os.name if os_name == "nt": @@ -76,22 +100,44 @@ def login(workspace=None, force=False): # default configuration location conf_location = os.getenv("ROBOFLOW_CONFIG_DIR", default=default_path) - if os.path.isfile(conf_location) and not force: - write_line("You are already logged into Roboflow. To make a different login,run roboflow.login(force=True).") - return None - # we could eventually return the workspace object here - # return Roboflow().workspace() - elif os.path.isfile(conf_location) and force: - os.remove(conf_location) + existing_config = {} + if os.path.isfile(conf_location): + try: + with open(conf_location) as f: + existing_config = json.load(f) + except json.JSONDecodeError: + # A forced login has historically replaced an unreadable config. + if not force: + raise + existing_config = {} + if not isinstance(existing_config, dict): + existing_config = {} + + # A config holding only preferences (e.g. from `auth set-region`) is not a session. + target_region = normalized_region or get_effective_region() + if has_credentials(existing_config) and not force: + if credentials_region(existing_config) == target_region: + write_line( + "You are already logged into Roboflow. To make a different login,run roboflow.login(force=True)." + ) + return None + # we could eventually return the workspace object here + # return Roboflow().workspace() + # Stored credentials were issued by the other platform and cannot authenticate here. + write_line(f"Stored credentials were issued by the {credentials_region(existing_config).upper()} platform.") + + # Resolve at call time so a region passed by the CLI is honored even though + # the module-level URL constants were resolved when roboflow was imported. + app_url = resolve_url("APP_URL", region=normalized_region) if workspace is None: - write_line("visit " + APP_URL + "/auth-cli to get your authentication token.") + write_line("visit " + app_url + "/auth-cli to get your authentication token.") else: - write_line("visit " + APP_URL + "/auth-cli/?workspace=" + workspace + " to get your authentication token.") + write_line("visit " + app_url + "/auth-cli/?workspace=" + workspace + " to get your authentication token.") token = getpass("Paste the authentication token here: ") - r_login = requests.get(APP_URL + "/query/cliAuthToken/" + token) + r_login = requests.get(app_url + "/query/cliAuthToken/" + token) if r_login.status_code == 200: r_login = r_login.json() @@ -102,16 +148,24 @@ def login(workspace=None, force=False): if not os.path.exists(os.path.dirname(conf_location)): os.makedirs(os.path.dirname(conf_location)) - r_login = {"workspaces": r_login} + # The previous credentials are replaced only once the new ones are in hand. + existing_config["workspaces"] = r_login # set first workspace as default workspace - default_workspace_id = list(r_login["workspaces"].keys())[0] - workspace = r_login["workspaces"][default_workspace_id] - r_login["RF_WORKSPACE"] = workspace["url"] + default_workspace_id = list(existing_config["workspaces"].keys())[0] + workspace = existing_config["workspaces"][default_workspace_id] + existing_config["RF_WORKSPACE"] = workspace["url"] + if normalized_region is not None: + existing_config["ROBOFLOW_REGION"] = normalized_region + existing_config[CREDENTIALS_REGION_KEY] = target_region # write config file with open(conf_location, "w") as f: - json.dump(r_login, f, indent=2) + json.dump(existing_config, f, indent=2) + + if normalized_region is not None: + # Constants bound at import time still point at the previous region. + refresh_region_urls() else: r_login.raise_for_status() @@ -145,12 +199,27 @@ def initialize_roboflow(the_workspace=None): return active_workspace +_ROBOFLOW_APP_HOSTS = ( + "universe.roboflow.com", + "app.roboflow.com", + "app.roboflow.eu", + # Staging platforms from the endpoint matrix in roboflow.config. + "universe.roboflow.one", + "app.roboflow.one", + "app.roboflow-eu.one", +) + + +def _is_roboflow_app_url(url): + return any(host in url for host in _ROBOFLOW_APP_HOSTS) + + def load_model(model_url): """High level function to load Roboflow models. Args: model_url: the model url to load. - Must be from either app.roboflow.com or universe.roboflow.com + Must be from app.roboflow.com, app.roboflow.eu or universe.roboflow.com Returns: the model object to use for inference @@ -158,13 +227,13 @@ def load_model(model_url): operate_workspace = initialize_roboflow() - if "universe.roboflow.com" in model_url or "app.roboflow.com" in model_url: + if _is_roboflow_app_url(model_url): parsed_url = urlparse(model_url) path_parts = parsed_url.path.split("/") project = path_parts[2] version = int(path_parts[-1]) else: - raise ValueError("Model URL must be from either app.roboflow.com or universe.roboflow.com") + raise ValueError("Model URL must be from app.roboflow.com, app.roboflow.eu or universe.roboflow.com") project = operate_workspace.project(project) version = project.version(version) @@ -179,7 +248,7 @@ def download_dataset(dataset_url, model_format, location=None): Args: dataset_url: the dataset url to download. - Must be from either app.roboflow.com or universe.roboflow.com + Must be from app.roboflow.com, app.roboflow.eu or universe.roboflow.com model_format: the format the dataset will be downloaded in location: the location the dataset will be downloaded to @@ -187,14 +256,14 @@ def download_dataset(dataset_url, model_format, location=None): The dataset object with location available as dataset.location """ - if "universe.roboflow.com" in dataset_url or "app.roboflow.com" in dataset_url: + if _is_roboflow_app_url(dataset_url): parsed_url = urlparse(dataset_url) path_parts = parsed_url.path.split("/") project = path_parts[2] version = int(path_parts[-1]) the_workspace = path_parts[1] else: - raise ValueError("Model URL must be from either app.roboflow.com or universe.roboflow.com") + raise ValueError("Model URL must be from app.roboflow.com, app.roboflow.eu or universe.roboflow.com") operate_workspace = initialize_roboflow(the_workspace=the_workspace) project = operate_workspace.project(project) diff --git a/roboflow/cli/handlers/_aliases.py b/roboflow/cli/handlers/_aliases.py index 301c2dab..5e7ceb0a 100644 --- a/roboflow/cli/handlers/_aliases.py +++ b/roboflow/cli/handlers/_aliases.py @@ -40,12 +40,15 @@ def login_alias( login_api_key: Annotated[ Optional[str], typer.Option("--api-key", help="API key (skip interactive login)") ] = None, + region: Annotated[ + Optional[str], typer.Option("--region", metavar="{us,eu}", help="Roboflow platform region") + ] = None, force: Annotated[bool, typer.Option("--force", "-f", help="Force re-login")] = False, ) -> None: """Log in to Roboflow (alias for 'auth login').""" from roboflow.cli.handlers.auth import _login - args = ctx_to_args(ctx, login_api_key=login_api_key, force=force) + args = ctx_to_args(ctx, login_api_key=login_api_key, region=region, force=force) _login(args) @app.command("whoami", hidden=True) diff --git a/roboflow/cli/handlers/auth.py b/roboflow/cli/handlers/auth.py index 374b7fc2..658524b8 100644 --- a/roboflow/cli/handlers/auth.py +++ b/roboflow/cli/handlers/auth.py @@ -1,4 +1,4 @@ -"""Auth commands: login, logout, status, set-workspace.""" +"""Auth commands: login, logout, status, set-region, set-workspace.""" from __future__ import annotations @@ -18,10 +18,19 @@ def login( login_workspace: Annotated[ Optional[str], typer.Option("--workspace", help="Set default workspace during login") ] = None, + region: Annotated[ + Optional[str], typer.Option("--region", metavar="{us,eu}", help="Roboflow platform region") + ] = None, force: Annotated[bool, typer.Option("--force", "-f", help="Force re-login even if already logged in")] = False, ) -> None: """Log in to Roboflow.""" - args = ctx_to_args(ctx, login_api_key=login_api_key, login_workspace=login_workspace, force=force) + args = ctx_to_args( + ctx, + login_api_key=login_api_key, + login_workspace=login_workspace, + region=region, + force=force, + ) _login(args) @@ -42,6 +51,16 @@ def set_workspace( _set_workspace(args) +@auth_app.command("set-region") +def set_region( + ctx: typer.Context, + region: Annotated[str, typer.Argument(metavar="{us,eu}", help="Roboflow platform region")], +) -> None: + """Set the Roboflow platform region.""" + args = ctx_to_args(ctx, region=region) + _set_region(args) + + @auth_app.command("logout") def logout(ctx: typer.Context) -> None: """Remove stored credentials.""" @@ -95,6 +114,41 @@ def _mask_key(key: str) -> str: return key[:2] + "*" * (len(key) - 4) + key[-2:] +def _validate_region(args, region: Optional[str]) -> Optional[str]: # noqa: ANN001 + """Normalize and validate a region supplied explicitly on the CLI.""" + if region is None: + return None + + normalized = region.lower() + if normalized not in {"us", "eu"}: + from roboflow.cli._output import output_error + + output_error( + args, + f"Invalid region '{region}'.", + hint="Region must be 'us' or 'eu'.", + exit_code=2, + ) + return normalized + + +def _region_status() -> tuple[dict[str, str], list[str]]: + """Return the effective region metadata in structured and text forms.""" + from roboflow.config import get_effective_environment, get_effective_region, get_region_warning, resolve_url + + region = get_effective_region() + environment = get_effective_environment() + api_url = resolve_url("API_URL", region=region) + app_url = resolve_url("APP_URL", region=region) + data = {"region": region, "environment": environment, "api_url": api_url, "app_url": app_url} + lines = [f"Region: {region}", f"Environment: {environment}", f"API URL: {api_url}", f"App URL: {app_url}"] + region_warning = get_region_warning() + if region_warning is not None: + data["region_warning"] = region_warning + lines.append(f"Warning: {region_warning}") + return data, lines + + def _print_completion_tip(args) -> None: # noqa: ANN001 """Nudge users towards shell completion after a successful login. @@ -111,17 +165,40 @@ def _login(args): # noqa: ANN001 api_key = getattr(args, "login_api_key", None) or getattr(args, "api_key", None) workspace_id = getattr(args, "login_workspace", None) or getattr(args, "workspace", None) + region = _validate_region(args, getattr(args, "region", None)) force = getattr(args, "force", False) + from roboflow.config import ( + CREDENTIALS_REGION_KEY, + credentials_region, + get_effective_region, + has_credentials, + region_conflict, + ) + + conflicting = region_conflict(region) if region is not None else None + if conflicting is not None: + # Requests follow the environment, so credentials from --region would go to the other platform. + output_error( + args, + f"ROBOFLOW_REGION={conflicting} in the environment overrides --region {region}.", + hint="Unset ROBOFLOW_REGION or make it match before logging in.", + exit_code=2, + ) + return + target_region = region or get_effective_region() + if api_key: # Non-interactive: validate key and fetch workspace info import requests - from roboflow.config import API_URL + from roboflow.config import resolve_url - resp = requests.post(API_URL + "/?api_key=" + api_key) + api_url = resolve_url("API_URL", region=region) + app_url = resolve_url("APP_URL", region=region) + resp = requests.post(api_url + "/?api_key=" + api_key) if resp.status_code == 401: - output_error(args, "Invalid API key.", hint="Check your key at app.roboflow.com/settings", exit_code=2) + output_error(args, "Invalid API key.", hint=f"Check your key at {app_url}/settings", exit_code=2) return if resp.status_code != 200: output_error(args, f"API error ({resp.status_code}).", exit_code=1) @@ -141,9 +218,9 @@ def _login(args): # noqa: ANN001 # Fetch workspace name from the API ws_name = ws_url try: - from roboflow.adapters import rfapi - - ws_json = rfapi.get_workspace(api_key, ws_url) + ws_resp = requests.get(f"{api_url}/{ws_url}?api_key={api_key}") + ws_resp.raise_for_status() + ws_json = ws_resp.json() ws_detail = ws_json.get("workspace", ws_json) ws_name = ws_detail.get("name", ws_url) except Exception: # noqa: BLE001 @@ -152,9 +229,15 @@ def _login(args): # noqa: ANN001 # Build config with workspace info config = _load_config() workspaces = config.get("workspaces", {}) + if has_credentials(config) and credentials_region(config) != target_region: + # Keys from the other platform cannot be used here; don't leave them selectable. + workspaces = {} workspaces[ws_url] = {"url": ws_url, "name": ws_name, "apiKey": api_key} config["workspaces"] = workspaces config["RF_WORKSPACE"] = ws_url + if region is not None: + config["ROBOFLOW_REGION"] = region + config[CREDENTIALS_REGION_KEY] = target_region _save_config(config) note = "" @@ -170,21 +253,23 @@ def _login(args): # noqa: ANN001 # Interactive flow import roboflow - conf_path = _get_config_path() - import os - - if os.path.isfile(conf_path) and not force: - # Already logged in — show status - config = _load_config() - ws = config.get("RF_WORKSPACE", "unknown") - output( - args, - {"status": "logged_in", "workspace": ws, "api_key": "****"}, - text=f"Already logged in. Default workspace: {ws}\nUse --force to re-login.", - ) - return - - roboflow.login(workspace=workspace_id, force=force) + config = _load_config() + if has_credentials(config) and not force: + if credentials_region(config) != target_region: + # Stored credentials belong to the platform that issued them; another + # region needs new ones, so treat it as a forced login. + force = True + else: + # Already logged in — show status + ws = config.get("RF_WORKSPACE", "unknown") + output( + args, + {"status": "logged_in", "workspace": ws, "api_key": "****"}, + text=f"Already logged in. Default workspace: {ws}\nUse --force to re-login.", + ) + return + + roboflow.login(workspace=workspace_id, force=force, region=region) # Re-read config after interactive login config = _load_config() ws = config.get("RF_WORKSPACE", "unknown") @@ -194,7 +279,6 @@ def _login(args): # noqa: ANN001 text=f"Logged in. Default workspace: {ws}", ) _print_completion_tip(args) - _print_completion_tip(args) def _status(args): # noqa: ANN001 @@ -202,6 +286,7 @@ def _status(args): # noqa: ANN001 from roboflow.cli._output import output, output_error + region_data, region_lines = _region_status() config = _load_config() workspaces = config.get("workspaces", {}) default_ws_url = config.get("RF_WORKSPACE") @@ -215,14 +300,19 @@ def _status(args): # noqa: ANN001 if explicit_api_key or (api_key and not default_ws_url): import requests - from roboflow.config import API_URL + from roboflow.config import resolve_url assert api_key is not None # guaranteed by the condition above - resp = requests.post(API_URL + "/?api_key=" + api_key) + resp = requests.post(resolve_url("API_URL", region=region_data["region"]) + "/?api_key=" + api_key) if resp.status_code == 200: ws_url = resp.json().get("workspace", "unknown") - data = {"url": ws_url, "name": ws_url, "apiKey": _mask_key(api_key)} - lines = [ + data = { + "url": ws_url, + "name": ws_url, + "apiKey": _mask_key(api_key), + **region_data, + } + lines = region_lines + [ f"Workspace: {ws_url}", f" URL: {ws_url}", f" API Key: {_mask_key(api_key)}", @@ -234,6 +324,21 @@ def _status(args): # noqa: ANN001 return if not workspaces and not default_ws_url and not api_key: + if getattr(args, "json", False): + import json + import sys + + payload = { + "error": { + "message": "Not logged in.", + "hint": "Run 'roboflow auth login' to authenticate.", + }, + **region_data, + } + print(json.dumps(payload), file=sys.stderr) + raise SystemExit(2) + + output(args, region_data, text="\n".join(region_lines)) output_error(args, "Not logged in.", hint="Run 'roboflow auth login' to authenticate.", exit_code=2) return # unreachable, but helps mypy @@ -249,7 +354,8 @@ def _status(args): # noqa: ANN001 display_key = api_key or default_ws.get("apiKey", "") masked = dict(default_ws) masked["apiKey"] = _mask_key(display_key) - lines = [ + masked.update(region_data) + lines = region_lines + [ f"Workspace: {masked.get('name', 'unknown')}", f" URL: {masked.get('url', 'unknown')}", f" API Key: {masked['apiKey']}", @@ -257,14 +363,58 @@ def _status(args): # noqa: ANN001 output(args, masked, text="\n".join(lines)) else: # RF_WORKSPACE is set but no matching workspace details - data = {"url": default_ws_url, "name": default_ws_url} + data = {"url": default_ws_url, "name": default_ws_url, **region_data} output( args, data, - text=f"Workspace: {default_ws_url}\n (no detailed info available)", + text="\n".join(region_lines + [f"Workspace: {default_ws_url}", " (no detailed info available)"]), ) +def _set_region(args): # noqa: ANN001 + from roboflow.cli._output import output + from roboflow.config import credentials_region, get_effective_region, has_credentials, resolve_url + + region = _validate_region(args, args.region) + assert region is not None + + config = _load_config() + stored_credentials = has_credentials(config) + issuing_region = credentials_region(config) + config["ROBOFLOW_REGION"] = region + _save_config(config) + + effective_region = get_effective_region() + api_url = resolve_url("API_URL") + app_url = resolve_url("APP_URL") + warning = None + if stored_credentials and issuing_region != region: + warning = ( + f"Stored credentials were issued by the {issuing_region.upper()} platform. " + "EU and US use separate authentication backends and API keys, so " + f"'roboflow auth login --force --region {region}' may be needed." + ) + environment_note = "" + if effective_region != region: + environment_note = ( + f"\nNote: ROBOFLOW_REGION overrides the saved choice; the effective region remains {effective_region}." + ) + output( + args, + { + "region": effective_region, + "configured_region": region, + "api_url": api_url, + "app_url": app_url, + "warning": warning, + }, + text=( + f"Region set to: {region}{environment_note}\nAPI URL: {api_url}\nApp URL: {app_url}" + + (f"\nWarning: {warning}" if warning else "") + ), + ) + + def _set_workspace(args): # noqa: ANN001 from roboflow.cli._output import output diff --git a/roboflow/config.py b/roboflow/config.py index 60147af6..f5bfd443 100644 --- a/roboflow/config.py +++ b/roboflow/config.py @@ -1,5 +1,75 @@ import json import os +import sys +import warnings + +URL_DEFAULTS = { + "API_URL": "https://api.roboflow.com", + "APP_URL": "https://app.roboflow.com", + "UNIVERSE_URL": "https://universe.roboflow.com", + "INSTANCE_SEGMENTATION_URL": "https://serverless.roboflow.com", + "SEMANTIC_SEGMENTATION_URL": "https://segment.roboflow.com", + "OBJECT_DETECTION_URL": "https://serverless.roboflow.com", + "SERVERLESS_URL": "https://serverless.roboflow.com", + "CLIP_FEATURIZE_URL": "CLIP FEATURIZE URL NOT IN ENV", + "OCR_URL": "OCR URL NOT IN ENV", + "DEDICATED_DEPLOYMENT_URL": "https://roboflow.cloud", +} + +SUPPORTED_REGIONS = ("us", "eu") +SUPPORTED_ENVIRONMENTS = ("prod", "staging") +DEFAULT_REGION = "us" +DEFAULT_ENVIRONMENT = "prod" + +# Per (region, environment) overrides of URL_DEFAULTS, which is US production. Hosts +# mirror the platform's environment configs and the dedicated-deployment ingresses. +SERVICE_URL_OVERRIDES: dict[tuple[str, str], dict[str, str]] = { + ("us", "prod"): {}, + ("us", "staging"): { + "API_URL": "https://api.roboflow.one", + "APP_URL": "https://app.roboflow.one", + "UNIVERSE_URL": "https://universe.roboflow.one", + "OBJECT_DETECTION_URL": "https://serverless.roboflow.one", + "INSTANCE_SEGMENTATION_URL": "https://serverless.roboflow.one", + "SERVERLESS_URL": "https://serverless.roboflow.one", + "SEMANTIC_SEGMENTATION_URL": "https://lambda-semantic-segmentation.staging.roboflow.com", + "DEDICATED_DEPLOYMENT_URL": "https://staging.roboflow.cloud", + }, + ("eu", "prod"): { + "API_URL": "https://api.roboflow.eu", + "APP_URL": "https://app.roboflow.eu", + "OBJECT_DETECTION_URL": "https://serverless.roboflow.eu", + "INSTANCE_SEGMENTATION_URL": "https://serverless.roboflow.eu", + "SERVERLESS_URL": "https://serverless.roboflow.eu", + "DEDICATED_DEPLOYMENT_URL": "https://eu.roboflow.cloud", + }, + ("eu", "staging"): { + "API_URL": "https://api.roboflow-eu.one", + "APP_URL": "https://app.roboflow-eu.one", + "UNIVERSE_URL": "https://universe.roboflow.one", + "OBJECT_DETECTION_URL": "https://serverless.roboflow-eu.one", + "INSTANCE_SEGMENTATION_URL": "https://serverless.roboflow-eu.one", + "SERVERLESS_URL": "https://serverless.roboflow-eu.one", + "DEDICATED_DEPLOYMENT_URL": "https://eu.staging.roboflow.cloud", + }, +} + +# Hosted services with no deployment in a (region, environment). Their fallback would +# send data to another region or environment, so callers refuse unless the URL is +# set explicitly. +UNAVAILABLE_URL_KEYS: dict[tuple[str, str], tuple[str, ...]] = { + ("eu", "prod"): ("SEMANTIC_SEGMENTATION_URL",), + ("eu", "staging"): ("SEMANTIC_SEGMENTATION_URL",), +} + +_UNSET = object() + + +class RegionWarning(UserWarning): + """Emitted when ROBOFLOW_REGION or ROBOFLOW_ENVIRONMENT holds an unrecognized value.""" + + +_WARNED_UNKNOWN_VALUES: set[tuple[str, str]] = set() def get_conditional_configuration_variable(key, default): @@ -42,6 +112,131 @@ def get_conditional_configuration_variable(key, default): return default +def _normalize_choice(setting: str, value, supported: tuple[str, ...], default: str) -> str: + normalized = value.strip().lower() if isinstance(value, str) else "" + if normalized in supported: + return normalized + + warning_key = (setting, repr(value)) + if warning_key not in _WARNED_UNKNOWN_VALUES: + _WARNED_UNKNOWN_VALUES.add(warning_key) + # This runs while roboflow is imported, before the CLI parses its flags, so + # mirror the CLI's --json detection: its stderr must stay machine-readable, + # and `auth status` reports the problem as a JSON field instead. + if not ("--json" in sys.argv or "-j" in sys.argv): + warnings.warn(_unknown_value_message(setting, value, default), RegionWarning, stacklevel=3) + return default + + +def _normalize_region(region) -> str: + return _normalize_choice("region", region, SUPPORTED_REGIONS, DEFAULT_REGION) + + +def _normalize_environment(environment) -> str: + return _normalize_choice("environment", environment, SUPPORTED_ENVIRONMENTS, DEFAULT_ENVIRONMENT) + + +def _unknown_value_message(setting: str, value, default: str) -> str: + return f"unknown Roboflow {setting} {value!r}; falling back to {default!r}." + + +def unknown_region_message(region) -> str: + return _unknown_value_message("region", region, DEFAULT_REGION) + + +def get_region_warning() -> str | None: + """Return the fallback warning(s) when the configured region or environment is not recognized.""" + messages = [] + for setting, key, supported, default in ( + ("region", "ROBOFLOW_REGION", SUPPORTED_REGIONS, DEFAULT_REGION), + ("environment", "ROBOFLOW_ENVIRONMENT", SUPPORTED_ENVIRONMENTS, DEFAULT_ENVIRONMENT), + ): + value = get_conditional_configuration_variable(key, default=default) + normalized = value.strip().lower() if isinstance(value, str) else "" + if normalized not in supported: + messages.append(_unknown_value_message(setting, value, default)) + return " ".join(messages) or None + + +def get_effective_region() -> str: + """Return the configured Roboflow region, defaulting safely to US.""" + region = get_conditional_configuration_variable("ROBOFLOW_REGION", default=DEFAULT_REGION) + return _normalize_region(region) + + +def get_effective_environment() -> str: + """Return the configured Roboflow environment, defaulting safely to production.""" + environment = get_conditional_configuration_variable("ROBOFLOW_ENVIRONMENT", default=DEFAULT_ENVIRONMENT) + return _normalize_environment(environment) + + +def resolve_url(key: str, region: str | None = None, environment: str | None = None) -> str: + """Resolve a Roboflow URL: explicit override, then region/environment default, then US production.""" + if key not in URL_DEFAULTS: + raise KeyError(f"Unknown Roboflow URL configuration key: {key}") + + explicit_url = get_conditional_configuration_variable(key, default=_UNSET) + if explicit_url is not _UNSET: + return explicit_url + + effective_region = get_effective_region() if region is None else _normalize_region(region) + effective_environment = get_effective_environment() if environment is None else _normalize_environment(environment) + return SERVICE_URL_OVERRIDES[(effective_region, effective_environment)].get(key, URL_DEFAULTS[key]) + + +def ensure_url_available_in_region(key: str) -> None: + """Raise if ``key`` has no deployment in the effective region/environment and no explicit override.""" + target = (get_effective_region(), get_effective_environment()) + if key not in UNAVAILABLE_URL_KEYS.get(target, ()): + return + if get_conditional_configuration_variable(key, default=_UNSET) is not _UNSET: + return + label = f"{target[0].upper()} {target[1]}" + raise RuntimeError( + f"{key} has no Roboflow {label} deployment; the default {resolve_url(key)} would send " + f"data outside {label}. Set {key} explicitly to override." + ) + + +def resolve_available_url(key: str) -> str: + """Resolve ``key`` for a request, refusing when the service is not deployed in the effective region.""" + ensure_url_available_in_region(key) + return resolve_url(key) + + +def region_conflict(region: str) -> str | None: + """Return the ROBOFLOW_REGION environment value when it would override ``region``, else None. + + The environment variable wins over every saved or explicit choice when URLs are + resolved, so authenticating against ``region`` would pair its credentials with + requests sent to another platform. + """ + value = os.getenv("ROBOFLOW_REGION") + if value is None or _normalize_region(value) == region: + return None + return value + + +CREDENTIALS_REGION_KEY = "ROBOFLOW_CREDENTIALS_REGION" + + +def has_credentials(config) -> bool: + """Whether a loaded config holds workspace credentials, not just preferences such as a region.""" + return isinstance(config, dict) and bool(config.get("workspaces")) + + +def credentials_region(config) -> str: + """Region whose platform issued the stored credentials. + + Recorded at login and independent of ROBOFLOW_REGION, which ``auth set-region`` + changes without touching credentials. Configs written before the key existed can + only hold US credentials. + """ + stored = config.get(CREDENTIALS_REGION_KEY) if isinstance(config, dict) else None + normalized = stored.strip().lower() if isinstance(stored, str) else "" + return normalized if normalized in SUPPORTED_REGIONS else DEFAULT_REGION + + CLASSIFICATION_MODEL = os.getenv("CLASSIFICATION_MODEL", "ClassificationModel") INSTANCE_SEGMENTATION_MODEL = "InstanceSegmentationModel" KEYPOINT_DETECTION_MODEL = "KeypointDetectionModel" @@ -49,22 +244,41 @@ def get_conditional_configuration_variable(key, default): SEMANTIC_SEGMENTATION_MODEL = "SemanticSegmentationModel" PREDICTION_OBJECT = os.getenv("PREDICTION_OBJECT", "Prediction") -API_URL = get_conditional_configuration_variable("API_URL", "https://api.roboflow.com") -APP_URL = get_conditional_configuration_variable("APP_URL", "https://app.roboflow.com") -UNIVERSE_URL = get_conditional_configuration_variable("UNIVERSE_URL", "https://universe.roboflow.com") +API_URL = resolve_url("API_URL") +APP_URL = resolve_url("APP_URL") +UNIVERSE_URL = resolve_url("UNIVERSE_URL") + +INSTANCE_SEGMENTATION_URL = resolve_url("INSTANCE_SEGMENTATION_URL") +SEMANTIC_SEGMENTATION_URL = resolve_url("SEMANTIC_SEGMENTATION_URL") +OBJECT_DETECTION_URL = resolve_url("OBJECT_DETECTION_URL") +SERVERLESS_URL = resolve_url("SERVERLESS_URL") + +CLIP_FEATURIZE_URL = resolve_url("CLIP_FEATURIZE_URL") +OCR_URL = resolve_url("OCR_URL") + +DEDICATED_DEPLOYMENT_URL = resolve_url("DEDICATED_DEPLOYMENT_URL") + -INSTANCE_SEGMENTATION_URL = get_conditional_configuration_variable( - "INSTANCE_SEGMENTATION_URL", "https://serverless.roboflow.com" -) -SEMANTIC_SEGMENTATION_URL = get_conditional_configuration_variable( - "SEMANTIC_SEGMENTATION_URL", "https://segment.roboflow.com" -) -OBJECT_DETECTION_URL = get_conditional_configuration_variable("OBJECT_DETECTION_URL", "https://serverless.roboflow.com") +def refresh_region_urls() -> None: + """Re-resolve the URL constants after the region changes in a running process. -CLIP_FEATURIZE_URL = get_conditional_configuration_variable("CLIP_FEATURIZE_URL", "CLIP FEATURIZE URL NOT IN ENV") -OCR_URL = get_conditional_configuration_variable("OCR_URL", "OCR URL NOT IN ENV") + Modules across the package bind these constants by value at import time + (``from roboflow.config import API_URL``), so a login that switches region + would otherwise keep talking to the previous platform until restart. Only + bindings still holding the previous default are replaced, so values a caller + patched deliberately are left alone. + """ + module_globals = globals() + previous_urls = {key: module_globals[key] for key in URL_DEFAULTS} + current_urls = {key: resolve_url(key) for key in URL_DEFAULTS} + for module_name, module in list(sys.modules.items()): + if module is None or not (module_name == "roboflow" or module_name.startswith("roboflow.")): + continue + namespace = vars(module) + for key, previous_url in previous_urls.items(): + if namespace.get(key, _UNSET) == previous_url: + namespace[key] = current_urls[key] -DEDICATED_DEPLOYMENT_URL = get_conditional_configuration_variable("DEDICATED_DEPLOYMENT_URL", "https://roboflow.cloud") DEMO_KEYS = ["coco-128-sample", "chess-sample-only-api-key"] diff --git a/roboflow/core/training.py b/roboflow/core/training.py index 96e02f7f..29d8c982 100644 --- a/roboflow/core/training.py +++ b/roboflow/core/training.py @@ -22,12 +22,12 @@ OBJECT_DETECTION_MODEL, OBJECT_DETECTION_URL, SEMANTIC_SEGMENTATION_MODEL, - SEMANTIC_SEGMENTATION_URL, TASK_CLS, TASK_OBB, TASK_POSE, TASK_SEG, TASK_SEM, + resolve_available_url, ) from roboflow.models.inference import InferenceModel from roboflow.util.model_processor import task_of_model_type @@ -35,7 +35,7 @@ def _serverless_base_url_for_task(task: str) -> str: if task == TASK_SEM: - return SEMANTIC_SEGMENTATION_URL + return resolve_available_url("SEMANTIC_SEGMENTATION_URL") return OBJECT_DETECTION_URL diff --git a/roboflow/models/classification.py b/roboflow/models/classification.py index 15c8be94..b917aa33 100644 --- a/roboflow/models/classification.py +++ b/roboflow/models/classification.py @@ -8,7 +8,7 @@ import requests from PIL import Image -from roboflow.config import CLASSIFICATION_MODEL +from roboflow.config import CLASSIFICATION_MODEL, SERVERLESS_URL from roboflow.models.inference import InferenceModel from roboflow.util.image_utils import check_image_url from roboflow.util.prediction import PredictionGroup @@ -51,7 +51,7 @@ def __init__( self.id = id self.name = name self.version = version - self.base_url = "https://serverless.roboflow.com/" + self.base_url = SERVERLESS_URL + "/" if self.name is not None and version is not None: self.__generate_url() diff --git a/roboflow/models/keypoint_detection.py b/roboflow/models/keypoint_detection.py index c3b7321e..8f99588b 100644 --- a/roboflow/models/keypoint_detection.py +++ b/roboflow/models/keypoint_detection.py @@ -8,7 +8,7 @@ import requests from PIL import Image -from roboflow.config import KEYPOINT_DETECTION_MODEL +from roboflow.config import KEYPOINT_DETECTION_MODEL, SERVERLESS_URL from roboflow.models.inference import InferenceModel from roboflow.util.image_utils import check_image_url from roboflow.util.prediction import PredictionGroup @@ -53,7 +53,7 @@ def __init__( self.confidence = confidence self.version = version self.colors = {} - self.base_url = "https://serverless.roboflow.com/" + self.base_url = SERVERLESS_URL + "/" if self.name is not None and version is not None: self.__generate_url() diff --git a/roboflow/models/object_detection.py b/roboflow/models/object_detection.py index 5793ec86..f8c14d6b 100644 --- a/roboflow/models/object_detection.py +++ b/roboflow/models/object_detection.py @@ -278,7 +278,7 @@ def predict( # type: ignore[override] def webcam( self, webcam_id=0, - inference_engine_url="https://serverless.roboflow.com/", + inference_engine_url=None, within_jupyter=False, confidence=40, overlap=30, @@ -291,7 +291,7 @@ def webcam( Args: webcam_id (int): Webcam ID (default 0) - inference_engine_url (str): Inference engine address to use (default https://serverless.roboflow.com) + inference_engine_url (str): Inference engine address to use (default: the region's serverless host) within_jupyter (bool): Whether or not to display the webcam within Jupyter notebook (default True) confidence (int): Confidence threshold for detections overlap (int): Overlap threshold for detections @@ -302,6 +302,9 @@ def webcam( os.environ["OPENCV_VIDEOIO_PRIORITY_MSMF"] = "0" + if inference_engine_url is None: + inference_engine_url = OBJECT_DETECTION_URL + "/" + # Generate url before predicting self.__generate_url( confidence=confidence, diff --git a/roboflow/models/semantic_segmentation.py b/roboflow/models/semantic_segmentation.py index 5dfd5659..3e959a13 100644 --- a/roboflow/models/semantic_segmentation.py +++ b/roboflow/models/semantic_segmentation.py @@ -1,4 +1,4 @@ -from roboflow.config import SEMANTIC_SEGMENTATION_MODEL, SEMANTIC_SEGMENTATION_URL +from roboflow.config import SEMANTIC_SEGMENTATION_MODEL, SEMANTIC_SEGMENTATION_URL, resolve_available_url from roboflow.models.inference import InferenceModel @@ -40,6 +40,10 @@ def predict(self, image_path: str, confidence: int = 50): # type: ignore[overri >>> prediction = model.predict("YOUR_IMAGE.jpg") """ # noqa: E501 // docs + # Resolve the destination now, with the same lookup the residency check uses, + # so the check always applies to the host that receives the image. + base_url = resolve_available_url("SEMANTIC_SEGMENTATION_URL") + self.api_url = f"{base_url}/{self.dataset_id}/{self.version}" return super().predict( image_path, confidence=confidence, diff --git a/roboflow/models/vlm.py b/roboflow/models/vlm.py index b24c3ceb..31f720e8 100644 --- a/roboflow/models/vlm.py +++ b/roboflow/models/vlm.py @@ -18,6 +18,7 @@ import requests from PIL import Image +from roboflow.config import SERVERLESS_URL from roboflow.models.inference import InferenceModel from roboflow.util.image_utils import check_image_url @@ -40,7 +41,7 @@ def __init__( self.id = id self.name = name self.version = version - self.base_url = local if local else "https://serverless.roboflow.com/" + self.base_url = local if local else SERVERLESS_URL + "/" self.colors = {} if colors is None else colors self.preprocessing = {} if preprocessing is None else preprocessing diff --git a/tests/cli/test_auth_region.py b/tests/cli/test_auth_region.py new file mode 100644 index 00000000..985a9bb4 --- /dev/null +++ b/tests/cli/test_auth_region.py @@ -0,0 +1,295 @@ +"""Region-specific tests for the auth CLI handler.""" + +import json +import os +import re +import tempfile +import unittest +from unittest import mock + +import responses +from typer.testing import CliRunner + +from roboflow.cli import app + +runner = CliRunner() + +_ANSI_RE = re.compile(r"\x1b\[[0-9;]*m") + + +def _strip_ansi(text: str) -> str: + return _ANSI_RE.sub("", text) + + +class TestAuthRegion(unittest.TestCase): + def setUp(self) -> None: + self.tempdir = tempfile.TemporaryDirectory() + self.config_path = os.path.join(self.tempdir.name, "config.json") + self.env_patch = mock.patch.dict( + os.environ, + {"ROBOFLOW_CONFIG_DIR": self.config_path}, + clear=False, + ) + self.env_patch.start() + for key in ("ROBOFLOW_REGION", "ROBOFLOW_ENVIRONMENT", "API_URL", "APP_URL", "ROBOFLOW_API_KEY"): + os.environ.pop(key, None) + + def tearDown(self) -> None: + self.env_patch.stop() + self.tempdir.cleanup() + + def _write_config(self, config: dict) -> None: + with open(self.config_path, "w") as config_file: + json.dump(config, config_file) + + def _read_config(self) -> dict: + with open(self.config_path) as config_file: + return json.load(config_file) + + def _write_logged_in_config(self) -> None: + self._write_config( + { + "workspaces": { + "eu-workspace": { + "url": "eu-workspace", + "name": "EU Workspace", + "apiKey": "eu-secret-key", + } + }, + "RF_WORKSPACE": "eu-workspace", + } + ) + + def test_login_and_alias_help_include_region(self) -> None: + auth_result = runner.invoke(app, ["auth", "login", "--help"]) + alias_result = runner.invoke(app, ["login", "--help"]) + + self.assertEqual(auth_result.exit_code, 0) + self.assertEqual(alias_result.exit_code, 0) + # Rich styles "--region" per segment when colors are forced (as in CI). + self.assertIn("--region", _strip_ansi(auth_result.output)) + self.assertIn("--region", _strip_ansi(alias_result.output)) + + def test_interactive_login_passes_normalized_region(self) -> None: + with mock.patch("roboflow.login") as login: + result = runner.invoke(app, ["auth", "login", "--region", "EU"]) + + self.assertEqual(result.exit_code, 0, result.output) + login.assert_called_once_with(workspace=None, force=False, region="eu") + + def test_login_alias_passes_normalized_region(self) -> None: + with mock.patch("roboflow.login") as login: + result = runner.invoke(app, ["login", "--region", "EU"]) + + self.assertEqual(result.exit_code, 0, result.output) + login.assert_called_once_with(workspace=None, force=False, region="eu") + + def test_login_with_new_region_reauthenticates_existing_user(self) -> None: + self._write_logged_in_config() + + with mock.patch("roboflow.login") as login: + result = runner.invoke(app, ["auth", "login", "--region", "eu"]) + + self.assertEqual(result.exit_code, 0, result.output) + login.assert_called_once_with(workspace=None, force=True, region="eu") + + def test_login_with_current_region_keeps_existing_session(self) -> None: + self._write_logged_in_config() + + with mock.patch("roboflow.login") as login: + result = runner.invoke(app, ["auth", "login", "--region", "us"]) + + self.assertEqual(result.exit_code, 0, result.output) + self.assertIn("Already logged in", result.output) + login.assert_not_called() + + def test_login_after_set_region_on_fresh_install_authenticates(self) -> None: + set_result = runner.invoke(app, ["auth", "set-region", "eu"]) + for command in (["auth", "login", "--region", "eu"], ["auth", "login"]): + with self.subTest(command=command), mock.patch("roboflow.login") as login: + result = runner.invoke(app, command) + + self.assertEqual(result.exit_code, 0, result.output) + self.assertNotIn("Already logged in", result.output) + login.assert_called_once() + self.assertEqual(set_result.exit_code, 0, set_result.output) + + def test_set_region_does_not_relabel_existing_credentials(self) -> None: + # US credentials, then a switch of the routing preference only. + self._write_logged_in_config() + runner.invoke(app, ["auth", "set-region", "eu"]) + + for command, region in ((["auth", "login", "--region", "eu"], "eu"), (["auth", "login"], None)): + with self.subTest(command=command), mock.patch("roboflow.login") as login: + result = runner.invoke(app, command) + + self.assertEqual(result.exit_code, 0, result.output) + login.assert_called_once_with(workspace=None, force=True, region=region) + + def test_login_region_conflicting_with_environment_is_refused(self) -> None: + os.environ["ROBOFLOW_REGION"] = "us" + + with mock.patch("roboflow.login") as login, responses.RequestsMock() as mocked: + for command in ( + ["auth", "login", "--region", "eu"], + ["auth", "login", "--api-key", "eu-key", "--region", "eu"], + ): + with self.subTest(command=command): + result = runner.invoke(app, command) + + self.assertEqual(result.exit_code, 2, result.output) + self.assertIn("ROBOFLOW_REGION=us", result.output) + self.assertEqual(len(mocked.calls), 0) + login.assert_not_called() + self.assertFalse(os.path.exists(self.config_path)) + + @responses.activate + def test_api_key_login_on_other_platform_replaces_workspaces(self) -> None: + self._write_logged_in_config() + responses.add( + responses.POST, "https://api.roboflow.eu/?api_key=eu-key", json={"workspace": "new-eu"}, status=200 + ) + responses.add( + responses.GET, + "https://api.roboflow.eu/new-eu?api_key=eu-key", + json={"workspace": {"name": "New EU"}}, + status=200, + ) + + result = runner.invoke(app, ["auth", "login", "--api-key", "eu-key", "--region", "eu"]) + + self.assertEqual(result.exit_code, 0, result.output) + config = self._read_config() + self.assertEqual(list(config["workspaces"]), ["new-eu"]) + self.assertEqual(config["ROBOFLOW_CREDENTIALS_REGION"], "eu") + + def test_set_region_without_credentials_has_no_warning(self) -> None: + result = runner.invoke(app, ["--json", "auth", "set-region", "eu"]) + + self.assertEqual(result.exit_code, 0, result.output) + self.assertIsNone(json.loads(result.stdout)["warning"]) + + def test_set_region_to_current_region_has_no_warning(self) -> None: + self._write_logged_in_config() + + result = runner.invoke(app, ["auth", "set-region", "us"]) + + self.assertEqual(result.exit_code, 0, result.output) + self.assertNotIn("Warning", result.output) + + @responses.activate + def test_api_key_login_uses_eu_api_and_persists_region(self) -> None: + responses.add( + responses.POST, + "https://api.roboflow.eu/?api_key=eu-key", + json={"workspace": "eu-workspace"}, + status=200, + ) + responses.add( + responses.GET, + "https://api.roboflow.eu/eu-workspace?api_key=eu-key", + json={"workspace": {"name": "EU Workspace"}}, + status=200, + ) + + result = runner.invoke( + app, + ["auth", "login", "--api-key", "eu-key", "--region", "eu"], + ) + + self.assertEqual(result.exit_code, 0, result.output) + self.assertEqual( + [call.request.url for call in responses.calls], + [ + "https://api.roboflow.eu/?api_key=eu-key", + "https://api.roboflow.eu/eu-workspace?api_key=eu-key", + ], + ) + config = self._read_config() + self.assertEqual(config["ROBOFLOW_REGION"], "eu") + self.assertEqual(config["RF_WORKSPACE"], "eu-workspace") + self.assertEqual(config["workspaces"]["eu-workspace"]["apiKey"], "eu-key") + + def test_set_region_then_status_shows_eu_endpoints(self) -> None: + self._write_logged_in_config() + + set_result = runner.invoke(app, ["auth", "set-region", "eu"]) + status_result = runner.invoke(app, ["auth", "status"]) + + self.assertEqual(set_result.exit_code, 0, set_result.output) + self.assertIn("Region set to: eu", set_result.output) + self.assertIn("API URL: https://api.roboflow.eu", set_result.output) + self.assertIn("App URL: https://app.roboflow.eu", set_result.output) + self.assertIn("separate authentication backends and API keys", set_result.output) + self.assertIn("roboflow auth login --force", set_result.output) + self.assertEqual(status_result.exit_code, 0, status_result.output) + self.assertIn("Region: eu", status_result.output) + self.assertIn("API URL: https://api.roboflow.eu", status_result.output) + self.assertIn("App URL: https://app.roboflow.eu", status_result.output) + self.assertEqual(self._read_config()["ROBOFLOW_REGION"], "eu") + + def test_status_json_includes_region_and_urls(self) -> None: + self._write_logged_in_config() + runner.invoke(app, ["auth", "set-region", "eu"]) + + result = runner.invoke(app, ["--json", "auth", "status"]) + + self.assertEqual(result.exit_code, 0, result.output) + payload = json.loads(result.stdout) + self.assertEqual(payload["region"], "eu") + self.assertEqual(payload["api_url"], "https://api.roboflow.eu") + self.assertEqual(payload["app_url"], "https://app.roboflow.eu") + + def test_status_json_includes_staging_environment(self) -> None: + self._write_logged_in_config() + os.environ["ROBOFLOW_REGION"] = "eu" + os.environ["ROBOFLOW_ENVIRONMENT"] = "staging" + + result = runner.invoke(app, ["--json", "auth", "status"]) + + self.assertEqual(result.exit_code, 0, result.output) + payload = json.loads(result.stdout) + self.assertEqual(payload["region"], "eu") + self.assertEqual(payload["environment"], "staging") + self.assertEqual(payload["api_url"], "https://api.roboflow-eu.one") + self.assertEqual(payload["app_url"], "https://app.roboflow-eu.one") + + def test_set_region_reports_environment_override_as_effective(self) -> None: + os.environ["ROBOFLOW_REGION"] = "us" + + result = runner.invoke(app, ["--json", "auth", "set-region", "eu"]) + + self.assertEqual(result.exit_code, 0, result.output) + payload = json.loads(result.stdout) + self.assertEqual(payload["configured_region"], "eu") + self.assertEqual(payload["region"], "us") + self.assertEqual(payload["api_url"], "https://api.roboflow.com") + self.assertEqual(payload["app_url"], "https://app.roboflow.com") + self.assertEqual(self._read_config()["ROBOFLOW_REGION"], "eu") + + def test_region_only_status_shows_endpoints_and_remains_not_logged_in(self) -> None: + set_result = runner.invoke(app, ["auth", "set-region", "eu"]) + status_result = runner.invoke(app, ["--json", "auth", "status"]) + + self.assertEqual(set_result.exit_code, 0, set_result.output) + self.assertEqual(status_result.exit_code, 2, status_result.output) + payload = json.loads(status_result.stderr) + self.assertEqual(payload["error"]["message"], "Not logged in.") + self.assertEqual(payload["region"], "eu") + self.assertEqual(payload["api_url"], "https://api.roboflow.eu") + self.assertEqual(payload["app_url"], "https://app.roboflow.eu") + + def test_set_region_rejects_invalid_value_without_mutating_config(self) -> None: + original = {"ROBOFLOW_REGION": "us", "preserved": True} + self._write_config(original) + + result = runner.invoke(app, ["auth", "set-region", "bogus"]) + + self.assertNotEqual(result.exit_code, 0) + self.assertIn("Invalid region 'bogus'", result.output) + self.assertIn("must be 'us' or 'eu'", result.output) + self.assertEqual(self._read_config(), original) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_login_region.py b/tests/test_login_region.py new file mode 100644 index 00000000..1da5d8b3 --- /dev/null +++ b/tests/test_login_region.py @@ -0,0 +1,272 @@ +"""Tests for region-aware interactive login.""" + +import io +import json +import os +import tempfile +import unittest +from contextlib import redirect_stdout +from unittest import mock + +import responses + +import roboflow +from roboflow.adapters import rfapi +from roboflow.config import refresh_region_urls + + +class TestLoginRegion(unittest.TestCase): + def setUp(self) -> None: + self.temporary_directory = tempfile.TemporaryDirectory() + self.config_path = os.path.join(self.temporary_directory.name, "config.json") + self.environment = mock.patch.dict( + os.environ, + { + "HOME": self.temporary_directory.name, + # config.py builds its default path from USERPROFILE on Windows. + "USERPROFILE": self.temporary_directory.name, + "ROBOFLOW_CONFIG_DIR": self.config_path, + }, + clear=True, + ) + self.environment.start() + + def tearDown(self) -> None: + self.environment.stop() + self.temporary_directory.cleanup() + # A region login rebinds package-wide URL constants; restore them. + refresh_region_urls() + + @responses.activate + def test_eu_login_uses_eu_app_and_persists_region(self) -> None: + token = "auth-token" + workspaces = { + "workspace-id": { + "url": "example-workspace", + "apiKey": "example-api-key", + } + } + responses.get( + f"https://app.roboflow.eu/query/cliAuthToken/{token}", + json=workspaces, + status=200, + ) + + output = io.StringIO() + with mock.patch.object(roboflow, "getpass", return_value=token), redirect_stdout(output): + roboflow.login(region="EU") + + self.assertIn("https://app.roboflow.eu/auth-cli", output.getvalue()) + self.assertEqual(responses.calls[0].request.url, f"https://app.roboflow.eu/query/cliAuthToken/{token}") + with open(self.config_path) as config_file: + config = json.load(config_file) + self.assertEqual(config["ROBOFLOW_REGION"], "eu") + self.assertEqual(config["workspaces"], workspaces) + self.assertEqual(config["RF_WORKSPACE"], "example-workspace") + + @responses.activate + def test_eu_login_switches_import_time_url_constants(self) -> None: + self.assertEqual(roboflow.API_URL, "https://api.roboflow.com") + token = "auth-token" + responses.get( + f"https://app.roboflow.eu/query/cliAuthToken/{token}", + json={"workspace-id": {"url": "example-workspace", "apiKey": "example-api-key"}}, + status=200, + ) + + with mock.patch.object(roboflow, "getpass", return_value=token), redirect_stdout(io.StringIO()): + roboflow.login(region="eu") + + self.assertEqual(roboflow.API_URL, "https://api.roboflow.eu") + self.assertEqual(roboflow.APP_URL, "https://app.roboflow.eu") + self.assertEqual(rfapi.API_URL, "https://api.roboflow.eu") + self.assertEqual(roboflow.config.API_URL, "https://api.roboflow.eu") + + @responses.activate + def test_forced_login_preserves_existing_region_and_other_config(self) -> None: + existing_config = { + "ROBOFLOW_REGION": "eu", + "API_URL": "https://custom-api.example.com", + "workspaces": {"old": {"url": "old-workspace", "apiKey": "old-key"}}, + "RF_WORKSPACE": "old-workspace", + } + with open(self.config_path, "w") as config_file: + json.dump(existing_config, config_file) + + token = "replacement-token" + workspaces = { + "new": { + "url": "new-workspace", + "apiKey": "new-key", + } + } + responses.get( + f"https://app.roboflow.eu/query/cliAuthToken/{token}", + json=workspaces, + status=200, + ) + + with mock.patch.object(roboflow, "getpass", return_value=token), redirect_stdout(io.StringIO()): + roboflow.login(force=True) + + with open(self.config_path) as config_file: + config = json.load(config_file) + self.assertEqual(config["ROBOFLOW_REGION"], "eu") + self.assertEqual(config["API_URL"], "https://custom-api.example.com") + self.assertEqual(config["workspaces"], workspaces) + self.assertEqual(config["RF_WORKSPACE"], "new-workspace") + + def _write_config(self, config: dict) -> None: + with open(self.config_path, "w") as config_file: + json.dump(config, config_file) + + def _read_config(self) -> dict: + with open(self.config_path) as config_file: + return json.load(config_file) + + @responses.activate + def test_region_only_config_is_not_a_session(self) -> None: + # What `auth set-region eu` writes on a fresh install. + self._write_config({"ROBOFLOW_REGION": "eu"}) + token = "auth-token" + workspaces = {"workspace-id": {"url": "example-workspace", "apiKey": "example-api-key"}} + responses.get(f"https://app.roboflow.eu/query/cliAuthToken/{token}", json=workspaces, status=200) + + output = io.StringIO() + with mock.patch.object(roboflow, "getpass", return_value=token), redirect_stdout(output): + roboflow.login() + + self.assertNotIn("already logged in", output.getvalue()) + config = self._read_config() + self.assertEqual(config["ROBOFLOW_REGION"], "eu") + self.assertEqual(config["ROBOFLOW_CREDENTIALS_REGION"], "eu") + self.assertEqual(config["workspaces"], workspaces) + + @responses.activate + def test_credentials_from_other_platform_are_replaced_without_force(self) -> None: + # US credentials, then `auth set-region eu`. + self._write_config( + { + "ROBOFLOW_REGION": "eu", + "workspaces": {"old": {"url": "us-workspace", "apiKey": "us-key"}}, + "RF_WORKSPACE": "us-workspace", + } + ) + token = "auth-token" + workspaces = {"new": {"url": "eu-workspace", "apiKey": "eu-key"}} + responses.get(f"https://app.roboflow.eu/query/cliAuthToken/{token}", json=workspaces, status=200) + + with mock.patch.object(roboflow, "getpass", return_value=token), redirect_stdout(io.StringIO()): + roboflow.login() + + config = self._read_config() + self.assertEqual(config["workspaces"], workspaces) + self.assertEqual(config["ROBOFLOW_CREDENTIALS_REGION"], "eu") + + def test_credentials_from_same_platform_keep_session(self) -> None: + original = { + "ROBOFLOW_CREDENTIALS_REGION": "us", + "workspaces": {"old": {"url": "us-workspace", "apiKey": "us-key"}}, + "RF_WORKSPACE": "us-workspace", + } + self._write_config(original) + + output = io.StringIO() + with mock.patch.object(roboflow, "getpass") as getpass, redirect_stdout(output): + roboflow.login() + + getpass.assert_not_called() + self.assertIn("already logged in", output.getvalue()) + self.assertEqual(self._read_config(), original) + + @responses.activate + def test_environment_region_is_recorded_as_credentials_region(self) -> None: + os.environ["ROBOFLOW_REGION"] = "eu" + refresh_region_urls() + token = "auth-token" + responses.get( + f"https://app.roboflow.eu/query/cliAuthToken/{token}", + json={"workspace-id": {"url": "example-workspace", "apiKey": "example-api-key"}}, + status=200, + ) + + with mock.patch.object(roboflow, "getpass", return_value=token), redirect_stdout(io.StringIO()): + roboflow.login() + + config = self._read_config() + self.assertEqual(config["ROBOFLOW_CREDENTIALS_REGION"], "eu") + self.assertNotIn("ROBOFLOW_REGION", config) + + def test_explicit_region_conflicting_with_environment_is_refused(self) -> None: + os.environ["ROBOFLOW_REGION"] = "us" + + with mock.patch.object(roboflow, "getpass") as getpass, responses.RequestsMock() as mocked: + with self.assertRaisesRegex(ValueError, "ROBOFLOW_REGION=us.*region='eu'"): + roboflow.login(region="eu") + self.assertEqual(len(mocked.calls), 0) + + getpass.assert_not_called() + self.assertFalse(os.path.exists(self.config_path)) + + @responses.activate + def test_failed_forced_login_keeps_existing_credentials(self) -> None: + original = { + "workspaces": {"old": {"url": "old-workspace", "apiKey": "old-key"}}, + "RF_WORKSPACE": "old-workspace", + } + self._write_config(original) + token = "bad-token" + responses.get(f"https://app.roboflow.com/query/cliAuthToken/{token}", status=500) + + with mock.patch.object(roboflow, "getpass", return_value=token), redirect_stdout(io.StringIO()): + with self.assertRaises(Exception): + roboflow.login(force=True) + + self.assertEqual(self._read_config(), original) + + def test_invalid_region_does_not_mutate_config(self) -> None: + original_config = {"ROBOFLOW_REGION": "eu", "marker": "unchanged"} + with open(self.config_path, "w") as config_file: + json.dump(original_config, config_file) + + with self.assertRaisesRegex(ValueError, "Invalid region 'bogus'.*us, eu"): + roboflow.login(force=True, region="bogus") + + with open(self.config_path) as config_file: + self.assertEqual(json.load(config_file), original_config) + + +class TestEuAppUrls(unittest.TestCase): + def test_download_dataset_accepts_eu_app_url(self) -> None: + with mock.patch.object(roboflow, "initialize_roboflow") as initialize: + workspace = initialize.return_value + roboflow.download_dataset("https://app.roboflow.eu/eu-ws/eu-project/3", "coco", location="/tmp/x") + + initialize.assert_called_once_with(the_workspace="eu-ws") + workspace.project.assert_called_once_with("eu-project") + workspace.project.return_value.version.assert_called_once_with(3) + + def test_load_model_accepts_eu_app_url(self) -> None: + with mock.patch.object(roboflow, "initialize_roboflow") as initialize: + roboflow.load_model("https://app.roboflow.eu/eu-ws/eu-project/2") + + initialize.return_value.project.assert_called_once_with("eu-project") + initialize.return_value.project.return_value.version.assert_called_once_with(2) + + def test_load_model_accepts_staging_app_urls(self) -> None: + for url in ( + "https://app.roboflow.one/ws/project/2", + "https://app.roboflow-eu.one/ws/project/2", + "https://universe.roboflow.one/ws/project/2", + ): + with self.subTest(url=url), mock.patch.object(roboflow, "initialize_roboflow") as initialize: + roboflow.load_model(url) + initialize.return_value.project.assert_called_once_with("project") + + def test_unknown_host_is_still_rejected(self) -> None: + with self.assertRaisesRegex(ValueError, "app.roboflow.eu"): + roboflow.download_dataset("https://example.com/ws/project/1", "coco") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_region_config.py b/tests/test_region_config.py new file mode 100644 index 00000000..b18902b6 --- /dev/null +++ b/tests/test_region_config.py @@ -0,0 +1,219 @@ +"""Tests for region-aware Roboflow URL configuration.""" + +import importlib +import json +import os +import subprocess +import sys +import tempfile +import unittest +import warnings +from pathlib import Path + +import roboflow.config as config_module + +URL_DEFAULTS = { + "API_URL": "https://api.roboflow.com", + "APP_URL": "https://app.roboflow.com", + "UNIVERSE_URL": "https://universe.roboflow.com", + "INSTANCE_SEGMENTATION_URL": "https://serverless.roboflow.com", + "SEMANTIC_SEGMENTATION_URL": "https://segment.roboflow.com", + "OBJECT_DETECTION_URL": "https://serverless.roboflow.com", + "SERVERLESS_URL": "https://serverless.roboflow.com", + "CLIP_FEATURIZE_URL": "CLIP FEATURIZE URL NOT IN ENV", + "OCR_URL": "OCR URL NOT IN ENV", + "DEDICATED_DEPLOYMENT_URL": "https://roboflow.cloud", +} + +REGION_ENVIRONMENT_KEYS = ("ROBOFLOW_CONFIG_DIR", "ROBOFLOW_REGION", "ROBOFLOW_ENVIRONMENT", *URL_DEFAULTS) + + +class TestRegionConfiguration(unittest.TestCase): + def setUp(self) -> None: + self.temp_directory = tempfile.TemporaryDirectory() + self.config_path = Path(self.temp_directory.name) / "config.json" + self.saved_environment = {key: os.environ[key] for key in REGION_ENVIRONMENT_KEYS if key in os.environ} + for key in REGION_ENVIRONMENT_KEYS: + os.environ.pop(key, None) + os.environ["ROBOFLOW_CONFIG_DIR"] = str(self.config_path) + self.config = importlib.reload(config_module) + + def tearDown(self) -> None: + for key in REGION_ENVIRONMENT_KEYS: + os.environ.pop(key, None) + os.environ.update(self.saved_environment) + importlib.reload(config_module) + self.temp_directory.cleanup() + + def _write_config(self, config: dict) -> None: + self.config_path.write_text(json.dumps(config)) + + def _reload_config(self): + self.config = importlib.reload(config_module) + return self.config + + def test_existing_us_url_defaults_are_unchanged(self) -> None: + self.assertEqual(self.config.get_effective_region(), "us") + for key, expected_url in URL_DEFAULTS.items(): + with self.subTest(key=key): + self.assertEqual(getattr(self.config, key), expected_url) + self.assertEqual(self.config.resolve_url(key), expected_url) + + def test_region_and_explicit_url_precedence(self) -> None: + self._write_config({"ROBOFLOW_REGION": "us"}) + os.environ["ROBOFLOW_REGION"] = "EU" + config = self._reload_config() + self.assertEqual(config.get_effective_region(), "eu") + self.assertEqual(config.API_URL, "https://api.roboflow.eu") + + os.environ.pop("ROBOFLOW_REGION") + self._write_config({"ROBOFLOW_REGION": "eU"}) + config = self._reload_config() + self.assertEqual(config.get_effective_region(), "eu") + self.assertEqual(config.API_URL, "https://api.roboflow.eu") + + os.environ["API_URL"] = "https://api.env.example" + config = self._reload_config() + self.assertEqual(config.API_URL, "https://api.env.example") + self.assertEqual(config.resolve_url("API_URL"), "https://api.env.example") + + os.environ.pop("API_URL") + self._write_config( + { + "ROBOFLOW_REGION": "eu", + "API_URL": "https://api.config.example", + } + ) + config = self._reload_config() + self.assertEqual(config.API_URL, "https://api.config.example") + self.assertEqual(config.resolve_url("API_URL"), "https://api.config.example") + + def test_eu_region_url_map(self) -> None: + self._write_config({"ROBOFLOW_REGION": "eu"}) + config = self._reload_config() + expected_urls = { + "API_URL": "https://api.roboflow.eu", + "APP_URL": "https://app.roboflow.eu", + "OBJECT_DETECTION_URL": "https://serverless.roboflow.eu", + "INSTANCE_SEGMENTATION_URL": "https://serverless.roboflow.eu", + "SERVERLESS_URL": "https://serverless.roboflow.eu", + "DEDICATED_DEPLOYMENT_URL": "https://eu.roboflow.cloud", + "UNIVERSE_URL": "https://universe.roboflow.com", + "SEMANTIC_SEGMENTATION_URL": "https://segment.roboflow.com", + } + for key, expected_url in expected_urls.items(): + with self.subTest(key=key): + self.assertEqual(getattr(config, key), expected_url) + self.assertEqual(config.resolve_url(key), expected_url) + + os.environ["ROBOFLOW_REGION"] = "us" + self.assertEqual( + config.resolve_url("API_URL", region="EU"), + "https://api.roboflow.eu", + ) + + def test_region_environment_url_matrix(self) -> None: + expected = { + ("us", "staging"): { + "API_URL": "https://api.roboflow.one", + "APP_URL": "https://app.roboflow.one", + "UNIVERSE_URL": "https://universe.roboflow.one", + "OBJECT_DETECTION_URL": "https://serverless.roboflow.one", + "INSTANCE_SEGMENTATION_URL": "https://serverless.roboflow.one", + "SERVERLESS_URL": "https://serverless.roboflow.one", + "SEMANTIC_SEGMENTATION_URL": "https://lambda-semantic-segmentation.staging.roboflow.com", + "DEDICATED_DEPLOYMENT_URL": "https://staging.roboflow.cloud", + }, + ("eu", "staging"): { + "API_URL": "https://api.roboflow-eu.one", + "APP_URL": "https://app.roboflow-eu.one", + "UNIVERSE_URL": "https://universe.roboflow.one", + "OBJECT_DETECTION_URL": "https://serverless.roboflow-eu.one", + "INSTANCE_SEGMENTATION_URL": "https://serverless.roboflow-eu.one", + "SERVERLESS_URL": "https://serverless.roboflow-eu.one", + "DEDICATED_DEPLOYMENT_URL": "https://eu.staging.roboflow.cloud", + }, + } + for (region, environment), urls in expected.items(): + os.environ["ROBOFLOW_REGION"] = region + os.environ["ROBOFLOW_ENVIRONMENT"] = environment + config = self._reload_config() + self.assertEqual(config.get_effective_environment(), environment) + for key, expected_url in urls.items(): + with self.subTest(region=region, environment=environment, key=key): + self.assertEqual(getattr(config, key), expected_url) + + def test_environment_is_read_from_config_file(self) -> None: + self._write_config({"ROBOFLOW_ENVIRONMENT": "Staging"}) + config = self._reload_config() + self.assertEqual(config.get_effective_environment(), "staging") + self.assertEqual(config.API_URL, "https://api.roboflow.one") + + def test_explicit_url_beats_environment(self) -> None: + os.environ["ROBOFLOW_ENVIRONMENT"] = "staging" + os.environ["API_URL"] = "https://localapi.roboflow.one" + config = self._reload_config() + self.assertEqual(config.API_URL, "https://localapi.roboflow.one") + self.assertEqual(config.APP_URL, "https://app.roboflow.one") + + def test_unknown_environment_warns_and_falls_back_to_prod(self) -> None: + os.environ["ROBOFLOW_ENVIRONMENT"] = "production" + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + config = self._reload_config() + self.assertEqual(config.get_effective_environment(), "prod") + self.assertEqual(config.API_URL, URL_DEFAULTS["API_URL"]) + + messages = [str(w.message) for w in caught if issubclass(w.category, config.RegionWarning)] + self.assertEqual(messages, ["unknown Roboflow environment 'production'; falling back to 'prod'."]) + self.assertEqual(config.get_region_warning(), messages[0]) + + def test_semantic_segmentation_availability_by_region_and_environment(self) -> None: + os.environ["ROBOFLOW_ENVIRONMENT"] = "staging" + config = self._reload_config() + config.ensure_url_available_in_region("SEMANTIC_SEGMENTATION_URL") + + os.environ["ROBOFLOW_REGION"] = "eu" + with self.assertRaisesRegex(RuntimeError, "EU staging"): + config.ensure_url_available_in_region("SEMANTIC_SEGMENTATION_URL") + + def test_unknown_region_warns_once_and_falls_back_to_us(self) -> None: + os.environ["ROBOFLOW_REGION"] = "bogus" + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + config = self._reload_config() + self.assertEqual(config.get_effective_region(), "us") + self.assertEqual(config.resolve_url("API_URL"), URL_DEFAULTS["API_URL"]) + + region_warnings = [w for w in caught if issubclass(w.category, config.RegionWarning)] + self.assertEqual(len(region_warnings), 1) + self.assertIn("unknown Roboflow region 'bogus'", str(region_warnings[0].message)) + self.assertIn("falling back to 'us'", str(region_warnings[0].message)) + self.assertEqual(config.get_region_warning(), "unknown Roboflow region 'bogus'; falling back to 'us'.") + + def test_valid_region_has_no_region_warning(self) -> None: + self._write_config({"ROBOFLOW_REGION": "EU"}) + config = self._reload_config() + self.assertIsNone(config.get_region_warning()) + + def test_unknown_region_is_not_printed_in_cli_json_mode(self) -> None: + env = {key: value for key, value in os.environ.items() if key != "ROBOFLOW_API_KEY"} + env["ROBOFLOW_REGION"] = "bogus" + env["PYTHONPATH"] = str(Path(__file__).resolve().parents[1]) + result = subprocess.run( + [sys.executable, "-m", "roboflow.roboflowpy", "--json", "auth", "status"], + capture_output=True, + text=True, + env=env, + check=False, + ) + + self.assertEqual(result.returncode, 2, result.stderr) + payload = json.loads(result.stderr) + self.assertEqual(payload["error"]["message"], "Not logged in.") + self.assertEqual(payload["region"], "us") + self.assertIn("unknown Roboflow region 'bogus'", payload["region_warning"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_region_models.py b/tests/test_region_models.py new file mode 100644 index 00000000..449a29d2 --- /dev/null +++ b/tests/test_region_models.py @@ -0,0 +1,80 @@ +"""Hosted inference models must stay inside the selected region.""" + +import os +import tempfile +import unittest +from unittest import mock + +import responses + +from roboflow.config import TASK_SEM, refresh_region_urls +from roboflow.core.training import TrainedModel, _serverless_base_url_for_task +from roboflow.models.classification import ClassificationModel +from roboflow.models.keypoint_detection import KeypointDetectionModel +from roboflow.models.semantic_segmentation import SemanticSegmentationModel +from roboflow.models.vlm import VLMModel + + +class TestRegionModels(unittest.TestCase): + def setUp(self) -> None: + self.temporary_directory = tempfile.TemporaryDirectory() + self.environment = mock.patch.dict( + os.environ, + { + "HOME": self.temporary_directory.name, + "USERPROFILE": self.temporary_directory.name, + "ROBOFLOW_CONFIG_DIR": os.path.join(self.temporary_directory.name, "config.json"), + "ROBOFLOW_REGION": "eu", + }, + clear=True, + ) + self.environment.start() + refresh_region_urls() + + def tearDown(self) -> None: + self.environment.stop() + self.temporary_directory.cleanup() + refresh_region_urls() + + def test_serverless_models_use_eu_host(self) -> None: + models = { + "classification": ClassificationModel("key", "ws/proj/1", version="1"), + "keypoint": KeypointDetectionModel("key", "ws/proj/1", version="1"), + "vlm": VLMModel("key", "ws/proj/1", version="1"), + } + for name, model in models.items(): + with self.subTest(model=name): + self.assertEqual(model.base_url, "https://serverless.roboflow.eu/") + + def test_semantic_segmentation_refuses_to_leave_eu(self) -> None: + model = SemanticSegmentationModel("key", "ws/proj/1") + with responses.RequestsMock() as mocked: + with self.assertRaisesRegex(RuntimeError, "SEMANTIC_SEGMENTATION_URL"): + model.predict("tests/images/rabbit.JPG") + self.assertEqual(len(mocked.calls), 0) + + def test_trained_semantic_segmentation_refuses_to_leave_eu(self) -> None: + model = TrainedModel("key", "ws", "proj", "ws/model-slug", model_type="yolo26-sem") + with self.assertRaisesRegex(RuntimeError, "SEMANTIC_SEGMENTATION_URL"): + model.predict("tests/images/rabbit.JPG") + + def test_explicit_semantic_segmentation_url_is_the_request_destination(self) -> None: + model = SemanticSegmentationModel("key", "ws/proj/1") + # Set after construction: the override must still be where the image goes. + os.environ["SEMANTIC_SEGMENTATION_URL"] = "https://segment.example" + destinations = [] + with mock.patch( + "roboflow.models.inference.InferenceModel.predict", + autospec=True, + side_effect=lambda self, *args, **kwargs: destinations.append(self.api_url) or "ok", + ): + self.assertEqual(model.predict("tests/images/rabbit.JPG"), "ok") + self.assertEqual(destinations, ["https://segment.example/proj/1"]) + + def test_trained_semantic_segmentation_uses_explicit_url(self) -> None: + os.environ["SEMANTIC_SEGMENTATION_URL"] = "https://segment.example" + self.assertEqual(_serverless_base_url_for_task(TASK_SEM), "https://segment.example") + + +if __name__ == "__main__": + unittest.main()