Skip to content

⚡ Bolt: Combine API calls and optimize regex for faster startup #170

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
abhimehro wants to merge 2 commits into from
Closed

⚡ Bolt: Combine API calls and optimize regex for faster startup #170

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
7961a8f
⚡ Bolt: Combine API calls and optimize regex for faster startup
google-labs-jules[bot] Feb 7, 2026
a57a7bc
⚡ Bolt: Combine API calls and optimize regex for faster startup
google-labs-jules[bot] Feb 7, 2026
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
6 changes: 4 additions & 2 deletions WARP.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -91,7 +91,8 @@ The entire tool lives in `main.py` and is structured into clear phases:
- `sanitize_for_log()` – Redacts `TOKEN` values from any log messages.

4. **Control D API helpers**
- `list_existing_folders()` – Returns a `{folder_name -> folder_id}` mapping for a profile.
- `verify_access_and_get_folders()` – Combines the API access check and fetching existing folders into a single request. Returns `{folder_name -> folder_id}` on success.
- `list_existing_folders()` – Helper that returns a `{folder_name -> folder_id}` mapping (used as fallback).
- `get_all_existing_rules()` – Collects all existing rule PKs from both the root and each folder, using a `ThreadPoolExecutor` to parallelize per-folder fetches while accumulating into a shared `set` guarded by a lock.
- `delete_folder()` – Deletes a folder by ID with error-logged failures.
- `create_folder()` – Creates a folder and tries to read its ID directly from the response; if that fails, it polls `GET /groups` with increasing waits (using `FOLDER_CREATION_DELAY`) until the new folder appears.
Expand All @@ -111,7 +112,8 @@ The entire tool lives in `main.py` and is structured into clear phases:
2. Builds a `plan_entry` summarizing folder names, rule counts, and per-action breakdown (for `rule_groups`), appending it to the shared `plan_accumulator`.
3. If `dry_run=True`, stops here after logging a summary message.
4. Otherwise, reuses a single `_api_client()` instance to:
- List and optionally delete existing folders with matching names (`--no-delete` skips this step).
- Verify access and list existing folders in one request (`verify_access_and_get_folders`).
- Optionally delete existing folders with matching names (`--no-delete` skips this step).
- If any deletions occurred, waits ~60 seconds (`countdown_timer`) to let Control D fully process the removals.
- Build the global `existing_rules` set.
- Sequentially process each folder (executor with `max_workers=1` to avoid rate-limit and ordering issues), calling `_process_single_folder()` for each.
Expand Down
128 changes: 84 additions & 44 deletions main.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -289,6 +289,10 @@
FOLDER_CREATION_DELAY = 5 # <--- CHANGED: Increased from 2 to 5 for patience
MAX_RESPONSE_SIZE = 10 * 1024 * 1024 # 10MB limit

# Regex patterns (compiled for performance)
PROFILE_ID_PATTERN = re.compile(r"^[a-zA-Z0-9_-]+$")
RULE_PATTERN = re.compile(r"^[a-zA-Z0-9.\-_:*\/]+$")


# --------------------------------------------------------------------------- #
# 2. Clients
Expand Down Expand Up @@ -398,7 +402,7 @@


def is_valid_profile_id_format(profile_id: str) -> bool:
if not re.match(r"^[a-zA-Z0-9_-]+$", profile_id):
if not PROFILE_ID_PATTERN.match(profile_id):
return False
if len(profile_id) > 64:
return False
Expand All @@ -408,7 +412,7 @@
def validate_profile_id(profile_id: str, log_errors: bool = True) -> bool:
if not is_valid_profile_id_format(profile_id):
if log_errors:
if not re.match(r"^[a-zA-Z0-9_-]+$", profile_id):
if not PROFILE_ID_PATTERN.match(profile_id):
log.error("Invalid profile ID format (contains unsafe characters)")
elif len(profile_id) > 64:
log.error("Invalid profile ID length (max 64 chars)")
Expand All @@ -427,7 +431,7 @@

