init

.gitignore

@@ -0,0 +1,4 @@
+.env
+sync_mapping.json
+__pycache__/
+*.pyc

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()

requirements.txt

@@ -0,0 +1,3 @@
+jira>=3.6.0
+requests>=2.31.0
+python-dotenv>=1.0.0

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"
+  }
+}

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()