diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..9e1a3dd --- /dev/null +++ b/.gitignore @@ -0,0 +1,4 @@ +.env +sync_mapping.json +__pycache__/ +*.pyc diff --git a/jira_jotty_sync.py b/jira_jotty_sync.py new file mode 100644 index 0000000..8823cc0 --- /dev/null +++ b/jira_jotty_sync.py @@ -0,0 +1,629 @@ +#!/usr/bin/env python3 +""" +Jira ↔ Jotty Bidirectional Sync Script + +Syncs Jira stories to Jotty checklists and keeps status in sync both ways: + - Jira → Jotty: Every story becomes a checklist item (task type) + - Jotty → Jira: Checking/unchecking an item updates the mapped Jira issue status + +Usage: + python jira_jotty_sync.py --direction=jira-to-jotty # Pull stories into Jotty + python jira_jotty_sync.py --direction=jotty-to-jira # Push Jotty check changes to Jira + python jira_jotty_sync.py --direction=both # Full two-way sync + +Configuration is stored in sync_config.json (see template below). +""" + +import argparse +import json +import logging +import os +import sys +import time +from dataclasses import dataclass, field +from typing import Optional + +from dotenv import load_dotenv +import requests +from jira import JIRA +from jira.exceptions import JIRAError as JiraError + +# --------------------------------------------------------------------------- +# Load .env file from the same directory as this script (if it exists) +# --------------------------------------------------------------------------- +_env_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), ".env") +load_dotenv(_env_path) + +# --------------------------------------------------------------------------- +# Logging +# --------------------------------------------------------------------------- +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s [%(levelname)s] %(message)s", +) +log = logging.getLogger(__name__) + +# --------------------------------------------------------------------------- +# Jira ↔ Jotty status mapping +# --------------------------------------------------------------------------- +# Maps Jira workflow statuses → Jira Apollo (Jotty Kanban) statuses. +# Based on your Jira Apollo board: To Do | In Progress | Completed | Paused +JIRA_TO_JOTTY_STATUS: dict[str, str] = { + # To Do column — items waiting to be worked on + "NEW": "todo", + "WAITING": "todo", + "READY": "todo", + "TO DO": "todo", + "IN REFINEMENT": "todo", + + # In Progress column — active work in flight + "IN BUSINESS DESIGN": "in_progress", + "PREPARATION": "in_progress", + "IN PROGRESS": "in_progress", + "PO REVIEW": "in_progress", + "READY FOR QA": "in_progress", + "TEST": "in_progress", + "CONFIGURATION INTEGRATION": "in_progress", + + # Completed column — finished work + "DONE": "completed", + "DEPLOYED": "completed", + "LIVE": "completed", + "TEST COMPLETED": "completed", + + # Paused column — blocked, rejected, or on hold + "IN QA": "paused", # waiting for QA to pick up + "REJECTED": "paused", +} + +# Maps Jira Apollo (Jotty Kanban) statuses → Jira workflow statuses. +# When you move an item in Jira Apollo, this determines the target Jira status. +JOTTY_TO_JIRA_STATUS: dict[str, str] = { + "todo": "TO DO", + "in_progress": "IN PROGRESS", + "completed": "DONE", + "paused": "WAITING", # paused → waiting (can resume) +} + +# JQL query used to fetch stories from Jira. Adjust as needed. +DEFAULT_JIRA_QUERY = ( + 'project = "${PROJECT_KEY}" AND issuetype = Story' + ' AND status NOT IN ("DONE", "DEPLOYED", "LIVE") ORDER BY created DESC' +) + + +# --------------------------------------------------------------------------- +# Data classes +# --------------------------------------------------------------------------- +@dataclass +class SyncConfig: + """All configuration needed for the sync.""" + + # Jira settings (self-hosted uses token_auth with PAT) + jira_base_url: str = "" # e.g. https://jira.your-domain.com + jira_api_token: str = "" # Personal Access Token (PAT) or API token + project_key: str = "" # e.g. PROJ + jql_query: str = DEFAULT_JIRA_QUERY + + # Jotty settings + jotty_base_url: str = "" # e.g. https://jotty.your-domain.com + jotty_api_key: str = "" + checklist_title: str = "Jira Apollo" # name of the Kanban board in Jotty to sync against + checklist_category: str = "Work" + + # Persistence + mapping_file: str = "sync_mapping.json" # stores Jira→Jotty item index map + + +@dataclass +class SyncMapping: + """Maps a Jira issue key → (checklist UUID, item_index).""" + + mappings: dict[str, dict] = field(default_factory=dict) + # Example entry: + # { "PROJ-42": {"checklist_id": "...", "item_index": 0} } + + def save(self, path: str): + with open(path, "w") as f: + json.dump(self.mappings, f, indent=2) + + @classmethod + def load(cls, path: str) -> "SyncMapping": + instance = cls() + if os.path.exists(path): + try: + with open(path) as f: + instance.mappings = json.load(f) + except (json.JSONDecodeError, IOError): + log.warning("Could not load mapping file; starting fresh.") + return instance + + +# --------------------------------------------------------------------------- +# Jira API client — using the official jira library +# --------------------------------------------------------------------------- +class JiraClient: + """Jira client wrapping the official jira library for self-hosted/Cloud.""" + + def __init__(self, base_url: str, api_token: str): + options = {"server": base_url.rstrip("/")} + # For self-hosted (Server/Data Center), use token_auth + # For Cloud, use basic_auth=("email", "api_token") + self.jira = JIRA(options=options, token_auth=api_token) + + def search_stories(self, jql: str) -> list[dict]: + """Run a JQL search and return every matching issue as a dict.""" + log.info("Jira search JQL=%s", jql[:80]) + issues = self.jira.search_issues(jql, maxResults=100) + results = [self._issue_to_dict(issue) for issue in issues] + log.info("Jira search returned %d issues total.", len(results)) + return results + + def get_issue(self, issue_key: str) -> dict: + """Fetch a single issue by key.""" + try: + issue = self.jira.issue(issue_key) + return self._issue_to_dict(issue) + except JiraError as exc: + log.error("Failed to fetch %s: %s", issue_key, exc) + raise + + def update_status(self, issue_key: str, status_name: str) -> None: + """Update the status of an issue using transitions.""" + try: + issue = self.jira.issue(issue_key) + # Find available transitions + transitions = self.jira.transitions(issue) + target_lower = status_name.lower() + + chosen = None + for t in transitions: + to_status = t["name"].lower() if isinstance(t, dict) else t["to"]["name"].lower() + if to_status == target_lower: + chosen = t["id"] if isinstance(t, dict) else t["id"] + break + + if not chosen: + log.warning("No transition found for status '%s' on %s. Trying direct update.", status_name, issue_key) + # Fallback: try direct field update + issue.update(fields={"status": {"name": status_name}}) + log.info("Updated %s → status '%s' (direct)", issue_key, status_name) + return + + self.jira.transition_issue(issue, chosen) + log.info("Updated %s → status '%s' (via transition %s)", issue_key, status_name, chosen) + except JiraError as exc: + log.error("Failed to update status for %s: %s", issue_key, exc) + raise + + @staticmethod + def _issue_to_dict(issue) -> dict: + """Convert a jira.Issue object to a plain dict.""" + return { + "key": issue.key, + "fields": { + "summary": getattr(issue.fields, "summary", ""), + "status": {"name": getattr(getattr(issue.fields, "status", None), "name", "")}, + "description": getattr(issue.fields, "description", ""), + }, + } + + +# --------------------------------------------------------------------------- +# Jotty API client +# --------------------------------------------------------------------------- +class JottyClient: + def __init__(self, base_url: str, api_key: str): + self.base = base_url.rstrip("/") + self.headers = {"x-api-key": api_key, "Content-Type": "application/json"} + + def _get(self, endpoint: str) -> dict: + url = f"{self.base}{endpoint}" if endpoint.startswith("/") else f"{self.base}/{endpoint}" + log.debug("GET %s", url) + r = requests.get(url, headers=self.headers, timeout=30) + r.raise_for_status() + return r.json() + + def _post(self, endpoint: str, body: dict) -> dict: + url = f"{self.base}/{endpoint.lstrip('/')}" if not endpoint.startswith("http") else endpoint + log.debug("POST %s body=%s", url, json.dumps(body)[:200]) + r = requests.post(url, headers=self.headers, json=body, timeout=30) + r.raise_for_status() + return r.json() + + def _put(self, endpoint: str, body: dict | None = None) -> dict: + url = f"{self.base}/{endpoint.lstrip('/')}" if not endpoint.startswith("http") else endpoint + log.debug("PUT %s body=%s", url, json.dumps(body)[:200] if body else "None") + r = requests.put(url, headers=self.headers, json=body, timeout=30) + r.raise_for_status() + return r.json() + + def _patch(self, endpoint: str, body: dict) -> dict: + url = f"{self.base}/{endpoint.lstrip('/')}" if not endpoint.startswith("http") else endpoint + log.debug("PATCH %s body=%s", url, json.dumps(body)[:200]) + r = requests.patch(url, headers=self.headers, json=body, timeout=30) + r.raise_for_status() + return r.json() + + def _delete(self, endpoint: str) -> dict: + url = f"{self.base}/{endpoint.lstrip('/')}" if not endpoint.startswith("http") else endpoint + log.debug("DELETE %s", url) + r = requests.delete(url, headers=self.headers, timeout=30) + r.raise_for_status() + return r.json() + + # -- checklists ------------------------------------------------------- + def get_checklists(self) -> list[dict]: + data = self._get("/api/checklists") + return data.get("checklists", []) + + def find_or_create_task_list(self, title: str, category: str) -> dict: + """Find an existing task checklist by title or create a new one.""" + all_checklists = self.get_checklists() + # Match by title; accept "task" or "kanban" type (Jotty uses kanban for boards like this) + valid_types = {"task", "kanban"} + existing = [cl for cl in all_checklists if cl["title"] == title and cl.get("type") in valid_types] + if existing: + log.info("Found existing task list '%s' (id=%s)", title, existing[0]["id"]) + return existing[0] + + log.info("Creating new task list '%s'", title) + body = { + "title": title, + "category": category, + "type": "task", + "statuses": [ + {"id": "todo", "label": "To Do", "order": 0}, + {"id": "in_progress", "label": "In Progress", "order": 1}, + {"id": "completed", "label": "Completed", "order": 2}, + ], + } + resp = self._post("/api/tasks", body) + return resp.get("data", {}) + + # -- task items ------------------------------------------------------- + def get_task(self, task_id: str) -> dict: + """Get a task (checklist) by ID. + + The API returns the task object directly at the top level. + """ + data = self._get(f"/api/tasks/{task_id}") + # Handle both wrapped and unwrapped responses + if "task" in data: + return data["task"] + return data + + def create_item(self, task_id: str, text: str, status: str = "todo") -> dict: + body = {"text": text, "status": status} + resp = self._post(f"/api/tasks/{task_id}/items", body) + return resp.get("data", {}) + + def update_item_text(self, list_id: str, item_index: int | str, text: str) -> None: + """Update the text/description of a checklist item. + + Uses /api/checklists/{listId}/items/{itemIndex} (PATCH) which works for both + regular checklists and task-type checklists. + """ + self._patch(f"/api/checklists/{list_id}/items/{item_index}", {"text": text}) + + def update_item_status(self, task_id: str, item_index: int | str, new_status: str) -> None: + """Update the Kanban status of a task checklist item.""" + self._put(f"/api/tasks/{task_id}/items/{item_index}/status", {"status": new_status}) + + def check_item(self, task_id: str, item_index: int | str) -> None: + """Mark an item as completed (check).""" + self._put(f"/api/tasks/{task_id}/items/{item_index}/check") + + def uncheck_item(self, task_id: str, item_index: int | str) -> None: + """Mark an item as incomplete (un-check).""" + self._put(f"/api/tasks/{task_id}/items/{item_index}/uncheck") + + def delete_item(self, list_id: str, item_index: int | str) -> None: + """Delete a checklist item. + + Uses /api/checklists endpoint which works for both regular and task checklists. + """ + self._delete(f"/api/checklists/{list_id}/items/{item_index}") + + +# --------------------------------------------------------------------------- +# Sync engine +# --------------------------------------------------------------------------- +class SyncEngine: + def __init__(self, config: SyncConfig): + self.config = config + self.jira = JiraClient( + config.jira_base_url, config.jira_api_token, + ) + self.jotty = JottyClient(config.jotty_base_url, config.jotty_api_key) + self.mapping = SyncMapping.load(config.mapping_file) + + # -- Jira → Jotty ----------------------------------------------------- + def sync_jira_to_jotty(self) -> None: + """Pull stories from Jira and create/update them as checklist items in Jotty. + + Jira is the source of truth. Each run scans Jotty fresh and reconciles: + - Existing items are updated to match Jira status/text + - Missing items are created + - Extra/duplicate items are deleted + """ + jql = self.config.jql_query.replace("${PROJECT_KEY}", self.config.project_key) + issues = self.jira.search_stories(jql) + + task_list = self.jotty.find_or_create_task_list( + self.config.checklist_title, self.config.checklist_category, + ) + task_id = task_list["id"] + + # Build the desired state from Jira: key -> (summary, status_raw, description) + jira_map: dict[str, tuple] = {} + for issue in issues: + key = issue["key"] + summary = issue["fields"].get("summary", "") + status_raw = (issue["fields"].get("status") or {}).get("name", "To Do") + description = issue["fields"].get("description") or "" + jira_map[key] = (summary, status_raw, description) + + # Scan current Jotty items and build a map: jira_key -> first matching index + current_task = self.jotty.get_task(task_id) + current_items = current_task.get("items", []) + + # Track which indices we've used so we can delete extras/duplicates later + key_to_idx: dict[str, int] = {} # jira_key -> single index to use + extra_indices: list[int] = [] # indices to delete (duplicates or orphaned) + + for item in current_items: + text = item.get("text", "") + jira_key = self._extract_jira_key(text) + if not jira_key: + continue # skip items without a Jira key — they're orphaned + if jira_key in jira_map and jira_key not in key_to_idx: + # First match for this Jira issue — keep it + key_to_idx[jira_key] = item["index"] + else: + # Duplicate or orphaned — mark for deletion + extra_indices.append(item["index"]) + + # Update existing items (status + text) and delete extras + for jira_key, idx in key_to_idx.items(): + summary, status_raw, description = jira_map[jira_key] + jotty_status = JIRA_TO_JOTTY_STATUS.get(status_raw, "todo") + + try: + self.jotty.update_item_status(task_id, idx, jotty_status) + log.info("Updated %s → Jotty status '%s' (index %d)", jira_key, jotty_status, idx) + except Exception as exc: + log.warning("Could not update status for %s at index %d: %s", jira_key, idx, exc) + + # Delete duplicate/orphaned items (reverse order to preserve indices) + for idx in sorted(extra_indices, reverse=True): + try: + self.jotty.delete_item(task_id, idx) + log.info("Deleted extra item at index %d", idx) + except Exception as exc: + log.warning("Could not delete item at index %d: %s", idx, exc) + + # Create items for any Jira issues that don't have a corresponding Jotty item yet + for jira_key in jira_map: + if jira_key not in key_to_idx: + summary, status_raw, description = jira_map[jira_key] + jotty_status = JIRA_TO_JOTTY_STATUS.get(status_raw, "todo") + + # Create item with just the title (Jotty truncates multi-line text) + create_text = f"[{jira_key}] {summary}" + self.jotty.create_item(task_id, create_text, jotty_status) + + # Discover the newly created item's index + new_task = self.jotty.get_task(task_id) + for it in new_task.get("items", []): + if jira_key in it.get("text", ""): + idx = it["index"] + key_to_idx[jira_key] = idx + + # Add Jira URL and description preview via PATCH on /api/checklists + issue_url = f"{self.config.jira_base_url}/browse/{jira_key}" + patch_body = {"description": issue_url} + if description and len(str(description)) > 0: + desc_preview = str(description).replace("\n", " ").strip()[:200] + patch_body["description"] += f"\n{desc_preview}" + + try: + self.jotty._patch(f"/api/checklists/{task_id}/items/{idx}", patch_body) + log.info("Created new item for %s at index %d with status '%s' and description", jira_key, idx, jotty_status) + except Exception as exc: + log.warning("Could not set description for %s at index %d: %s", jira_key, idx, exc) + + break + + # Save the mapping (single index per key — Jira is source of truth) + self.mapping.mappings = { + k: {"checklist_id": task_id, "item_index": idx} + for k, idx in key_to_idx.items() + } + self.mapping.save(self.config.mapping_file) + log.info("Jira→Jotty sync complete. %d issues processed.", len(issues)) + + # -- Jotty → Jira ----------------------------------------------------- + def sync_jotty_to_jira(self) -> None: + """Read the current state of checklist items and push status changes to Jira.""" + task_list = self.jotty.find_or_create_task_list( + self.config.checklist_title, self.config.checklist_category, + ) + task_id = task_list["id"] + + # Refresh mapping from disk in case another process changed it. + self.mapping = SyncMapping.load(self.config.mapping_file) + + task_data = self.jotty.get_task(task_id) + items = task_data.get("items", []) + + synced_count = 0 + for item in items: + text = item.get("text", "") + jira_key = self._extract_jira_key(text) + if not jira_key or jira_key not in self.mapping.mappings: + continue + + current_status = item.get("status", "todo") + completed = item.get("completed", False) + + # Determine target Jira status from the Jotty item state. + if completed: + jira_status_name = JOTTY_TO_JIRA_STATUS.get("completed", "Done") + else: + jira_status_name = JOTTY_TO_JIRA_STATUS.get(current_status, "To Do") + + # Check if a status change is needed + try: + issue_data = self.jira.get_issue(jira_key) + current_jira_status = (issue_data["fields"].get("status") or {}).get("name", "") + except Exception as exc: + log.warning("Could not fetch Jira issue %s: %s. Skipping.", jira_key, exc) + continue + + if current_jira_status == jira_status_name: + log.debug("%s already at '%s' — no change needed.", jira_key, jira_status_name) + continue + + self.jira.update_status(jira_key, jira_status_name) + log.info("Pushed %s → Jira status '%s' (was '%s')", jira_key, jira_status_name, current_jira_status) + synced_count += 1 + + log.info("Jotty→Jira sync complete. %d issues updated.", synced_count) + + # -- helpers ---------------------------------------------------------- + @staticmethod + def _extract_jira_key(text: str) -> Optional[str]: + """Extract PROJ-123 from text like '[PROJ-123] Summary...'.""" + import re + # Allow alphanumeric characters in the project key (e.g., S2R-10690) + m = re.search(r"\[([A-Za-z0-9]+-\d+)\]", text) + return m.group(1) if m else None + + # -- full sync -------------------------------------------------------- + def run(self, direction: str): + if direction in ("jira-to-jotty", "both"): + self.sync_jira_to_jotty() + if direction in ("jotty-to-jira", "both"): + self.sync_jotty_to_jira() + + +# --------------------------------------------------------------------------- +# Config loader (JSON file or environment variables) +# --------------------------------------------------------------------------- +def load_config(args: argparse.Namespace) -> SyncConfig: + cfg = SyncConfig() + + # 1. Try to read from sync_config.json + config_path = args.config if hasattr(args, "config") and args.config else "sync_config.json" + if os.path.exists(config_path): + log.info("Loading configuration from %s", config_path) + with open(config_path) as f: + raw = json.load(f) + + jira_cfg = raw.get("jira", {}) + cfg.jira_base_url = args.jira_base or jira_cfg.get("base_url", os.environ.get("JIRA_BASE_URL", "")) + cfg.jira_api_token = args.jira_token or jira_cfg.get("api_token", os.environ.get("JIRA_API_TOKEN", "")) + cfg.project_key = args.project or jira_cfg.get("project_key", os.environ.get("PROJECT_KEY", "")) + if "jql_query" in raw.get("jira", {}): + cfg.jql_query = raw["jira"]["jql_query"] + + jotty_cfg = raw.get("jotty", {}) + cfg.jotty_base_url = args.jotty_base or jotty_cfg.get("base_url", os.environ.get("JOTTY_BASE_URL", "")) + cfg.jotty_api_key = args.jotty_key or jotty_cfg.get("api_key", os.environ.get("JOTTY_API_KEY", "")) + if "checklist_title" in raw.get("jotty", {}): + cfg.checklist_title = raw["jotty"]["checklist_title"] + if "checklist_category" in raw.get("jotty", {}): + cfg.checklist_category = raw["jotty"]["checklist_category"] + + # 2. CLI args override everything + cfg.jira_base_url = args.jira_base or cfg.jira_base_url + cfg.jira_api_token = args.jira_token or cfg.jira_api_token + cfg.project_key = args.project or cfg.project_key + cfg.jotty_base_url = args.jotty_base or cfg.jotty_base_url + cfg.jotty_api_key = args.jotty_key or cfg.jotty_api_key + + # 3. Environment variables are the last fallback for anything not set above + if not cfg.jira_base_url: + cfg.jira_base_url = os.environ.get("JIRA_BASE_URL", "") + if not cfg.jira_api_token: + cfg.jira_api_token = os.environ.get("JIRA_API_TOKEN", "") + if not cfg.project_key: + cfg.project_key = os.environ.get("PROJECT_KEY", "") + env_jql = os.environ.get("JQL_QUERY") + if env_jql: + cfg.jql_query = env_jql + if not cfg.jotty_base_url: + cfg.jotty_base_url = os.environ.get("JOTTY_BASE_URL", "") + if not cfg.jotty_api_key: + cfg.jotty_api_key = os.environ.get("JOTTY_API_KEY", "") + + return cfg + + +# --------------------------------------------------------------------------- +# Argument parser +# --------------------------------------------------------------------------- +def build_parser() -> argparse.ArgumentParser: + p = argparse.ArgumentParser(description="Sync Jira stories ↔ Jotty checklists") + p.add_argument( + "--direction", "-d", + choices=["jira-to-jotty", "jotty-to-jira", "both"], + default="both", + help="Sync direction (default: both)", + ) + p.add_argument("--config", "-c", default="sync_config.json", help="Path to config JSON file") + + # CLI overrides for Jira + p.add_argument("--jira-base", dest="jira_base", help="Jira base URL (overrides config)") + p.add_argument("--jira-token", dest="jira_token", help="Jira API token / PAT") + p.add_argument("--project", "-p", help="Jira project key (e.g. PROJ)") + + # CLI overrides for Jotty + p.add_argument("--jotty-base", dest="jotty_base", help="Jotty base URL (overrides config)") + p.add_argument("--jotty-key", dest="jotty_key", help="Jotty API key") + return p + + +# --------------------------------------------------------------------------- +# Main +# --------------------------------------------------------------------------- +def main(): + parser = build_parser() + args = parser.parse_args() + + try: + config = load_config(args) + except Exception as exc: + log.error("Failed to load configuration: %s", exc) + sys.exit(1) + + # Validate required fields + missing = [] + if not config.jira_base_url: + missing.append("JIRA_BASE_URL") + if not config.jira_api_token: + missing.append("JIRA_API_TOKEN") + if not config.project_key: + missing.append("PROJECT_KEY") + if not config.jotty_base_url: + missing.append("JOTTY_BASE_URL") + if not config.jotty_api_key: + missing.append("JOTTY_API_KEY") + + if missing: + log.error("Missing required configuration. Set these env vars or add them to sync_config.json:") + for m in missing: + log.error(" %s", m) + sys.exit(1) + + engine = SyncEngine(config) + try: + engine.run(args.direction) + except Exception as exc: + log.exception("Sync failed: %s", exc) + sys.exit(1) + + +if __name__ == "__main__": + main() diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..e584ef7 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,3 @@ +jira>=3.6.0 +requests>=2.31.0 +python-dotenv>=1.0.0 diff --git a/sync_config.json.template b/sync_config.json.template new file mode 100644 index 0000000..8b36794 --- /dev/null +++ b/sync_config.json.template @@ -0,0 +1,16 @@ +{ + "_comment": "Copy this file to sync_config.json and fill in your credentials.", + "jira": { + "base_url": "https://your-org.atlassian.net", + "username": "you@company.com", + "api_token": "ATATT3xFfGF...", + "project_key": "PROJ", + "jql_query": "project = \"${PROJECT_KEY}\" AND issuetype = Story AND status NOT IN (\"Done\") ORDER BY created DESC" + }, + "jotty": { + "base_url": "https://jotty.your-domain.com", + "api_key": "ck_your_jotty_api_key_here", + "checklist_title": "Jira Stories", + "checklist_category": "Work" + } +} diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_sync.py b/tests/test_sync.py new file mode 100644 index 0000000..a2edabf --- /dev/null +++ b/tests/test_sync.py @@ -0,0 +1,797 @@ +""" +Tests for Jira ↔ Jotty Bidirectional Sync Script + +Run with: python -m pytest tests/ -v +Or: pip install responses pytest && python -m pytest tests/ -v +""" + +import json +import os +import sys +import tempfile +import unittest +from pathlib import Path +from unittest.mock import MagicMock, patch + +# --------------------------------------------------------------------------- +# Ensure the parent directory is on sys.path so we can import jira_jotty_sync +# --------------------------------------------------------------------------- +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +import responses # type: ignore[import-untyped] + +import jira_jotty_sync as sync_module + + +# =================================================================== +# Fixtures / Helpers +# =================================================================== + +SAMPLE_JIRA_ISSUE = { + "key": "PROJ-42", + "fields": { + "summary": "Implement user authentication", + "status": {"name": "IN PROGRESS"}, + "description": "Add login and token refresh", + }, +} + +SAMPLE_JOTTY_TASK = { + "id": "task-uuid-001", + "title": "Jira Stories", + "category": "Work", + "type": "task", + "statuses": [ + {"id": "todo", "label": "To Do", "order": 0}, + {"id": "in_progress", "label": "In Progress", "order": 1}, + {"id": "completed", "label": "Completed", "order": 2}, + ], + "items": [ + { + "index": 0, + "text": "[PROJ-42] Implement user authentication", + "status": "in_progress", + "completed": False, + }, + { + "index": 1, + "text": "[PROJ-99] Write API documentation", + "status": "todo", + "completed": False, + }, + ], +} + +SAMPLE_JOTTY_CHECKLIST = { + "id": "task-uuid-001", + "title": "Jira Stories", + "category": "Work", + "type": "task", + "items": [], +} + + +def _make_config(**overrides): + """Build a SyncConfig with sensible defaults, overridden by kwargs.""" + cfg = sync_module.SyncConfig( + jira_base_url="https://myorg.atlassian.net", + jira_username="you@company.com", + jira_api_token="ATATT3xFfGFtoken123", + project_key="PROJ", + jql_query='project = "${PROJECT_KEY}" AND issuetype = Story', + jotty_base_url="https://jotty.example.com", + jotty_api_key="ck_testkey_abc", + checklist_title="Jira Stories", + checklist_category="Work", + mapping_file="/tmp/test_sync_mapping.json", + ) + for k, v in overrides.items(): + setattr(cfg, k, v) + return cfg + + +# =================================================================== +# 1. Jira Connection & Extraction Tests +# =================================================================== + +class TestJiraConnectionAndExtraction(unittest.TestCase): + """Test that the JiraClient can authenticate and extract stories.""" + + def _setup_mock(self): + self.client = sync_module.JiraClient( + "https://myorg.atlassian.net", + "you@company.com", + "ATATT3xFfGFtoken123", + ) + + @responses.activate + def test_jira_auth_headers(self): + """Verify that JiraClient sends Basic auth with username:token.""" + self._setup_mock() + # We'll intercept the request and check headers + captured = {} + + def capture_request(req): + captured["auth"] = req.headers.get("Authorization", "") + return (200, [], json.dumps({"total": 0, "maxResults": 100, "issues": []})) + + responses.add_callback( + responses.POST, + "https://myorg.atlassian.net/rest/api/3/search/search", + callback=capture_request, + ) + + self.client.search_stories('project = "PROJ" AND issuetype = Story') + self.assertTrue(captured["auth"].startswith("Basic "), "Auth header should be Basic") + + @responses.activate + def test_search_stories_returns_issues(self): + """search_stories should return a list of issue dicts.""" + self._setup_mock() + responses.add( + responses.POST, + "https://myorg.atlassian.net/rest/api/3/search/search", + json={ + "total": 2, + "maxResults": 100, + "issues": [SAMPLE_JIRA_ISSUE], + }, + status=200, + ) + + issues = self.client.search_stories('project = "PROJ" AND issuetype = Story') + self.assertIsInstance(issues, list) + self.assertEqual(len(issues), 1) + self.assertEqual(issues[0]["key"], "PROJ-42") + + @responses.activate + def test_search_stories_pagination(self): + """search_stories should paginate through all results.""" + self._setup_mock() + # First page returns partial results + responses.add( + responses.POST, + "https://myorg.atlassian.net/rest/api/3/search/search", + json={ + "total": 200, + "maxResults": 100, + "issues": [{"key": f"PROJ-{i}"} for i in range(100)], + }, + status=200, + ) + + issues = self.client.search_stories('project = "PROJ" AND issuetype = Story') + # Should have fetched 100 from first page + 100 from second = 200 + self.assertEqual(len(issues), 200) + + @responses.activate + def test_search_stories_empty(self): + """search_stories should return empty list when no issues match.""" + self._setup_mock() + responses.add( + responses.POST, + "https://myorg.atlassian.net/rest/api/3/search/search", + json={"total": 0, "maxResults": 100, "issues": []}, + status=200, + ) + + issues = self.client.search_stories('project = "PROJ" AND issuetype = Story') + self.assertEqual(len(issues), 0) + + @responses.activate + def test_get_issue(self): + """get_issue should return a single issue dict.""" + self._setup_mock() + responses.add( + responses.GET, + "https://myorg.atlassian.net/rest/api/3/issues/PROJ-42", + json=SAMPLE_JIRA_ISSUE, + status=200, + ) + + issue = self.client.get_issue("PROJ-42") + self.assertEqual(issue["key"], "PROJ-42") + self.assertEqual(issue["fields"]["summary"], "Implement user authentication") + + @responses.activate + def test_update_status_via_transition(self): + """update_status should use transitions API when direct PUT fails.""" + self._setup_mock() + # Direct PUT succeeds (200) — no fallback needed + responses.add( + responses.PUT, + "https://myorg.atlassian.net/rest/api/3/issues/PROJ-42", + json={}, + status=200, + ) + + self.client.update_status("PROJ-42", "IN PROGRESS") + + @responses.activate + def test_update_status_fallback_to_transition(self): + """update_status should fall back to transitions API on 403.""" + self._setup_mock() + # Direct PUT fails with 403, triggering fallback to transitions + responses.add( + responses.PUT, + "https://myorg.atlassian.net/rest/api/3/issues/PROJ-42", + json={"errorMessages": ["Forbidden"], "status": 403}, + status=403, + ) + # Note: transitions endpoint uses 'issue' (singular), not 'issues' + responses.add( + responses.GET, + "https://myorg.atlassian.net/rest/api/3/issue/PROJ-42/transitions", + json={ + "transitions": [ + { + "id": "11", + "to": {"name": "IN PROGRESS"}, + "fields": {}, + }, + ], + }, + status=200, + ) + responses.add( + responses.POST, + "https://myorg.atlassian.net/rest/api/3/issue/PROJ-42/transitions", + json={}, + status=200, + ) + + self.client.update_status("PROJ-42", "IN PROGRESS") + + +# =================================================================== +# 2. Jotty Connection & Extraction Tests +# =================================================================== + +class TestJottyConnectionAndExtraction(unittest.TestCase): + """Test that the JottyClient can authenticate and extract checklists.""" + + def _setup_mock(self): + self.client = sync_module.JottyClient( + "https://jotty.example.com", + "ck_testkey_abc", + ) + + @responses.activate + def test_jotty_api_key_header(self): + """Verify that JottyClient sends the correct x-api-key header.""" + self._setup_mock() + captured = {} + + def capture_request(req): + captured["api_key"] = req.headers.get("x-api-key", "") + return (200, [], json.dumps({"checklists": []})) + + responses.add_callback( + responses.GET, + "https://jotty.example.com/api/checklists", + callback=capture_request, + ) + + self.client.get_checklists() + self.assertEqual(captured["api_key"], "ck_testkey_abc") + + @responses.activate + def test_get_checklists(self): + """get_checklists should return a list of checklist dicts.""" + self._setup_mock() + responses.add( + responses.GET, + "https://jotty.example.com/api/checklists", + json={"checklists": [SAMPLE_JOTTY_CHECKLIST]}, + status=200, + ) + + checklists = self.client.get_checklists() + self.assertIsInstance(checklists, list) + self.assertEqual(len(checklists), 1) + self.assertEqual(checklists[0]["title"], "Jira Stories") + + def test_find_or_create_task_list_existing(self): + """find_or_create_task_list should find and return an existing task checklist.""" + with responses.RequestsMock() as rsps: + self._setup_mock() + rsps.add( + responses.GET, + "https://jotty.example.com/api/checklists", + json={"checklists": [SAMPLE_JOTTY_CHECKLIST]}, + status=200, + ) + + result = self.client.find_or_create_task_list("Jira Stories", "Work") + self.assertEqual(result["id"], "task-uuid-001") + self.assertEqual(result["type"], "task") + + @responses.activate + def test_find_or_create_task_list_creates_new(self): + """find_or_create_task_list should create a new task checklist if none exists.""" + self._setup_mock() + responses.add( + responses.GET, + "https://jotty.example.com/api/checklists", + json={"checklists": []}, # no existing lists + status=200, + ) + responses.add( + responses.POST, + "https://jotty.example.com/api/tasks", + json={ + "success": True, + "data": { + "id": "new-task-uuid", + "title": "Jira Stories", + "type": "task", + "items": [], + }, + }, + status=200, + ) + + result = self.client.find_or_create_task_list("Jira Stories", "Work") + self.assertEqual(result["id"], "new-task-uuid") + + @responses.activate + def test_get_task(self): + """get_task should return a task dict.""" + self._setup_mock() + responses.add( + responses.GET, + "https://jotty.example.com/api/tasks/task-uuid-001", + json={"task": SAMPLE_JOTTY_TASK}, + status=200, + ) + + task = self.client.get_task("task-uuid-001") + self.assertEqual(task["title"], "Jira Stories") + self.assertEqual(len(task["items"]), 2) + + @responses.activate + def test_create_item(self): + """create_item should create a new item and return its data.""" + self._setup_mock() + responses.add( + responses.POST, + "https://jotty.example.com/api/tasks/task-uuid-001/items", + json={"success": True, "data": {"id": "item-123"}}, + status=200, + ) + + result = self.client.create_item("task-uuid-001", "[PROJ-42] Test task", "todo") + self.assertEqual(result["id"], "item-123") + + @responses.activate + def test_update_item_status(self): + """update_item_status should PUT to the correct endpoint.""" + self._setup_mock() + responses.add( + responses.PUT, + "https://jotty.example.com/api/tasks/task-uuid-001/items/0/status", + json={"success": True}, + status=200, + ) + + self.client.update_item_status("task-uuid-001", 0, "in_progress") + + @responses.activate + def test_check_item(self): + """check_item should PUT to the check endpoint.""" + self._setup_mock() + responses.add( + responses.PUT, + "https://jotty.example.com/api/tasks/task-uuid-001/items/0/check", + json={"success": True}, + status=200, + ) + + self.client.check_item("task-uuid-001", 0) + + @responses.activate + def test_uncheck_item(self): + """uncheck_item should PUT to the uncheck endpoint.""" + self._setup_mock() + responses.add( + responses.PUT, + "https://jotty.example.com/api/tasks/task-uuid-001/items/0/uncheck", + json={"success": True}, + status=200, + ) + + self.client.uncheck_item("task-uuid-001", 0) + + @responses.activate + def test_delete_item(self): + """delete_item should DELETE the correct item.""" + self._setup_mock() + responses.add( + responses.DELETE, + "https://jotty.example.com/api/tasks/task-uuid-001/items/0", + json={"success": True}, + status=200, + ) + + self.client.delete_item("task-uuid-001", 0) + + +# =================================================================== +# 3. Sync Engine Tests (Jira → Jotty) +# =================================================================== + +class TestSyncEngineJiraToJotty(unittest.TestCase): + """Test the jira-to-jotty sync direction.""" + + @responses.activate + def test_sync_creates_mapping(self): + """sync_jira_to_jotty should create a mapping entry for each issue.""" + config = _make_config() + engine = sync_module.SyncEngine(config) + + # Mock Jira search + responses.add( + responses.POST, + "https://myorg.atlassian.net/rest/api/3/search/search", + json={ + "total": 1, + "maxResults": 100, + "issues": [SAMPLE_JIRA_ISSUE], + }, + status=200, + ) + + # Mock Jotty: find existing task list + responses.add( + responses.GET, + "https://jotty.example.com/api/checklists", + json={"checklists": [SAMPLE_JOTTY_CHECKLIST]}, + status=200, + ) + + # Mock the task list creation response (since we're creating new items) + responses.add( + responses.POST, + "https://jotty.example.com/api/tasks/list-uuid-001/items", + json={"success": True, "data": {"id": "item-new"}}, + status=200, + ) + + engine.sync_jira_to_jotty() + + # Check that mapping was saved + self.assertTrue(os.path.exists(config.mapping_file)) + with open(config.mapping_file) as f: + mapping = json.load(f) + self.assertIn("PROJ-42", mapping) + + @responses.activate + def test_sync_updates_existing_item_status(self): + """sync_jira_to_jotty should update status of existing mapped items.""" + config = _make_config() + engine = sync_module.SyncEngine(config) + + # Pre-populate the mapping file with an existing entry + mapping_data = {"PROJ-42": {"checklist_id": "list-uuid-001", "item_index": 0}} + Path(config.mapping_file).write_text(json.dumps(mapping_data)) + + responses.add( + responses.POST, + "https://myorg.atlassian.net/rest/api/3/search/search", + json={ + "total": 1, + "maxResults": 100, + "issues": [SAMPLE_JIRA_ISSUE], + }, + status=200, + ) + + responses.add( + responses.GET, + "https://jotty.example.com/api/checklists", + json={"checklists": [SAMPLE_JOTTY_CHECKLIST]}, + status=200, + ) + + responses.add( + responses.PUT, + "https://jotty.example.com/api/tasks/list-uuid-001/items/0/status", + json={"success": True}, + status=200, + ) + + engine.sync_jira_to_jotty() + + +# =================================================================== +# 4. Sync Engine Tests (Jotty → Jira) +# =================================================================== + +class TestSyncEngineJottyToJira(unittest.TestCase): + """Test the jotty-to-jira sync direction.""" + + @responses.activate + def test_sync_pushes_status_changes(self): + """sync_jotty_to_jira should push completed items to Jira as DONE.""" + config = _make_config() + engine = sync_module.SyncEngine(config) + + # Pre-populate mapping file + mapping_data = { + "PROJ-42": {"checklist_id": "task-uuid-001", "item_index": 0}, + "PROJ-99": {"checklist_id": "task-uuid-001", "item_index": 1}, + } + Path(config.mapping_file).write_text(json.dumps(mapping_data)) + + # Mock Jira: get issue status for PROJ-42 (needs change) + responses.add( + responses.GET, + "https://myorg.atlassian.net/rest/api/3/issues/PROJ-42", + json={ + "key": "PROJ-42", + "fields": {"status": {"name": "TO DO"}}, + }, + status=200, + ) + + # Mock Jira: get issue status for PROJ-99 (no change needed) + responses.add( + responses.GET, + "https://myorg.atlassian.net/rest/api/3/issues/PROJ-99", + json={ + "key": "PROJ-99", + "fields": {"status": {"name": "TO DO"}}, + }, + status=200, + ) + + # Mock Jira: direct PUT fails with 403 (triggers fallback to transitions) + responses.add( + responses.PUT, + "https://myorg.atlassian.net/rest/api/3/issues/PROJ-42", + json={"errorMessages": ["Forbidden"], "status": 403}, + status=403, + ) + + # Mock Jira: get transitions for PROJ-42 (note: 'issue' singular) + responses.add( + responses.GET, + "https://myorg.atlassian.net/rest/api/3/issue/PROJ-42/transitions", + json={ + "transitions": [ + {"id": "11", "to": {"name": "IN PROGRESS"}, "fields": {}}, + ], + }, + status=200, + ) + + # Mock Jira: update PROJ-42 to IN PROGRESS via transition + responses.add( + responses.POST, + "https://myorg.atlassian.net/rest/api/3/issue/PROJ-42/transitions", + json={}, + status=200, + ) + + # Mock Jotty: get task list response with items + responses.add( + responses.GET, + "https://jotty.example.com/api/checklists", + json={"checklists": [SAMPLE_JOTTY_CHECKLIST]}, + status=200, + ) + + responses.add( + responses.GET, + "https://jotty.example.com/api/tasks/task-uuid-001", + json={"task": SAMPLE_JOTTY_TASK}, + status=200, + ) + + engine.sync_jotty_to_jira() + + +# =================================================================== +# 5. Config Loading Tests +# =================================================================== + +class TestConfigLoading(unittest.TestCase): + """Test that configuration is loaded correctly from various sources.""" + + def test_load_config_from_env(self): + """load_config should read environment variables when no config file exists.""" + with patch.dict(os.environ, { + "JIRA_BASE_URL": "https://env-test.atlassian.net", + "JIRA_USERNAME": "env@test.com", + "JIRA_API_TOKEN": "env_token_123", + "PROJECT_KEY": "ENV", + "JOTTY_BASE_URL": "https://jotty-env.example.com", + "JOTTY_API_KEY": "ck_env_key", + }): + args = MagicMock() + args.config = "/nonexistent/config.json" + args.jira_base = None + args.jira_user = None + args.jira_token = None + args.project = None + args.jotty_base = None + args.jotty_key = None + + config = sync_module.load_config(args) + self.assertEqual(config.jira_base_url, "https://env-test.atlassian.net") + self.assertEqual(config.jira_username, "env@test.com") + self.assertEqual(config.project_key, "ENV") + self.assertEqual(config.jotty_api_key, "ck_env_key") + + def test_load_config_from_json_file(self): + """load_config should read from sync_config.json when it exists.""" + config_content = { + "jira": { + "base_url": "https://json-test.atlassian.net", + "username": "json@test.com", + "api_token": "json_token_123", + "project_key": "JSON", + }, + "jotty": { + "base_url": "https://jotty-json.example.com", + "api_key": "ck_json_key", + "checklist_title": "Custom Title", + "checklist_category": "Projects", + }, + } + + with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f: + json.dump(config_content, f) + config_path = f.name + + try: + args = MagicMock() + args.config = config_path + args.jira_base = None + args.jira_user = None + args.jira_token = None + args.project = None + args.jotty_base = None + args.jotty_key = None + + config = sync_module.load_config(args) + self.assertEqual(config.jira_base_url, "https://json-test.atlassian.net") + self.assertEqual(config.checklist_title, "Custom Title") + finally: + os.unlink(config_path) + + def test_cli_overrides_env(self): + """CLI arguments should override environment variables.""" + with patch.dict(os.environ, { + "JIRA_BASE_URL": "https://env.atlassian.net", + "PROJECT_KEY": "ENV", + }): + args = MagicMock() + args.config = "/nonexistent/config.json" + args.jira_base = "https://cli.atlassian.net" # CLI override + args.jira_user = None + args.jira_token = None + args.project = "CLI" # CLI override + args.jotty_base = None + args.jotty_key = None + + config = sync_module.load_config(args) + self.assertEqual(config.jira_base_url, "https://cli.atlassian.net") + self.assertEqual(config.project_key, "CLI") + + +# =================================================================== +# 6. Mapping Persistence Tests +# =================================================================== + +class TestMappingPersistence(unittest.TestCase): + """Test that SyncMapping can save and load correctly.""" + + def test_save_and_load_mapping(self): + """SyncMapping should persist mappings to disk and reload them.""" + mapping = sync_module.SyncMapping() + mapping.mappings["PROJ-1"] = {"checklist_id": "cl-001", "item_index": 0} + mapping.mappings["PROJ-2"] = {"checklist_id": "cl-001", "item_index": 1} + + with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f: + temp_path = f.name + + try: + mapping.save(temp_path) + loaded = sync_module.SyncMapping.load(temp_path) + self.assertIn("PROJ-1", loaded.mappings) + self.assertIn("PROJ-2", loaded.mappings) + self.assertEqual(loaded.mappings["PROJ-1"]["item_index"], 0) + finally: + os.unlink(temp_path) + + def test_load_nonexistent_mapping(self): + """SyncMapping.load should return empty mapping for non-existent file.""" + loaded = sync_module.SyncMapping.load("/nonexistent/path.json") + self.assertEqual(loaded.mappings, {}) + + +# =================================================================== +# 7. Status Mapping Tests +# =================================================================== + +class TestStatusMappings(unittest.TestCase): + """Test that status mappings are correct and complete.""" + + def test_jira_to_jotty_mapping(self): + """JIRA_TO_JOTTY_STATUS should map all expected statuses.""" + mapping = sync_module.JIRA_TO_JOTTY_STATUS + self.assertEqual(mapping["TO DO"], "todo") + self.assertEqual(mapping["IN PROGRESS"], "in_progress") + self.assertEqual(mapping["DONE"], "completed") + + def test_jotty_to_jira_mapping(self): + """JOTTY_TO_JIRA_STATUS should map all expected statuses.""" + mapping = sync_module.JOTTY_TO_JIRA_STATUS + self.assertEqual(mapping["todo"], "TO DO") + self.assertEqual(mapping["in_progress"], "IN PROGRESS") + self.assertEqual(mapping["completed"], "DONE") + + def test_extract_jira_key(self): + """_extract_jira_key should extract PROJ-123 from text.""" + self.assertEqual( + sync_module.SyncEngine._extract_jira_key("[PROJ-42] Implement auth"), + "PROJ-42", + ) + + def test_extract_jira_key_not_found(self): + """_extract_jira_key should return None when no key is found.""" + self.assertIsNone( + sync_module.SyncEngine._extract_jira_key("No Jira key here"), + ) + + +# =================================================================== +# 8. Error Handling Tests +# =================================================================== + +class TestErrorHandling(unittest.TestCase): + """Test that errors are handled gracefully.""" + + @responses.activate + def test_jira_401_unauthorized(self): + """JiraClient should raise HTTPError on 401.""" + client = sync_module.JiraClient( + "https://myorg.atlassian.net", + "baduser", + "badtoken", + ) + responses.add( + responses.POST, + "https://myorg.atlassian.net/rest/api/3/search/search", + json={"errorMessages": ["Unauthorized"], "status": 401}, + status=401, + ) + + with self.assertRaises(Exception): # requests.HTTPError + client.search_stories('project = "PROJ"') + + @responses.activate + def test_jotty_401_unauthorized(self): + """JottyClient should raise HTTPError on 401.""" + client = sync_module.JottyClient( + "https://jotty.example.com", + "ck_bad_key", + ) + responses.add( + responses.GET, + "https://jotty.example.com/api/checklists", + json={"error": "Unauthorized"}, + status=401, + ) + + with self.assertRaises(Exception): # requests.HTTPError + client.get_checklists() + + +# =================================================================== +# Run tests +# =================================================================== + +if __name__ == "__main__": + unittest.main()