Skip to content
Merged
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
2 changes: 1 addition & 1 deletion benchmark_retry_jitter.py
Original file line number Diff line number Diff line change
Expand Up @@ -95,7 +95,7 @@ def main():

print(f" First retry window: {min_time:.2f}s to {max_time:.2f}s (spread: {max_time - min_time:.2f}s)")
print(f" Average first retry: {avg_time:.2f}s")
print(f" Retries distributed over time → reduced peak load on server")
print(" Retries distributed over time → reduced peak load on server")
print()

# Calculate approximate load reduction based on bucketed concurrency
Expand Down
209 changes: 50 additions & 159 deletions main.py
Original file line number Diff line number Diff line change
Expand Up @@ -271,196 +271,84 @@
return safe


def print_plan_details(plan_entry: Dict[str, Any]) -> None:
"""Pretty-print the folder-level breakdown during a dry-run."""
profile = sanitize_for_log(plan_entry.get("profile", "unknown"))
folders = plan_entry.get("folders", [])

if USE_COLORS:
print(f"\n{Colors.HEADER}📝 Plan Details for {profile}:{Colors.ENDC}")
else:
print(f"\nPlan Details for {profile}:")

if not folders:
if USE_COLORS:
print(f" {Colors.WARNING}No folders to sync.{Colors.ENDC}")
else:
print(" No folders to sync.")
return

# Calculate max width for alignment
max_name_len = max(
# Use the same default ("Unknown") as when printing, so alignment is accurate
(len(sanitize_for_log(f.get("name", "Unknown"))) for f in folders),
default=0,
)
max_rules_len = max((len(f"{f.get('rules', 0):,}") for f in folders), default=0)

for folder in sorted(folders, key=lambda f: f.get("name", "Unknown")):
name = sanitize_for_log(folder.get("name", "Unknown"))
rules_count = folder.get("rules", 0)
formatted_rules = f"{rules_count:,}"

# Determine action (Block/Allow)
action_text = ""
action_color = ""
action_label = ""

# Check for multiple rule groups first
if "rule_groups" in folder and folder["rule_groups"]:
actions = {rg.get("action") for rg in folder["rule_groups"]}
if len(actions) > 1:
action_label = "Mixed"
action_color = Colors.WARNING
action_text = f"({action_color}⚠️ {action_label}{Colors.ENDC})" if USE_COLORS else f"[{action_label}]"
else:
# All groups have same action
action_val = list(actions)[0]
if action_val == 0:
action_label = "Block"
action_color = Colors.FAIL
action_text = f"({action_color}⛔ {action_label}{Colors.ENDC})" if USE_COLORS else f"[{action_label}]"
elif action_val == 1:
action_label = "Allow"
action_color = Colors.GREEN
action_text = f"({action_color}✅ {action_label}{Colors.ENDC})" if USE_COLORS else f"[{action_label}]"

# Fallback to single action if not set
if not action_text and "action" in folder:
action_val = folder["action"]
if action_val == 0:
action_label = "Block"
action_color = Colors.FAIL
action_text = f"({action_color}⛔ {action_label}{Colors.ENDC})" if USE_COLORS else f"[{action_label}]"
elif action_val == 1:
action_label = "Allow"
action_color = Colors.GREEN
action_text = f"({action_color}✅ {action_label}{Colors.ENDC})" if USE_COLORS else f"[{action_label}]"

Comment on lines +304 to +339

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

The logic to determine the action text for 'Block' and 'Allow' is duplicated for handling rule_groups and the fallback action. This can be refactored to reduce duplication and improve maintainability by using a more data-driven approach.

By consolidating the logic to determine action_val first and then using a map to generate the display text, you can make the code more concise and easier to extend in the future.

        # Determine action (Block/Allow)
        action_text = ""
        action_color = ""
        action_label = ""

        action_map = {
            0: ("Block", Colors.FAIL, "⛔"),
            1: ("Allow", Colors.GREEN, "✅"),
        }

        action_val = None
        # Check for multiple rule groups first
        if "rule_groups" in folder and folder["rule_groups"]:
            actions = {rg.get("action") for rg in folder["rule_groups"]}
            if len(actions) > 1:
                action_label = "Mixed"
                action_color = Colors.WARNING
                action_text = f"({action_color}⚠️  {action_label}{Colors.ENDC})" if USE_COLORS else f"[{action_label}]"
            elif actions:
                action_val = list(actions)[0]
        # Fallback to single action if not set
        elif "action" in folder:
            action_val = folder.get("action")

        if action_val in action_map:
            action_label, action_color, icon = action_map[action_val]
            action_text = f"({action_color}{icon} {action_label}{Colors.ENDC})" if USE_COLORS else f"[{action_label}]"

