users deactivate and reactivate commands (#317)

* users deactivate and reactivate commands

* d>r

Co-authored-by: tim.abramson <tim.abramson@code42.com>

Author: Juliya Smith

CHANGELOG.md

@@ -8,6 +8,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 The intended audience of this file is for py42 consumers -- as such, changes that don't affect
 how a consumer would use the library (e.g. adding unit tests, updating documentation, etc) are not captured here.
 
+## Unreleased
+
+### Added
+
+- New commands:
+  - `code42 users deactivate`
+  - `code42 users reactivate`
+  - `code42 users bulk deactivate`
+  - `code42 users bulk reactivate`
+
 ## 1.9.0 - 2021-08-19
 
 ### Added

src/code42cli/click_ext/groups.py

@@ -3,6 +3,7 @@
 from collections import OrderedDict
 
 import click
+from py42.exceptions import Py42ActiveLegalHoldError
 from py42.exceptions import Py42CaseAlreadyHasEventError
 from py42.exceptions import Py42CaseNameExistsError
 from py42.exceptions import Py42DescriptionLimitExceededError
@@ -75,6 +76,7 @@ def invoke(self, ctx):
             Py42InvalidEmailError,
             Py42InvalidPasswordError,
             Py42InvalidUsernameError,
+            Py42ActiveLegalHoldError,
         ) as err:
             self.logger.log_error(err)
             raise Code42CLIError(str(err))

src/code42cli/cmds/users.py

@@ -36,12 +36,9 @@ def users(state):
     help="Limits results to only deactivated users.",
     cls=incompatible_with("active"),
 )
