Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
93 changes: 54 additions & 39 deletions main.py
Original file line number Diff line number Diff line change
Expand Up @@ -1478,53 +1478,68 @@ def main():
folder_urls = args.folder_url if args.folder_url else DEFAULT_FOLDER_URLS

# Interactive prompts for missing config
if not args.dry_run and sys.stdin.isatty():
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 (or just paste the URL).{Colors.ENDC}"
)
try:
if not args.dry_run and sys.stdin.isatty():
if not profile_ids:
print(f"{Colors.CYAN}ℹ Profile ID is missing.{Colors.ENDC}")
print(
f"{Colors.CYAN} Find it in your profile URL at "
f"https://controld.com/dashboard/profiles{Colors.ENDC}"
)
print(
f"{Colors.CYAN} (You can also paste the full profile URL).{Colors.ENDC}"
)

Check warning

Code scanning / Pylint (reported by Codacy)

Missing function docstring

Missing function docstring
def validate_profile_input(value: str) -> bool:
"""Validate comma-separated profile IDs or URLs."""
ids = [extract_profile_id(p) for p in value.split(",") if p.strip()]
return bool(ids) and all(
validate_profile_id(pid, log_errors=False) for pid in ids
)

def validate_profile_input(value: str) -> bool:
ids = [extract_profile_id(p) for p in value.split(",") if p.strip()]
return bool(ids) and all(
validate_profile_id(pid, log_errors=False) for pid in ids
p_input = get_validated_input(
f"{Colors.BOLD}Enter Control D Profile ID:{Colors.ENDC} ",
validate_profile_input,

Check warning

Code scanning / Pylint (reported by Codacy)

Line too long (127/100)

Line too long (127/100)
"Invalid ID(s) or URL(s). Must be a valid Profile ID or "
"a Control D Profile URL. Comma-separate for multiple.",
)
profile_ids = [
extract_profile_id(p) for p in p_input.split(",") if p.strip()
]

p_input = get_validated_input(
f"{Colors.BOLD}Enter Control D Profile ID:{Colors.ENDC} ",
validate_profile_input,
"Invalid ID(s) or URL(s). Must be a valid Profile ID or a Control D Profile URL. Comma-separate for multiple.",
)
profile_ids = [
extract_profile_id(p) 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}"
)
if not TOKEN:
print(f"{Colors.CYAN}ℹ API Token is missing.{Colors.ENDC}")
print(

Check warning

Code scanning / Pylint (reported by Codacy)

Line too long (115/100)

Line too long (115/100)
f"{Colors.CYAN} You can generate one at: https://controld.com/account/manage-account{Colors.ENDC}"
)