if USE_COLORS:
print(
f" • {Colors.BOLD}{name:<{max_name_len}}{Colors.ENDC} : {formatted_rules:>{max_rules_len}} rules"
f" • {Colors.BOLD}{name:<{max_name_len}}{Colors.ENDC} : {formatted_rules:>{max_rules_len}} rules {action_text}"
)
else:
print(
f" - {name:<{max_name_len}} : {formatted_rules:>{max_rules_len}} rules"
f" - {name:<{max_name_len}} : {formatted_rules:>{max_rules_len}} rules {action_text}"
)

print("")

Check notice on line 349 in main.py

View check run for this annotation

codefactor.io / CodeFactor

main.py#L274-L349

Complex Method


def print_summary_table(results: List[Dict[str, Any]], dry_run: bool) -> None:
"""Prints a nicely formatted summary table."""
# Determine the width for the Profile ID column (min 25)
max_profile_len = max((len(r["profile"]) for r in results), default=25)
profile_col_width = max(25, max_profile_len)

# Calculate widths
col_widths = {
"profile": profile_col_width,
"folders": 10,
"rules": 10,
"duration": 10,
"status": 15,
}

if USE_COLORS:
# Unicode Box Drawing
chars = {
"tl": "┌", "tm": "┬", "tr": "┐",
"bl": "└", "bm": "┴", "br": "┘",
"ml": "├", "mm": "┼", "mr": "┤",
"v": "│", "h": "─",
}
else:
# ASCII Fallback
chars = {
"tl": "+", "tm": "+", "tr": "+",
"bl": "+", "bm": "+", "br": "+",
"ml": "+", "mm": "+", "mr": "+",
"v": "|", "h": "-",
}

def _print_separator(left, mid, right):
segments = [chars["h"] * (width + 2) for width in col_widths.values()]
print(f"{chars[left]}{chars[mid].join(segments)}{chars[right]}")

def _print_row(profile, folders, rules, duration, status, is_header=False):
v = chars["v"]

# 1. Pad raw strings first (so padding is calculated on visible chars)
p_val = f"{profile:<{col_widths['profile']}}"
f_val = f"{folders:>{col_widths['folders']}}"
r_val = f"{rules:>{col_widths['rules']}}"
d_val = f"{duration:>{col_widths['duration']}}"
s_val = f"{status:<{col_widths['status']}}"

# 2. Wrap in color codes if needed
if is_header and USE_COLORS:
p_val = f"{Colors.BOLD}{p_val}{Colors.ENDC}"
f_val = f"{Colors.BOLD}{f_val}{Colors.ENDC}"
r_val = f"{Colors.BOLD}{r_val}{Colors.ENDC}"
d_val = f"{Colors.BOLD}{d_val}{Colors.ENDC}"
s_val = f"{Colors.BOLD}{s_val}{Colors.ENDC}"

print(
f"{v} {p_val} {v} {f_val} {v} {r_val} {v} {d_val} {v} {s_val} {v}"
)

title_text = "DRY RUN SUMMARY" if dry_run else "SYNC SUMMARY"
title_color = Colors.CYAN if dry_run else Colors.HEADER

total_width = (
1 + (col_widths["profile"] + 2) + 1 +
(col_widths["folders"] + 2) + 1 +
(col_widths["rules"] + 2) + 1 +
(col_widths["duration"] + 2) + 1 +
(col_widths["status"] + 2) + 1
)

print("\n" + (f"{title_color}{title_text:^{total_width}}{Colors.ENDC}" if USE_COLORS else f"{title_text:^{total_width}}"))

