🎨 Palette: Add validation to interactive prompts

.Jules/palette.md

@@ -9,3 +9,7 @@
 ## 2024-05-24 - Fail Fast & Friendly
 **Learning:** In CLI tools involving APIs, cascade failures (hundreds of "Failed to X") caused by basic auth issues (401/403) are overwhelming and confusing. A dedicated "Pre-flight Check" that validates credentials *before* attempting the main workload allows for specific, actionable error messages (e.g. "Check your token at [URL]") instead of generic HTTP errors.
 **Action:** Implement a `check_api_access()` step at the start of any CLI workflow to validate permissions and provide human-readable guidance on failure.
+
+## 2024-05-25 - Immediate CLI Input Validation
+**Learning:** Validating user input immediately during the interactive prompt loop (e.g. checking regex or length) prevents user frustration from long-running processes failing late due to simple typos. Immediate feedback loops ("❌ Invalid format, try again") are far superior to "Job failed after 20s".
+**Action:** Wrap `input()` calls in a `get_valid_input` helper that enforces constraints before proceeding.

main.py

@@ -247,15 +247,37 @@
 
     return True
 
-def validate_profile_id(profile_id: str) -> bool:
+def validate_profile_id(profile_id: str, log_errors: bool = True) -> bool:
     if not re.match(r"^[a-zA-Z0-9_-]+$", profile_id):
-        log.error("Invalid profile ID format (contains unsafe characters)")
+        if log_errors:
+            log.error("Invalid profile ID format (contains unsafe characters)")
         return False
     if len(profile_id) > 64:
-        log.error("Invalid profile ID length (max 64 chars)")
+        if log_errors:
+            log.error("Invalid profile ID length (max 64 chars)")
         return False
     return True
 
