Users command: option to report on legal hold membership

CHANGELOG.md

@@ -8,6 +8,12 @@ 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 command 'code42 users list --include-legal-hold-membership' to print the legal hold matter name and ID for any user on legal hold.
+
 ## 1.9.0 - 2021-08-19
 
 ### Added

src/code42cli/cmds/users.py

@@ -1,5 +1,6 @@
 import click
 from pandas import DataFrame
+from pandas import json_normalize
 
 from code42cli.bulk import generate_template_cmd_factory
 from code42cli.bulk import run_bulk_process
@@ -62,9 +63,17 @@ def username_option(help, required=False):
 @role_name_option("Limit results to only users having the specified role.")
 @active_option
 @inactive_option
+@click.option(
+    "--include-legal-hold-membership",
+    default=False,
+    is_flag=True,
+    help="Include legal hold membership in output.",
+)
 @format_option
 @sdk_options()
-def list_users(state, org_uid, role_name, active, inactive, format):
+def list_users(
+    state, org_uid, role_name, active, inactive, include_legal_hold_membership, format
+):
     """List users in your Code42 environment."""
     if inactive:
         active = False
@@ -75,6 +84,8 @@ def list_users(state, org_uid, role_name, active, inactive, format):
         else None
     )
     df = _get_users_dataframe(state.sdk, columns, org_uid, role_id, active)
+    if include_legal_hold_membership:
+        df = _add_legal_hold_membership_to_user_dataframe(state.sdk, df)
     if df.empty:
         click.echo("No results found.")
     else:
@@ -302,6 +313,44 @@ def _get_users_dataframe(sdk, columns, org_uid, role_id, active):
     return DataFrame.from_records(users_list, columns=columns)
 
 
+def _add_legal_hold_membership_to_user_dataframe(sdk, df):
+    columns = ["legalHold.legalHoldUid", "legalHold.name", "user.userUid"]
+
+    custodians = list(_get_all_active_hold_memberships(sdk))
+    if len(custodians) == 0:
+        return df
+
+    legal_hold_member_dataframe = (
+        json_normalize(custodians)[columns]
+        .groupby(["user.userUid"])
+        .agg(",".join)
+        .rename(
+            {
+                "legalHold.legalHoldUid": "legalHoldUid",
+                "legalHold.name": "legalHoldName",
+            },
+            axis=1,
+        )
+    )
+    df = df.merge(
+        legal_hold_member_dataframe,
+        how="left",
+        left_on="userUid",
+        right_on="user.userUid",
+    )
+
+    return df
+
+
+def _get_all_active_hold_memberships(sdk):
+    for page in sdk.legalhold.get_all_matters(active=True):
+        for matter in page["legalHolds"]:
+            for _page in sdk.legalhold.get_all_matter_custodians(
+                legal_hold_uid=matter["legalHoldUid"], active=True
+            ):
+                yield from _page["legalHoldMemberships"]
+
+
 def _update_user(
     sdk,
     user_id,

tests/cmds/test_users.py

@@ -34,6 +34,42 @@
         }
     ]
 }
+TEST_MATTER_RESPONSE = {
+    "legalHolds": [
+        {"legalHoldUid": "123456789", "name": "Legal Hold #1", "active": True},
+        {"legalHoldUid": "987654321", "name": "Legal Hold #2", "active": True},
+    ]
+}
+TEST_CUSTODIANS_RESPONSE = {
+    "legalHoldMemberships": [
+        {
+            "legalHoldMembershipUid": "99999",
+            "active": True,
+            "creationDate": "2020-07-16T08:50:23.405Z",
+            "legalHold": {"legalHoldUid": "123456789", "name": "Legal Hold #1"},
+            "user": {
+                "userUid": "911162111513111325",
+                "username": "[email protected]",
+                "email": "[email protected]",
+                "userExtRef": None,
+            },
+        },
+        {
+            "legalHoldMembershipUid": "11111",
+            "active": True,
+            "creationDate": "2020-07-16T08:50:23.405Z",
+            "legalHold": {"legalHoldUid": "987654321", "name": "Legal Hold #2"},
+            "user": {
+                "userUid": "911162111513111325",
+                "username": "[email protected]",
+                "email": "[email protected]",
+                "userExtRef": None,
+            },
+        },
+    ]
+}
+TEST_EMPTY_CUSTODIANS_RESPONSE = {"legalHoldMemberships": []}
+TEST_EMPTY_MATTERS_RESPONSE = {"legalHolds": []}
 TEST_EMPTY_USERS_RESPONSE = {"users": []}
 TEST_USERNAME = TEST_USERS_RESPONSE["users"][0]["username"]
 TEST_USER_ID = TEST_USERS_RESPONSE["users"][0]["userId"]
