diff --git a/codecov.yml b/codecov.yml
index 017ad2df..ee47c3f2 100644
--- a/codecov.yml
+++ b/codecov.yml
@@ -9,5 +9,6 @@ ignore:
- "src/IntuneCD/document_entra.py"
- "src/IntuneCD/run_documentation.py"
- "src/IntuneCD/intunecdlib/get_accesstoken.py"
+ - "src/IntuneCD/intunecdlib/logger.py"
- "src/IntuneCD/backup/Intune/backup_autopilotDevices.py"
- "src/IntuneCD/backup/Intune/backup_activationLock.py"
diff --git a/index.html b/index.html
index 70e850f7..ccb2c3a3 100644
--- a/index.html
+++ b/index.html
@@ -152,7 +152,7 @@
Get started now!
-
Follow on Twitter!
+
Follow on X!
@@ -182,7 +182,7 @@
Join IntuneCD Slack!
⋅
Contact
--->
-
© IntuneCD 2023. All Rights Reserved.
+
© IntuneCD 2024. All Rights Reserved.
diff --git a/setup.cfg b/setup.cfg
index 25f670c2..0dc7b0b8 100644
--- a/setup.cfg
+++ b/setup.cfg
@@ -1,6 +1,6 @@
[metadata]
name = IntuneCD
-version = 2.1.1
+version = 2.1.2
author = Tobias Almén
author_email = almenscorner@outlook.com
description = Tool to backup and update configurations in Intune
diff --git a/src/IntuneCD/backup/Intune/backup_compliancePartner.py b/src/IntuneCD/backup/Intune/backup_compliancePartner.py
index 26557416..23412f7f 100644
--- a/src/IntuneCD/backup/Intune/backup_compliancePartner.py
+++ b/src/IntuneCD/backup/Intune/backup_compliancePartner.py
@@ -17,7 +17,7 @@
# Get all Compliance Partners and save them in specified path
-def savebackup(path, output, token, append_id):
+def savebackup(path, output, exclude, token, append_id):
"""
Saves all Compliance Partners in Intune to a JSON or YAML file.
@@ -44,6 +44,12 @@ def savebackup(path, output, token, append_id):
fname = clean_filename(partner["displayName"])
if append_id:
fname = f"{fname}__{graph_id}"
+
+ if (
+ partner.get("lastHeartbeatDateTime")
+ and "CompliancePartnerHeartbeat" in exclude
+ ):
+ partner.pop("lastHeartbeatDateTime", None)
# Save Compliance policy as JSON or YAML depending on configured
# value in "-o"
save_output(output, configpath, fname, partner)
diff --git a/src/IntuneCD/backup/Intune/backup_profiles.py b/src/IntuneCD/backup/Intune/backup_profiles.py
index 4f30bf65..83033094 100644
--- a/src/IntuneCD/backup/Intune/backup_profiles.py
+++ b/src/IntuneCD/backup/Intune/backup_profiles.py
@@ -112,6 +112,8 @@ def savebackup(path, output, exclude, token, prefix, append_id, ignore_omasettin
)
decoded_oma["value"] = oma_value
omas.append(decoded_oma)
+ else:
+ omas.append(setting)
else:
omas = profile["omaSettings"]
diff --git a/src/IntuneCD/backup_intune.py b/src/IntuneCD/backup_intune.py
index 55eb4176..1bbc6c79 100644
--- a/src/IntuneCD/backup_intune.py
+++ b/src/IntuneCD/backup_intune.py
@@ -111,7 +111,7 @@ def backup_intune(results, path, output, exclude, token, prefix, append_id, args
if "CompliancePartner" not in exclude:
from .backup.Intune.backup_compliancePartner import savebackup
- results.append(savebackup(path, output, token, append_id))
+ results.append(savebackup(path, output, exclude, token, append_id))
if "ManagementPartner" not in exclude:
from .backup.Intune.backup_managementPartner import savebackup
diff --git a/src/IntuneCD/intunecdlib/get_accesstoken.py b/src/IntuneCD/intunecdlib/get_accesstoken.py
index 20f7d2bf..3ec5937c 100644
--- a/src/IntuneCD/intunecdlib/get_accesstoken.py
+++ b/src/IntuneCD/intunecdlib/get_accesstoken.py
@@ -91,7 +91,7 @@ def obtain_accesstoken_cert(TENANT_NAME, CLIENT_ID, THUMBPRINT, KEY_FILE):
return token
-def obtain_accesstoken_interactive(TENANT_NAME, CLIENT_ID):
+def obtain_accesstoken_interactive(TENANT_NAME, CLIENT_ID, scopes):
"""
This function is used to get an access token to MS Graph interactivly.
@@ -109,17 +109,6 @@ def obtain_accesstoken_interactive(TENANT_NAME, CLIENT_ID):
token = None
- # Set the requited scopes
- scopes = [
- "DeviceManagementApps.ReadWrite.All",
- "DeviceManagementConfiguration.ReadWrite.All",
- "DeviceManagementManagedDevices.Read.All",
- "DeviceManagementServiceConfig.ReadWrite.All",
- "Group.Read.All",
- "Policy.ReadWrite.ConditionalAccess",
- "Policy.Read.All",
- ]
-
try:
# Get the token interactively
token = app.acquire_token_interactive(
diff --git a/src/IntuneCD/intunecdlib/get_authparams.py b/src/IntuneCD/intunecdlib/get_authparams.py
index 9cf5c466..0132b74b 100644
--- a/src/IntuneCD/intunecdlib/get_authparams.py
+++ b/src/IntuneCD/intunecdlib/get_authparams.py
@@ -15,7 +15,7 @@
)
-def getAuth(mode, localauth, certauth, interactiveauth, entra, tenant):
+def getAuth(mode, localauth, certauth, interactiveauth, scopes, entra, tenant):
"""
This function authenticates to MS Graph and returns the access token.
@@ -42,7 +42,7 @@ def getAuth(mode, localauth, certauth, interactiveauth, entra, tenant):
if not all([TENANT_NAME, CLIENT_ID]):
raise ValueError("One or more os.environ variables not set")
- return obtain_accesstoken_interactive(TENANT_NAME, CLIENT_ID)
+ return obtain_accesstoken_interactive(TENANT_NAME, CLIENT_ID, scopes)
if mode:
if mode == "devtoprod":
diff --git a/src/IntuneCD/intunecdlib/graph_batch.py b/src/IntuneCD/intunecdlib/graph_batch.py
index 488607a6..a56fccf0 100644
--- a/src/IntuneCD/intunecdlib/graph_batch.py
+++ b/src/IntuneCD/intunecdlib/graph_batch.py
@@ -11,6 +11,7 @@
import time
from .graph_request import makeapirequestPost
+from .logger import log
def create_batch_request(batch, batch_id, method, url, extra_url) -> tuple:
@@ -41,6 +42,7 @@ def handle_responses(
"""Handle the responses from the batch request.
Args:
+ initial_request_data (list): List of initial requests
request_data (list): List of responses from the batch request
responses (list): List of responses from the batch request
retry_pool (list): List of failed requests
@@ -53,10 +55,13 @@ def handle_responses(
failed_batch_requests = []
if resp["status"] == 200:
responses.append(resp["body"])
+ retry_pool = [req for req in retry_pool if req["id"] != int(resp["id"])]
elif resp["status"] in [429, 503]:
if initial_request_data:
failed_batch_requests = [
- i for i in initial_request_data if i["id"] == int(resp["id"])
+ i
+ for i in initial_request_data
+ if i["id"] == int(resp["id"]) and i not in retry_pool
]
retry_pool += failed_batch_requests
@@ -65,44 +70,101 @@ def handle_responses(
return responses, retry_pool, wait_time
-def batch_request(data, url, extra_url, token, method="GET") -> list:
- """_summary_
+def create_batch_list(data, batch_count) -> list:
+ """Create a list of batches from the data.
Args:
- data (list): List of IDs
+ data (list): List of objects
+ batch_count (int): Number of objects to include in each batch
+
+ Returns:
+ list: List of batches
+ """
+ return [data[i : i + batch_count] for i in range(0, len(data), batch_count)]
+
+
+def process_batch(
+ batch,
+ batch_id,
+ method,
+ url,
+ extra_url,
+ token,
+ initial_request_data,
+ responses,
+ retry_pool,
+) -> tuple:
+ """Process the batch request.
+
+ Args:
+ batch (list): List of objects
+ batch_id (str): ID for the batch request
+ method (str): HTTP method to use
url (str): MS graph endpoint for the object
extra_url (str): Used if anything extra is needed for the url such as /assignments or ?$filter
token (str): OAuth token used for authentication
- method (str, optional): HTTP method to use. Defaults to "GET".
+ initial_request_data (list): List of initial requests
+ responses (list): List of responses from the batch request
+ retry_pool (list): List of failed requests
Returns:
- list: List of responses from the batch request
+ tuple: Tuple containing the batch ID, responses, retry pool and wait time
"""
- responses = []
- batch_id = 1
- batch_count = 20
- retry_pool = []
- wait_time = 0
- batch_list = [data[i : i + batch_count] for i in range(0, len(data), batch_count)]
- initial_request_data = []
+ query_data, batch_id = create_batch_request(batch, batch_id, method, url, extra_url)
+ json_data = json.dumps(query_data)
+ request = makeapirequestPost(
+ "https://graph.microsoft.com/beta/$batch", token, jdata=json_data
+ )
+ request_data = sorted(request["responses"], key=lambda item: item.get("id"))
+ initial_request_data += query_data["requests"]
+ responses, retry_pool, wait_time = handle_responses(
+ initial_request_data, request_data, responses, retry_pool
+ )
+ return batch_id, responses, retry_pool, wait_time
+
+
+def retry_failed_requests(
+ retry_pool,
+ wait_time,
+ max_retries,
+ max_wait_time,
+ token,
+ initial_request_data,
+ responses,
+ batch_count,
+) -> tuple:
+ """Retry failed requests.
- for batch in batch_list:
- query_data, batch_id = create_batch_request(
- batch, batch_id, method, url, extra_url
- )
- json_data = json.dumps(query_data)
- request = makeapirequestPost(
- "https://graph.microsoft.com/beta/$batch", token, jdata=json_data
- )
- request_data = sorted(request["responses"], key=lambda item: item.get("id"))
- initial_request_data += query_data["requests"]
- responses, retry_pool, wait_time = handle_responses(
- initial_request_data, request_data, responses, retry_pool
- )
+ Args:
+ retry_pool (list): List of failed requests
+ wait_time (int): Time to wait before retrying
+ max_retries (int): Maximum number of retries
+ max_wait_time (int): Maximum time to wait before retrying
+ token (str): OAuth token used for authentication
+ initial_request_data (list): List of initial requests
+ responses (list): List of responses from the batch request
+ batch_count (int): Number of objects to include in each batch
- if retry_pool:
+ Returns:
+ list: List of responses from the batch request
+ """
+ retry_count = 0
+ failed_retry_requests = []
+ while retry_count < max_retries and retry_pool:
+ log(
+ "retry_failed_requests",
+ f"Retrying failed requests, retry pool count: {str(len(retry_pool))}",
+ )
if wait_time > 0:
+ log("retry_failed_requests", f"Sleeping for {str(wait_time)} seconds...")
time.sleep(wait_time)
+ wait_time = min(wait_time * 2, max_wait_time)
+ else:
+ log(
+ "retry_failed_requests",
+ "No wait time in headers, sleeping for 20 seconds...",
+ )
+ time.sleep(20)
batch_list = [
retry_pool[i : i + batch_count]
for i in range(0, len(retry_pool), batch_count)
@@ -114,9 +176,70 @@ def batch_request(data, url, extra_url, token, method="GET") -> list:
"https://graph.microsoft.com/beta/$batch", token, jdata=json_data
)
request_data = sorted(request["responses"], key=lambda item: item.get("id"))
- responses, _, _ = handle_responses(
+ responses, retry_pool, _ = handle_responses(
initial_request_data, request_data, responses, retry_pool
)
+ failed_retry_requests = [
+ r for r in request["responses"] if r["status"] != 200
+ ]
+ retry_count += 1
+ log("retry_failed_requests", f"Retry count: {str(retry_count)}")
+ if retry_pool and retry_count == max_retries:
+ break
+ log(
+ "retry_failed_requests",
+ f"Failed requests after {str(retry_count)} retries: {str(len(failed_retry_requests))}",
+ )
+ return responses
+
+
+def batch_request(data, url, extra_url, token, method="GET") -> list:
+ """Batch request to the Graph API.
+
+ Args:
+ data (list): List of objects
+ url (str): MS graph endpoint for the object
+ extra_url (str): Used if anything extra is needed for the url such as /assignments or ?$filter
+ token (str): OAuth token used for authentication
+ method (str): HTTP method to use
+
+ Returns:
+ list: List of responses from the batch request
+ """
+ responses = []
+ batch_id = 1
+ batch_count = 20
+ retry_pool = []
+ wait_time = 0
+ initial_request_data = []
+
+ batch_list = create_batch_list(data, batch_count)
+ for batch in batch_list:
+ batch_id, responses, retry_pool, wait_time = process_batch(
+ batch,
+ batch_id,
+ method,
+ url,
+ extra_url,
+ token,
+ initial_request_data,
+ responses,
+ retry_pool,
+ )
+
+ max_retries = 10
+ max_wait_time = 60
+ if retry_pool:
+ responses = retry_failed_requests(
+ retry_pool,
+ wait_time,
+ max_retries,
+ max_wait_time,
+ token,
+ initial_request_data,
+ responses,
+ batch_count,
+ )
return responses
diff --git a/src/IntuneCD/intunecdlib/logger.py b/src/IntuneCD/intunecdlib/logger.py
new file mode 100644
index 00000000..b7601e0e
--- /dev/null
+++ b/src/IntuneCD/intunecdlib/logger.py
@@ -0,0 +1,23 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+
+"""
+This module is used to log messages if verbose is set to True.
+"""
+
+import os
+import time
+
+verbose = os.getenv("VERBOSE")
+
+
+def log(function, msg):
+ """Prints a message to the console if the VERBOSE environment variable is set to True.
+
+ Args:
+ function (str): The name of the function that is calling the log function.
+ msg (str): The message to print to the console.
+ """
+ if verbose:
+ msg = f"[{time.asctime()}] - [{function}] - {msg}"
+ print(msg)
diff --git a/src/IntuneCD/run_backup.py b/src/IntuneCD/run_backup.py
index 60e68509..b57d8d28 100644
--- a/src/IntuneCD/run_backup.py
+++ b/src/IntuneCD/run_backup.py
@@ -138,6 +138,7 @@ def start():
"ScopeTags",
"VPPusedLicenseCount",
"GPlaySyncTime",
+ "CompliancePartnerHeartbeat",
],
nargs="+",
)
@@ -182,9 +183,20 @@ def start():
help="When set, backs up Activation Lock Bypass Codes",
action="store_true",
)
+ parser.add_argument(
+ "--scopes",
+ help="The scopes to use when obtaining an access token interactively separated by space. Only used when using interactive auth. Default is: DeviceManagementApps.ReadWrite.All, DeviceManagementConfiguration.ReadWrite.All, DeviceManagementManagedDevices.Read.All, DeviceManagementServiceConfig.ReadWrite.All, DeviceManagementRBAC.ReadWrite.All, Group.Read.All, Policy.ReadWrite.ConditionalAccess, Policy.Read.All",
+ nargs="+",
+ )
+ parser.add_argument(
+ "-v", "--verbose", help="Prints verbose output", action="store_true"
+ )
args = parser.parse_args()
+ if args.verbose:
+ os.environ["VERBOSE"] = "True"
+
def devtoprod():
return "devtoprod"
@@ -202,11 +214,24 @@ def selected_mode(argument):
else:
args.mode = selected_mode(args.mode)
+ if not args.scopes:
+ args.scopes = [
+ "DeviceManagementApps.ReadWrite.All",
+ "DeviceManagementConfiguration.ReadWrite.All",
+ "DeviceManagementManagedDevices.Read.All",
+ "DeviceManagementServiceConfig.ReadWrite.All",
+ "DeviceManagementRBAC.ReadWrite.All",
+ "Group.Read.All",
+ "Policy.ReadWrite.ConditionalAccess",
+ "Policy.Read.All",
+ ]
+
token = getAuth(
args.mode,
args.localauth,
args.certauth,
args.interactiveauth,
+ args.scopes,
args.entrabackup,
tenant="DEV",
)
@@ -284,6 +309,9 @@ def run_backup(path, output, exclude, token, prefix, append_id):
else:
print("Please enter a valid output format, json or yaml")
+ if "VERBOSE" in os.environ:
+ del os.environ["VERBOSE"]
+
if __name__ == "__main__":
start()
diff --git a/src/IntuneCD/run_update.py b/src/IntuneCD/run_update.py
index d4c24ff6..6d984317 100644
--- a/src/IntuneCD/run_update.py
+++ b/src/IntuneCD/run_update.py
@@ -158,9 +158,20 @@ def start():
help="When this parameter is set, the script will also update Entra configurations",
action="store_true",
)
+ parser.add_argument(
+ "--scopes",
+ help="The scopes to use when obtaining an access token interactively separated by space. Only used when using interactive auth. Default is: DeviceManagementApps.ReadWrite.All, DeviceManagementConfiguration.ReadWrite.All, DeviceManagementManagedDevices.Read.All, DeviceManagementServiceConfig.ReadWrite.All, DeviceManagementRBAC.ReadWrite.All, Group.Read.All, Policy.ReadWrite.ConditionalAccess, Policy.Read.All",
+ nargs="+",
+ )
+ parser.add_argument(
+ "-v", "--verbose", help="Prints verbose output", action="store_true"
+ )
args = parser.parse_args()
+ if args.verbose:
+ os.environ["VERBOSE"] = "True"
+
def devtoprod():
return "devtoprod"
@@ -178,11 +189,24 @@ def selected_mode(argument):
else:
args.mode = selected_mode(args.mode)
+ if not args.scopes:
+ args.scopes = [
+ "DeviceManagementApps.ReadWrite.All",
+ "DeviceManagementConfiguration.ReadWrite.All",
+ "DeviceManagementManagedDevices.Read.All",
+ "DeviceManagementServiceConfig.ReadWrite.All",
+ "DeviceManagementRBAC.ReadWrite.All",
+ "Group.Read.All",
+ "Policy.ReadWrite.ConditionalAccess",
+ "Policy.Read.All",
+ ]
+
token = getAuth(
args.mode,
args.localauth,
args.certauth,
args.interactiveauth,
+ args.scopes,
args.entraupdate,
tenant="PROD",
)
@@ -291,6 +315,9 @@ def run_update(path, token, assignment, exclude, report, create_groups, remove):
args.remove,
)
+ if "VERBOSE" in os.environ:
+ del os.environ["VERBOSE"]
+
if __name__ == "__main__":
start()
diff --git a/src/IntuneCD/update/Intune/update_appleEnrollmentProfile.py b/src/IntuneCD/update/Intune/update_appleEnrollmentProfile.py
index be2ad06e..4e93ef60 100644
--- a/src/IntuneCD/update/Intune/update_appleEnrollmentProfile.py
+++ b/src/IntuneCD/update/Intune/update_appleEnrollmentProfile.py
@@ -70,8 +70,14 @@ def update(path, token, report):
profile_data["value"][0], repo_data, ignore_order=True
).get("values_changed", {})
+ skip_diff = DeepDiff(
+ profile_data["value"][0]["enabledSkipKeys"],
+ repo_data["enabledSkipKeys"],
+ ignore_order=True,
+ )
+
# If any changed values are found, push them to Intune
- if diff and report is False:
+ if diff or skip_diff and report is False:
request_data = json.dumps(repo_data)
q_param = None
makeapirequestPatch(
diff --git a/tests/Backup/Intune/test_backup_compliancePartner.py b/tests/Backup/Intune/test_backup_compliancePartner.py
index 83693389..49d62361 100644
--- a/tests/Backup/Intune/test_backup_compliancePartner.py
+++ b/tests/Backup/Intune/test_backup_compliancePartner.py
@@ -21,11 +21,13 @@ def setUp(self):
self.directory = TempDirectory()
self.directory.create()
self.token = "token"
+ self.exclude = []
self.append_id = False
self.saved_path = f"{self.directory.path}/Partner Connections/Compliance/test."
self.expected_data = {
"@odata.type": "#microsoft.graph.complianceManagementPartner",
"partnerState": "unavailable",
+ "lastHeartbeatDateTime": "2022-01-28T12:28:48.975089Z",
"displayName": "test",
"macOsOnboarded": True,
"macOsEnrollmentAssignments": [
@@ -41,6 +43,7 @@ def setUp(self):
"@odata.type": "#microsoft.graph.complianceManagementPartner",
"id": "0",
"partnerState": "unavailable",
+ "lastHeartbeatDateTime": "2022-01-28T12:28:48.975089Z",
"displayName": "test",
"macOsOnboarded": True,
"macOsEnrollmentAssignments": [
@@ -66,7 +69,9 @@ def tearDown(self):
def test_backup_yml(self):
"""The folder should be created, the file should have the expected contents, and the count should be 1."""
- self.count = savebackup(self.directory.path, "yaml", self.token, self.append_id)
+ self.count = savebackup(
+ self.directory.path, "yaml", self.exclude, self.token, self.append_id
+ )
with open(self.saved_path + "yaml", "r", encoding="utf-8") as f:
data = json.dumps(yaml.safe_load(f))
@@ -81,7 +86,9 @@ def test_backup_yml(self):
def test_backup_json(self):
"""The folder should be created, the file should have the expected contents, and the count should be 1."""
- self.count = savebackup(self.directory.path, "json", self.token, self.append_id)
+ self.count = savebackup(
+ self.directory.path, "json", self.exclude, self.token, self.append_id
+ )
with open(self.saved_path + "json", "r", encoding="utf-8") as f:
self.saved_data = json.load(f)
@@ -96,13 +103,17 @@ def test_backup_with_no_return_data(self):
"""The count should be 0 if no data is returned."""
self.makeapirequest.return_value = {"value": [{"partnerState": "unknown"}]}
- self.count = savebackup(self.directory.path, "json", self.token, self.append_id)
+ self.count = savebackup(
+ self.directory.path, "json", self.exclude, self.token, self.append_id
+ )
self.assertEqual(0, self.count["config_count"])
def test_backup_append_id(self):
"""The folder should be created, the file should have the expected contents, and the count should be 1."""
- self.count = savebackup(self.directory.path, "json", self.token, True)
+ self.count = savebackup(
+ self.directory.path, "json", self.exclude, self.token, True
+ )
self.assertTrue(
Path(
@@ -110,6 +121,20 @@ def test_backup_append_id(self):
).exists()
)
+ def test_backup_exclude_lastHeartbeatDateTime(self):
+ """The lastHeartbeatDateTime should be excluded from the saved data."""
+
+ self.exclude.append("CompliancePartnerHeartbeat")
+
+ self.count = savebackup(
+ self.directory.path, "json", self.exclude, self.token, self.append_id
+ )
+
+ with open(self.saved_path + "json", "r", encoding="utf-8") as f:
+ saved_data = json.load(f)
+
+ self.assertNotIn("lastHeartbeatDateTime", saved_data)
+
if __name__ == "__main__":
unittest.main()
diff --git a/tests/Update/Intune/test_update_appleEnrollmentProfile.py b/tests/Update/Intune/test_update_appleEnrollmentProfile.py
index d640ae3f..43b08a57 100644
--- a/tests/Update/Intune/test_update_appleEnrollmentProfile.py
+++ b/tests/Update/Intune/test_update_appleEnrollmentProfile.py
@@ -32,6 +32,7 @@ def setUp(self):
"id": "0",
"displayName": "test",
"testvalue": "test",
+ "enabledSkipKeys": ["test"],
}
]
}
@@ -40,6 +41,7 @@ def setUp(self):
"id": "0",
"displayName": "test",
"testvalue": "test1",
+ "enabledSkipKeys": ["test"],
}
self.makeapirequest_patch = patch(
@@ -79,10 +81,11 @@ def test_update_with_multiple_diffs(self):
self.repo_data["testvalue2"] = "test2"
self.mem_data["value"][0]["testvalue"] = "test"
self.mem_data["value"][0]["testvalue2"] = "test1"
+ self.mem_data["value"][0]["enabledSkipKeys"] = ["test1"]
self.count = update(self.directory.path, self.token, report=False)
- self.assertEqual(self.count[0].count, 2)
+ self.assertEqual(self.count[0].count, 3)
self.assertEqual(self.makeapirequestPatch.call_count, 1)
def test_update_with_no_diffs(self):
diff --git a/tests/test_get_authparams.py b/tests/test_get_authparams.py
index 50077734..3aebd323 100644
--- a/tests/test_get_authparams.py
+++ b/tests/test_get_authparams.py
@@ -69,6 +69,7 @@ def test_get_auth_devtoprod_env_dev_app(self, _, __, ___, ____):
localauth=None,
certauth=None,
interactiveauth=None,
+ scopes=[],
entra=False,
tenant="DEV",
)
@@ -89,6 +90,7 @@ def test_get_auth_devtoprod_env_prod(self, _, __, ___, ____):
localauth=None,
certauth=None,
interactiveauth=None,
+ scopes=[],
entra=False,
tenant="PROD",
)
@@ -101,6 +103,7 @@ def test_get_auth_devtoprod_localauth_dev(self, _, __, ___, ____):
localauth=self.auth_dev_json,
certauth=None,
interactiveauth=None,
+ scopes=[],
entra=False,
tenant="DEV",
)
@@ -113,6 +116,7 @@ def test_get_auth_devtoprod_localauth_prod(self, _, __, ___, ____):
localauth=self.auth_prod_json,
certauth=None,
interactiveauth=None,
+ scopes=[],
entra=False,
tenant="PROD",
)
@@ -129,6 +133,7 @@ def test_get_auth_standalone_env(self, _, __, ___, ____):
localauth=None,
certauth=None,
interactiveauth=None,
+ scopes=[],
entra=False,
tenant=None,
)
@@ -141,6 +146,7 @@ def test_get_auth_standalone_localauth(self, _, __, ___, ____):
localauth=self.auth_json,
certauth=None,
interactiveauth=None,
+ scopes=[],
entra=False,
tenant=None,
)
@@ -157,6 +163,7 @@ def test_get_auth_devtoprod_missing_env_dev(self, _, __, ___, ____):
localauth=None,
certauth=None,
interactiveauth=None,
+ scopes=[],
entra=False,
tenant="DEV",
)
@@ -172,6 +179,7 @@ def test_get_auth_devtoprod_missing_env_prod(self, _, __, ___, ____):
localauth=None,
certauth=None,
interactiveauth=None,
+ scopes=[],
entra=False,
tenant="PROD",
)
@@ -185,6 +193,7 @@ def test_get_auth_standalone_missing_env(self, _, __, ___, ____):
localauth=None,
certauth=None,
interactiveauth=None,
+ scopes=[],
entra=False,
tenant=None,
)
@@ -205,6 +214,7 @@ def test_get_auth_cert(self, _, __, ___, ____):
localauth=None,
certauth=True,
interactiveauth=None,
+ scopes=[],
tenant=None,
entra=False,
)
@@ -222,6 +232,7 @@ def test_get_auth_cert_missing_env(self, _, __, ___, ____):
localauth=None,
certauth=True,
interactiveauth=None,
+ scopes=[],
entra=False,
tenant=None,
)
@@ -242,6 +253,7 @@ def test_get_auth_interactive(self, _, __, ___, ____):
localauth=None,
certauth=None,
interactiveauth=True,
+ scopes=[],
tenant=None,
entra=False,
)
@@ -259,6 +271,7 @@ def test_get_auth_interactive_missing_env(self, _, __, ___, ____):
localauth=None,
certauth=None,
interactiveauth=True,
+ scopes=[],
entra=False,
tenant=None,
)
diff --git a/tests/test_graph_batch.py b/tests/test_graph_batch.py
index 262ed23e..a80dc1ad 100644
--- a/tests/test_graph_batch.py
+++ b/tests/test_graph_batch.py
@@ -5,6 +5,7 @@
This module tests the graph_batch module.
"""
+import os
import unittest
from unittest.mock import patch
@@ -22,6 +23,7 @@ class TestGraphBatch(unittest.TestCase):
def setUp(self):
self.token = "token"
+ os.environ["VERBOSE"] = "True"
self.batch_request_data = ["1", "2", "3"]
self.batch_assignment_data = {"value": [{"id": "0"}]}
self.batch_intents_data = {
@@ -119,6 +121,7 @@ def tearDown(self):
self.batch_intents.stop()
self.batch_assignment.stop()
self.batch_request.stop()
+ del os.environ["VERBOSE"]
def test_batch_request(self):
"""The batch request function should return the expected result."""
@@ -162,7 +165,53 @@ def test_batch_request_429(self):
{
"id": "2",
"status": 200,
- "headers": {"Retry-After": 1},
+ "headers": {},
+ "body": {
+ "odata.count": 1,
+ "value": [{"id": "1", "displayName": "test"}],
+ },
+ },
+ ]
+ },
+ {
+ "responses": [
+ {
+ "id": "1",
+ "status": 200,
+ "headers": {},
+ "body": {
+ "odata.count": 1,
+ "value": [{"id": "0", "displayName": "test"}],
+ },
+ }
+ ]
+ },
+ )
+ self.result = batch_request(self.batch_request_data, "test", "test", self.token)
+
+ self.assertEqual(self.makeapirequestPost.call_count, 2)
+ self.assertEqual(self.result, self.expected_result)
+
+ def test_batch_request_503(self):
+ """The batch request function should return the expected result."""
+
+ self.expected_result = [
+ {"odata.count": 1, "value": [{"displayName": "test", "id": "1"}]},
+ {"odata.count": 1, "value": [{"displayName": "test", "id": "0"}]},
+ ]
+ self.makeapirequestPost.side_effect = (
+ {
+ "responses": [
+ {
+ "id": "1",
+ "status": 503,
+ "headers": {},
+ "body": {},
+ },
+ {
+ "id": "2",
+ "status": 200,
+ "headers": {},
"body": {
"odata.count": 1,
"value": [{"id": "1", "displayName": "test"}],