_print_separator("tl", "tm", "tr")
# Header row - pad manually then print
_print_row("Profile ID", "Folders", "Rules", "Duration", "Status", is_header=True)
_print_separator("ml", "mm", "mr")

total_folders = 0
total_rules = 0
total_duration = 0.0
success_count = 0

for res in results:
# Profile
p_val = f"{res['profile']:<{col_widths['profile']}}"

# Folders
f_val = f"{res['folders']:>{col_widths['folders']}}"

# Rules
r_val = f"{res['rules']:>{col_widths['rules']},}"

# Duration
d_val = f"{res['duration']:>{col_widths['duration']-1}.1f}s"

# Status
status_label = res["status_label"]
s_val_raw = f"{status_label:<{col_widths['status']}}"
if USE_COLORS:
status_color = Colors.GREEN if res["success"] else Colors.FAIL
s_val = f"{status_color}{s_val_raw}{Colors.ENDC}"
else:
s_val = s_val_raw

# Delegate the actual row printing to the shared helper to avoid
# duplicating table border/spacing logic here.
_print_row(p_val, f_val, r_val, d_val, s_val)

total_folders += res["folders"]
total_rules += res["rules"]
total_duration += res["duration"]
if res["success"]:
success_count += 1

_print_separator("ml", "mm", "mr")

# Total Row
total = len(results)
all_success = success_count == total

if dry_run:
total_status_text = "✅ Ready" if all_success else "❌ Errors"
else:
total_status_text = "✅ All Good" if all_success else "❌ Errors"

p_val = f"{'TOTAL':<{col_widths['profile']}}"
if USE_COLORS:
p_val = f"{Colors.BOLD}{p_val}{Colors.ENDC}"

f_val = f"{total_folders:>{col_widths['folders']}}"
r_val = f"{total_rules:>{col_widths['rules']},}"
d_val = f"{total_duration:>{col_widths['duration']-1}.1f}s"

s_val_raw = f"{total_status_text:<{col_widths['status']}}"
if USE_COLORS:
status_color = Colors.GREEN if all_success else Colors.FAIL
s_val = f"{status_color}{s_val_raw}{Colors.ENDC}"
else:
s_val = s_val_raw

print(
f"{chars['v']} {p_val} "
f"{chars['v']} {f_val} "
f"{chars['v']} {r_val} "
f"{chars['v']} {d_val} "
f"{chars['v']} {s_val} {chars['v']}"
)

_print_separator("bl", "bm", "br")


def _get_progress_bar_width() -> int:
Expand Down Expand Up @@ -1486,7 +1374,7 @@

_cache_stats["misses"] += 1

except httpx.HTTPStatusError as e:
except httpx.HTTPStatusError:
# Re-raise with original exception (don't catch and re-raise)
raise

Expand Down Expand Up @@ -2458,21 +2346,24 @@
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}"
def print_line(left_char, mid_char, right_char):
return f"{Colors.BOLD}{left_char}{mid_char.join('─' * (x+2) for x in w)}{right_char}{Colors.ENDC}"

def print_row(cols):
return f"{Colors.BOLD}│{Colors.ENDC} {cols[0]:<{w[0]}} {Colors.BOLD}│{Colors.ENDC} {cols[1]:>{w[1]}} {Colors.BOLD}│{Colors.ENDC} {cols[2]:>{w[2]}} {Colors.BOLD}│{Colors.ENDC} {cols[3]:>{w[3]}} {Colors.BOLD}│{Colors.ENDC} {cols[4]:<{w[4]}} {Colors.BOLD}│{Colors.ENDC}"