+def get_valid_input(prompt: str, validator: callable, error_msg: str, is_password: bool = False) -> str:
+    """
+    Prompts the user for input until it passes validation.
+    """
+    import getpass
+    while True:
+        if is_password:
+            value = getpass.getpass(prompt).strip()
+        else:
+            value = input(prompt).strip()
+
+        if not value:
+            print(f"{Colors.FAIL}❌ Value cannot be empty.{Colors.ENDC}")
+            continue
+
+        if validator(value):
+            return value
+
+        print(f"{Colors.FAIL}❌ {error_msg}{Colors.ENDC}")
+
 def is_valid_rule(rule: str) -> bool:
     """
     Validates that a rule is safe to use.
@@ -376,9 +398,9 @@
             log.critical(f"{Colors.FAIL}❌ Authentication Failed: The API Token is invalid.{Colors.ENDC}")
             log.critical(f"{Colors.FAIL}   Please check your token at: https://controld.com/account/manage-account{Colors.ENDC}")
         elif code == 403:
             log.critical(f"{Colors.FAIL}🚫 Access Denied: Token lacks permission for Profile {profile_id}.{Colors.ENDC}")
         elif code == 404:
             log.critical(f"{Colors.FAIL}🔍 Profile Not Found: The ID '{profile_id}' does not exist.{Colors.ENDC}")
             log.critical(f"{Colors.FAIL}   Please verify the Profile ID from your Control D Dashboard URL.{Colors.ENDC}")
         else:
             log.error(f"API Access Check Failed ({code}): {e}")
@@ -854,7 +876,7 @@
         return success_count == len(folder_data_list)
 
     except Exception as e:
         log.error(f"Unexpected error during sync for profile {profile_id}: {sanitize_for_log(e)}")
         return False
 
 # --------------------------------------------------------------------------- #
@@ -866,7 +888,7 @@
     parser.add_argument("--folder-url", action="append", help="Folder JSON URL(s)", default=None)
     parser.add_argument("--dry-run", action="store_true", help="Plan only")
     parser.add_argument("--no-delete", action="store_true", help="Do not delete existing folders")
     parser.add_argument("--plan-json", help="Write plan to JSON file", default=None)
     return parser.parse_args()
 
 def main():
@@ -881,17 +903,34 @@
         if not profile_ids:
             print(f"{Colors.CYAN}ℹ Profile ID is missing.{Colors.ENDC}")
             print(f"{Colors.CYAN}  You can find this in the URL of your profile in the Control D Dashboard.{Colors.ENDC}")
-            p_input = input(f"{Colors.BOLD}Enter Control D Profile ID:{Colors.ENDC} ").strip()
-            if p_input:
-                profile_ids = [p.strip() for p in p_input.split(",") if p.strip()]
+
+            def validate_profiles(text: str) -> bool:
+                ids = [p.strip() for p in text.split(",") if p.strip()]
+                if not ids: return False
+                return all(validate_profile_id(pid, log_errors=False) for pid in ids)
+
+            p_input = get_valid_input(
+                f"{Colors.BOLD}Enter Control D Profile ID:{Colors.ENDC} ",
+                validate_profiles,
+                "Invalid Profile ID(s). Must be alphanumeric (max 64 chars)."
+            )
+            profile_ids = [p.strip() for p in p_input.split(",") if p.strip()]
 
         if not TOKEN:
             print(f"{Colors.CYAN}ℹ API Token is missing.{Colors.ENDC}")
             print(f"{Colors.CYAN}  You can generate one at: https://controld.com/account/manage-account{Colors.ENDC}")
-            import getpass
-            t_input = getpass.getpass(f"{Colors.BOLD}Enter Control D API Token:{Colors.ENDC} ").strip()
-            if t_input:
-                TOKEN = t_input
+
+            def validate_token(text: str) -> bool:
+                # Basic sanity check
+                return len(text) > 8
+
+            t_input = get_valid_input(
+                f"{Colors.BOLD}Enter Control D API Token:{Colors.ENDC} ",
+                validate_token,
+                "Token seems too short. Please check your API token.",
+                is_password=True
+            )
+            TOKEN = t_input
 
     if not profile_ids and not args.dry_run:
         log.error("PROFILE missing and --dry-run not set. Provide --profiles or set PROFILE env.")
@@ -925,7 +964,7 @@
                 })
                 continue
 
             log.info("Starting sync for profile %s", profile_id)
             status = sync_profile(
                 profile_id,
                 folder_urls,
@@ -1015,11 +1054,11 @@
         status_color = Colors.GREEN if res['success'] else Colors.FAIL
 
         print(
             f"{res['profile']:<{profile_col_width}} | "
             f"{res['folders']:>10} | "
             f"{res['rules']:>10,} | "
             f"{res['duration']:>9.1f}s | "
             f"{status_color}{res['status_label']:<15}{Colors.ENDC}"
         )
         total_folders += res['folders']
         total_rules += res['rules']
@@ -1045,12 +1084,12 @@
     total_status_color = Colors.GREEN if all_success else Colors.FAIL
 
     print(
         f"{Colors.BOLD}"
         f"{'TOTAL':<{profile_col_width}} | "
         f"{total_folders:>10} | "
         f"{total_rules:>10,} | "
         f"{total_duration:>9.1f}s | "
         f"{total_status_color}{total_status_text:<15}{Colors.ENDC}"
     )
     print("=" * table_width + "\n")
 

test_main.py

@@ -65,7 +65,7 @@
 
 # Case 3: push_rules updates data dictionary with pre-calculated batch keys correctly
 def test_push_rules_updates_data_with_batch_keys(monkeypatch):
     m = reload_main_with_env(monkeypatch)
     mock_client = MagicMock()
     mock_post_form = MagicMock()
     monkeypatch.setattr(m, "_api_post_form", mock_post_form)
@@ -255,7 +255,7 @@
     assert "Authentication Failed" in args
 
 def test_check_api_access_403(monkeypatch):
     m = reload_main_with_env(monkeypatch)
     mock_client = MagicMock()
 
     # Simulate 403 response
@@ -272,7 +272,7 @@
     assert "Access Denied" in str(mock_log.critical.call_args)
 
 def test_check_api_access_404(monkeypatch):
     m = reload_main_with_env(monkeypatch)
     mock_client = MagicMock()
 
     # Simulate 404 response
@@ -289,7 +289,7 @@
     assert "Profile Not Found" in str(mock_log.critical.call_args_list)
 
 def test_check_api_access_generic_http_error(monkeypatch):
     m = reload_main_with_env(monkeypatch)
     mock_client = MagicMock()
 
     # Simulate 500 response
@@ -317,5 +317,58 @@
     monkeypatch.setattr(m, "log", mock_log)
 
     assert m.check_api_access(mock_client, "profile") is False
     assert mock_log.error.called
     assert "Network failure" in str(mock_log.error.call_args)
+
+# Case 8: validate_profile_id respects log_errors flag
+def test_validate_profile_id_log_errors(monkeypatch):
+    m = reload_main_with_env(monkeypatch)
+    mock_log = MagicMock()
+    monkeypatch.setattr(m, "log", mock_log)
+
+    # Invalid ID with logging enabled (default)
+    assert m.validate_profile_id("invalid spaces") is False
+    assert mock_log.error.called
+
+    mock_log.reset_mock()
+
+    # Invalid ID with logging disabled
+    assert m.validate_profile_id("invalid spaces", log_errors=False) is False
+    assert not mock_log.error.called
+
+# Case 9: get_valid_input retries on invalid input and returns valid input
+def test_get_valid_input_retry(monkeypatch, capsys):
+    m = reload_main_with_env(monkeypatch)
+
+    # Mock input to return invalid first, then valid
+    # First call: empty string -> "Value cannot be empty"
+    # Second call: "invalid" -> Validator fails -> Error message
+    # Third call: "valid" -> Validator passes
+    input_mock = MagicMock(side_effect=["", "invalid", "valid"])
+    monkeypatch.setattr('builtins.input', input_mock)
+
+    validator = lambda x: x == "valid"
+
+    result = m.get_valid_input("Prompt: ", validator, "Error message")
+
+    assert result == "valid"
+    assert input_mock.call_count == 3
+
+    # Check output for error messages
+    captured = capsys.readouterr()
+    assert "Value cannot be empty" in captured.out
+    assert "Error message" in captured.out
+
+# Case 10: get_valid_input works with getpass
+def test_get_valid_input_password(monkeypatch):
+    m = reload_main_with_env(monkeypatch)
+
+    getpass_mock = MagicMock(return_value="secret")
+    monkeypatch.setattr('getpass.getpass', getpass_mock)
+
+    validator = lambda x: True
+
+    result = m.get_valid_input("Password: ", validator, "Error", is_password=True)
+
+    assert result == "secret"
+    getpass_mock.assert_called_once()