# Strict whitelist to prevent injection
# ^[a-zA-Z0-9.\-_:*\/]+$
if not re.match(r"^[a-zA-Z0-9.\-_:*\/]+$", rule):
if not RULE_PATTERN.match(rule):
return False

return True
Expand Down Expand Up @@ -516,7 +520,7 @@
return response
except (httpx.HTTPError, httpx.TimeoutException) as e:
if attempt == max_retries - 1:
if hasattr(e, "response") and e.response is not None:

Check notice

Code scanning / Pylintpython3 (reported by Codacy)

Use lazy % formatting in logging functions Note

Use lazy % formatting in logging functions
log.debug(f"Response content: {sanitize_for_log(e.response.text)}")
raise
wait_time = delay * (2**attempt)
Expand Down Expand Up @@ -583,44 +587,6 @@
return _cache[url]


def check_api_access(client: httpx.Client, profile_id: str) -> bool:
"""
Verifies API access and Profile existence before starting heavy work.
Returns True if access is good, False otherwise (with helpful logs).
"""
url = f"{API_BASE}/{profile_id}/groups"
try:
# We use a raw request here to avoid the automatic retries of _retry_request
# for auth errors, which are permanent.
resp = client.get(url)
resp.raise_for_status()
return True
except httpx.HTTPStatusError as e:
code = e.response.status_code
if code == 401:
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}")
return False
except httpx.RequestError as e:
log.error(f"Network Error during access check: {e}")
return False


def list_existing_folders(client: httpx.Client, profile_id: str) -> Dict[str, str]:
Expand All @@ -629,14 +595,87 @@
folders = data.get("body", {}).get("groups", [])
return {
f["group"].strip(): f["PK"]
for f in folders

Check warning

Code scanning / Pylint (reported by Codacy)

Variable name "e" doesn't conform to snake_case naming style Warning

Variable name "e" doesn't conform to snake_case naming style

Check warning

Code scanning / Pylintpython3 (reported by Codacy)

Variable name "e" doesn't conform to snake_case naming style Warning

Variable name "e" doesn't conform to snake_case naming style
if f.get("group") and f.get("PK")
}
except (httpx.HTTPError, KeyError) as e:

Check warning

Code scanning / Prospector (reported by Codacy)

Use lazy % formatting in logging functions (logging-fstring-interpolation) Warning

Use lazy % formatting in logging functions (logging-fstring-interpolation)

Check notice

Code scanning / Pylintpython3 (reported by Codacy)

Use lazy % formatting in logging functions Note

Use lazy % formatting in logging functions
log.error(f"Failed to list existing folders: {sanitize_for_log(e)}")
return {}

Check warning

Code scanning / Prospector (reported by Codacy)

Use lazy % formatting in logging functions (logging-fstring-interpolation) Warning

Use lazy % formatting in logging functions (logging-fstring-interpolation)

Check notice

Code scanning / Pylintpython3 (reported by Codacy)

Use lazy % formatting in logging functions Note

Use lazy % formatting in logging functions

Check warning

Code scanning / Pylint (reported by Codacy)

Line too long (127/100) Warning

Line too long (127/100)
def verify_access_and_get_folders(
client: httpx.Client, profile_id: str

Check warning

Code scanning / Pylint (reported by Codacy)

Wrong hanging indentation before block (add 4 spaces). Warning

Wrong hanging indentation before block (add 4 spaces).
) -> Optional[Dict[str, str]]:

Check warning

Code scanning / Prospector (reported by Codacy)

Use lazy % formatting in logging functions (logging-fstring-interpolation) Warning

Use lazy % formatting in logging functions (logging-fstring-interpolation)

Check notice

Code scanning / Pylintpython3 (reported by Codacy)

Use lazy % formatting in logging functions Note

