mirror of
https://github.com/domainaware/parsedmarc.git
synced 2026-08-02 13:42:19 +00:00
Add MS Graph well-known folder fallback for root listing failures (#689)
* Add MS Graph well-known folder fallback for root listing failures * Resolve test merge cleanup for MS Graph folder fallback
This commit is contained in:
@@ -93,6 +93,15 @@ def _generate_credential(auth_method: str, token_path: Path, **kwargs):
|
||||
|
||||
|
||||
class MSGraphConnection(MailboxConnection):
|
||||
_WELL_KNOWN_FOLDERS = {
|
||||
"inbox": "inbox",
|
||||
"archive": "archive",
|
||||
"drafts": "drafts",
|
||||
"sentitems": "sentitems",
|
||||
"deleteditems": "deleteditems",
|
||||
"junkemail": "junkemail",
|
||||
}
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
auth_method: str,
|
||||
@@ -267,8 +276,25 @@ class MSGraphConnection(MailboxConnection):
|
||||
parent_folder_id = folder_id
|
||||
return self._find_folder_id_with_parent(path_parts[-1], parent_folder_id)
|
||||
else:
|
||||
# Shared mailboxes can fail root listing; try well-known folders first.
|
||||
well_known_folder_id = self._get_well_known_folder_id(folder_name)
|
||||
if well_known_folder_id:
|
||||
return well_known_folder_id
|
||||
return self._find_folder_id_with_parent(folder_name, None)
|
||||
|
||||
def _get_well_known_folder_id(self, folder_name: str) -> Optional[str]:
|
||||
folder_key = folder_name.lower().replace(" ", "").replace("-", "")
|
||||
alias = self._WELL_KNOWN_FOLDERS.get(folder_key)
|
||||
if alias is None:
|
||||
return None
|
||||
|
||||
url = f"/users/{self.mailbox_name}/mailFolders/{alias}?$select=id,displayName"
|
||||
folder_resp = self._client.get(url)
|
||||
if folder_resp.status_code != 200:
|
||||
return None
|
||||
payload = folder_resp.json()
|
||||
return payload.get("id")
|
||||
|
||||
def _find_folder_id_with_parent(
|
||||
self, folder_name: str, parent_folder_id: Optional[str]
|
||||
):
|
||||
@@ -279,6 +305,10 @@ class MSGraphConnection(MailboxConnection):
|
||||
filter = f"?$filter=displayName eq '{folder_name}'"
|
||||
folders_resp = self._request_with_retries("get", url + filter)
|
||||
if folders_resp.status_code != 200:
|
||||
if parent_folder_id is None:
|
||||
well_known_folder_id = self._get_well_known_folder_id(folder_name)
|
||||
if well_known_folder_id:
|
||||
return well_known_folder_id
|
||||
raise RuntimeWarning(f"Failed to list folders.{folders_resp.json()}")
|
||||
folders: list = folders_resp.json()["value"]
|
||||
matched_folders = [
|
||||
|
||||
@@ -468,6 +468,8 @@ hosts = localhost
|
||||
self.assertEqual(ctx.exception.code, 1)
|
||||
mock_save_aggregate.assert_called_once()
|
||||
mock_save_forensic_opensearch.assert_called_once()
|
||||
|
||||
|
||||
class _FakeGraphResponse:
|
||||
def __init__(self, status_code, payload=None, text=""):
|
||||
self.status_code = status_code
|
||||
@@ -477,6 +479,7 @@ class _FakeGraphResponse:
|
||||
def json(self):
|
||||
return self._payload
|
||||
|
||||
|
||||
class _BreakLoop(BaseException):
|
||||
pass
|
||||
|
||||
@@ -1210,5 +1213,52 @@ since = 2d
|
||||
|
||||
self.assertEqual(system_exit.exception.code, 1)
|
||||
self.assertEqual(mock_watch_inbox.call_args.kwargs.get("since"), "2d")
|
||||
|
||||
class _FakeGraphClient:
|
||||
def get(self, url, params=None):
|
||||
if "/mailFolders/inbox?$select=id,displayName" in url:
|
||||
return _FakeGraphResponse(200, {"id": "inbox-id", "displayName": "Inbox"})
|
||||
|
||||
if "/mailFolders?$filter=displayName eq 'Inbox'" in url:
|
||||
return _FakeGraphResponse(
|
||||
404,
|
||||
{
|
||||
"error": {
|
||||
"code": "ErrorItemNotFound",
|
||||
"message": "Default folder Root not found.",
|
||||
}
|
||||
},
|
||||
)
|
||||
|
||||
if "/mailFolders?$filter=displayName eq 'Custom'" in url:
|
||||
return _FakeGraphResponse(
|
||||
404,
|
||||
{
|
||||
"error": {
|
||||
"code": "ErrorItemNotFound",
|
||||
"message": "Default folder Root not found.",
|
||||
}
|
||||
},
|
||||
)
|
||||
|
||||
return _FakeGraphResponse(404, {"error": {"code": "NotFound"}})
|
||||
|
||||
|
||||
class TestMSGraphFolderFallback(unittest.TestCase):
|
||||
def testWellKnownFolderFallback(self):
|
||||
connection = MSGraphConnection.__new__(MSGraphConnection)
|
||||
connection.mailbox_name = "shared@example.com"
|
||||
connection._client = _FakeGraphClient()
|
||||
|
||||
folder_id = connection._find_folder_id_from_folder_path("Inbox")
|
||||
self.assertEqual(folder_id, "inbox-id")
|
||||
|
||||
def testUnknownFolderStillFails(self):
|
||||
connection = MSGraphConnection.__new__(MSGraphConnection)
|
||||
connection.mailbox_name = "shared@example.com"
|
||||
connection._client = _FakeGraphClient()
|
||||
|
||||
with self.assertRaises(RuntimeWarning):
|
||||
connection._find_folder_id_from_folder_path("Custom")
|
||||
if __name__ == "__main__":
|
||||
unittest.main(verbosity=2)
|
||||
|
||||
Reference in New Issue
Block a user