print(f"\n{line('┌', '─', '┐')}")
print(f"\n{print_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("├", "┼", "┤"))
print(f"{print_line('├', '┬', '┤')}\n{print_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(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(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")
print(f"{print_line('├', '┼', '┤')}\n{print_row(['TOTAL', str(t_f), f'{t_r:,}', f'{t_d:.1f}s', f'{t_col}{t_status}{Colors.ENDC}'])}")
print(f"{print_line('└', '┴', '┘')}\n")


def print_success_message(profile_ids: List[str]) -> None:
Expand Down
2 changes: 1 addition & 1 deletion test_main.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
import importlib
import os
import sys
from unittest.mock import MagicMock, call, patch
from unittest.mock import MagicMock, patch

import httpx
import pytest
Expand Down
1 change: 0 additions & 1 deletion tests/test_bootstrapping.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,4 @@
import os
import sys
from unittest.mock import MagicMock, patch

import pytest
Expand Down
1 change: 0 additions & 1 deletion tests/test_content_type.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,6 @@
from unittest.mock import patch, MagicMock
import sys
import os
import json

# Add root to path to import main
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
Expand Down
1 change: 0 additions & 1 deletion tests/test_disk_cache.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,6 @@
"""
import json
import os
import platform
import tempfile
import time
import unittest
Expand Down
2 changes: 1 addition & 1 deletion tests/test_env_permissions.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,7 @@

import os
import stat
from unittest.mock import MagicMock, patch, call
from unittest.mock import MagicMock, patch


# Set TOKEN before importing main to avoid issues with load_dotenv()
Expand Down
3 changes: 1 addition & 2 deletions tests/test_fix_env.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,6 @@
import os
import stat
import pytest
from unittest.mock import MagicMock, patch
from unittest.mock import patch
import fix_env

def test_fix_env_skips_symlink(tmp_path):
Expand Down
1 change: 0 additions & 1 deletion tests/test_folder_validation.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,5 @@
from unittest.mock import MagicMock

import pytest

import main

Expand Down
3 changes: 1 addition & 2 deletions tests/test_hostname_validation.py
Original file line number Diff line number Diff line change
@@ -1,8 +1,7 @@

import socket
from unittest.mock import MagicMock, patch
from unittest.mock import patch

import pytest
import main

def test_validate_hostname_caching():
Expand Down
1 change: 0 additions & 1 deletion tests/test_performance_regression.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,6 @@
import sys
import os

import pytest
import httpx

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
Expand Down
17 changes: 10 additions & 7 deletions tests/test_plan_details.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,6 @@

import importlib
import os
import sys
from unittest.mock import patch


Expand All @@ -18,8 +17,9 @@ def test_print_plan_details_no_colors(capsys):
plan_entry = {
"profile": "test_profile",
"folders": [
{"name": "Folder B", "rules": 5},
{"name": "Folder A", "rules": 10},
{"name": "Folder B", "rules": 5, "action": 0},
{"name": "Folder A", "rules": 10, "action": 1},
{"name": "Folder C", "rules": 3, "rule_groups": [{"action": 0}, {"action": 1}]},
],
}
m.print_plan_details(plan_entry)
Expand All @@ -29,10 +29,12 @@ def test_print_plan_details_no_colors(capsys):

assert "Plan Details for test_profile:" in output
# Match exact output including alignment spaces
assert " - Folder A : 10 rules" in output
assert " - Folder B : 5 rules" in output
# Verify alphabetical ordering (A before B)
assert " - Folder A : 10 rules [Allow]" in output
assert " - Folder B : 5 rules [Block]" in output
assert " - Folder C : 3 rules [Mixed]" in output
# Verify alphabetical ordering (A before B before C)
assert output.index("Folder A") < output.index("Folder B")
assert output.index("Folder B") < output.index("Folder C")


def test_print_plan_details_empty_folders(capsys):
Expand Down Expand Up @@ -65,7 +67,7 @@ def test_print_plan_details_with_colors(capsys):

plan_entry = {
"profile": "test_profile",
"folders": [{"name": "Folder A", "rules": 10}],
"folders": [{"name": "Folder A", "rules": 10, "action": 1}],
}
m.print_plan_details(plan_entry)

Expand All @@ -75,3 +77,4 @@ def test_print_plan_details_with_colors(capsys):
assert "📝 Plan Details for test_profile:" in output
assert "Folder A" in output
assert "10 rules" in output
assert "✅ Allow" in output
3 changes: 1 addition & 2 deletions tests/test_push_rules_perf.py
Original file line number Diff line number Diff line change
@@ -1,12 +1,11 @@

import unittest
from unittest.mock import MagicMock, patch, ANY
from unittest.mock import MagicMock, patch
import sys
import os

# Add root to path to import main
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
import main

class TestPushRulesPerf(unittest.TestCase):
def setUp(self):
Expand Down
Loading
Loading