Use lazy % formatting in logging functions
"""
Combines access check and fetching existing folders into a single request.
Returns:
Dict of {folder_name: folder_id} if successful.

Check warning

Code scanning / Prospector (reported by Codacy)

Use lazy % formatting in logging functions (logging-fstring-interpolation) Warning

Use lazy % formatting in logging functions (logging-fstring-interpolation)

Check notice

Code scanning / Pylintpython3 (reported by Codacy)

Use lazy % formatting in logging functions Note

Use lazy % formatting in logging functions
None if access is denied or request fails after retries.
"""
url = f"{API_BASE}/{profile_id}/groups"

Check warning

Code scanning / Prospector (reported by Codacy)

Use lazy % formatting in logging functions (logging-fstring-interpolation) Warning

Use lazy % formatting in logging functions (logging-fstring-interpolation)

Check notice

Code scanning / Pylintpython3 (reported by Codacy)

Use lazy % formatting in logging functions Note

Use lazy % formatting in logging functions

Check warning

Code scanning / Pylint (reported by Codacy)

Line too long (111/100) Warning

Line too long (111/100)
for attempt in range(MAX_RETRIES):
try:
# Don't use _api_get because we need custom error handling for 4xx
resp = client.get(url)
resp.raise_for_status()

Check warning

Code scanning / Pylint (reported by Codacy)

Variable name "e" doesn't conform to snake_case naming style Warning

Variable name "e" doesn't conform to snake_case naming style

Check warning

Code scanning / Pylintpython3 (reported by Codacy)

Variable name "e" doesn't conform to snake_case naming style Warning

Variable name "e" doesn't conform to snake_case naming style

# Success! Parse and return
try:
data = resp.json()
folders = data.get("body", {}).get("groups", [])
return {
f["group"].strip(): f["PK"]
for f in folders
if f.get("group") and f.get("PK")
}
except (KeyError, ValueError) as e:

Check warning

Code scanning / Pylint (reported by Codacy)

Variable name "e" doesn't conform to snake_case naming style Warning

Variable name "e" doesn't conform to snake_case naming style

Check warning

Code scanning / Pylintpython3 (reported by Codacy)

Variable name "e" doesn't conform to snake_case naming style Warning

Variable name "e" doesn't conform to snake_case naming style
log.error(f"Failed to parse folders data: {e}")

Check warning

Code scanning / Prospector (reported by Codacy)

Use lazy % formatting in logging functions (logging-fstring-interpolation) Warning

Use lazy % formatting in logging functions (logging-fstring-interpolation)

Check notice

Code scanning / Pylintpython3 (reported by Codacy)

Use lazy % formatting in logging functions Note

Use lazy % formatting in logging functions
return None

except httpx.HTTPStatusError as e:
code = e.response.status_code
if code in (401, 403, 404):
if code == 401:
log.critical(
f"{Colors.FAIL}❌ Authentication Failed: The API Token is invalid.{Colors.ENDC}"

Check warning

Code scanning / Pylint (reported by Codacy)

Line too long (103/100) Warning

Line too long (103/100)

Check warning

Code scanning / Pylintpython3 (reported by Codacy)

Line too long (103/100) Warning

Line too long (103/100)
)
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}"
)
return None
Comment on lines 638 to 657
Copy link

@gemini-code-assist gemini-code-assist bot Feb 7, 2026

Choose a reason for hiding this comment

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

medium

This block for handling specific HTTP status codes (401, 403, 404) is quite large and contains detailed, user-facing log messages. To improve the readability and maintainability of verify_access_and_get_folders, consider extracting this logic into a separate helper function. This would make the main function's flow, which is focused on retries and data fetching, easier to follow.

You could define a new helper function like this elsewhere in the file:

def _log_api_access_error(code: int, profile_id: str) -> None:
    """Logs a critical, user-facing error for common API access issues."""
    if code == 401:
        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}"
        )
            if code in (401, 403, 404):
                _log_api_access_error(code, profile_id)
                return None


# For 5xx errors, retry
if attempt == MAX_RETRIES - 1:
log.error(f"API Request Failed ({code}): {e}")
return None

Comment on lines 636 to 663
Copy link

Copilot AI Feb 7, 2026

Choose a reason for hiding this comment

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