-
-
-user_uid_option = click.option(
+user_id_option = click.option(
     "--user-id", help="The unique identifier of the user to be modified.", required=True
 )
-
 org_id_option = click.option(
     "--org-id",
     help="The identifier for the organization to which the user will be moved.",
@@ -101,7 +98,7 @@ def remove_role(state, username, role_name):
 
 
 @users.command(name="update")
-@user_uid_option
+@user_id_option
 @click.option("--username", help="The new username for the user.")
 @click.option("--password", help="The new password for the user.")
 @click.option("--email", help="The new email for the user.")
@@ -137,6 +134,24 @@ def update_user(
     )
 
 
+@users.command()
+@click.argument("username")
+@sdk_options()
+def deactivate(state, username):
+    """Deactivate a user."""
+    sdk = state.sdk
+    _deactivate_user(sdk, username)
+
+
+@users.command()
+@click.argument("username")
+@sdk_options()
+def reactivate(state, username):
+    """Reactivate a user."""
+    sdk = state.sdk
+    _reactivate_user(sdk, username)
+
+
 _bulk_user_update_headers = [
     "user_id",
     "username",
@@ -259,19 +274,100 @@ def handle_row(**row):
     formatter.echo_formatted_list(result_rows)
 
 
+_bulk_user_activation_headers = ["username"]
+
+
+@bulk.command(
+    name="deactivate",
+    help=f"Deactivate a list of users from the provided CSV in format: {','.join(_bulk_user_activation_headers)}",
+)
+@read_csv_arg(headers=_bulk_user_activation_headers)
+@format_option
+@sdk_options()
+def bulk_deactivate(state, csv_rows, format):
+    """Deactivate a list of users."""
+
+    # Initialize the SDK before starting any bulk processes
+    # to prevent multiple instances and having to enter 2fa multiple times.
+    sdk = state.sdk
+
+    csv_rows[0]["deactivated"] = "False"
+    formatter = OutputFormatter(format, {key: key for key in csv_rows[0].keys()})
+    stats = create_worker_stats(len(csv_rows))
+
+    def handle_row(**row):
+        try:
+            _deactivate_user(
+                sdk, **{key: row[key] for key in row.keys() if key != "deactivated"}
+            )
+            row["deactivated"] = "True"
+        except Exception as err:
+            row["deactivated"] = f"False: {err}"
+            stats.increment_total_errors()
+        return row
+
+    result_rows = run_bulk_process(
+        handle_row,
+        csv_rows,
+        progress_label="Deactivating users:",
+        stats=stats,
+        raise_global_error=False,
+    )
+    formatter.echo_formatted_list(result_rows)
+
+
+@bulk.command(
+    name="reactivate",
+    help=f"Reactivate a list of users from the provided CSV in format: {','.join(_bulk_user_activation_headers)}",
+)
+@read_csv_arg(headers=_bulk_user_activation_headers)
+@format_option
+@sdk_options()
+def bulk_reactivate(state, csv_rows, format):
+    """Reactivate a list of users."""
+
+    # Initialize the SDK before starting any bulk processes
+    # to prevent multiple instances and having to enter 2fa multiple times.
+    sdk = state.sdk
+
+    csv_rows[0]["reactivated"] = "False"
+    formatter = OutputFormatter(format, {key: key for key in csv_rows[0].keys()})
+    stats = create_worker_stats(len(csv_rows))
+
+    def handle_row(**row):
+        try:
+            _reactivate_user(
+                sdk, **{key: row[key] for key in row.keys() if key != "reactivated"}
+            )
+            row["reactivated"] = "True"
+        except Exception as err:
+            row["reactivated"] = f"False: {err}"
+            stats.increment_total_errors()
+        return row
+
+    result_rows = run_bulk_process(
+        handle_row,
+        csv_rows,
+        progress_label="Reactivating users:",
+        stats=stats,
+        raise_global_error=False,
+    )
+    formatter.echo_formatted_list(result_rows)
+
+
 def _add_user_role(sdk, username, role_name):
-    user_id = _get_user_id(sdk, username)
+    user_id = _get_legacy_user_id(sdk, username)
     _get_role_id(sdk, role_name)  # function provides role name validation
     sdk.users.add_role(user_id, role_name)
 
 
 def _remove_user_role(sdk, role_name, username):
-    user_id = _get_user_id(sdk, username)
+    user_id = _get_legacy_user_id(sdk, username)
     _get_role_id(sdk, role_name)  # function provides role name validation
     sdk.users.remove_role(user_id, role_name)
 
 
-def _get_user_id(sdk, username):
+def _get_legacy_user_id(sdk, username):
     if not username:
         # py42 returns all users when passing `None` to `get_by_username()`.
         raise click.BadParameter("Username is required.")
@@ -326,11 +422,21 @@ def _update_user(
 
 
 def _change_organization(sdk, username, org_id):
-    user_id = _get_user_id(sdk, username)
+    user_id = _get_legacy_user_id(sdk, username)
     org_id = _get_org_id(sdk, org_id)
     return sdk.users.change_org_assignment(user_id=int(user_id), org_id=int(org_id))
 
 
 def _get_org_id(sdk, org_id):
     org = sdk.orgs.get_by_uid(org_id)
     return org["orgId"]
+
+
+def _deactivate_user(sdk, username):
+    user_id = _get_legacy_user_id(sdk, username)
+    sdk.users.deactivate(user_id)
+
+
+def _reactivate_user(sdk, username):
+    user_id = _get_legacy_user_id(sdk, username)
+    sdk.users.reactivate(user_id)

tests/cmds/test_users.py

@@ -1,4 +1,5 @@
 import pytest
+from py42.exceptions import Py42ActiveLegalHoldError
 from py42.exceptions import Py42InvalidEmailError
 from py42.exceptions import Py42InvalidPasswordError
 from py42.exceptions import Py42InvalidUsernameError
@@ -130,6 +131,23 @@ def update_user_success(cli_state, update_user_response):
     cli_state.sdk.users.update_user.return_value = update_user_response
 
 
+@pytest.fixture
+def deactivate_user_success(mocker, cli_state):
+    cli_state.sdk.users.deactivate.return_value = create_mock_response(mocker)
+
+
+@pytest.fixture
+def deactivate_user_legal_hold_failure(mocker, cli_state):
+    cli_state.sdk.users.deactivate.side_effect = Py42ActiveLegalHoldError(
+        create_mock_http_error(mocker, status=400), "user", TEST_USER_ID
+    )
+
+
+@pytest.fixture
+def reactivate_user_success(mocker, cli_state):
+    cli_state.sdk.users.deactivate.return_value = create_mock_response(mocker)
+
+
 @pytest.fixture
 def change_org_success(cli_state, change_org_response):
     cli_state.sdk.users.change_org_assignment.return_value = change_org_response
@@ -433,6 +451,33 @@ def test_update_when_py42_raises_invalid_password_outputs_error_message(
     assert "Error: Invalid password." in result.output
 
 
+def test_deactivate_calls_deactivate_with_correct_parameters(
+    runner, cli_state, get_user_id_success, deactivate_user_success
+):
+    command = ["users", "deactivate", "test@example.com"]
+    runner.invoke(cli, command, obj=cli_state)
+    cli_state.sdk.users.deactivate.assert_called_once_with(TEST_USER_ID)
+
+
+def test_deactivate_when_user_on_legal_hold_outputs_expected_error_text(
+    runner, cli_state, get_user_id_success, deactivate_user_legal_hold_failure
+):
+    command = ["users", "deactivate", "test@example.com"]
+    result = runner.invoke(cli, command, obj=cli_state)
+    assert (
+        "Error: Cannot deactivate the user with ID 1234 as the user is involved in a legal hold matter."
+        in result.output
+    )
+
+
+def test_reactivate_calls_reactivate_with_correct_parameters(
+    runner, cli_state, get_user_id_success, deactivate_user_success
+):
+    command = ["users", "reactivate", "test@example.com"]
+    runner.invoke(cli, command, obj=cli_state)
+    cli_state.sdk.users.reactivate.assert_called_once_with(TEST_USER_ID)
+
+
 def test_bulk_update_uses_expected_arguments_when_only_some_are_passed(
     runner, mocker, cli_state
 ):
@@ -655,3 +700,120 @@ def test_bulk_move_uses_handle_than_when_called_and_row_has_missing_username_err
     assert worker_stats.increment_total_errors.call_count == 1
     # Ensure it does not try to get the username for the None user.
     assert not cli_state.sdk.users.get_by_username.call_count
+
+
+def test_bulk_deactivate_uses_expected_arguments(runner, mocker, cli_state):
+    bulk_processor = mocker.patch(f"{_NAMESPACE}.run_bulk_process")
+    with runner.isolated_filesystem():
+        with open("test_bulk_deactivate.csv", "w") as csv:
+            csv.writelines(["username\n", f"{TEST_USERNAME}\n"])
+        runner.invoke(
+            cli,
+            ["users", "bulk", "deactivate", "test_bulk_deactivate.csv"],
+            obj=cli_state,
+        )
+    assert bulk_processor.call_args[0][1] == [
+        {"username": TEST_USERNAME, "deactivated": "False"}
+    ]
+    bulk_processor.assert_called_once()
+
+
+def test_bulk_deactivate_ignores_blank_lines(runner, mocker, cli_state):
+    bulk_processor = mocker.patch(f"{_NAMESPACE}.run_bulk_process")
+    with runner.isolated_filesystem():
+        with open("test_bulk_deactivate.csv", "w") as csv:
+            csv.writelines(["username\n\n\n", f"{TEST_USERNAME}\n\n\n"])
+        runner.invoke(
+            cli,
+            ["users", "bulk", "deactivate", "test_bulk_deactivate.csv"],
+            obj=cli_state,
+        )
+    assert bulk_processor.call_args[0][1] == [
+        {"username": TEST_USERNAME, "deactivated": "False"}
+    ]
+    bulk_processor.assert_called_once()
+
+
+def test_bulk_deactivate_uses_handler_that_when_encounters_error_increments_total_errors(
+    runner, mocker, cli_state, worker_stats, get_users_response
+):
+    lines = ["username\n", f"{TEST_USERNAME}\n"]
+
+    def _get(username, *args, **kwargs):
+        if username == "test@example.com":
+            raise Exception("TEST")
+        return get_users_response
+
+    cli_state.sdk.users.get_by_username.side_effect = _get
+    bulk_processor = mocker.patch(f"{_NAMESPACE}.run_bulk_process")
+    with runner.isolated_filesystem():
+        with open("test_bulk_deactivate.csv", "w") as csv:
+            csv.writelines(lines)
+        runner.invoke(
+            cli,
+            ["users", "bulk", "deactivate", "test_bulk_deactivate.csv"],
+            obj=cli_state,
+        )
+    handler = bulk_processor.call_args[0][0]
+    handler(username="test@example.com")
+    handler(username="not.test@example.com")
+    assert worker_stats.increment_total_errors.call_count == 1
+
+
+def test_bulk_reactivate_uses_expected_arguments(runner, mocker, cli_state):
+    bulk_processor = mocker.patch(f"{_NAMESPACE}.run_bulk_process")
+    with runner.isolated_filesystem():
+        with open("test_bulk_reactivate.csv", "w") as csv:
+            csv.writelines(["username\n", f"{TEST_USERNAME}\n"])
+        runner.invoke(
+            cli,
+            ["users", "bulk", "reactivate", "test_bulk_reactivate.csv"],
+            obj=cli_state,
+        )
+    assert bulk_processor.call_args[0][1] == [
+        {"username": TEST_USERNAME, "reactivated": "False"}
+    ]
+    bulk_processor.assert_called_once()
+
+
+def test_bulk_reactivate_ignores_blank_lines(runner, mocker, cli_state):
+    bulk_processor = mocker.patch(f"{_NAMESPACE}.run_bulk_process")
+    with runner.isolated_filesystem():
+        with open("test_bulk_reactivate.csv", "w") as csv:
+            csv.writelines(["username\n\n\n", f"{TEST_USERNAME}\n\n\n"])
+        runner.invoke(
+            cli,
+            ["users", "bulk", "reactivate", "test_bulk_reactivate.csv"],
+            obj=cli_state,
+        )
+    assert bulk_processor.call_args[0][1] == [
+        {"username": TEST_USERNAME, "reactivated": "False"}
+    ]
+    bulk_processor.assert_called_once()
+
+
+def test_bulk_reactivate_uses_handler_that_when_encounters_error_increments_total_errors(
+    runner, mocker, cli_state, worker_stats, get_users_response
+):
+    lines = ["username\n", f"{TEST_USERNAME}\n"]
+
+    def _get(username, *args, **kwargs):
+        if username == "test@example.com":
+            raise Exception("TEST")
+
+        return get_users_response
+
+    cli_state.sdk.users.get_by_username.side_effect = _get
+    bulk_processor = mocker.patch(f"{_NAMESPACE}.run_bulk_process")
+    with runner.isolated_filesystem():
+        with open("test_bulk_reactivate.csv", "w") as csv:
+            csv.writelines(lines)
+        runner.invoke(
+            cli,
+            ["users", "bulk", "reactivate", "test_bulk_reactivate.csv"],
+            obj=cli_state,
+        )
+    handler = bulk_processor.call_args[0][0]
+    handler(username="test@example.com")
+    handler(username="not.test@example.com")
+    assert worker_stats.increment_total_errors.call_count == 1