🎨 Palette: Enhance CLI progress bars and summary table

main.py

@@ -1654,98 +1654,98 @@
         return False
 
 
 def create_folder(
     client: httpx.Client, profile_id: str, name: str, do: int, status: int
 ) -> Optional[str]:
     """
     Create a new folder and return its ID.
     Attempts to read ID from response first, then falls back to polling.
     """
     try:
         # 1. Send the Create Request
         response = _api_post(
             client,
             f"{API_BASE}/{profile_id}/groups",
             data={"name": name, "do": do, "status": status},
         )
 
         # OPTIMIZATION: Try to grab ID directly from response to avoid the wait loop
         try:
             resp_data = response.json()
             body = resp_data.get("body", {})
 
             # Check if it returned a single group object
             if isinstance(body, dict) and "group" in body and "PK" in body["group"]:
                 pk = str(body["group"]["PK"])
                 if not validate_folder_id(pk, log_errors=False):
                     log.error(f"API returned invalid folder ID: {sanitize_for_log(pk)}")
                     return None
                 log.info(
                     "Created folder %s (ID %s) [Direct]",
                     sanitize_for_log(name),
                     sanitize_for_log(pk),
                 )
                 return pk
 
             # Check if it returned a list containing our group
             if isinstance(body, dict) and "groups" in body:
                 for grp in body["groups"]:
                     if grp.get("group") == name:
                         pk = str(grp["PK"])
                         if not validate_folder_id(pk, log_errors=False):
                             log.error(f"API returned invalid folder ID: {sanitize_for_log(pk)}")
                             continue
                         log.info(
                             "Created folder %s (ID %s) [Direct]",
                             sanitize_for_log(name),
                             sanitize_for_log(pk),
                         )
                         return pk
         except Exception as e:
             log.debug(
                 f"Could not extract ID from POST response: " f"{sanitize_for_log(e)}"
             )
 
         # 2. Fallback: Poll for the new folder (The Robust Retry Logic)
         for attempt in range(MAX_RETRIES + 1):
             try:
                 data = _api_get(client, f"{API_BASE}/{profile_id}/groups").json()
                 groups = data.get("body", {}).get("groups", [])
 
                 for grp in groups:
                     if grp["group"].strip() == name.strip():
                         pk = str(grp["PK"])
                         if not validate_folder_id(pk, log_errors=False):
                             log.error(f"API returned invalid folder ID: {sanitize_for_log(pk)}")
                             return None
                         log.info(
                             "Created folder %s (ID %s) [Polled]",
                             sanitize_for_log(name),
                             sanitize_for_log(pk),
                         )
                         return pk
             except Exception as e:
                 log.warning(
                     f"Error fetching groups on attempt {attempt}: {sanitize_for_log(e)}"
                 )
 
             if attempt < MAX_RETRIES:
                 wait_time = FOLDER_CREATION_DELAY * (attempt + 1)
                 log.info(
                     f"Folder '{sanitize_for_log(name)}' not found yet. Retrying in {wait_time}s..."
                 )
                 time.sleep(wait_time)
 
         log.error(
             f"Folder {sanitize_for_log(name)} was not found after creation and retries."
         )
         return None
 
     except (httpx.HTTPError, KeyError) as e:
         log.error(
             f"Failed to create folder {sanitize_for_log(name)}: {sanitize_for_log(e)}"
         )
         return None
 
 
 def push_rules(
@@ -1897,7 +1897,7 @@
                     existing_rules.update(result)
 
                 render_progress_bar(
                     successful_batches,
                     total_batches,
                     progress_label,
                 )
@@ -2191,6 +2191,47 @@
 # --------------------------------------------------------------------------- #
 # 5. Entry-point
 # --------------------------------------------------------------------------- #
+def print_summary_table(
+    sync_results: List[Dict[str, Any]], success_count: int, total: int, dry_run: bool
+) -> None:
+    # 1. Setup Data
+    max_p = max((len(r["profile"]) for r in sync_results), default=25)
+    w = [max(25, max_p), 10, 12, 10, 15]
+
+    t_f, t_r, t_d = sum(r["folders"] for r in sync_results), sum(r["rules"] for r in sync_results), sum(r["duration"] for r in sync_results)
+    all_ok = success_count == total
+    t_status = ("✅ Ready" if dry_run else "✅ All Good") if all_ok else "❌ Errors"
+    t_col = Colors.GREEN if all_ok else Colors.FAIL
+
+    # 2. Render
+    if not USE_COLORS:
+        # Simple ASCII Fallback
+        header = f"{'Profile ID':<{w[0]}} | {'Folders':>{w[1]}} | {'Rules':>{w[2]}} | {'Duration':>{w[3]}} | {'Status':<{w[4]}}"
+        sep = "-" * len(header)
+        print(f"\n{('DRY RUN' if dry_run else 'SYNC') + ' SUMMARY':^{len(header)}}\n{sep}\n{header}\n{sep}")
+        for r in sync_results:
+            print(f"{r['profile']:<{w[0]}} | {r['folders']:>{w[1]}} | {r['rules']:>{w[2]},} | {r['duration']:>{w[3]-1}.1f}s | {r['status_label']:<{w[4]}}")
+        print(f"{sep}\n{'TOTAL':<{w[0]}} | {t_f:>{w[1]}} | {t_r:>{w[2]},} | {t_d:>{w[3]-1}.1f}s | {t_status:<{w[4]}}\n{sep}\n")
+        return
+
+    # Unicode Table
+    def line(l, m, r): return f"{Colors.BOLD}{l}{m.join('─' * (x+2) for x in w)}{r}{Colors.ENDC}"
+    def row(c): return f"{Colors.BOLD}│{Colors.ENDC} {c[0]:<{w[0]}} {Colors.BOLD}│{Colors.ENDC} {c[1]:>{w[1]}} {Colors.BOLD}│{Colors.ENDC} {c[2]:>{w[2]}} {Colors.BOLD}│{Colors.ENDC} {c[3]:>{w[3]}} {Colors.BOLD}│{Colors.ENDC} {c[4]:<{w[4]}} {Colors.BOLD}│{Colors.ENDC}"
+
+    print(f"\n{line('┌', '─', '┐')}")
+    title = f"{'DRY RUN' if dry_run else 'SYNC'} SUMMARY"
+    print(f"{Colors.BOLD}│{Colors.CYAN if dry_run else Colors.HEADER}{title:^{sum(w) + 14}}{Colors.ENDC}{Colors.BOLD}│{Colors.ENDC}")
+    print(f"{line('├', '┬', '┤')}\n{row([f'{Colors.HEADER}Profile ID{Colors.ENDC}', f'{Colors.HEADER}Folders{Colors.ENDC}', f'{Colors.HEADER}Rules{Colors.ENDC}', f'{Colors.HEADER}Duration{Colors.ENDC}', f'{Colors.HEADER}Status{Colors.ENDC}'])}")
+    print(line("├", "┼", "┤"))
+
+    for r in sync_results:
+        sc = Colors.GREEN if r["success"] else Colors.FAIL
+        print(row([r["profile"], str(r["folders"]), f"{r['rules']:,}", f"{r['duration']:.1f}s", f"{sc}{r['status_label']}{Colors.ENDC}"]))
+
+    print(f"{line('├', '┼', '┤')}\n{row(['TOTAL', str(t_f), f'{t_r:,}', f'{t_d:.1f}s', f'{t_col}{t_status}{Colors.ENDC}'])}")
+    print(f"{line('└', '┴', '┘')}\n")
+
+
 def parse_args() -> argparse.Namespace:
     """
     Parses command-line arguments for the Control D sync tool.

tests/test_content_type.py

@@ -67,10 +67,6 @@ def test_reject_text_html(self, mock_stream):
         self.assertIn("Invalid Content-Type", str(cm.exception))
 
     @patch('main._gh.stream')
-    def test_reject_xml(self, mock_stream):
-        """Test that application/xml is rejected."""
-        mock_response = MagicMock()
-        mock_response.status_code = 200
     def test_reject_xml(self, mock_stream):
         """Test that application/xml is rejected."""
         mock_response = MagicMock()