verify_access_and_get_folders() retries on any HTTPStatusError that isn’t 401/403/404, which includes non-retriable 4xx responses (e.g., 400/409). This can cause long backoffs for permanent client errors and contradicts the “For 5xx errors, retry” comment. Consider explicitly retrying only for 5xx (and possibly 429 with Retry-After), and for other 4xx log once and return None immediately.

Copilot uses AI. Check for mistakes.
except httpx.RequestError as e:
if attempt == MAX_RETRIES - 1:
log.error(f"Network Error: {e}")
return None

# Wait before retry (exponential backoff)
wait_time = RETRY_DELAY * (2**attempt)
log.warning(
f"Request failed (attempt {attempt + 1}/{MAX_RETRIES}). Retrying in {wait_time}s..."
)
Comment on lines +671 to +673
Copy link

Copilot AI Feb 7, 2026

Choose a reason for hiding this comment

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

The retry warning log (Request failed... Retrying in ...) doesn’t include the underlying exception/status code, which makes diagnosing repeated failures harder. Consider including the caught exception (sanitized) or at least the HTTP status code in this warning message.

Copilot uses AI. Check for mistakes.
time.sleep(wait_time)

return None
Copy link

@gemini-code-assist gemini-code-assist bot Feb 7, 2026

Choose a reason for hiding this comment

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

medium

This return None statement appears to be unreachable. The for loop above will always exit via a return statement within its body, either on success from the try block or on failure from the except blocks (especially on the last attempt). Removing this line will make the code cleaner and avoid confusion.



