Feat: command for group migration (#49)

littlepay/commands/groups.py

@@ -43,6 +43,9 @@ def groups(args: Namespace = None) -> int:
     elif command == "unlink":
         for group in groups:
             return_code += unlink_product(client, group.id, args.product_id)
+    elif command == "migrate":
+        for group in groups:
+            return_code += migrate_group(client, group.id, getattr(args, "force", False))
 
     groups = list(groups)
     if csv_output and command != "products":
@@ -141,3 +144,30 @@ def unlink_product(client: Client, group_id: str, product_id: str) -> int:
         return_code = RESULT_FAILURE
 
     return return_code
+
+
+def migrate_group(client: Client, group_id: str, force: bool = False) -> int:
+    config = Config()
+    print_active_message(config, "Migrating group", f"[{group_id}]")
+    return_code = RESULT_SUCCESS
+
+    if force is True:
+        confirm = "yes"
+    else:
+        try:
+            confirm = input("❔ Are you sure? (yes/no): ")
+        except EOFError:
+            confirm = "no"
+
+    if confirm.lower().startswith("y"):
+        print("Migrating group...")
+        try:
+            client.migrate_concession_group(group_id)
+            print("✅ Migrated")
+        except HTTPError as err:
+            print(f"❌ Error: {err}")
+            return_code = RESULT_FAILURE
+    else:
+        print("Canceled...")
+
+    return return_code

littlepay/main.py

@@ -64,6 +64,13 @@ def _maincmd(name, help):
     groups_link = _subcmd(groups_commands, "link", help="Link one or more concession groups to a product")
     groups_link.add_argument("product_id", help="The ID of the product to link to")
 
+    groups_migrate = _subcmd(
+        groups_commands, "migrate", help="Migrate a group from the old Customer Group format to the current format"
+    )
+    groups_migrate.add_argument(
+        "--force", action="store_true", default=False, help="Don't ask for confirmation before migration"
+    )
+
     groups_products = _subcmd(groups_commands, "products", help="List products for one or more concession groups")
     groups_products.add_argument(
         "--csv", action="store_true", default=False, help="Output results in simple CSV format", dest="csv"

tests/commands/test_groups.py

@@ -296,3 +296,79 @@ def test_groups_group_command__unlink_HTTPError(mock_client, capfd):
     assert "Unlinking group <-> product" in capture.out
     assert "Error" in capture.out
     assert "Unlinked" not in capture.out
+
+
+@pytest.mark.parametrize("sample_input", ["y", "Y", "yes", "Yes", "YES"])
+def test_groups_group_command__migrate_confirm(mock_client, capfd, mock_input, sample_input):
+    mock_input(sample_input)
+
+    args = Namespace(group_command="migrate")
+    res = groups(args)
+    capture = capfd.readouterr()
+
+    for group in GROUP_RESPONSES:
+        mock_client.migrate_concession_group.assert_any_call(group.id)
+
+    assert res == RESULT_SUCCESS
+    assert "Migrating group" in capture.out
+    assert "Migrated" in capture.out
+    assert "Matching groups" in capture.out
+
+
+def test_groups_group_command__migrate_confirm_error(capfd, mock_input):
+    _input = mock_input(None)
+    _input.side_effect = EOFError
+
+    args = Namespace(group_command="migrate")
+    res = groups(args)
+    capture = capfd.readouterr()
+
+    assert res == RESULT_SUCCESS
+    assert "Migrating group" in capture.out
+    assert "Canceled" in capture.out
+    assert "Matching groups" in capture.out
+
+
+@pytest.mark.parametrize("sample_input", ["n", "N", "no", "No", "NO"])
+def test_groups_group_command__migrate_decline(capfd, mock_input, sample_input):
+    mock_input(sample_input)
+
+    args = Namespace(group_command="migrate")
+    res = groups(args)
+    capture = capfd.readouterr()
+
+    assert res == RESULT_SUCCESS
+    assert "Migrating group" in capture.out
+    assert "Canceled" in capture.out
+    assert "Matching groups" in capture.out
+
+
+def test_groups_group_command__migrate_force(mock_client, capfd, mock_input):
+    _input = mock_input("no")
+
+    args = Namespace(group_command="migrate", force=True)
+    res = groups(args)
+    capture = capfd.readouterr()
+
+    for group in GROUP_RESPONSES:
+        mock_client.migrate_concession_group.assert_any_call(group.id)
+
+    assert res == RESULT_SUCCESS
+    assert _input.called is False
+    assert "Migrating group" in capture.out
+    assert "Migrated" in capture.out
+    assert "Matching groups" in capture.out
+
+
+def test_groups_group_command__migrate_HTTPError(mock_client, capfd, mock_input):
+    mock_client.migrate_concession_group.side_effect = HTTPError
+    mock_input("y")
+
+    args = Namespace(group_command="migrate", group_id="1234")
+    res = groups(args)
+    capture = capfd.readouterr()
+
+    assert res == RESULT_FAILURE
+    assert "Migrating group" in capture.out
+    assert "Error" in capture.out
+    assert "Matching groups" in capture.out

tests/test_main.py

@@ -131,6 +131,25 @@ def test_main_groups_link(mock_commands_groups):
     assert call_args.product_id == "1234"
 
 
+def test_main_groups_migrate(mock_commands_groups):
+    result = main(argv=["groups", "migrate"])
+
+    assert result == RESULT_SUCCESS
+    mock_commands_groups.assert_called_once()
+    call_args = mock_commands_groups.call_args.args[0]
+    assert call_args.group_command == "migrate"
+
+
+def test_main_groups_migrate_force(mock_commands_groups):
+    result = main(argv=["groups", "migrate", "--force"])
+
+    assert result == RESULT_SUCCESS
+    mock_commands_groups.assert_called_once()
+    call_args = mock_commands_groups.call_args.args[0]
+    assert call_args.group_command == "migrate"
+    assert call_args.force is True
+
+
 def test_main_groups_products(mock_commands_groups):
     result = main(argv=["groups", "products"])