GoogleDriveApiModule
Common Google Drive code that will be appended into the Google Drive integrations when it's deployed.
python · ApiModules
Details
| ID | GoogleDriveApiModule |
|---|---|
| Language | python |
| From Version | 6.10.0 |
| Docker Image | demisto/googleapi-python3:1.0.0.8516410 |
| Tags | infra server |
README
To use the common Google Drive integration logic, run the following command to import the GoogleDriveApiModule.
def main():
run_google_drive_integration()
from GoogleDriveApiModule import * # noqa: E402
if __name__ in ["builtins", "__main__"]:
main()
Then, the run_google_drive_integration entry point and the GSuiteClient (re-exported from GSuiteApiModule) will be available for usage. For the canonical consumer, see the Google Drive integration.
from unittest.mock import MagicMock, patch import pytest from CommonServerPython import * from GoogleDriveApiModule import HR_MESSAGES, MESSAGES, OUTPUT_PREFIX, GSuiteClient with open("test_data/service_account_json.txt") as f: TEST_JSON = f.read() MOCKER_HTTP_METHOD = "GSuiteApiModule.GSuiteClient.http_request" @pytest.fixture def gsuite_client(): headers = {"Content-Type": "application/json"} return GSuiteClient(GSuiteClient.safe_load_non_strict_json(TEST_JSON), verify=False, proxy=False, headers=headers) def test_test_function(mocker, gsuite_client): """ Scenario: Call to test-module should return 'ok' if API call succeeds. Given: - client object When: - Calling test function. Then: - Ensure 'ok' should be return. """ from GoogleDriveApiModule import GSuiteClient, test_module mocker.patch.object(GSuiteClient, "set_authorized_http") mocker.patch.object(GSuiteClient, "http_request") assert test_module(gsuite_client, {}, {}) == "ok" @patch(MOCKER_HTTP_METHOD) def test_drive_create_command_success(mocker_http_request, gsuite_client): """ Scenario: For google-drive-create command success. Given: - Command args. When: - Calling google-drive-create command with the parameters provided. Then: - Ensure command's raw_response, outputs, readable_output, outputs_key_field, outputs_prefix should be as expected. """ from GoogleDriveApiModule import drive_create_command with open("test_data/drive_create_response.json", encoding="utf-8") as data: response_data = json.load(data) mocker_http_request.return_value = response_data result = drive_create_command(gsuite_client, {}) assert result.raw_response == response_data assert result.outputs == response_data assert result.readable_output.startswith("### " + HR_MESSAGES["DRIVE_CREATE_SUCCESS"]) assert result.outputs_key_field == "id" assert result.outputs_prefix == OUTPUT_PREFIX["GOOGLE_DRIVE_HEADER"] @patch(MOCKER_HTTP_METHOD) def test_drive_create_command_failure(mocker_http_request, gsuite_client): """ Scenario: For google-drive-create command failure. Given: - Command args and a non-working google api integration. When: - Calling google-drive-create command with the parameters provided. Then: - Ensure command's error response is as expected. """ mocker_http_request.side_effect = ValueError("SOME_ERROR") from GoogleDriveApiModule import drive_create_command with pytest.raises(Exception, match="SOME_ERROR"): drive_create_command(gsuite_client, {}) @patch(MOCKER_HTTP_METHOD) def test_drive_changes_list_command_success(mocker_http_request, gsuite_client): """ Scenario: For google-drive-changes-list command successful run. Given: - Command args. When: - Calling google-drive-changes-list command with the parameters provided. Then: - Ensure command's raw_response, outputs and readable_output should be as expected. """ from GoogleDriveApiModule import drive_changes_list_command with open("test_data/drive_changes_response.json", encoding="utf-8") as data: mock_response = json.load(data) with open("test_data/drive_changes_drive_context.json", encoding="utf-8") as data: expected_res = json.load(data) mocker_http_request.return_value = mock_response with open("test_data/drive_changes_hr.txt") as data: expected_hr = data.read() args = {"user_id": "user@test.com", "page_token": "1"} result = drive_changes_list_command(gsuite_client, args) assert result.raw_response == mock_response assert result.outputs == expected_res assert result.readable_output == expected_hr @patch(MOCKER_HTTP_METHOD) def test_drive_changes_list_command_wrong_argument(mocker_http_request, gsuite_client): """ Scenario: Wrong argument given google-drive-changes-list command. Given: - Command args. When: - Calling google-drive-changes-list command with the parameters provided. Then: - Ensure command should raise Exception as expected. """ from GoogleDriveApiModule import drive_changes_list_command message = "message" mocker_http_request.side_effect = Exception(message) args = {"page_token": "1", "user_id": "user@test.comm", "fields": "advance"} with pytest.raises(Exception, match=message): drive_changes_list_command(gsuite_client, args) def test_prepare_params_for_drive_changes_list(): """ Scenario: Arguments given for google-drive-changes-list command. Given: - Command args. When: - Calling prepare_params_for_drive_changes_list with command arguments. Then: - Ensure prepared arguments should be returned or return valid value error. """ from GoogleDriveApiModule import prepare_params_for_drive_changes_list fields = "fields" arguments = { "page_token": "1", "drive_id": "driveId", "include_corpus_removals": "false", "include_items_from_all_drives": "false", "include_removed": "false", "restrict_to_my_drive": "false", "supports_all_drives": "false", fields: "advance", "page_size": "1", "spaces": "drive,appDataFolder", "include_permissions_for_view": "published", } expected_arguments = { "pageToken": "1", "driveId": "driveId", "includeCorpusRemovals": False, "includeItemsFromAllDrives": False, "includeRemoved": False, "restrictToMyDrive": False, "supportsAllDrives": False, fields: "*", "pageSize": 1, "spaces": "drive,appDataFolder", "includePermissionsForView": "published", } assert prepare_params_for_drive_changes_list(arguments) == expected_arguments arguments = {fields: "some"} with pytest.raises(ValueError) as e: prepare_params_for_drive_changes_list(arguments) assert MESSAGES["DRIVE_CHANGES_FIELDS"].format(fields) == str(e.value) arguments = {"page_size": "-1"} with pytest.raises(ValueError) as e: prepare_params_for_drive_changes_list(arguments) assert MESSAGES["INTEGER_ERROR"].format("page_size") == str(e.value) def test_prepare_body_for_drive_activity(): """ Scenario: Arguments given for prepare_body_for_drive_activity method. Given: - args. When: - Calling prepare_body_for_drive_activity with command arguments. Then: - Ensure method should return dict. """ from GoogleDriveApiModule import prepare_body_for_drive_activity args = { "folder_name": "items/1", "item_name": "items/2", "filter": 'time >= "2020-09-17T13:19:10.197Z"', "time_range": "5 days", "action_detail_case_include": "RENAME", "action_detail_case_remove": "CREATE", "page_token": "token123", } expected_body = { "ancestorName": args.get("folder_name"), "itemName": args.get("item_name"), "pageToken": args.get("page_token"), "filter": args.get("filter"), } assert expected_body == prepare_body_for_drive_activity(args) @patch(MOCKER_HTTP_METHOD) def test_drive_activity_list_command_success(mocker_http_request, gsuite_client): """ Scenario: For google-drive-activity-list command successful run. Given: - Command args. When: - Calling google-drive-activity-list command with the parameters provided. Then: - Ensure command's raw_response, outputs should be as expected. """ from GoogleDriveApiModule import drive_activity_list_command with open("test_data/drive_activity_response.json", encoding="utf-8") as data: mock_response = json.load(data) with open("test_data/drive_activity_context.json", encoding="utf-8") as data: expected_res = json.load(data) mocker_http_request.return_value = mock_response args = {} result = drive_activity_list_command(gsuite_client, args) assert result.raw_response == mock_response assert result.outputs == expected_res @patch(MOCKER_HTTP_METHOD) def test_drive_activity_list_command_human_readable(mocker_http_request, gsuite_client): """ Scenario: For google-drive-activity-list command successful run. Given: - Command args. When: - Calling google-drive-activity-list command with the parameters provided. Then: - Ensure command's human redable should be as expected. """ from GoogleDriveApiModule import drive_activity_list_command with open("test_data/drive_activity_primary_activities.json", encoding="utf-8") as data: mock_response = json.load(data) with open("test_data/drive_activity_list_hr.txt") as data: expected_hr = data.read() mocker_http_request.return_value = mock_response args = {} result = drive_activity_list_command(gsuite_client, args) assert result.readable_output == expected_hr @patch(MOCKER_HTTP_METHOD) def test_drive_activity_list_command_no_records(mocker_http_request, gsuite_client): """ Scenario: For google-drive-activity-list command when no records found. Given: - Command args. When: - Calling google-drive-activity-list command with the parameters provided. Then: - Ensure command's readable_output. """ from GoogleDriveApiModule import drive_activity_list_command mocker_http_request.return_value = {"activities": []} args = {} result = drive_activity_list_command(gsuite_client, args) assert result.readable_output == "No Drive Activity found." def test_validate_params_for_fetch_incidents_error(): """ Scenario: Parameters provided for fetch-incidents. Given: - Configuration parameters. When: - Calling validate_params_for_fetch_incidents with parameters. Then: - Ensure parameters validation. """ from GoogleDriveApiModule import validate_params_for_fetch_incidents params = {"isFetch": True, "drive_item_search_value": "create", "max_fetch": "abc", "user_id": "helo"} with pytest.raises(ValueError, match=MESSAGES["FETCH_INCIDENT_REQUIRED_ARGS"]): validate_params_for_fetch_incidents(params) params.pop("drive_item_search_value") params["drive_item_search_field"] = "create" validate_params_for_fetch_incidents(params) with pytest.raises(ValueError, match=MESSAGES["MAX_INCIDENT_ERROR"]): params.pop("drive_item_search_value") validate_params_for_fetch_incidents(params) def test_validate_params_for_fetch_incidents_requires_user_id_in_legacy_mode(mocker): """ Given: A legacy (non-UCP) configuration with no User ID. When: validate_params_for_fetch_incidents is called. Then: A ValueError about the required User ID is raised. """ from GoogleDriveApiModule import validate_params_for_fetch_incidents mocker.patch("GoogleDriveApiModule.should_use_ucp_auth", return_value=False) with pytest.raises(ValueError, match=MESSAGES["USER_ID_REQUIRED"]): validate_params_for_fetch_incidents({"isFetch": True}) def test_validate_params_for_fetch_incidents_user_id_optional_in_ucp_mode(mocker): """ Given: A UCP (ConnectUs) configuration with no User ID. When: validate_params_for_fetch_incidents is called. Then: No User ID error is raised (the subject comes from the connection profile). """ from GoogleDriveApiModule import validate_params_for_fetch_incidents mocker.patch("GoogleDriveApiModule.should_use_ucp_auth", return_value=True) # Should not raise USER_ID_REQUIRED; max_fetch/first_fetch defaults are applied. params: dict = {"isFetch": True} validate_params_for_fetch_incidents(params) assert params["max_fetch"] == 10 def test_prepare_args_for_fetch_incidents(): """ Scenario: Prepare request body for fetch-incidents. Given: - Configuration parameters. When: - Calling prepare_args_for_fetch_incidents with parameters. Then: - Ensure body preparation. """ from GoogleDriveApiModule import prepare_args_for_fetch_incidents params = { "action_detail_case_include": ["create", "edit"], } assert prepare_args_for_fetch_incidents(0, params) == { "filter": "time > 0 AND detail.action_detail_case: (CREATE EDIT)", "pageSize": 100, } with pytest.raises(ValueError, match=MESSAGES["FETCH_INCIDENT_REQUIRED_ARGS"]): prepare_args_for_fetch_incidents(0, {"drive_item_search_value": "a"}) def test_fetch_incidents(gsuite_client, mocker): """ Scenario: fetch_incidents called with valid arguments. Given: - Configuration parameters. When: - Calling fetch_incidents with parameters. Then: - Ensure successful execution of fetch_incidents. """ from GoogleDriveApiModule import fetch_incidents params = { "drive_item_search_field": "create", "drive_item_search_value": "create", "action_detail_case_include": ["create", "edit"], "user_id": "user@domain.io", } with open("test_data/fetch_incidents_response.json") as file: fetch_incidents_response = json.load(file) mocker.patch(MOCKER_HTTP_METHOD, return_value=fetch_incidents_response) with open("test_data/fetch_incidents_output.json") as file: fetch_incidents_output = json.load(file) mocker.patch(MOCKER_HTTP_METHOD, return_value=fetch_incidents_response) params["first_fetch"] = "10 day" params["max_incidents"] = 10 fetch_incident = fetch_incidents(gsuite_client, {}, params) assert fetch_incident[0] == fetch_incidents_output["incidents"] def test_main_fetch_incidents(mocker): """ Given working service integration When fetch-incidents is called from main() Then demistomock.incidents and demistomock.setLastRun should be called with respected values. :param args: Mocker objects. :return: None """ from GoogleDriveApiModule import demisto, run_google_drive_integration with open("test_data/fetch_incidents_output.json") as file: fetch_incidents_output = json.load(file) mocker.patch.object(demisto, "command", return_value="fetch-incidents") mocker.patch.object(demisto, "incidents") mocker.patch.object(demisto, "setLastRun") mocker.patch.object( demisto, "params", return_value={ "user_service_account_json": TEST_JSON, "max_incidents": 1, "first_fetch": "10 minutes", "isFetch": True, "user_id": "hellod", }, ) mocker.patch( "GoogleDriveApiModule.fetch_incidents", return_value=(fetch_incidents_output["incidents"], fetch_incidents_output["last_fetch"]), ) run_google_drive_integration() demisto.incidents.assert_called_once_with(fetch_incidents_output["incidents"]) demisto.setLastRun.assert_called_once_with(fetch_incidents_output["last_fetch"]) def test_flatten_move_keys_for_fetch_incident(): """ Scenario: Move action parents dictionary should be flatten. Given: - Move list. When: - Calling flatten_move_keys_for_fetch_incident with parameters. Then: - Ensure Dict is flatten as expected. """ from GoogleDriveApiModule import flatten_move_keys_for_fetch_incident name = "drive name" title = "drive title" output = { "driveitemname": name, "driveitemtitle": title, "driveitemisdrivefile": True, "driveitemfoldertype": "folder type", "drivename": name, "drivetitle": title, } move = { "addedParents": [ { "driveItem": {"name": name, "title": title, "driveFile": {}, "driveFolder": {"type": "folder type"}}, "drive": {"name": name, "title": title}, } ] } move["removedParents"] = move["addedParents"] flatten_move_keys_for_fetch_incident(move) assert move == {"addedParents": [output], "removedParents": [output]} def test_flatten_comment_mentioned_user_keys_for_fetch_incident(): """ Scenario: Move action parents dictionary should be flatten. Given: - Comment mentioned users list. When: - Calling flatten_comment_mentioned_user_keys_for_fetch_incident with parameters. Then: - Ensure Dict is flatten as expected. """ from GoogleDriveApiModule import flatten_comment_mentioned_user_keys_for_fetch_incident mentioned_users = { "mentionedUsers": [ {"knownUser": {"personName": "person name", "isCurrentUser": True}, "deletedUser": {}, "unknownUser": {}} ] } output = { "mentionedUsers": [{"personName": "person name", "isCurrentUser": True, "isDeletedUser": True, "isUnknownUser": True}] } flatten_comment_mentioned_user_keys_for_fetch_incident(mentioned_users) assert mentioned_users == output class TestDriveMethods: @patch(MOCKER_HTTP_METHOD) def test_drives_list_command_success(self, mocker_http_request, gsuite_client): """ Scenario: For google-drive-drives-list command successful run. Given: - Command args. When: - Calling google-drive-drives-list command with the parameters provided. Then: - Ensure command's raw_response, outputs should be as expected. """ from GoogleDriveApiModule import drives_list_command with open("test_data/drives_list_response.json", encoding="utf-8") as data: mock_response = json.load(data) mocker_http_request.return_value = mock_response args = {"use_domain_admin_access": True} result = drives_list_command(gsuite_client, args) assert "GoogleDrive.Drive" in result.outputs assert result.outputs.get("GoogleDrive.Drive").get("PageToken") == "myNextPageToken" assert len(result.outputs["GoogleDrive.Drive"].get("Drive")) == 4 assert result.raw_response == mock_response assert result.readable_output.startswith("### Total Retrieved Drive(s): ") assert HR_MESSAGES["LIST_COMMAND_SUCCESS"].format("Drive(s)", 4) in result.readable_output @patch(MOCKER_HTTP_METHOD) def test_drives_list_command_failure(self, mocker_http_request, gsuite_client): """ Scenario: For google-drives-list command failure. Given: - Command args and a non-working google api integration. When: - Calling google-drives-list command with the parameters provided. Then: - Ensure command's error response is as expected. """ mocker_http_request.side_effect = DemistoException("SOME_ERROR") from GoogleDriveApiModule import drives_list_command args = {"use_domain_admin_access": True} with pytest.raises(DemistoException, match="SOME_ERROR"): drives_list_command(gsuite_client, args) @patch(MOCKER_HTTP_METHOD) def test_drive_get_command_success(self, mocker_http_request, gsuite_client): """ Scenario: For google-drive-drive-get command successful run. Given: - Command args. When: - Calling google-drive-drive-get command with the parameters provided. Then: - Ensure command's raw_response, outputs should be as expected. """ from GoogleDriveApiModule import drive_get_command with open("test_data/drive_get_response.json", encoding="utf-8") as data: mock_response = json.load(data) mocker_http_request.return_value = mock_response args = {"use_domain_admin_access": True} result = drive_get_command(gsuite_client, args) assert "GoogleDrive.Drive" in result.outputs assert result.outputs.get("GoogleDrive.Drive").get("Drive").get("id") == "17" assert result.raw_response == mock_response assert HR_MESSAGES["LIST_COMMAND_SUCCESS"].format("Drive(s)", 1) in result.readable_output assert "17" in result.readable_output @patch(MOCKER_HTTP_METHOD) def test_drive_get_command_failure(self, mocker_http_request, gsuite_client): """ Scenario: For google-drive-get command failure. Given: - Command args and a non-working google api integration. When: - Calling google-drive-get command with the parameters provided. Then: - Ensure command's error response is as expected. """ mocker_http_request.side_effect = ValueError("SOME_ERROR") from GoogleDriveApiModule import drive_get_command args = {"use_domain_admin_access": True} with pytest.raises(ValueError, match="SOME_ERROR"): drive_get_command(gsuite_client, args) class TestFileMethods: @patch(MOCKER_HTTP_METHOD) def test_files_list_command_success(self, mocker_http_request, gsuite_client): """ Scenario: For google-drive-files-list command successful run. Given: - Command args. When: - Calling google-drive-files-list command with the parameters provided. Then: - Ensure command's raw_response, outputs should be as expected. """ from GoogleDriveApiModule import files_list_command with open("test_data/files_list_response.json", encoding="utf-8") as data: mock_response = json.load(data) mocker_http_request.return_value = mock_response args = {"use_domain_admin_access": True} result = files_list_command(gsuite_client, args) assert "GoogleDrive.File" in result.outputs assert result.outputs.get("GoogleDrive.File").get("PageToken") == "myNextPageToken" assert len(result.outputs["GoogleDrive.File"].get("File")) == 2 assert result.raw_response == mock_response assert result.readable_output.startswith("### Total Retrieved File(s): ") assert HR_MESSAGES["LIST_COMMAND_SUCCESS"].format("File(s)", 2) in result.readable_output @patch(MOCKER_HTTP_METHOD) def test_files_list_command_failure(self, mocker_http_request, gsuite_client): """ Scenario: For google-files-list command failure. Given: - Command args and a non-working google api integration. When: - Calling google-files-list command with the parameters provided. Then: - Ensure command's error response is as expected. """ mocker_http_request.side_effect = DemistoException("SOME_ERROR") from GoogleDriveApiModule import files_list_command args = {"use_domain_admin_access": True} with pytest.raises(DemistoException, match="SOME_ERROR"): files_list_command(gsuite_client, args) @patch(MOCKER_HTTP_METHOD) def test_file_get_command_success(self, mocker_http_request, gsuite_client): """ Scenario: For google-drive-file-get command successful run. Given: - Command args. When: - Calling google-drive-file-get command with the parameters provided. Then: - Ensure command's raw_response, outputs should be as expected. """ from GoogleDriveApiModule import file_get_command with open("test_data/file_get_response.json", encoding="utf-8") as data: mock_response = json.load(data) mocker_http_request.return_value = mock_response args = {"use_domain_admin_access": True} result = file_get_command(gsuite_client, args) assert "GoogleDrive.File" in result.outputs assert result.outputs.get("GoogleDrive.File").get("File").get("id") == "17" assert result.raw_response == mock_response assert HR_MESSAGES["LIST_COMMAND_SUCCESS"].format("File(s)", 1) in result.readable_output assert "17" in result.readable_output @patch(MOCKER_HTTP_METHOD) def test_file_get_command_failure(self, mocker_http_request, gsuite_client): """ Scenario: For google-file-get command failure. Given: - Command args and a non-working google api integration. When: - Calling google-file-get command with the parameters provided. Then: - Ensure command's error response is as expected. """ mocker_http_request.side_effect = ValueError("SOME_ERROR") from GoogleDriveApiModule import file_get_command args = {"use_domain_admin_access": True} with pytest.raises(ValueError, match="SOME_ERROR"): file_get_command(gsuite_client, args) class TestFilePermissionMethods: @patch(MOCKER_HTTP_METHOD) def test_file_permission_list_command_success(self, mocker_http_request, gsuite_client): """ Scenario: For google-drive-file-permission-list command successful run. Given: - Command args. When: - Calling google-drive-file-permission-list command with the parameters provided. Then: - Ensure command's raw_response, outputs should be as expected. """ from GoogleDriveApiModule import file_permission_list_command with open("test_data/file_permission_list_response.json", encoding="utf-8") as data: mock_response = json.load(data) mocker_http_request.return_value = mock_response args = {"use_domain_admin_access": True} result = file_permission_list_command(gsuite_client, args) assert "GoogleDrive.FilePermission" in result.outputs assert len(result.outputs["GoogleDrive.FilePermission"]) == 1 assert result.raw_response == mock_response assert result.readable_output.startswith("### Total") assert HR_MESSAGES["LIST_COMMAND_SUCCESS"].format("Permission(s)", 1) in result.readable_output @patch(MOCKER_HTTP_METHOD) def test_list_labels(self, mocker_http_request, gsuite_client): """ Scenario: For google-drive-list-labels command successful run. Given: - Command args. When: - Calling google-drive-list-labels command with the parameters provided. Then: - Ensure command's raw_response, outputs should be as expected. """ from GoogleDriveApiModule import get_labels_command with open("test_data/list_labels_response.json", encoding="utf-8") as data: mock_response = json.load(data) mocker_http_request.return_value = mock_response result = get_labels_command(gsuite_client, {}) assert "GoogleDrive.Labels" in result.outputs assert len(result.outputs["GoogleDrive.Labels"]["labels"]) == 2 @patch(MOCKER_HTTP_METHOD) def test_modify_labels_command(self, mocker_http_request, gsuite_client): """ Scenario: For google-drive-modify-label command successful run. Given: - Command args. When: - Calling google-drive-modify-label command with the parameters provided. Then: - Ensure command's raw_response, outputs should be as expected. """ from GoogleDriveApiModule import modify_label_command with open("test_data/modify_label_command_response.json", encoding="utf-8") as data: mock_response = json.load(data) mocker_http_request.return_value = mock_response args = {"field_id": "test", "selection_label_id": "test", "label_id": "test", "file_id": "test"} result = modify_label_command(gsuite_client, args) assert "modifiedLabels" in result.outputs.get("GoogleDrive.Labels") assert ( result.outputs.get("GoogleDrive.Labels").get("modifiedLabels")[0].get("id") == "vFmXsMA1fQMz1BdE59YSkisZV4DiKdpxxLQRNNEbbFcb" ) assert result.raw_response == mock_response assert HR_MESSAGES["MODIFY_LABEL_SUCCESS"].format(args.get("file_id")) in result.readable_output @patch(MOCKER_HTTP_METHOD) def test_file_permission_list_command_failure(self, mocker_http_request, gsuite_client): """ Scenario: For google-file-permission-list command failure. Given: - Command args and a non-working google api integration. When: - Calling google-file-permission-list command with the parameters provided. Then: - Ensure command's error response is as expected. """ mocker_http_request.side_effect = ValueError("SOME_ERROR") from GoogleDriveApiModule import file_permission_list_command args = {"use_domain_admin_access": True} with pytest.raises(ValueError, match="SOME_ERROR"): file_permission_list_command(gsuite_client, args) @patch(MOCKER_HTTP_METHOD) def test_file_permission_create_command_success(self, mocker_http_request, gsuite_client): """ Scenario: For google-drive-file-permission-create command successful run. Given: - Command args. When: - Calling google-drive-file-permission-create command with the parameters provided. Then: - Ensure command's raw_response, outputs should be as expected. """ from GoogleDriveApiModule import file_permission_create_command with open("test_data/file_permission_create_response.json", encoding="utf-8") as data: mock_response = json.load(data) mocker_http_request.return_value = mock_response args = {"use_domain_admin_access": True} result = file_permission_create_command(gsuite_client, args) assert "GoogleDrive.FilePermission" in result.outputs assert result.outputs.get("GoogleDrive.FilePermission").get("FilePermission").get("id") == "17" assert result.raw_response == mock_response assert HR_MESSAGES["LIST_COMMAND_SUCCESS"].format("Permission(s)", 1) in result.readable_output assert "17" in result.readable_output @patch(MOCKER_HTTP_METHOD) def test_file_permission_create_command_failure(self, mocker_http_request, gsuite_client): """ Scenario: For google-file-permission-create command failure. Given: - Command args and a non-working google api integration. When: - Calling google-file-permission-create command with the parameters provided. Then: - Ensure command's error response is as expected. """ mocker_http_request.side_effect = ValueError("SOME_ERROR") from GoogleDriveApiModule import file_permission_create_command args = {"use_domain_admin_access": True} with pytest.raises(ValueError, match="SOME_ERROR"): file_permission_create_command(gsuite_client, args) @patch(MOCKER_HTTP_METHOD) def test_file_permission_update_command_success(self, mocker_http_request, gsuite_client): """ Scenario: For google-drive-file-permission-update command successful run. Given: - Command args. When: - Calling google-drive-file-permission-update command with the parameters provided. Then: - Ensure command's raw_response, outputs should be as expected. """ from GoogleDriveApiModule import file_permission_update_command with open("test_data/file_permission_create_response.json", encoding="utf-8") as data: mock_response = json.load(data) mocker_http_request.return_value = mock_response args = {"use_domain_admin_access": True} result = file_permission_update_command(gsuite_client, args) assert "GoogleDrive.FilePermission" in result.outputs assert result.outputs.get("GoogleDrive.FilePermission").get("FilePermission").get("id") == "17" assert result.raw_response == mock_response assert HR_MESSAGES["LIST_COMMAND_SUCCESS"].format("Permission(s)", 1) in result.readable_output assert "17" in result.readable_output @patch(MOCKER_HTTP_METHOD) def test_file_permission_update_command_failure(self, mocker_http_request, gsuite_client): """ Scenario: For google-file-permission-update command failure. Given: - Command args and a non-working google api integration. When: - Calling google-file-permission-update command with the parameters provided. Then: - Ensure command's error response is as expected. """ mocker_http_request.side_effect = ValueError("SOME_ERROR") from GoogleDriveApiModule import file_permission_update_command args = {"use_domain_admin_access": True} with pytest.raises(ValueError, match="SOME_ERROR"): file_permission_update_command(gsuite_client, args) def test_upload_file_with_parent_command_success(self, mocker, gsuite_client): """ Scenario: For google-drive-file-upload command with the 'parent' arg. Given: - Command args. When: - Calling google-drive-file-upload command with the parent arg. Then: - Ensure parent arg send as expected by Google API (in array). """ import demistomock as demisto import GoogleDriveApiModule from GoogleDriveApiModule import file_upload_command mocker.patch("googleapiclient.http.HttpRequest.execute") mocker.patch("GoogleDriveApiModule.handle_response_file_single") mocker.patch("GoogleDriveApiModule.assign_params", return_value={}) mocker.patch.object( demisto, "getFilePath", return_value={"id": "test_id", "path": "test_data/drive_changes_hr.txt", "name": "drive_changes_hr.txt"}, ) args = {"parent": "test_parent", "entry_id": "test_entry_id", "file_name": "test_file_name"} file_upload_command(gsuite_client, args) assert GoogleDriveApiModule.assign_params.call_args[1]["parents"] == ["test_parent"] def test_file_copy_command(self, mocker, gsuite_client): """ Given: - A request to copy a Drive file. When: - Calling google-drive-file-copy. Then: - Copy the Drive file. """ from GoogleDriveApiModule import file_copy_command mocker.patch( "GoogleDriveApiModule.copy_file_http_request", return_value={ "id": "test_id", "kind": "drive#file", "mimeType": "application/octet-stream", "name": "TEST COPY", }, ) results = file_copy_command( gsuite_client, args={ "file_id": "test_file_id", "copy_title": "test_copy_title", "supports_all_drives": "true", "user_id": "test_user_id", }, ) assert results.outputs == { "id": "test_id", "kind": "drive#file", "mimeType": "application/octet-stream", "name": "TEST COPY", } assert results.readable_output == ( "### File copied successfully.\n" "|Id|Kind|Mimetype|Name|\n" "|---|---|---|---|\n" "| test_id | drive#file | application/octet-stream | TEST COPY |\n" ) def test_file_copy_command_error(self, mocker, gsuite_client): """ Given: - A request to copy a Drive file with an error. When: - Calling google-drive-file-copy. Then: - Return an error gracefully. """ from GoogleDriveApiModule import errors, file_copy_command def raise_error(): raise errors.HttpError(resp=type("MockRequest", (), {"status": 400, "reason": "Bad Request"}), content=b"Bad Request") mocker.patch("googleapiclient.http.HttpRequest.execute", side_effect=raise_error) with pytest.raises(DemistoException, match="Status Code: 400"): file_copy_command( gsuite_client, args={ "file_id": "test_file_id", "copy_title": "test_copy_title", "supports_all_drives": "true", "user_id": "test_user_id", }, ) @patch(MOCKER_HTTP_METHOD) def test_drive_get_file_parents_success(self, mocker_http_request, gsuite_client): """ Scenario: For file_get_parents command successful run. Given: - Command args. When: - Calling google-drive-get-file-parents command with the parameters provided. Then: - Ensure command's raw_response, outputs should be as expected. """ from GoogleDriveApiModule import file_get_parents with open("test_data/get_parents_list.txt", encoding="utf-8") as data: mock_response = json.load(data) mocker_http_request.return_value = mock_response args = {"use_domain_admin_access": True, "file_id": "test", "user_id": "test"} result: CommandResults = file_get_parents(gsuite_client, args) assert len(result.outputs.get("GoogleDrive.File.Parents", [])) == 1 # type: ignore assert result.raw_response == mock_response @patch(MOCKER_HTTP_METHOD) def test_file_move_command_success(self, mocker_http_request, gsuite_client): """ Scenario: For google-drive-file-move command successful run. Given: - Command args. When: - Calling google-drive-file-move command with the parameters provided. Then: - Ensure command's outputs and readable_output should be as expected. """ from GoogleDriveApiModule import file_move_command with open("test_data/file_move_response.json", encoding="utf-8") as data: mock_response = json.load(data) mocker_http_request.return_value = mock_response args = { "file_id": "1234567890abcdef", "add_parent_id": "quarantine_folder_id_123", "remove_parent_id": "original_folder_id_789", "user_id": "admin@example.com", } result: CommandResults = file_move_command(gsuite_client, args) assert result.outputs_prefix == "GoogleDrive.File" assert result.outputs_key_field == "id" assert result.outputs["id"] == "1234567890abcdef" assert result.outputs["parents"] == ["quarantine_folder_id_123"] assert "moved successfully" in result.readable_output @patch(MOCKER_HTTP_METHOD) def test_file_create_command_success(self, mocker_http_request, gsuite_client): """ Scenario: For google-drive-file-create command successful run for folder creation. Given: - Command args. When: - Calling google-drive-file-create command with the parameters provided. Then: - Ensure command's outputs and readable_output should be as expected. """ from GoogleDriveApiModule import file_create_command with open("test_data/file_create_response.json", encoding="utf-8") as data: mock_response = json.load(data) mocker_http_request.return_value = mock_response args = { "file_name": "Quarantine Folder", "mime_type": "application/vnd.google-apps.folder", "user_id": "admin@example.com", "parent": "root", } result: CommandResults = file_create_command(gsuite_client, args) assert result.outputs_prefix == "GoogleDrive.File" assert result.outputs_key_field == "id" assert result.outputs["id"] == "new_folder_id_456" assert result.outputs["mimeType"] == "application/vnd.google-apps.folder" assert "Created" in result.readable_output # Verify the request body and params _, call_kwargs = mocker_http_request.call_args assert call_kwargs["body"]["name"] == "Quarantine Folder" assert call_kwargs["body"]["parents"] == ["root"] # supports_all_drives defaults to False when not provided assert call_kwargs["params"]["supportsAllDrives"] is False @patch(MOCKER_HTTP_METHOD) def test_file_create_tombstone_command_success(self, mocker_http_request, gsuite_client): """ Scenario: For google-drive-file-create command successful run for tombstone placeholder creation. Given: - Command args for creating a tombstone placeholder file. When: - Calling google-drive-file-create command with the parameters provided. Then: - Ensure command's outputs should be as expected. """ from GoogleDriveApiModule import file_create_command mock_response = { "kind": "drive#file", "id": "tombstone_id_789", "name": "This file has been quarantined", "mimeType": "application/vnd.google-apps.document", "parents": ["original_folder_id"], "description": "This file was quarantined by security policy. Contact admin for details.", } mocker_http_request.return_value = mock_response args = { "file_name": "This file has been quarantined", "mime_type": "application/vnd.google-apps.document", "user_id": "admin@example.com", "parent": "original_folder_id", "description": "This file was quarantined by security policy. Contact admin for details.", } result: CommandResults = file_create_command(gsuite_client, args) assert result.outputs["id"] == "tombstone_id_789" assert result.outputs["mimeType"] == "application/vnd.google-apps.document" @patch(MOCKER_HTTP_METHOD) def test_file_create_command_supports_all_drives(self, mocker_http_request, gsuite_client): """ Scenario: For google-drive-file-create command run with supports_all_drives="true". Given: - Command args including supports_all_drives="true". When: - Calling google-drive-file-create command with the parameters provided. Then: - Ensure supportsAllDrives is forwarded to the API as True. """ from GoogleDriveApiModule import file_create_command mock_response = { "kind": "drive#file", "id": "shared_drive_folder_id", "name": "Shared Drive Folder", "mimeType": "application/vnd.google-apps.folder", "parents": ["shared_drive_root_id"], } mocker_http_request.return_value = mock_response args = { "file_name": "Shared Drive Folder", "mime_type": "application/vnd.google-apps.folder", "user_id": "admin@example.com", "parent": "shared_drive_root_id", "supports_all_drives": "true", } result: CommandResults = file_create_command(gsuite_client, args) assert result.outputs["id"] == "shared_drive_folder_id" _, call_kwargs = mocker_http_request.call_args assert call_kwargs["params"]["supportsAllDrives"] is True assert call_kwargs["body"]["name"] == "Shared Drive Folder" assert call_kwargs["body"]["parents"] == ["shared_drive_root_id"] @patch(MOCKER_HTTP_METHOD) def test_file_create_command_without_content_uses_metadata_only_path(self, mocker_http_request, gsuite_client): """ Scenario: google-drive-file-create is called without the content argument. Given: - Command args that do not include content. When: - Calling google-drive-file-create command. Then: - Ensure the metadata-only http_request path is used unchanged and no media is uploaded, preserving backward compatibility for existing playbooks. """ from GoogleDriveApiModule import file_create_command mock_response = { "kind": "drive#file", "id": "metadata_only_id", "name": "Placeholder.txt", "mimeType": "text/plain", } mocker_http_request.return_value = mock_response args = { "file_name": "Placeholder.txt", "mime_type": "text/plain", "user_id": "admin@example.com", } with patch("GoogleDriveApiModule.discovery.build") as mock_discovery_build: result: CommandResults = file_create_command(gsuite_client, args) # The content path must not be reached when content is not supplied. mock_discovery_build.assert_not_called() mocker_http_request.assert_called_once() _, call_kwargs = mocker_http_request.call_args assert call_kwargs["method"] == "POST" assert call_kwargs["body"] == {"name": "Placeholder.txt", "mimeType": "text/plain"} assert call_kwargs["params"]["supportsAllDrives"] is False assert result.outputs_prefix == "GoogleDrive.File" assert result.outputs["id"] == "metadata_only_id" @patch(MOCKER_HTTP_METHOD) def test_file_create_command_with_content_success(self, mocker_http_request, gsuite_client): """ Scenario: google-drive-file-create is called with the content argument. Given: - Command args including content and a non-folder mime_type. When: - Calling google-drive-file-create command. Then: - Ensure the multipart upload path is used, the media body carries the UTF-8 encoded content, and the outputs contract is unchanged. """ from GoogleDriveApiModule import file_create_command mock_response = { "kind": "drive#file", "id": "file_with_content_id", "name": "Notes.txt", "mimeType": "text/plain", "parents": ["root"], } mock_create = MagicMock() mock_create.execute.return_value = mock_response mock_drive_service = MagicMock() mock_drive_service.files.return_value.create.return_value = mock_create args = { "file_name": "Notes.txt", "mime_type": "text/plain", "user_id": "admin@example.com", "parent": "root", "description": "Investigation notes", "content": "hello world", } with patch("GoogleDriveApiModule.discovery.build", return_value=mock_drive_service): result: CommandResults = file_create_command(gsuite_client, args) # The metadata-only path must not be used when content is supplied. mocker_http_request.assert_not_called() _, create_kwargs = mock_drive_service.files.return_value.create.call_args assert create_kwargs["body"] == { "name": "Notes.txt", "mimeType": "text/plain", "parents": ["root"], "description": "Investigation notes", } assert create_kwargs["supportsAllDrives"] is False assert create_kwargs["fields"] == "*" media = create_kwargs["media_body"] assert media.mimetype() == "text/plain" assert media.getbytes(0, media.size()) == b"hello world" assert result.outputs_prefix == "GoogleDrive.File" assert result.outputs_key_field == "id" assert result.outputs["id"] == "file_with_content_id" assert "Created" in result.readable_output @patch(MOCKER_HTTP_METHOD) def test_file_create_command_with_content_and_folder_mime_type(self, mocker_http_request, gsuite_client): """ Scenario: google-drive-file-create is called with content while mime_type is a folder. Given: - Command args including content and the default folder mime_type. When: - Calling google-drive-file-create command. Then: - Ensure a DemistoException is raised, since folders cannot hold content, and that no API call is made. """ from GoogleDriveApiModule import file_create_command args = { "file_name": "Quarantine Folder", "user_id": "admin@example.com", "content": "hello world", } with patch("GoogleDriveApiModule.discovery.build") as mock_discovery_build: # noqa: SIM117 with pytest.raises(DemistoException, match="folders cannot hold content"): file_create_command(gsuite_client, args) mock_discovery_build.assert_not_called() mocker_http_request.assert_not_called() @patch(MOCKER_HTTP_METHOD) def test_file_create_command_content_at_max_length(self, mocker_http_request, gsuite_client): """ Scenario: google-drive-file-create is called with content exactly at the length limit. Given: - Command args with content of exactly MAX_CONTENT_LENGTH characters. When: - Calling google-drive-file-create command. Then: - Ensure the upload proceeds, confirming the boundary is inclusive. """ from GoogleDriveApiModule import MAX_CONTENT_LENGTH, file_create_command mock_response = {"kind": "drive#file", "id": "max_len_id", "name": "max.txt", "mimeType": "text/plain"} mock_create = MagicMock() mock_create.execute.return_value = mock_response mock_drive_service = MagicMock() mock_drive_service.files.return_value.create.return_value = mock_create content = "a" * MAX_CONTENT_LENGTH args = {"file_name": "max.txt", "mime_type": "text/plain", "content": content} with patch("GoogleDriveApiModule.discovery.build", return_value=mock_drive_service): result: CommandResults = file_create_command(gsuite_client, args) assert result.outputs["id"] == "max_len_id" _, create_kwargs = mock_drive_service.files.return_value.create.call_args media = create_kwargs["media_body"] assert media.size() == MAX_CONTENT_LENGTH mocker_http_request.assert_not_called() @patch(MOCKER_HTTP_METHOD) def test_file_create_command_content_exceeds_max_length(self, mocker_http_request, gsuite_client): """ Scenario: google-drive-file-create is called with content over the length limit. Given: - Command args with content of MAX_CONTENT_LENGTH + 1 characters. When: - Calling google-drive-file-create command. Then: - Ensure a DemistoException is raised before any API call is made, and that the message reports the actual length. """ from GoogleDriveApiModule import MAX_CONTENT_LENGTH, file_create_command over_limit = MAX_CONTENT_LENGTH + 1 args = {"file_name": "too-long.txt", "mime_type": "text/plain", "content": "a" * over_limit} with patch("GoogleDriveApiModule.discovery.build") as mock_discovery_build: # noqa: SIM117 with pytest.raises(DemistoException, match=f"must not exceed {MAX_CONTENT_LENGTH} characters, but got {over_limit}"): file_create_command(gsuite_client, args) mock_discovery_build.assert_not_called() mocker_http_request.assert_not_called() @patch(MOCKER_HTTP_METHOD) def test_file_delete_command_soft_delete_true(self, mocker_http_request, gsuite_client): """ Scenario: google-drive-file-delete invoked with soft_delete=true. Given: - file_id, user_id, and soft_delete=true. When: - Calling file_delete_command. Then: - The HTTP call is a PATCH (not DELETE) with body {"trashed": True}, and the response's trashed field is surfaced under GoogleDrive.File.File. """ from GoogleDriveApiModule import file_delete_command mock_response = { "kind": "drive#file", "id": "file_id_123", "name": "Quarterly Report.docx", "mimeType": "application/vnd.openxmlformats-officedocument.wordprocessingml.document", "trashed": True, "trashedTime": "2026-05-29T12:00:00.000Z", } mocker_http_request.return_value = mock_response args = {"file_id": "file_id_123", "user_id": "owner@example.com", "soft_delete": "true"} result: CommandResults = file_delete_command(gsuite_client, args) _, call_kwargs = mocker_http_request.call_args assert call_kwargs["method"] == "PATCH" assert call_kwargs["body"] == {"trashed": True} assert "drive/v3/files/file_id_123" in call_kwargs["url_suffix"] file_ctx = result.outputs.get("GoogleDrive.File").get("File") assert file_ctx.get("trashed") is True assert file_ctx.get("trashedTime") == "2026-05-29T12:00:00.000Z" @patch(MOCKER_HTTP_METHOD) def test_file_delete_command_soft_delete_default_hard_delete(self, mocker_http_request, gsuite_client): """ Scenario: google-drive-file-delete invoked without soft_delete. Given: - file_id, user_id, and no soft_delete argument (default behavior). When: - Calling file_delete_command. Then: - The HTTP call is a DELETE (existing behavior preserved bit-for-bit), no PATCH body is sent, and the legacy output shape is emitted. """ from GoogleDriveApiModule import file_delete_command mocker_http_request.return_value = None args = {"file_id": "file_id_456", "user_id": "owner@example.com"} result: CommandResults = file_delete_command(gsuite_client, args) _, call_kwargs = mocker_http_request.call_args assert call_kwargs["method"] == "DELETE" # The legacy DELETE path must not send a body. assert "body" not in call_kwargs or call_kwargs.get("body") is None file_ctx = result.outputs.get("GoogleDrive.File").get("File") assert file_ctx.get("id") == "file_id_456" assert "trashed" not in file_ctx @patch(MOCKER_HTTP_METHOD) def test_file_permission_create_command_transfer_ownership(self, mocker_http_request, gsuite_client): """ Scenario: google-drive-file-permission-create invoked with transfer_ownership=true. Given: - role=owner, type=user, email_address set, and transfer_ownership=true. When: - Calling file_permission_create_command. Then: - The transferOwnership query parameter is forwarded to the vendor; when the argument is omitted on a second call, the parameter is absent. """ from GoogleDriveApiModule import file_permission_create_command with open("test_data/file_permission_create_response.json", encoding="utf-8") as data: mock_response = json.load(data) mocker_http_request.return_value = mock_response args_with = { "file_id": "file_id_789", "user_id": "admin@example.com", "role": "owner", "type": "user", "email_address": "newowner@example.com", "transfer_ownership": "true", } file_permission_create_command(gsuite_client, args_with) _, call_kwargs_with = mocker_http_request.call_args assert call_kwargs_with["params"].get("transferOwnership") == "true" args_without = { "file_id": "file_id_789", "user_id": "admin@example.com", "role": "reader", "type": "user", "email_address": "viewer@example.com", } file_permission_create_command(gsuite_client, args_without) _, call_kwargs_without = mocker_http_request.call_args assert "transferOwnership" not in call_kwargs_without["params"] @patch(MOCKER_HTTP_METHOD) def test_file_permission_create_command_move_to_new_owners_root(self, mocker_http_request, gsuite_client): """ Scenario: google-drive-file-permission-create invoked with move_to_new_owners_root. Given: - role=owner, transfer_ownership=true, and move_to_new_owners_root=true. When: - Calling file_permission_create_command. Then: - The moveToNewOwnersRoot query parameter is forwarded to the vendor; when the argument is omitted on a second call, the parameter is absent. """ from GoogleDriveApiModule import file_permission_create_command with open("test_data/file_permission_create_response.json", encoding="utf-8") as data: mock_response = json.load(data) mocker_http_request.return_value = mock_response args_with = { "file_id": "file_id_789", "user_id": "admin@example.com", "role": "owner", "type": "user", "email_address": "newowner@example.com", "transfer_ownership": "true", "move_to_new_owners_root": "true", } file_permission_create_command(gsuite_client, args_with) _, call_kwargs_with = mocker_http_request.call_args assert call_kwargs_with["params"].get("moveToNewOwnersRoot") == "true" args_without = { "file_id": "file_id_789", "user_id": "admin@example.com", "role": "owner", "type": "user", "email_address": "newowner@example.com", "transfer_ownership": "true", } file_permission_create_command(gsuite_client, args_without) _, call_kwargs_without = mocker_http_request.call_args assert "moveToNewOwnersRoot" not in call_kwargs_without["params"] @patch(MOCKER_HTTP_METHOD) def test_file_permission_delete_command_ignore_not_found_404_treated_success(self, mocker_http_request, gsuite_client): """ Scenario: google-drive-file-permission-delete with ignore_not_found=true on a permission Not Found. Given: - The shared GSuite client raises DemistoException carrying the documented "Permission not found" reason from Drive. - ignore_not_found=true. When: - Calling file_permission_delete_command. Then: - The exception is swallowed and the result surfaces alreadyRemoved=True. - With ignore_not_found omitted, the same Not Found still raises (default behavior preserved). """ from GoogleDriveApiModule import file_permission_delete_command args = { "file_id": "file_id_999", "user_id": "admin@example.com", "permission_id": "perm_id_111", "ignore_not_found": "true", } # Real-world payload observed from production. mocker_http_request.side_effect = DemistoException("Not found. Reason: Permission not found: 12849315382336496719.") result: CommandResults = file_permission_delete_command(gsuite_client, args) perm_ctx = result.outputs.get("GoogleDrive.FilePermission").get("FilePermission") assert perm_ctx.get("alreadyRemoved") is True assert perm_ctx.get("id") == "perm_id_111" assert perm_ctx.get("fileId") == "file_id_999" # Default behavior preserved: with ignore_not_found omitted, the same # Not Found still raises. args_default = dict(args) args_default.pop("ignore_not_found") with pytest.raises(DemistoException, match=r"(?i)permission not found"): file_permission_delete_command(gsuite_client, args_default) @patch(MOCKER_HTTP_METHOD) def test_file_permission_delete_command_ignore_not_found_file_not_found_still_raises( self, mocker_http_request, gsuite_client ): """ Scenario: bogus file_id with ignore_not_found=true must still raise. Given: - The shared GSuite client raises DemistoException carrying "File not found" (the parent file does not exist, not the permission). - ignore_not_found=true. When: - Calling file_permission_delete_command. Then: - The exception still raises; only "permission not found" is swallowed so operator typos in file_id are not silently masked. """ from GoogleDriveApiModule import file_permission_delete_command mocker_http_request.side_effect = DemistoException("Not found. Reason: File not found: bogus_file_id_123.") args = { "file_id": "bogus_file_id_123", "user_id": "admin@example.com", "permission_id": "perm_id_111", "ignore_not_found": "true", } with pytest.raises(DemistoException, match=r"(?i)file not found"): file_permission_delete_command(gsuite_client, args) @patch(MOCKER_HTTP_METHOD) def test_file_permission_delete_command_ignore_not_found_other_error_still_raises(self, mocker_http_request, gsuite_client): """ Scenario: google-drive-file-permission-delete with ignore_not_found=true on non-Not Found. Given: - The shared GSuite client raises a Forbidden DemistoException (not a 404). - ignore_not_found=true. When: - Calling file_permission_delete_command. Then: - The exception is still raised; only Not Found is swallowed. """ from GoogleDriveApiModule import file_permission_delete_command mocker_http_request.side_effect = DemistoException("HTTP Connection error occurred. Status: 403. Reason: Forbidden") args = { "file_id": "file_id_999", "user_id": "admin@example.com", "permission_id": "perm_id_222", "ignore_not_found": "true", } with pytest.raises(DemistoException, match="Status: 403"): file_permission_delete_command(gsuite_client, args) @patch(MOCKER_HTTP_METHOD) def test_file_permission_list_command_inherited_output(self, mocker_http_request, gsuite_client): """ Scenario: google-drive-file-permissions-list surfaces permissionDetails. Given: - A vendor response containing permissionDetails with inherited=true. When: - Calling file_permission_list_command. Then: - The permissionDetails field is surfaced under GoogleDrive.FilePermission.FilePermission so playbooks can filter on it. """ from GoogleDriveApiModule import file_permission_list_command mock_response = { "kind": "drive#permissionList", "permissions": [ { "id": "perm_inherited", "type": "user", "role": "writer", "emailAddress": "shared@example.com", "permissionDetails": [ { "permissionType": "file", "role": "writer", "inheritedFrom": "parent_folder_id", "inherited": True, } ], } ], } mocker_http_request.return_value = mock_response args = {"file_id": "file_id_xyz", "user_id": "owner@example.com", "supports_all_drives": "true"} result: CommandResults = file_permission_list_command(gsuite_client, args) perm_ctx = result.outputs.get("GoogleDrive.FilePermission").get("FilePermission") # The list command surfaces an array; the helper may emit a single dict # when there is one permission. Normalize for the assertion. perms = perm_ctx if isinstance(perm_ctx, list) else [perm_ctx] assert perms[0].get("permissionDetails")[0].get("inherited") is True assert perms[0].get("permissionDetails")[0].get("inheritedFrom") == "parent_folder_id" assert perms[0].get("permissionDetails")[0].get("permissionType") == "file" """ get-file-content command tests """ @pytest.mark.parametrize( "url, expected_id", [ ("https://docs.google.com/document/d/test-file-id/edit", "test-file-id"), ("https://docs.google.com/spreadsheets/d/test-file-id/edit#gid=0", "test-file-id"), ("https://docs.google.com/presentation/d/test-file-id/edit", "test-file-id"), ("https://docs.google.com/forms/d/test-file-id/edit", "test-file-id"), ("https://drive.google.com/file/d/test-file-id/view", "test-file-id"), ("https://drive.google.com/open?id=test-file-id", "test-file-id"), ], ) def test_extract_file_id_from_url(url: str, expected_id: str): """ Given: A Google Drive file URL in various supported formats. When: Calling _extract_file_id_from_url to parse the URL. Then: The correct file ID is extracted from the URL. """ from GoogleDriveApiModule import _extract_file_id_from_url assert _extract_file_id_from_url(url) == expected_id def test_extract_file_id_from_url_invalid(): """ Given: An unsupported URL format that does not contain a Google Drive file ID. When: Calling _extract_file_id_from_url to parse the URL. Then: A ValueError is raised with a descriptive error message. """ from GoogleDriveApiModule import _extract_file_id_from_url with pytest.raises(ValueError, match="Could not extract file ID from URL"): _extract_file_id_from_url("https://example.com/not-a-drive-url") @pytest.mark.parametrize( "file_name, mime_type, expected", [ ("notes.csv", None, True), ("report.docx", None, True), ("data.json", None, True), ("archive.zip", None, False), ("", "application/vnd.google-apps.document", True), ("", "application/vnd.google-apps.spreadsheet", True), ("", "application/pdf", False), ("", None, False), ], ) def test_is_approved_file(file_name, mime_type, expected): """ Given: A file name and/or MIME type (including the case where the file has no name). When: Calling is_approved_file. Then: Approval is decided by extension first and falls back to the MIME type mapping. """ from GoogleDriveApiModule import is_approved_file assert is_approved_file(file_name, mime_type) is expected @pytest.mark.parametrize( "mime, expected", [ ("text/csv", True), ("text/plain", True), ("application/json", True), ("image/svg+xml", True), ("application/pdf", False), ("image/png", False), ("", False), ], ) def test_is_text_mime(mime, expected): """ Given: A MIME type. When: Calling _is_text_mime. Then: Text-decodable MIME types return True and binary types return False. """ from GoogleDriveApiModule import _is_text_mime assert _is_text_mime(mime) is expected @patch(MOCKER_HTTP_METHOD) def test_get_file_content_command_google_doc(mocker_http_request, gsuite_client, mocker): """ Given: A Google Docs URL pointing to a Google Workspace document (exported as Markdown). When: Calling the get-file-content command with the URL. Then: Content holds the decoded Markdown text and Type is 'text/markdown' (no ';base64'). """ from GoogleDriveApiModule import get_file_content_command mock_metadata = { "id": "test-file-id", "name": "test_document", "mimeType": "application/vnd.google-apps.document", "description": "Document used by unit tests.", "webViewLink": "https://docs.google.com/document/d/test-file-id/edit", } mocker_http_request.return_value = mock_metadata mocker.patch.object(demisto, "callingContext", {"context": {"User": {"email": "user@example.com"}}}) markdown_bytes = b"# Title\n\nSome **bold** text.\n" mock_drive_service = MagicMock() mock_drive_service.files().export().execute.return_value = markdown_bytes with patch("GoogleDriveApiModule.discovery.build", return_value=mock_drive_service): args = {"url": "https://docs.google.com/document/d/test-file-id/edit"} result = get_file_content_command(gsuite_client, args) assert result.outputs_prefix == "FileContent" assert result.outputs["Id"] == "test-file-id" assert result.outputs["Title"] == "test_document" assert result.outputs["Type"] == "text/markdown" assert result.outputs["Content"] == "# Title\n\nSome **bold** text.\n" @patch(MOCKER_HTTP_METHOD) def test_get_file_content_command_google_sheet_text_export(mocker_http_request, gsuite_client, mocker): """ Given: A Google Sheets URL pointing to a Google Workspace spreadsheet (exported as CSV). When: Calling the get-file-content command with the URL. Then: Content is the decoded CSV text and Type is 'text/csv' (no ';base64' suffix). """ from GoogleDriveApiModule import get_file_content_command mock_metadata = { "id": "test-sheet-id", "name": "test_sheet", "mimeType": "application/vnd.google-apps.spreadsheet", "webViewLink": "https://docs.google.com/spreadsheets/d/test-sheet-id/edit", } mocker_http_request.return_value = mock_metadata mocker.patch.object(demisto, "callingContext", {"context": {"User": {"email": "user@example.com"}}}) csv_bytes = b"a,b,c\n1,2,3\n" mock_drive_service = MagicMock() mock_drive_service.files().export().execute.return_value = csv_bytes with patch("GoogleDriveApiModule.discovery.build", return_value=mock_drive_service): args = {"url": "https://docs.google.com/spreadsheets/d/test-sheet-id/edit"} result = get_file_content_command(gsuite_client, args) assert result.outputs["Type"] == "text/csv" assert result.outputs["Content"] == "a,b,c\n1,2,3\n" @patch(MOCKER_HTTP_METHOD) def test_get_file_content_command_regular_text_file(mocker_http_request, gsuite_client, mocker): """ Given: A Google Drive URL pointing to a regular CSV file uploaded directly to Drive. When: Calling the get-file-content command with the URL. Then: The command downloads raw bytes via get_media and returns decoded text inline in Content. """ from GoogleDriveApiModule import get_file_content_command mock_metadata = { "id": "test-csv-id", "name": "test_notes.csv", "mimeType": "text/csv", "webViewLink": "https://drive.google.com/file/d/test-csv-id/view", } mocker_http_request.return_value = mock_metadata mocker.patch.object(demisto, "callingContext", {"context": {"User": {"email": "user@example.com"}}}) mock_drive_service = MagicMock() mock_drive_service.files().get_media().execute.return_value = b"hello,world\n" with patch("GoogleDriveApiModule.discovery.build", return_value=mock_drive_service): args = {"url": "https://drive.google.com/file/d/test-csv-id/view"} result = get_file_content_command(gsuite_client, args) assert result.outputs["Type"] == "text/csv" assert result.outputs["Content"] == "hello,world\n" @patch(MOCKER_HTTP_METHOD) def test_get_file_content_command_binary_file_base64(mocker_http_request, gsuite_client, mocker): """ Given: A Google Drive URL pointing to a regular binary file (a .docx uploaded to Drive). When: Calling the get-file-content command with the URL. Then: Content holds the base64-encoded bytes and Type is suffixed with ';base64' so downstream consumers know Content must be base64-decoded. """ import base64 as _b64 from GoogleDriveApiModule import get_file_content_command docx_mime = "application/vnd.openxmlformats-officedocument.wordprocessingml.document" mock_metadata = { "id": "test-docx-id", "name": "report.docx", "mimeType": docx_mime, "webViewLink": "https://drive.google.com/file/d/test-docx-id/view", } mocker_http_request.return_value = mock_metadata mocker.patch.object(demisto, "callingContext", {"context": {"User": {"email": "user@example.com"}}}) raw_bytes = b"PK\x03\x04binary-docx-bytes\x00\xff" mock_drive_service = MagicMock() mock_drive_service.files().get_media().execute.return_value = raw_bytes with patch("GoogleDriveApiModule.discovery.build", return_value=mock_drive_service): args = {"url": "https://drive.google.com/file/d/test-docx-id/view"} result = get_file_content_command(gsuite_client, args) assert result.outputs["Type"] == f"{docx_mime};base64" # Content is base64 and decodes back to the exact original bytes. assert result.outputs["Content"] == _b64.b64encode(raw_bytes).decode("ascii") assert _b64.b64decode(result.outputs["Content"]) == raw_bytes @patch(MOCKER_HTTP_METHOD) def test_get_file_content_command_unapproved_format(mocker_http_request, gsuite_client, mocker): """ Given: A Drive URL pointing to a file whose format is not approved (e.g. a PDF). When: Calling the get-file-content command. Then: A ValueError is raised and the message lists the supported formats. """ from GoogleDriveApiModule import get_file_content_command mock_metadata = { "id": "test-pdf-id", "name": "report.pdf", "mimeType": "application/pdf", } mocker_http_request.return_value = mock_metadata mocker.patch.object(demisto, "callingContext", {"context": {"User": {"email": "user@example.com"}}}) with pytest.raises(ValueError, match="File format not approved.*Supported formats"): get_file_content_command(gsuite_client, {"url": "https://drive.google.com/file/d/test-pdf-id/view"}) @patch(MOCKER_HTTP_METHOD) def test_get_file_content_command_exceeds_size_limit(mocker_http_request, gsuite_client, mocker): """ Given: A Drive URL pointing to an approved file whose reported size exceeds the 5 MB limit. When: Calling the get-file-content command. Then: A ValueError is raised indicating the file exceeds the maximum allowed size. """ from GoogleDriveApiModule import MAX_FILE_SIZE, get_file_content_command mock_metadata = { "id": "big-id", "name": "big.csv", "mimeType": "text/csv", "size": str(MAX_FILE_SIZE + 1), } mocker_http_request.return_value = mock_metadata mocker.patch.object(demisto, "callingContext", {"context": {"User": {"email": "user@example.com"}}}) with pytest.raises(ValueError, match="exceeds the maximum allowed size"): get_file_content_command(gsuite_client, {"url": "https://drive.google.com/file/d/big-id/view"}) def test_get_file_content_command_missing_url(gsuite_client): """ Given: No URL argument provided to the get-file-content command. When: Calling the get-file-content command without a URL. Then: A ValueError is raised indicating the url argument is required. """ from GoogleDriveApiModule import get_file_content_command with pytest.raises(ValueError, match="'url' argument is required"): get_file_content_command(gsuite_client, {}) def test_get_file_content_command_missing_user_email(gsuite_client, mocker): """ Given: A valid URL but no logged-in user email available in the calling context. When: Calling the get-file-content command. Then: A ValueError is raised indicating the logged-in user's email could not be determined. """ from GoogleDriveApiModule import get_file_content_command mocker.patch.object(demisto, "callingContext", {"context": {"User": {}}}) with pytest.raises(ValueError, match="Could not determine the email of the logged-in user"): get_file_content_command(gsuite_client, {"url": "https://drive.google.com/file/d/test-file-id/view"}) @patch(MOCKER_HTTP_METHOD) def test_get_file_content_command_access_denied(mocker_http_request, gsuite_client, mocker): """ Given: A valid URL and logged-in user email, but the API request fails with a token refresh/access_denied error (file not shared with the user). When: Calling the get-file-content command. Then: A DemistoException is raised with an informative message asking the user to verify the file is shared with their email, while preserving the original error. """ from GoogleDriveApiModule import get_file_content_command mocker.patch.object(demisto, "callingContext", {"context": {"User": {"email": "user@example.com"}}}) mocker_http_request.side_effect = DemistoException( "Failed to generate/refresh token. Subject email or service account credentials are invalid. " "Reason: access_denied: Requested client not authorized." ) with pytest.raises(DemistoException, match="Ensure the file is shared with") as exc_info: get_file_content_command(gsuite_client, {"url": "https://drive.google.com/file/d/test-file-id/view"}) # The original error must be preserved (chained and/or embedded in the message). assert "access_denied" in str(exc_info.value) or exc_info.value.__cause__ is not None @patch(MOCKER_HTTP_METHOD) def test_get_file_content_command_other_demisto_exception_propagates(mocker_http_request, gsuite_client, mocker): """ Given: A valid URL and logged-in user email, but the API request fails with an unrelated error (not an auth/refresh issue). When: Calling the get-file-content command. Then: The original DemistoException is propagated unchanged. """ from GoogleDriveApiModule import get_file_content_command mocker.patch.object(demisto, "callingContext", {"context": {"User": {"email": "user@example.com"}}}) mocker_http_request.side_effect = DemistoException("Some unrelated network failure.") with pytest.raises(DemistoException, match="Some unrelated network failure"): get_file_content_command(gsuite_client, {"url": "https://drive.google.com/file/d/test-file-id/view"})