diff --git a/.gitignore b/.gitignore index dd4ac37..f3d7307 100644 --- a/.gitignore +++ b/.gitignore @@ -1,5 +1,6 @@ .dccache __pycache__ +.pytest_cache/ # Build results bin/ diff --git a/README.md b/README.md index a787f1f..ee3a174 100644 --- a/README.md +++ b/README.md @@ -48,6 +48,7 @@ This is a list of examples and scripts compiled by the Snyk Customer Experience - [How to generate a list (CSV) of all user and service accounts in a group](userlist) - [Pysnyk examples](https://github.com/snyk-labs/pysnyk/tree/master/examples) - [Tool for bulk project deletion](bulk-delete) +- [How to convert a CSV of repos into an import-projects.json for snyk-api-import](csv-to-snyk-api-import) # Guides diff --git a/csv-to-snyk-api-import/README.md b/csv-to-snyk-api-import/README.md new file mode 100644 index 0000000..0e8eb12 --- /dev/null +++ b/csv-to-snyk-api-import/README.md @@ -0,0 +1,106 @@ +# csv-to-snyk-api-import + +Converts a CSV of repositories into an `import-projects.json` file for the [snyk-api-import](https://github.com/snyk/snyk-api-import) CLI tool. + +## Features + +- **Auto-detects integration type** from repo URL (`github.com` → `github`, `dev.azure.com` → `azure-repos`, `bitbucket.org` → `bitbucket-cloud`). +- **Interactive prompt for unknown hosts** — self-hosted instances (GitHub Enterprise, Bitbucket Server, GitLab, etc.) trigger a prompt asking you to select the correct integration type. +- **Resolves or creates Snyk organisations** — provide an existing `snyk_org_id` or a `snyk_org_name` and the script will find-or-create the org from a template. +- **Looks up integration IDs** automatically via the Snyk v1 API. +- **Generates `import-projects.json`** in the exact format snyk-api-import expects. +- **Prompts to run the import** after generating the file. + +## Prerequisites + +- Python 3.10+ +- [snyk-api-import](https://github.com/snyk/snyk-api-import/releases) CLI (on your `PATH` if you want to run imports directly) + +## Installation + +```bash +pip3 install -r requirements.txt +``` + +## Environment Variables + +| Variable | Required | Description | +|---|---|---| +| `SNYK_TOKEN` | **yes** | Your Snyk API token | +| `SNYK_GROUP_ID` | **yes** | Snyk group ID (needed for org creation) | +| `SNYK_TEMPLATE_ORG_ID` | no | Template org ID to clone settings from when creating new orgs | +| `SNYK_LOG_PATH` | no | Log directory for snyk-api-import (defaults to `./logs`) | +| `SNYK_API` | no | Snyk API base URL (defaults to `https://api.snyk.io`) | + +## CSV Format + +| Column | Required | Description | +|---|---|---| +| `repo_url` | **yes** | Full repository URL | +| `snyk_org_id` | one of these | Existing Snyk org public ID | +| `snyk_org_name` | one of these | Org name — will be found or created | +| `branch` | no | Branch to scan (defaults to `main`) | +| `exclusion_globs` | no | Comma-separated folder names to exclude | + +Either `snyk_org_id` or `snyk_org_name` must be provided per row (both is also fine). + +See [`sample.csv`](sample.csv) for an example. + +## Usage + +```bash +# Generate import-projects.json from your CSV +python3 csv_to_snyk_import.py input.csv + +# Specify a custom output filename +python3 csv_to_snyk_import.py input.csv my-import.json +``` + +The script will: + +1. Read and validate the CSV. +2. Prompt for the integration type of any unrecognised hostnames (e.g. self-hosted Bitbucket Server, GitLab, GitHub Enterprise). +3. Resolve or create Snyk organisations as needed. +4. Look up integration IDs for each org. +5. Write `import-projects.json`. +6. Ask if you want to run `snyk-api-import` immediately. + +### Supported SCM Platforms + +| Platform | Auto-detected | Integration type | +|---|---|---| +| GitHub.com | yes | `github` | +| Azure DevOps | yes | `azure-repos` | +| Bitbucket Cloud | yes | `bitbucket-cloud` | +| GitHub Enterprise | prompted | `github-enterprise` | +| Bitbucket Server / Data Center | prompted | `bitbucket-server` | +| GitLab | prompted | `gitlab` | + +## Running Tests + +```bash +pytest -v test_converter.py +``` + +All API calls are mocked — tests run entirely offline. + +## Generated JSON Format + +The output follows the [snyk-api-import format](https://github.com/snyk/snyk-api-import/blob/master/docs/import.md): + +```json +{ + "targets": [ + { + "orgId": "org-abc-123", + "integrationId": "int-xyz-789", + "target": { + "owner": "my-org", + "name": "my-repo", + "branch": "main" + }, + "exclusionGlobs": "fixtures, tests" + } + ] +} +``` diff --git a/csv-to-snyk-api-import/converter.py b/csv-to-snyk-api-import/converter.py new file mode 100644 index 0000000..a1b4602 --- /dev/null +++ b/csv-to-snyk-api-import/converter.py @@ -0,0 +1,280 @@ +""" +Pure transformation logic: URL parsing, CSV reading/validation, and +import-projects.json assembly. No network calls live here. +""" + +import csv +from urllib.parse import urlparse + +# --------------------------------------------------------------------------- +# Constants +# --------------------------------------------------------------------------- + +DEFAULT_BRANCH = "main" + +# Mapping of hostname patterns to Snyk integration type strings. +# Order matters – first match wins. +HOST_TO_INTEGRATION_TYPE = { + "github.com": "github", + "dev.azure.com": "azure-repos", + "bitbucket.org": "bitbucket-cloud", +} + +# All integration types the tool knows how to build targets for. +KNOWN_INTEGRATION_TYPES = ( + "github", + "github-enterprise", + "azure-repos", + "bitbucket-cloud", + "bitbucket-server", + "gitlab", +) + + +# --------------------------------------------------------------------------- +# URL helpers +# --------------------------------------------------------------------------- + + +def detect_integration_type(repo_url: str) -> str | None: + """Return the Snyk integration type string for a given repo URL. + + Supports github.com, dev.azure.com, and bitbucket.org. + Returns ``None`` for unrecognised hostnames so the caller can + prompt the user interactively. + + Raises ValueError for URLs that cannot be parsed. + """ + parsed = urlparse(repo_url) + hostname = parsed.hostname + if not hostname: + raise ValueError(f"Cannot parse hostname from URL: {repo_url}") + + for pattern, integration_type in HOST_TO_INTEGRATION_TYPE.items(): + if hostname == pattern or hostname.endswith(f".{pattern}"): + return integration_type + + return None + + +def hostname_from_url(repo_url: str) -> str: + """Extract the hostname from a URL. Raises ValueError if missing.""" + hostname = urlparse(repo_url).hostname + if not hostname: + raise ValueError(f"Cannot parse hostname from URL: {repo_url}") + return hostname + + +def parse_repo_url(repo_url: str, integration_type: str | None = None) -> dict: + """Parse a repo URL into a snyk-api-import ``target`` dict. + + The returned dict shape depends on the integration type: + + GitHub / GitHub Enterprise / Bitbucket Cloud: + {"owner": ..., "name": ...} + + Azure DevOps: + {"owner": "/", "name": ""} + + Bitbucket Server: + {"repoSlug": ..., "projectKey": ...} + + GitLab: + {"owner": ..., "name": ...} (same shape as GitHub) + """ + parsed = urlparse(repo_url) + hostname = parsed.hostname + if not hostname: + raise ValueError(f"Cannot parse hostname from URL: {repo_url}") + + path_parts = [p for p in parsed.path.strip("/").split("/") if p] + + if integration_type is None: + integration_type = detect_integration_type(repo_url) + + if integration_type == "azure-repos": + return _parse_azure_url(repo_url, path_parts) + + if integration_type == "bitbucket-server": + return _parse_bitbucket_server_url(repo_url, path_parts) + + # GitHub, GitHub Enterprise, Bitbucket Cloud, GitLab all use owner/name. + if len(path_parts) < 2: + raise ValueError( + f"URL must follow the pattern " + f"https:////: {repo_url}" + ) + name = path_parts[1] + # Strip .git suffix common in clone URLs + if name.endswith(".git"): + name = name[:-4] + return { + "owner": path_parts[0], + "name": name, + } + + +def _parse_azure_url(repo_url: str, path_parts: list[str]) -> dict: + """Parse an Azure DevOps URL: ///_git/""" + if len(path_parts) < 4 or path_parts[2].lower() != "_git": + raise ValueError( + f"Azure DevOps URL must follow the pattern " + f"https://dev.azure.com///_git/: {repo_url}" + ) + return { + "owner": f"{path_parts[0]}/{path_parts[1]}", + "name": path_parts[3], + } + + +def _parse_bitbucket_server_url(repo_url: str, path_parts: list[str]) -> dict: + """Parse a Bitbucket Server URL. + + Supports two common patterns: + /scm//.git + /projects//repos/ + """ + # Clone URL style: /scm/PROJECT/repo.git + if len(path_parts) >= 3 and path_parts[0].lower() == "scm": + name = path_parts[2] + if name.endswith(".git"): + name = name[:-4] + return { + "repoSlug": name, + "projectKey": path_parts[1], + } + + # Browse URL style: /projects/PROJECT/repos/repo + if ( + len(path_parts) >= 4 + and path_parts[0].lower() == "projects" + and path_parts[2].lower() == "repos" + ): + return { + "repoSlug": path_parts[3], + "projectKey": path_parts[1], + } + + raise ValueError( + f"Bitbucket Server URL must follow one of these patterns:\n" + f" https:///scm//.git\n" + f" https:///projects//repos/\n" + f"Got: {repo_url}" + ) + + +# --------------------------------------------------------------------------- +# CSV helpers +# --------------------------------------------------------------------------- + + +def read_csv(csv_path: str) -> list[dict]: + """Read the CSV file and return a list of row dicts.""" + with open(csv_path, newline="", encoding="utf-8") as fh: + reader = csv.DictReader(fh) + rows = list(reader) + if not rows: + raise ValueError(f"CSV file is empty: {csv_path}") + return rows + + +def validate_row(row: dict, index: int) -> None: + """Validate a single CSV row. Raises ValueError on problems.""" + repo_url = (row.get("repo_url") or "").strip() + if not repo_url: + raise ValueError(f"Row {index}: 'repo_url' is required") + + org_id = (row.get("snyk_org_id") or "").strip() + org_name = (row.get("snyk_org_name") or "").strip() + if not org_id and not org_name: + raise ValueError( + f"Row {index}: at least one of 'snyk_org_id' or 'snyk_org_name' is required" + ) + + +def validate_rows(rows: list[dict]) -> None: + """Validate all rows; raises on the first error.""" + for i, row in enumerate(rows, start=1): + validate_row(row, i) + + +# --------------------------------------------------------------------------- +# Target building +# --------------------------------------------------------------------------- + + +def _org_id_for_row(row: dict, name_to_org_id: dict[str, str]) -> str: + org_id = (row.get("snyk_org_id") or "").strip() + if org_id: + return org_id + org_name = (row.get("snyk_org_name") or "").strip() + return name_to_org_id[org_name] + + +def build_target(row: dict, integration_type: str | None = None) -> dict: + """Build a single ``target`` dict from a CSV row.""" + repo_url = row["repo_url"].strip() + return parse_repo_url(repo_url, integration_type) + + +def build_import_target( + row: dict, + name_to_org_id: dict[str, str], + integration_cache: dict[tuple[str, str], str], + host_to_integration: dict[str, str] | None = None, +) -> dict: + """Build a single import-target entry for the import-projects.json file.""" + repo_url = row["repo_url"].strip() + org_id = _org_id_for_row(row, name_to_org_id) + integration_type = resolve_integration_type_for_url( + repo_url, host_to_integration or {} + ) + integration_id = integration_cache[(org_id, integration_type)] + + target = build_target(row, integration_type) + branch = (row.get("branch") or "").strip() or DEFAULT_BRANCH + target["branch"] = branch + + entry: dict = { + "orgId": org_id, + "integrationId": integration_id, + "target": target, + } + + exclusion_globs = (row.get("exclusion_globs") or "").strip() + if exclusion_globs: + entry["exclusionGlobs"] = exclusion_globs + + return entry + + +def resolve_integration_type_for_url( + repo_url: str, + host_to_integration: dict[str, str], +) -> str: + """Return the integration type for *repo_url*. + + First tries automatic detection; if that returns ``None`` (unknown host) + falls back to the user-supplied *host_to_integration* mapping. + Raises ``KeyError`` if the hostname has not been resolved. + """ + detected = detect_integration_type(repo_url) + if detected is not None: + return detected + hostname = hostname_from_url(repo_url) + return host_to_integration[hostname] + + +def build_import_json( + rows: list[dict], + name_to_org_id: dict[str, str], + integration_cache: dict[tuple[str, str], str], + host_to_integration: dict[str, str] | None = None, +) -> dict: + """Build the full import-projects.json structure.""" + h2i = host_to_integration or {} + targets = [ + build_import_target(row, name_to_org_id, integration_cache, h2i) + for row in rows + ] + return {"targets": targets} diff --git a/csv-to-snyk-api-import/csv_to_snyk_import.py b/csv-to-snyk-api-import/csv_to_snyk_import.py new file mode 100644 index 0000000..c0afb23 --- /dev/null +++ b/csv-to-snyk-api-import/csv_to_snyk_import.py @@ -0,0 +1,175 @@ +""" +CLI entry point – reads args, orchestrates the pipeline, writes JSON output, +and optionally kicks off snyk-api-import. +""" + +import json +import os +import shutil +import subprocess +import sys + +from converter import ( + KNOWN_INTEGRATION_TYPES, + build_import_json, + detect_integration_type, + hostname_from_url, + read_csv, + validate_rows, +) +from orchestrator import resolve_integration_ids, resolve_org_ids + +DEFAULT_LOG_PATH = "./logs" + + +def resolve_unknown_hosts(rows: list[dict]) -> dict[str, str]: + """Find repo URLs whose hostname can't be auto-detected and ask the user + to choose an integration type. Returns a mapping of hostname → type. + """ + unknown_hosts: dict[str, str] = {} # hostname → first example URL + for row in rows: + url = (row.get("repo_url") or "").strip() + if not url: + continue + if detect_integration_type(url) is not None: + continue + host = hostname_from_url(url) + unknown_hosts.setdefault(host, url) + + if not unknown_hosts: + return {} + + numbered = list(KNOWN_INTEGRATION_TYPES) + menu = "\n".join(f" {i + 1}) {t}" for i, t in enumerate(numbered)) + + host_to_integration: dict[str, str] = {} + for host, example_url in unknown_hosts.items(): + print(f"\nUnrecognised hostname: {host}") + print(f" Example URL: {example_url}") + print(f"Which Snyk integration type should be used?\n{menu}") + while True: + choice = input("Enter number: ").strip() + if choice.isdigit() and 1 <= int(choice) <= len(numbered): + selected = numbered[int(choice) - 1] + host_to_integration[host] = selected + print(f" → {host} mapped to '{selected}'") + break + print(f" Invalid choice. Enter a number between 1 and {len(numbered)}.") + + return host_to_integration + + +def main(argv: list[str] | None = None) -> None: + args = argv if argv is not None else sys.argv[1:] + + if not args: + print("Usage: python3 csv_to_snyk_import.py [output.json]") + sys.exit(1) + + csv_path = os.path.realpath(args[0]) + # Output file is always written to the current directory – strip any + # directory components to prevent path-traversal via CLI arguments. + output_filename = os.path.basename( + args[1] if len(args) > 1 else "import-projects.json" + ) + if not output_filename: + output_filename = "import-projects.json" + output_path = os.path.join(os.getcwd(), output_filename) + + # --- Environment --- + group_id = os.environ.get("SNYK_GROUP_ID", "") + template_org_id = os.environ.get("SNYK_TEMPLATE_ORG_ID") + log_path = os.environ.get("SNYK_LOG_PATH", DEFAULT_LOG_PATH) + + missing: list[str] = [] + if not os.environ.get("SNYK_TOKEN"): + missing.append("SNYK_TOKEN") + if missing: + for var in missing: + print(f"Error: {var} environment variable is not set.", file=sys.stderr) + sys.exit(1) + + # --- Phase 1: Read & validate CSV --- + print(f"Reading CSV: {csv_path}") + rows = read_csv(csv_path) + validate_rows(rows) + print(f" {len(rows)} rows read and validated.") + + # Check SNYK_GROUP_ID if any row needs org creation. + needs_org_creation = any( + not (row.get("snyk_org_id") or "").strip() + and (row.get("snyk_org_name") or "").strip() + for row in rows + ) + if needs_org_creation and not group_id: + print( + "Error: SNYK_GROUP_ID environment variable is required because " + "one or more rows use 'snyk_org_name' (org creation).", + file=sys.stderr, + ) + sys.exit(1) + + try: + # --- Phase 2: Resolve unknown hostnames --- + host_to_integration = resolve_unknown_hosts(rows) + + # --- Phase 3: Org resolution / creation --- + print("Resolving Snyk organisations …") + name_to_org_id = resolve_org_ids(rows, group_id, template_org_id) + if name_to_org_id: + print(f" Resolved {len(name_to_org_id)} org name(s) to IDs.") + + # --- Phase 4: Integration lookup --- + print("Looking up integration IDs …") + integration_cache = resolve_integration_ids( + rows, name_to_org_id, host_to_integration + ) + + # --- Phase 5: Build JSON --- + print("Building import-projects.json …") + import_data = build_import_json( + rows, name_to_org_id, integration_cache, host_to_integration + ) + except (RuntimeError, ValueError, KeyError) as exc: + print(f"\nError: {exc}", file=sys.stderr) + sys.exit(1) + + with open(output_path, "w", encoding="utf-8") as fh: + json.dump(import_data, fh, indent=2) + + target_count = len(import_data["targets"]) + unique_orgs = {t["orgId"] for t in import_data["targets"]} + print( + f"\nGenerated {output_path} with {target_count} target(s) " + f"across {len(unique_orgs)} org(s)." + ) + + # --- Phase 5: Prompt to run import --- + import_cmd = f"snyk-api-import import --file={output_path}" + + if shutil.which("snyk-api-import") is None: + print("\nsnyk-api-import is not installed or not on your PATH.") + print("Install it from: https://github.com/snyk/snyk-api-import/releases") + print("\nOnce installed, run:") + print(f" {import_cmd}") + return + + answer = input("\nRun snyk-api-import now? [y/N] ").strip().lower() + if answer in ("y", "yes"): + env = os.environ.copy() + env.setdefault("SNYK_LOG_PATH", log_path) + # snyk-api-import expects SNYK_API to include /v1 at the end. + snyk_api = env.get("SNYK_API", "https://api.snyk.io").rstrip("/") + if not snyk_api.endswith("/v1"): + snyk_api += "/v1" + env["SNYK_API"] = snyk_api + cmd = ["snyk-api-import", "import", f"--file={output_path}"] + print(f"Running: {' '.join(cmd)}") + subprocess.run(cmd, env=env, check=False) + else: + print("Skipped. You can run the import later with:") + print(f" {import_cmd}") + + +if __name__ == "__main__": + main() diff --git a/csv-to-snyk-api-import/orchestrator.py b/csv-to-snyk-api-import/orchestrator.py new file mode 100644 index 0000000..6ba13cd --- /dev/null +++ b/csv-to-snyk-api-import/orchestrator.py @@ -0,0 +1,103 @@ +""" +Orchestration helpers that combine API calls with the converter logic: +org resolution / creation and integration-ID lookup. +""" + +from converter import ( + _org_id_for_row, + resolve_integration_type_for_url, +) +from snyk_api import create_org, get_integration_id, list_orgs + + +def resolve_org_ids( + rows: list[dict], + group_id: str, + template_org_id: str | None, +) -> dict[str, str]: + """Ensure every row has a usable org ID. + + Returns a mapping of ``snyk_org_name -> snyk_org_id`` for any names that + had to be resolved or created. Rows that already carry an org ID are + left untouched (and not included in the returned mapping). + """ + # Collect unique names that need resolution. + names_to_resolve: set[str] = set() + for row in rows: + org_id = (row.get("snyk_org_id") or "").strip() + org_name = (row.get("snyk_org_name") or "").strip() + if not org_id and org_name: + names_to_resolve.add(org_name) + + if not names_to_resolve: + return {} + + # Fetch existing orgs once. + existing_orgs = list_orgs() + name_to_id: dict[str, str] = {} + for org in existing_orgs: + org_name_existing = org.get("name", "") + if org_name_existing in names_to_resolve: + name_to_id[org_name_existing] = org["id"] + + # Create any that don't exist yet. + for name in sorted(names_to_resolve): + if name not in name_to_id: + print(f" Creating Snyk org '{name}' …") + try: + new_org = create_org(name, group_id, template_org_id) + except (RuntimeError, ValueError) as exc: + raise RuntimeError( + f"Failed to create org '{name}': {exc}" + ) from exc + name_to_id[name] = new_org["id"] + print(f" Created org '{name}' → {new_org['id']}") + + return name_to_id + + +def resolve_integration_ids( + rows: list[dict], + name_to_org_id: dict[str, str], + host_to_integration: dict[str, str] | None = None, +) -> dict[tuple[str, str], str]: + """For every unique (org_id, integration_type) pair, look up the + integration ID. Returns a mapping of (org_id, type) → integration_id. + + Collects all missing integrations and raises a single error listing + every (org, type) pair that needs attention. + """ + h2i = host_to_integration or {} + pairs: set[tuple[str, str]] = set() + for row in rows: + org_id = _org_id_for_row(row, name_to_org_id) + integration_type = resolve_integration_type_for_url( + row["repo_url"].strip(), h2i + ) + pairs.add((org_id, integration_type)) + + result: dict[tuple[str, str], str] = {} + missing: list[tuple[str, str]] = [] + for org_id, integration_type in sorted(pairs): + print(f" Looking up integration '{integration_type}' for org {org_id} …") + int_id = get_integration_id(org_id, integration_type) + if int_id is None: + print(f" ✗ Not found") + missing.append((org_id, integration_type)) + else: + result[(org_id, integration_type)] = int_id + print(f" Found integration {int_id}") + + if missing: + lines = "\n".join( + f" • org {org_id} → {itype}" for org_id, itype in missing + ) + raise RuntimeError( + f"The following integration(s) are not configured:\n{lines}\n\n" + f"To fix this, go to each org's Settings → Integrations in the " + f"Snyk UI and enable the required integration.\n" + f"If you are using SNYK_TEMPLATE_ORG_ID to create orgs, make sure " + f"the template org has all needed integrations configured." + ) + + return result diff --git a/csv-to-snyk-api-import/requirements.txt b/csv-to-snyk-api-import/requirements.txt new file mode 100644 index 0000000..51d6286 --- /dev/null +++ b/csv-to-snyk-api-import/requirements.txt @@ -0,0 +1,2 @@ +requests>=2.31.0,<3.0.0 +pytest>=8.0.0,<9.0.0 diff --git a/csv-to-snyk-api-import/sample.csv b/csv-to-snyk-api-import/sample.csv new file mode 100644 index 0000000..0f41792 --- /dev/null +++ b/csv-to-snyk-api-import/sample.csv @@ -0,0 +1,7 @@ +repo_url,snyk_org_id,snyk_org_name,branch,exclusion_globs +https://github.com/my-org/my-app,org-abc-123,,main, +https://github.com/my-org/my-lib,org-abc-123,,develop,"fixtures, tests" +https://dev.azure.com/my-org/my-project/_git/my-service,,My New Org,main, +https://github.com/my-org/another-repo,,My New Org,, +https://bitbucket.org/my-workspace/my-bb-repo,org-abc-123,,main, +https://bb.mycompany.com/scm/PROJ/my-server-repo.git,,My New Org,main, diff --git a/csv-to-snyk-api-import/snyk_api.py b/csv-to-snyk-api-import/snyk_api.py new file mode 100644 index 0000000..aa092c7 --- /dev/null +++ b/csv-to-snyk-api-import/snyk_api.py @@ -0,0 +1,94 @@ +""" +Thin wrappers around the Snyk v1 REST API. +""" + +import os + +import requests + +DEFAULT_API_BASE = "https://api.snyk.io" + + +def _api_base() -> str: + base = os.environ.get("SNYK_API", DEFAULT_API_BASE).rstrip("/") + # Users may set SNYK_API with a trailing /v1 (required by snyk-api-import). + # Strip it here so our own URL construction doesn't double up to /v1/v1. + if base.endswith("/v1"): + base = base[:-3] + return base + + +def _auth_header() -> dict: + token = os.environ.get("SNYK_TOKEN") + if not token: + raise EnvironmentError("SNYK_TOKEN environment variable is not set") + return {"Authorization": f"token {token}"} + + +def list_orgs() -> list[dict]: + """Return all Snyk organisations the token has access to.""" + url = f"{_api_base()}/v1/orgs" + resp = requests.get(url, headers=_auth_header(), timeout=30) + if not resp.ok: + detail = resp.text + try: + detail = resp.json() + except Exception: + pass + raise RuntimeError( + f"Failed to list organisations (HTTP {resp.status_code}): {detail}\n" + f"Check that SNYK_TOKEN is valid and has the correct permissions." + ) + return resp.json().get("orgs", []) + + +def create_org(name: str, group_id: str, source_org_id: str | None = None) -> dict: + """Create a new Snyk organisation, optionally cloned from a template.""" + if not group_id: + raise ValueError( + "SNYK_GROUP_ID is required to create organisations. " + "Set it via the SNYK_GROUP_ID environment variable." + ) + url = f"{_api_base()}/v1/org" + body: dict = {"name": name, "groupId": group_id} + if source_org_id: + body["sourceOrgId"] = source_org_id + resp = requests.post(url, json=body, headers=_auth_header(), timeout=30) + if not resp.ok: + detail = resp.text + try: + detail = resp.json() + except Exception: + pass + raise RuntimeError( + f"Failed to create org '{name}' (HTTP {resp.status_code}): {detail}" + ) + return resp.json() + + +def get_integration_id(org_id: str, integration_type: str) -> str | None: + """Look up the integration ID for an org + type. + + Returns ``None`` when the integration is not configured (HTTP 404) + so the caller can collect all failures and report them together. + Raises on unexpected errors (auth, server, etc.). + """ + url = f"{_api_base()}/v1/org/{org_id}/integrations/{integration_type}" + resp = requests.get(url, headers=_auth_header(), timeout=30) + if resp.status_code == 404: + return None + if not resp.ok: + detail = resp.text + try: + detail = resp.json() + except Exception: + pass + raise RuntimeError( + f"Error looking up '{integration_type}' integration for org {org_id} " + f"(HTTP {resp.status_code}): {detail}" + ) + data = resp.json() + int_id = data.get("id") + if not int_id: + return None + return int_id diff --git a/csv-to-snyk-api-import/test_converter.py b/csv-to-snyk-api-import/test_converter.py new file mode 100644 index 0000000..e083149 --- /dev/null +++ b/csv-to-snyk-api-import/test_converter.py @@ -0,0 +1,557 @@ +""" +Tests for the CSV → import-projects.json transformation logic. + +Focuses on pure functions in converter.py. +All Snyk API calls are mocked so tests run entirely offline. +""" + +import csv +import json +import os +import tempfile +from unittest.mock import patch + +import pytest + +from converter import ( + DEFAULT_BRANCH, + KNOWN_INTEGRATION_TYPES, + build_import_json, + build_import_target, + build_target, + detect_integration_type, + hostname_from_url, + parse_repo_url, + read_csv, + resolve_integration_type_for_url, + validate_row, + validate_rows, +) +from orchestrator import ( + resolve_integration_ids, + resolve_org_ids, +) + + +# ====================================================================== +# Helpers +# ====================================================================== + +def _write_csv(path: str, rows: list[dict]) -> None: + """Write a list of dicts as a CSV file with a header row.""" + if not rows: + # Write an empty file (no header, no data) + with open(path, "w", newline="", encoding="utf-8") as fh: + pass + return + fieldnames = list(rows[0].keys()) + with open(path, "w", newline="", encoding="utf-8") as fh: + writer = csv.DictWriter(fh, fieldnames=fieldnames) + writer.writeheader() + writer.writerows(rows) + + +def _make_row(**overrides) -> dict: + """Return a minimal valid CSV row dict, with optional overrides.""" + base = { + "repo_url": "https://github.com/my-org/my-repo", + "snyk_org_id": "org-id-123", + "snyk_org_name": "", + "branch": "", + "exclusion_globs": "", + } + base.update(overrides) + return base + + +# ====================================================================== +# URL Parsing Tests +# ====================================================================== + + +class TestParseRepoUrl: + def test_github_standard(self): + target = parse_repo_url("https://github.com/my-org/my-repo") + assert target == {"owner": "my-org", "name": "my-repo"} + + def test_github_with_trailing_slash(self): + target = parse_repo_url("https://github.com/my-org/my-repo/") + assert target == {"owner": "my-org", "name": "my-repo"} + + def test_github_with_extra_path_segments(self): + target = parse_repo_url("https://github.com/my-org/my-repo/tree/main") + assert target == {"owner": "my-org", "name": "my-repo"} + + def test_azure_devops_standard(self): + target = parse_repo_url( + "https://dev.azure.com/my-org/my-project/_git/my-repo" + ) + assert target == {"owner": "my-org/my-project", "name": "my-repo"} + + def test_azure_devops_with_trailing_slash(self): + target = parse_repo_url( + "https://dev.azure.com/my-org/my-project/_git/my-repo/" + ) + assert target == {"owner": "my-org/my-project", "name": "my-repo"} + + def test_github_enterprise(self): + target = parse_repo_url( + "https://github.mycompany.com/eng/service-api", "github-enterprise" + ) + assert target == {"owner": "eng", "name": "service-api"} + + def test_bitbucket_cloud(self): + target = parse_repo_url("https://bitbucket.org/workspace/my-repo") + assert target == {"owner": "workspace", "name": "my-repo"} + + def test_bitbucket_server_scm_url(self): + target = parse_repo_url( + "https://bb.company.com/scm/PROJ/my-repo.git", "bitbucket-server" + ) + assert target == {"repoSlug": "my-repo", "projectKey": "PROJ"} + + def test_bitbucket_server_projects_url(self): + target = parse_repo_url( + "https://bb.company.com/projects/PROJ/repos/my-repo", "bitbucket-server" + ) + assert target == {"repoSlug": "my-repo", "projectKey": "PROJ"} + + def test_bitbucket_server_invalid_url(self): + with pytest.raises(ValueError, match="Bitbucket Server URL"): + parse_repo_url( + "https://bb.company.com/unknown/path", "bitbucket-server" + ) + + def test_git_suffix_stripped(self): + target = parse_repo_url( + "https://github.com/my-org/my-repo.git", "github" + ) + assert target == {"owner": "my-org", "name": "my-repo"} + + def test_invalid_url_no_host(self): + with pytest.raises(ValueError, match="Cannot parse hostname"): + parse_repo_url("not-a-url") + + def test_github_missing_repo(self): + with pytest.raises(ValueError, match="URL must follow"): + parse_repo_url("https://github.com/only-org") + + def test_azure_missing_git_segment(self): + with pytest.raises(ValueError, match="Azure DevOps URL must follow"): + parse_repo_url("https://dev.azure.com/my-org/my-project/bad/my-repo") + + def test_azure_too_short(self): + with pytest.raises(ValueError, match="Azure DevOps URL must follow"): + parse_repo_url("https://dev.azure.com/my-org") + + +# ====================================================================== +# Integration Type Detection Tests +# ====================================================================== + + +class TestDetectIntegrationType: + def test_github_dot_com(self): + assert detect_integration_type("https://github.com/o/r") == "github" + + def test_azure_devops(self): + url = "https://dev.azure.com/o/p/_git/r" + assert detect_integration_type(url) == "azure-repos" + + def test_bitbucket_cloud(self): + assert detect_integration_type("https://bitbucket.org/ws/repo") == "bitbucket-cloud" + + def test_unknown_host_returns_none(self): + url = "https://github.example.com/o/r" + assert detect_integration_type(url) is None + + def test_unknown_host_generic(self): + url = "https://my-git-server.internal/o/r" + assert detect_integration_type(url) is None + + def test_no_hostname_raises(self): + with pytest.raises(ValueError, match="Cannot parse hostname"): + detect_integration_type("://missing") + + +# ====================================================================== +# CSV Validation Tests +# ====================================================================== + + +class TestValidateRow: + def test_valid_with_org_id(self): + row = _make_row(snyk_org_id="abc", snyk_org_name="") + validate_row(row, 1) # should not raise + + def test_valid_with_org_name(self): + row = _make_row(snyk_org_id="", snyk_org_name="My Org") + validate_row(row, 1) + + def test_valid_with_both(self): + row = _make_row(snyk_org_id="abc", snyk_org_name="My Org") + validate_row(row, 1) + + def test_missing_both_org_fields(self): + row = _make_row(snyk_org_id="", snyk_org_name="") + with pytest.raises(ValueError, match="at least one of"): + validate_row(row, 1) + + def test_missing_repo_url(self): + row = _make_row(repo_url="") + with pytest.raises(ValueError, match="repo_url"): + validate_row(row, 1) + + +class TestValidateRows: + def test_all_valid(self): + rows = [_make_row(), _make_row(repo_url="https://dev.azure.com/o/p/_git/r")] + validate_rows(rows) + + def test_empty_csv(self): + with tempfile.NamedTemporaryFile( + mode="w", suffix=".csv", delete=False, newline="" + ) as fh: + path = fh.name + try: + with pytest.raises(ValueError, match="empty"): + read_csv(path) + finally: + os.unlink(path) + + def test_malformed_csv_missing_required_column(self): + """A CSV that has no repo_url column at all.""" + with tempfile.NamedTemporaryFile( + mode="w", suffix=".csv", delete=False, newline="", encoding="utf-8" + ) as fh: + fh.write("some_col,another_col\nval1,val2\n") + path = fh.name + try: + rows = read_csv(path) + with pytest.raises(ValueError, match="repo_url"): + validate_rows(rows) + finally: + os.unlink(path) + + +class TestReadCsv: + def test_reads_valid_file(self): + with tempfile.NamedTemporaryFile( + mode="w", suffix=".csv", delete=False, newline="", encoding="utf-8" + ) as fh: + _write_csv(fh.name, [_make_row()]) + path = fh.name + try: + rows = read_csv(path) + assert len(rows) == 1 + assert rows[0]["repo_url"] == "https://github.com/my-org/my-repo" + finally: + os.unlink(path) + + def test_empty_file_raises(self): + with tempfile.NamedTemporaryFile( + mode="w", suffix=".csv", delete=False, newline="" + ) as fh: + path = fh.name + try: + with pytest.raises(ValueError, match="empty"): + read_csv(path) + finally: + os.unlink(path) + + +# ====================================================================== +# Target JSON Generation Tests +# ====================================================================== + +# Shared fixtures for target-building tests. +_NAME_TO_ORG_ID: dict[str, str] = {"My Org": "resolved-org-id"} +_INTEGRATION_CACHE: dict[tuple[str, str], str] = { + ("org-id-123", "github"): "int-gh-001", + ("org-id-123", "azure-repos"): "int-az-001", + ("org-id-123", "bitbucket-cloud"): "int-bb-001", + ("org-id-123", "bitbucket-server"): "int-bbs-001", + ("resolved-org-id", "github"): "int-gh-002", + ("resolved-org-id", "azure-repos"): "int-az-002", +} +_HOST_TO_INTEGRATION: dict[str, str] = { + "bb.company.com": "bitbucket-server", +} + + +class TestBuildTarget: + def test_github(self): + row = _make_row() + assert build_target(row) == {"owner": "my-org", "name": "my-repo"} + + def test_azure(self): + row = _make_row(repo_url="https://dev.azure.com/o/p/_git/r") + assert build_target(row) == {"owner": "o/p", "name": "r"} + + def test_bitbucket_cloud(self): + row = _make_row(repo_url="https://bitbucket.org/ws/repo") + assert build_target(row) == {"owner": "ws", "name": "repo"} + + def test_bitbucket_server(self): + row = _make_row(repo_url="https://bb.company.com/scm/PROJ/repo.git") + assert build_target(row, "bitbucket-server") == { + "repoSlug": "repo", + "projectKey": "PROJ", + } + + +class TestBuildImportTarget: + def test_github_defaults(self): + row = _make_row() + entry = build_import_target(row, _NAME_TO_ORG_ID, _INTEGRATION_CACHE) + assert entry == { + "orgId": "org-id-123", + "integrationId": "int-gh-001", + "target": {"owner": "my-org", "name": "my-repo", "branch": "main"}, + } + + def test_custom_branch(self): + row = _make_row(branch="develop") + entry = build_import_target(row, _NAME_TO_ORG_ID, _INTEGRATION_CACHE) + assert entry["target"]["branch"] == "develop" + + def test_default_branch_when_blank(self): + row = _make_row(branch="") + entry = build_import_target(row, _NAME_TO_ORG_ID, _INTEGRATION_CACHE) + assert entry["target"]["branch"] == DEFAULT_BRANCH + + def test_exclusion_globs_included(self): + row = _make_row(exclusion_globs="fixtures, tests") + entry = build_import_target(row, _NAME_TO_ORG_ID, _INTEGRATION_CACHE) + assert entry["exclusionGlobs"] == "fixtures, tests" + + def test_exclusion_globs_omitted_when_blank(self): + row = _make_row(exclusion_globs="") + entry = build_import_target(row, _NAME_TO_ORG_ID, _INTEGRATION_CACHE) + assert "exclusionGlobs" not in entry + + def test_azure_target(self): + row = _make_row(repo_url="https://dev.azure.com/o/p/_git/r") + entry = build_import_target(row, _NAME_TO_ORG_ID, _INTEGRATION_CACHE) + assert entry["orgId"] == "org-id-123" + assert entry["integrationId"] == "int-az-001" + assert entry["target"] == {"owner": "o/p", "name": "r", "branch": "main"} + + def test_org_resolved_by_name(self): + row = _make_row(snyk_org_id="", snyk_org_name="My Org") + entry = build_import_target(row, _NAME_TO_ORG_ID, _INTEGRATION_CACHE) + assert entry["orgId"] == "resolved-org-id" + assert entry["integrationId"] == "int-gh-002" + + def test_bitbucket_cloud_target(self): + row = _make_row(repo_url="https://bitbucket.org/ws/repo") + entry = build_import_target(row, _NAME_TO_ORG_ID, _INTEGRATION_CACHE) + assert entry["orgId"] == "org-id-123" + assert entry["integrationId"] == "int-bb-001" + assert entry["target"] == { + "owner": "ws", + "name": "repo", + "branch": "main", + } + + def test_bitbucket_server_target(self): + row = _make_row(repo_url="https://bb.company.com/scm/PROJ/repo.git") + entry = build_import_target( + row, _NAME_TO_ORG_ID, _INTEGRATION_CACHE, _HOST_TO_INTEGRATION + ) + assert entry["orgId"] == "org-id-123" + assert entry["integrationId"] == "int-bbs-001" + assert entry["target"]["repoSlug"] == "repo" + assert entry["target"]["projectKey"] == "PROJ" + assert entry["target"]["branch"] == "main" + + +class TestBuildImportJson: + def test_single_github_row(self): + rows = [_make_row()] + result = build_import_json(rows, _NAME_TO_ORG_ID, _INTEGRATION_CACHE) + assert "targets" in result + assert len(result["targets"]) == 1 + assert result["targets"][0]["orgId"] == "org-id-123" + + def test_single_azure_row(self): + rows = [_make_row(repo_url="https://dev.azure.com/o/p/_git/r")] + result = build_import_json(rows, _NAME_TO_ORG_ID, _INTEGRATION_CACHE) + assert len(result["targets"]) == 1 + assert result["targets"][0]["target"]["owner"] == "o/p" + + def test_mixed_sources_multiple_orgs(self): + rows = [ + _make_row(), + _make_row( + repo_url="https://dev.azure.com/o/p/_git/r", + snyk_org_id="", + snyk_org_name="My Org", + ), + ] + result = build_import_json(rows, _NAME_TO_ORG_ID, _INTEGRATION_CACHE) + assert len(result["targets"]) == 2 + org_ids = {t["orgId"] for t in result["targets"]} + assert org_ids == {"org-id-123", "resolved-org-id"} + + def test_multiple_rows_same_org(self): + rows = [ + _make_row(repo_url="https://github.com/o/repo1"), + _make_row(repo_url="https://github.com/o/repo2"), + ] + result = build_import_json(rows, _NAME_TO_ORG_ID, _INTEGRATION_CACHE) + assert len(result["targets"]) == 2 + assert all(t["orgId"] == "org-id-123" for t in result["targets"]) + + def test_large_csv(self): + """100+ rows should all appear in the output.""" + rows = [ + _make_row(repo_url=f"https://github.com/org/repo-{i}") + for i in range(150) + ] + result = build_import_json(rows, _NAME_TO_ORG_ID, _INTEGRATION_CACHE) + assert len(result["targets"]) == 150 + + def test_output_is_valid_json(self): + rows = [_make_row()] + result = build_import_json(rows, _NAME_TO_ORG_ID, _INTEGRATION_CACHE) + # Roundtrip through JSON to ensure serialisability. + dumped = json.dumps(result) + loaded = json.loads(dumped) + assert loaded == result + + +# ====================================================================== +# End-to-End Transformation Tests (mocked API calls) +# ====================================================================== + + +class TestEndToEnd: + """Full CSV file → import-projects.json, with API calls mocked.""" + + @patch("orchestrator.get_integration_id") + @patch("orchestrator.list_orgs") + @patch("orchestrator.create_org") + def test_full_pipeline(self, mock_create, mock_list, mock_get_int): + # Simulate: "Team Alpha" already exists, "Team Beta" needs creation. + mock_list.return_value = [ + {"name": "Team Alpha", "id": "alpha-id"}, + ] + mock_create.return_value = {"id": "beta-id", "name": "Team Beta"} + + mock_get_int.side_effect = lambda org_id, itype: { + ("alpha-id", "github"): "int-alpha-gh", + ("beta-id", "azure-repos"): "int-beta-az", + ("existing-org", "github"): "int-existing-gh", + }[(org_id, itype)] + + rows = [ + { + "repo_url": "https://github.com/acme/app", + "snyk_org_id": "", + "snyk_org_name": "Team Alpha", + "branch": "main", + "exclusion_globs": "", + }, + { + "repo_url": "https://dev.azure.com/acme/proj/_git/svc", + "snyk_org_id": "", + "snyk_org_name": "Team Beta", + "branch": "develop", + "exclusion_globs": "tests, fixtures", + }, + { + "repo_url": "https://github.com/acme/lib", + "snyk_org_id": "existing-org", + "snyk_org_name": "", + "branch": "", + "exclusion_globs": "", + }, + ] + + name_to_id = resolve_org_ids(rows, "group-1", "template-1") + assert name_to_id == {"Team Alpha": "alpha-id", "Team Beta": "beta-id"} + mock_create.assert_called_once_with("Team Beta", "group-1", "template-1") + + int_cache = resolve_integration_ids(rows, name_to_id) + result = build_import_json(rows, name_to_id, int_cache) + + assert len(result["targets"]) == 3 + + # Target 1: GitHub → Team Alpha + t0 = result["targets"][0] + assert t0["orgId"] == "alpha-id" + assert t0["integrationId"] == "int-alpha-gh" + assert t0["target"] == {"owner": "acme", "name": "app", "branch": "main"} + assert "exclusionGlobs" not in t0 + + # Target 2: Azure → Team Beta + t1 = result["targets"][1] + assert t1["orgId"] == "beta-id" + assert t1["integrationId"] == "int-beta-az" + assert t1["target"] == { + "owner": "acme/proj", + "name": "svc", + "branch": "develop", + } + assert t1["exclusionGlobs"] == "tests, fixtures" + + # Target 3: GitHub → existing org + t2 = result["targets"][2] + assert t2["orgId"] == "existing-org" + assert t2["integrationId"] == "int-existing-gh" + assert t2["target"]["branch"] == DEFAULT_BRANCH + + @patch("orchestrator.get_integration_id") + @patch("orchestrator.list_orgs") + def test_csv_file_to_json_file(self, mock_list, mock_get_int): + """Read from a real CSV file, produce a real JSON file.""" + mock_list.return_value = [] + mock_get_int.return_value = "int-id-999" + + row_data = [ + { + "repo_url": "https://github.com/org-a/repo-1", + "snyk_org_id": "org-aaa", + "snyk_org_name": "", + "branch": "release", + "exclusion_globs": "node_modules", + }, + { + "repo_url": "https://github.com/org-a/repo-2", + "snyk_org_id": "org-aaa", + "snyk_org_name": "", + "branch": "", + "exclusion_globs": "", + }, + ] + + with tempfile.NamedTemporaryFile( + mode="w", suffix=".csv", delete=False, newline="", encoding="utf-8" + ) as csv_fh: + _write_csv(csv_fh.name, row_data) + csv_path = csv_fh.name + + json_path = csv_path.replace(".csv", ".json") + + try: + rows = read_csv(csv_path) + validate_rows(rows) + name_to_id: dict[str, str] = {} + int_cache = resolve_integration_ids(rows, name_to_id) + result = build_import_json(rows, name_to_id, int_cache) + + with open(json_path, "w", encoding="utf-8") as jf: + json.dump(result, jf, indent=2) + + with open(json_path, encoding="utf-8") as jf: + loaded = json.load(jf) + + assert loaded["targets"][0]["target"]["branch"] == "release" + assert loaded["targets"][0]["exclusionGlobs"] == "node_modules" + assert loaded["targets"][1]["target"]["branch"] == DEFAULT_BRANCH + assert "exclusionGlobs" not in loaded["targets"][1] + finally: + os.unlink(csv_path) + if os.path.exists(json_path): + os.unlink(json_path)