t_input = get_validated_input(
f"{Colors.BOLD}Enter Control D API Token:{Colors.ENDC} ",
lambda x: len(x) > 8,
"Token seems too short. Please check your API token.",
is_password=True,
t_input = get_validated_input(
f"{Colors.BOLD}Enter Control D API Token:{Colors.ENDC} ",
lambda x: len(x) > 8,
"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."
)
TOKEN = t_input
exit(1)

if not profile_ids and not args.dry_run:
log.error(
"PROFILE missing and --dry-run not set. Provide --profiles or set PROFILE env."
)
exit(1)
if not TOKEN and not args.dry_run:
log.error(

Check warning

Code scanning / Prospector (reported by Codacy)

Consider using sys.exit() (consider-using-sys-exit)

Consider using sys.exit() (consider-using-sys-exit)

Check warning

Code scanning / Pylintpython3 (reported by Codacy)

Consider using sys.exit()

Consider using sys.exit()
"TOKEN missing and --dry-run not set. Set TOKEN env for live sync."
)
exit(1)

if not TOKEN and not args.dry_run:
log.error("TOKEN missing and --dry-run not set. Set TOKEN env for live sync.")
exit(1)
warm_up_cache(folder_urls)

warm_up_cache(folder_urls)
except (KeyboardInterrupt, EOFError):
sys.stderr.write(
f"\n{Colors.WARNING}⚠️ Sync cancelled by user.{Colors.ENDC}\n"
)
exit(130)

plan: List[Dict[str, Any]] = []
success_count = 0
Expand Down
59 changes: 57 additions & 2 deletions test_main.py
Original file line number Diff line number Diff line change
Expand Up @@ -244,10 +244,65 @@ def test_interactive_prompts_show_hints(monkeypatch, capsys):
captured = capsys.readouterr()
stdout = captured.out

assert "You can find this in the URL of your profile" in stdout
# Verify both hint URLs are displayed
assert "https://controld.com/dashboard/profiles" in stdout
assert "https://controld.com/account/manage-account" in stdout


def test_keyboard_interrupt_during_setup(monkeypatch, capsys):
"""Test that KeyboardInterrupt during setup phase exits gracefully with code 130."""
# Ensure environment is clean
monkeypatch.delenv("PROFILE", raising=False)
monkeypatch.delenv("TOKEN", raising=False)

# Prevent dotenv from loading .env file
import dotenv

monkeypatch.setattr(dotenv, "load_dotenv", lambda: None)

# Reload main with isatty=True to trigger interactive mode
m = reload_main_with_env(monkeypatch, isatty=True)

# Mock sys.stdin.isatty to return True
monkeypatch.setattr("sys.stdin.isatty", lambda: True)

# Mock input to raise KeyboardInterrupt (simulating Ctrl+C)
monkeypatch.setattr(
"builtins.input", MagicMock(side_effect=KeyboardInterrupt())
)

# Mock parse_args
mock_args = MagicMock()
mock_args.profiles = None
mock_args.folder_url = None
mock_args.dry_run = False
mock_args.no_delete = False
mock_args.plan_json = None
monkeypatch.setattr(m, "parse_args", lambda: mock_args)

# Mock exit to capture the exit code
exit_code = None

def mock_exit(code):
nonlocal exit_code
exit_code = code
raise SystemExit(code)

monkeypatch.setattr("builtins.exit", mock_exit)

# Run main and expect SystemExit
with pytest.raises(SystemExit):
m.main()

# Verify exit code is 130 (standard for Ctrl+C)
assert exit_code == 130

# Verify user-friendly cancellation message
captured = capsys.readouterr()
stderr = captured.err
assert "Sync cancelled by user" in stderr


# Case 7: verify_access_and_get_folders handles success and errors correctly
def test_verify_access_and_get_folders_success(monkeypatch):
m = reload_main_with_env(monkeypatch)
Expand Down Expand Up @@ -451,7 +506,7 @@ def test_interactive_input_extracts_id(monkeypatch, capsys):

# Verify prompt text update
captured = capsys.readouterr()
assert "(or just paste the URL)" in captured.out
assert "(You can also paste the full profile URL)" in captured.out


# Case 10: validate_profile_id respects log_errors flag
Expand Down
20 changes: 13 additions & 7 deletions tests/test_plan_details.py
Original file line number Diff line number Diff line change
@@ -1,21 +1,25 @@
"""Tests for the print_plan_details dry-run output function."""

import sys
from unittest.mock import patch

import main


# Helper to get current main module (handles reloading by other tests)
def get_main():

Check warning

Code scanning / Pylintpython3 (reported by Codacy)

Missing function or method docstring

Missing function or method docstring

Check warning

Code scanning / Pylint (reported by Codacy)

Missing function docstring

Missing function docstring
return sys.modules.get("main", main)
def test_print_plan_details_no_colors(capsys):
"""Test print_plan_details output when colors are disabled."""
with patch("main.USE_COLORS", False):
current_main = get_main()
with patch.object(current_main, "USE_COLORS", False):
plan_entry = {
"profile": "test_profile",
"folders": [
{"name": "Folder B", "rules": 5},
{"name": "Folder A", "rules": 10},
],
}
main.print_plan_details(plan_entry)
current_main.print_plan_details(plan_entry)

captured = capsys.readouterr()
output = captured.out
Expand All @@ -29,9 +33,10 @@ def test_print_plan_details_no_colors(capsys):

def test_print_plan_details_empty_folders(capsys):
"""Test print_plan_details with no folders."""
with patch("main.USE_COLORS", False):
current_main = get_main()
with patch.object(current_main, "USE_COLORS", False):
plan_entry = {"profile": "test_profile", "folders": []}
main.print_plan_details(plan_entry)
current_main.print_plan_details(plan_entry)

captured = capsys.readouterr()
output = captured.out
Expand All @@ -42,12 +47,13 @@ def test_print_plan_details_empty_folders(capsys):

def test_print_plan_details_with_colors(capsys):
"""Test print_plan_details output when colors are enabled."""
with patch("main.USE_COLORS", True):
current_main = get_main()
with patch.object(current_main, "USE_COLORS", True):
plan_entry = {
"profile": "test_profile",
"folders": [{"name": "Folder A", "rules": 10}],
}
main.print_plan_details(plan_entry)
current_main.print_plan_details(plan_entry)

captured = capsys.readouterr()
output = captured.out
Expand Down
Loading