@@ -68,10 +104,6 @@
 }
 
 
-def get_all_users_generator():
-    yield TEST_USERS_RESPONSE
-
-
 @pytest.fixture
 def update_user_response(mocker):
     return create_mock_response(mocker)
@@ -103,7 +135,10 @@ def get_org_success(cli_state, get_org_response):
 
 
 @pytest.fixture
-def get_all_users_success(cli_state):
+def get_all_users_success(mocker, cli_state):
+    def get_all_users_generator():
+        yield create_mock_response(mocker, data=TEST_USERS_RESPONSE)
+
     cli_state.sdk.users.get_all.return_value = get_all_users_generator()
 
 
@@ -120,6 +155,42 @@ def get_user_id_failure(mocker, cli_state):
     )
 
 
+@pytest.fixture
+def get_custodian_failure(mocker, cli_state):
+    def empty_custodian_list_generator():
+        yield TEST_EMPTY_CUSTODIANS_RESPONSE
+
+    cli_state.sdk.legalhold.get_all_matter_custodians.return_value = (
+        empty_custodian_list_generator()
+    )
+
+
+@pytest.fixture
+def get_matter_failure(cli_state):
+    def empty_matter_list_generator():
+        yield TEST_EMPTY_MATTERS_RESPONSE
+
+    cli_state.sdk.legalhold.get_all_matters.return_value = empty_matter_list_generator()
+
+
+@pytest.fixture
+def get_all_matter_success(mocker, cli_state):
+    def matter_list_generator():
+        yield create_mock_response(mocker, data=TEST_MATTER_RESPONSE)
+
+    cli_state.sdk.legalhold.get_all_matters.return_value = matter_list_generator()
+
+
+@pytest.fixture
+def get_all_custodian_success(mocker, cli_state):
+    def custodian_list_generator():
+        yield create_mock_response(mocker, data=TEST_CUSTODIANS_RESPONSE)
+
+    cli_state.sdk.legalhold.get_all_matter_custodians.return_value = (
+        custodian_list_generator()
+    )
+
+
 @pytest.fixture
 def get_available_roles_success(cli_state, get_available_roles_response):
     cli_state.sdk.users.get_available_roles.return_value = get_available_roles_response
@@ -241,6 +312,66 @@ def test_list_users_when_given_excluding_active_and_inactive_uses_active_equals_
     )
 
 
+def test_list_legal_hold_flag_reports_none_for_users_not_on_legal_hold(
+    runner,
+    cli_state,
+    get_all_users_success,
+    get_custodian_failure,
+    get_all_matter_success,
+):
+    result = runner.invoke(
+        cli,
+        ["users", "list", "--include-legal-hold-membership", "-f", "CSV"],
+        obj=cli_state,
+    )
+
+    assert "Legal Hold #1,Legal Hold #2" not in result.output
+    assert "123456789,987654321" not in result.output
+    assert "legalHoldUid" not in result.output
+    assert "[email protected]" in result.output
+
+
+def test_list_legal_hold_flag_reports_none_if_no_matters_exist(
+    runner, cli_state, get_all_users_success, get_custodian_failure, get_matter_failure
+):
+    result = runner.invoke(
+        cli, ["users", "list", "--include-legal-hold-membership"], obj=cli_state
+    )
+
+    assert "Legal Hold #1,Legal Hold #2" not in result.output
+    assert "123456789,987654321" not in result.output
+    assert "legalHoldUid" not in result.output
+    assert "[email protected]" in result.output
+
+
+def test_list_legal_hold_values_not_included_for_legal_hold_user_if_legal_hold_flag_not_passed(
+    runner,
+    cli_state,
+    get_all_users_success,
+    get_all_custodian_success,
+    get_all_matter_success,
+):
+    result = runner.invoke(cli, ["users", "list"], obj=cli_state)
+    assert "Legal Hold #1,Legal Hold #2" not in result.output
+    assert "123456789,987654321" not in result.output
+    assert "[email protected]" in result.output
+
+
+def test_list_include_legal_hold_membership_merges_in_and_concats_legal_hold_info(
+    runner,
+    cli_state,
+    get_all_users_success,
+    get_all_custodian_success,
+    get_all_matter_success,
+):
+    result = runner.invoke(
+        cli, ["users", "list", "--include-legal-hold-membership"], obj=cli_state
+    )
+
+    assert "Legal Hold #1,Legal Hold #2" in result.output
+    assert "123456789,987654321" in result.output
+
+
 def test_add_user_role_adds(
     runner, cli_state, get_user_id_success, get_available_roles_success
 ):