def get_all_existing_rules(
client: httpx.Client,
profile_id: str,
Expand Down Expand Up @@ -704,7 +743,7 @@
if not validate_folder_data(js, url):
raise KeyError(f"Invalid folder data from {sanitize_for_log(url)}")
return js

Check warning

Code scanning / Prospector (reported by Codacy)

Use lazy % formatting in logging functions (logging-fstring-interpolation) Warning

Use lazy % formatting in logging functions (logging-fstring-interpolation)

def warm_up_cache(urls: Sequence[str]) -> None:
urls = list(set(urls))
Expand Down Expand Up @@ -1142,11 +1181,12 @@
# Initial client for getting existing state AND processing folders
# Optimization: Reuse the same client session to keep TCP connections alive
with _api_client() as client:
# Check for API access problems first (401/403/404)
if not check_api_access(client, profile_id):
# Check for API access problems first (401/403/404) AND get existing folders in one go
existing_folders = verify_access_and_get_folders(client, profile_id)
if existing_folders is None:
# Access check failed (logged already)
return False

existing_folders = list_existing_folders(client, profile_id)
if not no_delete:
deletion_occurred = False
for folder_data in folder_data_list:
Expand Down
52 changes: 39 additions & 13 deletions test_main.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -248,16 +248,29 @@
assert "https://controld.com/account/manage-account" in stdout


# Case 7: check_api_access handles success and errors correctly
def test_check_api_access_success(monkeypatch):
# Case 7: verify_access_and_get_folders handles success and errors correctly
def test_verify_access_and_get_folders_success(monkeypatch):

Check warning

Code scanning / Pylint (reported by Codacy)

Missing function docstring Warning test

Missing function docstring

Check warning

Code scanning / Pylintpython3 (reported by Codacy)

Missing function or method docstring Warning test

Missing function or method docstring
m = reload_main_with_env(monkeypatch)
mock_client = MagicMock()

# Mock successful JSON response
mock_resp = MagicMock()
mock_resp.json.return_value = {
"body": {
"groups": [
{"group": "Folder A", "PK": "id_a"},
{"group": "Folder B", "PK": "id_b"},
]
}
}
mock_client.get.return_value = mock_resp
mock_client.get.return_value.raise_for_status.return_value = None

assert m.check_api_access(mock_client, "valid_profile") is True
folders = m.verify_access_and_get_folders(mock_client, "valid_profile")
assert folders == {"Folder A": "id_a", "Folder B": "id_b"}

Check notice

Code scanning / Bandit

Use of assert detected. The enclosed code will be removed when compiling to optimised byte code. Note test

Use of assert detected. The enclosed code will be removed when compiling to optimised byte code.

Check notice

Code scanning / Bandit (reported by Codacy)

Use of assert detected. The enclosed code will be removed when compiling to optimised byte code. Note test

Use of assert detected. The enclosed code will be removed when compiling to optimised byte code.


def test_check_api_access_401(monkeypatch):
def test_verify_access_and_get_folders_401(monkeypatch):

Check warning

Code scanning / Pylint (reported by Codacy)

Missing function docstring Warning test

Missing function docstring

Check warning

Code scanning / Pylintpython3 (reported by Codacy)

Missing function or method docstring Warning test

Missing function or method docstring
m = reload_main_with_env(monkeypatch)
mock_client = MagicMock()

Expand All @@ -273,14 +286,14 @@
mock_log = MagicMock()
monkeypatch.setattr(m, "log", mock_log)

assert m.check_api_access(mock_client, "invalid_token") is False
assert m.verify_access_and_get_folders(mock_client, "invalid_token") is None

Check notice

Code scanning / Bandit

Use of assert detected. The enclosed code will be removed when compiling to optimised byte code. Note test

Use of assert detected. The enclosed code will be removed when compiling to optimised byte code.

Check notice

Code scanning / Bandit (reported by Codacy)

Use of assert detected. The enclosed code will be removed when compiling to optimised byte code. Note test

Use of assert detected. The enclosed code will be removed when compiling to optimised byte code.
assert mock_log.critical.call_count >= 1
# Check for authentication failed message
args = str(mock_log.critical.call_args_list)
assert "Authentication Failed" in args


def test_check_api_access_403(monkeypatch):
def test_verify_access_and_get_folders_403(monkeypatch):

Check warning

Code scanning / Pylint (reported by Codacy)

Missing function docstring Warning test

Missing function docstring

Check warning

Code scanning / Pylintpython3 (reported by Codacy)

Missing function or method docstring Warning test

Missing function or method docstring
m = reload_main_with_env(monkeypatch)
mock_client = MagicMock()

Expand All @@ -295,12 +308,12 @@
mock_log = MagicMock()
monkeypatch.setattr(m, "log", mock_log)

assert m.check_api_access(mock_client, "forbidden_profile") is False
assert m.verify_access_and_get_folders(mock_client, "forbidden_profile") is None

Check notice

Code scanning / Bandit

Use of assert detected. The enclosed code will be removed when compiling to optimised byte code. Note test

Use of assert detected. The enclosed code will be removed when compiling to optimised byte code.

Check notice

Code scanning / Bandit (reported by Codacy)

Use of assert detected. The enclosed code will be removed when compiling to optimised byte code. Note test

Use of assert detected. The enclosed code will be removed when compiling to optimised byte code.
assert mock_log.critical.call_count == 1
assert "Access Denied" in str(mock_log.critical.call_args)


def test_check_api_access_404(monkeypatch):
def test_verify_access_and_get_folders_404(monkeypatch):

Check warning

Code scanning / Pylintpython3 (reported by Codacy)

Missing function or method docstring Warning test

Missing function or method docstring

Check warning

Code scanning / Pylint (reported by Codacy)

Missing function docstring Warning test

Missing function docstring
m = reload_main_with_env(monkeypatch)
mock_client = MagicMock()

Expand All @@ -315,12 +328,12 @@
mock_log = MagicMock()
monkeypatch.setattr(m, "log", mock_log)

assert m.check_api_access(mock_client, "missing_profile") is False
assert m.verify_access_and_get_folders(mock_client, "missing_profile") is None

Check notice

Code scanning / Bandit

Use of assert detected. The enclosed code will be removed when compiling to optimised byte code. Note test

Use of assert detected. The enclosed code will be removed when compiling to optimised byte code.

Check notice

Code scanning / Bandit (reported by Codacy)

Use of assert detected. The enclosed code will be removed when compiling to optimised byte code. Note test

Use of assert detected. The enclosed code will be removed when compiling to optimised byte code.
assert mock_log.critical.call_count >= 1
assert "Profile Not Found" in str(mock_log.critical.call_args_list)


def test_check_api_access_generic_http_error(monkeypatch):
def test_verify_access_and_get_folders_500_retry(monkeypatch):

Check warning

Code scanning / Pylintpython3 (reported by Codacy)

Missing function or method docstring Warning test

Missing function or method docstring

Check warning

Code scanning / Pylint (reported by Codacy)

Missing function docstring Warning test

Missing function docstring
m = reload_main_with_env(monkeypatch)
mock_client = MagicMock()

Expand All @@ -335,12 +348,19 @@
mock_log = MagicMock()
monkeypatch.setattr(m, "log", mock_log)

assert m.check_api_access(mock_client, "profile") is False
# Speed up sleep
monkeypatch.setattr(m, "RETRY_DELAY", 0.001)
monkeypatch.setattr("time.sleep", lambda x: None)
monkeypatch.setattr(m, "MAX_RETRIES", 2)

assert m.verify_access_and_get_folders(mock_client, "profile") is None

Check notice

Code scanning / Bandit

Use of assert detected. The enclosed code will be removed when compiling to optimised byte code. Note test

Use of assert detected. The enclosed code will be removed when compiling to optimised byte code.

Check notice

Code scanning / Bandit (reported by Codacy)

Use of assert detected. The enclosed code will be removed when compiling to optimised byte code. Note test

Use of assert detected. The enclosed code will be removed when compiling to optimised byte code.
# Should have retried
assert mock_client.get.call_count == 2

Check notice

Code scanning / Bandit

Use of assert detected. The enclosed code will be removed when compiling to optimised byte code. Note test

Use of assert detected. The enclosed code will be removed when compiling to optimised byte code.

Check notice

Code scanning / Bandit (reported by Codacy)

Use of assert detected. The enclosed code will be removed when compiling to optimised byte code. Note test

Use of assert detected. The enclosed code will be removed when compiling to optimised byte code.
assert mock_log.error.called
assert "500" in str(mock_log.error.call_args)


def test_check_api_access_network_error(monkeypatch):
def test_verify_access_and_get_folders_network_error(monkeypatch):

Check warning

Code scanning / Pylintpython3 (reported by Codacy)

Missing function or method docstring Warning test

Missing function or method docstring

Check warning

Code scanning / Pylint (reported by Codacy)

Missing function docstring Warning test

Missing function docstring
m = reload_main_with_env(monkeypatch)
mock_client = MagicMock()

Expand All @@ -351,7 +371,13 @@
mock_log = MagicMock()
monkeypatch.setattr(m, "log", mock_log)

assert m.check_api_access(mock_client, "profile") is False
# Speed up sleep
monkeypatch.setattr(m, "RETRY_DELAY", 0.001)
monkeypatch.setattr("time.sleep", lambda x: None)
monkeypatch.setattr(m, "MAX_RETRIES", 2)

assert m.verify_access_and_get_folders(mock_client, "profile") is None

Check notice

Code scanning / Bandit

Use of assert detected. The enclosed code will be removed when compiling to optimised byte code. Note test

Use of assert detected. The enclosed code will be removed when compiling to optimised byte code.

Check notice

Code scanning / Bandit (reported by Codacy)

Use of assert detected. The enclosed code will be removed when compiling to optimised byte code. Note test

Use of assert detected. The enclosed code will be removed when compiling to optimised byte code.
assert mock_client.get.call_count == 2

Check notice

Code scanning / Bandit

Use of assert detected. The enclosed code will be removed when compiling to optimised byte code. Note test

Use of assert detected. The enclosed code will be removed when compiling to optimised byte code.

Check notice

Code scanning / Bandit (reported by Codacy)

Use of assert detected. The enclosed code will be removed when compiling to optimised byte code. Note test

Use of assert detected. The enclosed code will be removed when compiling to optimised byte code.
assert mock_log.error.called
assert "Network failure" in str(mock_log.error.call_args)

Expand Down
Loading