Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Display multiple motds in CLI and deactivate old ones #505

Merged
merged 18 commits into from
Aug 5, 2022
1 change: 1 addition & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -172,3 +172,4 @@ Please add a _short_ line describing the PR you make, if the PR implements a spe
## Summer 2022

- Check for DDS_CLI_ENV = "test-instance" in order to allow testing of features before production ([#506](https://github.com/ScilifelabDataCentre/dds_cli/pull/506))
- List all active motds instead of latest and new command for deactivating motds ([#505](https://github.com/ScilifelabDataCentre/dds_cli/pull/505))
56 changes: 53 additions & 3 deletions dds_cli/__main__.py
Original file line number Diff line number Diff line change
Expand Up @@ -91,9 +91,12 @@
highlight=False,
)

motd = dds_cli.motd_manager.MotdManager.get_latest_motd()
if motd:
dds_cli.utils.stderr_console.print(f"[bold]Important information:[/bold] {motd} \n")
if sys.argv[1] != "motd":
motds = dds_cli.motd_manager.MotdManager.list_all_active_motds(table=False)
if motds:
dds_cli.utils.stderr_console.print(f"[bold]Important information:[/bold]")
for motd in motds:
dds_cli.utils.stderr_console.print(f"{motd['Created']} - {motd['Message']} \n")

# -- dds -- #
@click.group()
Expand Down Expand Up @@ -1778,3 +1781,50 @@ def add_new_motd(click_ctx, message):
) as err:
LOG.error(err)
sys.exit(1)


# -- dds motd ls -- #
@motd_group_command.command(name="ls", no_args_is_help=False)
@click.pass_obj
def list_active_motds(click_ctx):
"""List all active MOTDs.
Only usable by Super Admins.
"""
try:
with dds_cli.motd_manager.MotdManager(
no_prompt=click_ctx.get("NO_PROMPT", False),
token_path=click_ctx.get("TOKEN_PATH"),
) as lister:
lister.list_all_active_motds(table=True)
except (
dds_cli.exceptions.AuthenticationError,
dds_cli.exceptions.ApiResponseError,
dds_cli.exceptions.ApiRequestError,
dds_cli.exceptions.DDSCLIException,
) as err:
LOG.error(err)
sys.exit(1)


# -- dds motd deactivate-- #
@motd_group_command.command(name="deactivate")
@click.argument("motd_id", metavar="[MOTD_ID]", nargs=1, type=int, required=True)
@click.pass_obj
def deactivate_motd(click_ctx, motd_id):
"""Deactivate Message Of The Day.
Only usable by Super Admins.
"""
try:
with dds_cli.motd_manager.MotdManager(
no_prompt=click_ctx.get("NO_PROMPT", False),
token_path=click_ctx.get("TOKEN_PATH"),
) as data_putter:
data_putter.deactivate_motd(motd_id)
except (
dds_cli.exceptions.AuthenticationError,
dds_cli.exceptions.ApiResponseError,
dds_cli.exceptions.ApiRequestError,
dds_cli.exceptions.DDSCLIException,
) as err:
LOG.error(err)
sys.exit(1)
61 changes: 53 additions & 8 deletions dds_cli/motd_manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -58,7 +58,7 @@ def __init__(

# Only method "add"can use the MotdManager class
if self.method not in ["add"]:
raise dds_cli.exceptions.AuthenticationError(f"Unauthorized method: '{self.method}'")
raise dds_cli.exceptions.InvalidMethodError(f"Unauthorized method: '{self.method}'")

def add_new_motd(self, message):
"""Add a new motd."""
Expand All @@ -70,18 +70,63 @@ def add_new_motd(self, message):
error_message="Failed adding a new MOTD",
)

LOG.info("A new MOTD was added to the database")
response_message = response_json.get(
"message", "No response. Cannot confirm MOTD creation."
)
LOG.info(response_message)

@staticmethod
def get_latest_motd():
"""Get the latest MOTD from dabase"""
def list_all_active_motds(table=False):
"""Get all active MOTDs."""
try:
response_json, _ = dds_cli.utils.perform_request(
endpoint=DDSEndpoint.MOTD,
response, _ = dds_cli.utils.perform_request(
endpoint=dds_cli.DDSEndpoint.MOTD,
method="get",
error_message="Failed getting MOTD",
error_message="Failed getting MOTDs from API",
)
except:
pass
else:
return response_json.get("message")
# Get items from response
motd = response.get("motds")
if not motd:
message = response.get("message", "No motds or info message returned from API.")
LOG.info(message)
else:
motds, keys = dds_cli.utils.get_required_in_response(
keys=["motds", "keys"], response=response
)
# Sort the active MOTDs according to date created
motds = dds_cli.utils.sort_items(items=motds, sort_by="Created")

# when called from "dds modt ls" with table=True
if table:
# Create table
table = dds_cli.utils.create_table(
title="Active MOTDs.",
columns=keys,
rows=motds,
ints_as_string=True,
caption="Active MOTDs.",
)

# Print out table
dds_cli.utils.print_or_page(item=table)
else:
# on every dds call
return motds

def deactivate_motd(self, motd_id):
"""Deactivate specific MOTD."""
response_json, _ = dds_cli.utils.perform_request(
endpoint=DDSEndpoint.MOTD,
headers=self.token,
method="put",
json={"motd_id": motd_id},
error_message="Failed deactivating the MOTD",
)

response_message = response_json.get(
"message", "No response. Cannot confirm MOTD deactivation."
)
LOG.info(response_message)
229 changes: 229 additions & 0 deletions tests/test_motd_manager.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,229 @@
import pytest
from requests_mock.mocker import Mocker
from dds_cli import DDSEndpoint
from typing import Dict, List
from dds_cli import motd_manager
from _pytest.capture import CaptureFixture
from _pytest.logging import LogCaptureFixture
import logging
from dds_cli.exceptions import ApiResponseError, InvalidMethodError

# init


def test_init_motdmanager_incorrect_method():
"""Init with incorrect method."""
method = "rm"
with pytest.raises(InvalidMethodError) as err:
_: motd_manager.MotdManager = motd_manager.MotdManager(
method=method, authenticate=False, no_prompt=True
)

assert f"Unauthorized method: '{method}'" in str(err.value)


def test_init_motdmanager():
"""Create manager."""
motdmanager: motd_manager.MotdManager = motd_manager.MotdManager(
authenticate=False, no_prompt=True
)
assert isinstance(motdmanager, motd_manager.MotdManager)


# list_all_active_motds
def test_list_all_active_motds_no_motds(caplog: LogCaptureFixture):
"""No motds returned."""
test_dicts: List[Dict] = [{}, {"message": "Test message when no motds."}, {"motds": {}}]
with caplog.at_level(logging.INFO):
# Iterate through possible alternative responses
for retd in test_dicts:
# Create mocker
with Mocker() as mock:

# Create mocked request - real request not executed
mock.get(DDSEndpoint.MOTD, status_code=200, json=retd)

with motd_manager.MotdManager(authenticate=False, no_prompt=True) as mtdm:
mtdm.list_all_active_motds(table=True) # Run active motds listing

assert (
"dds_cli.motd_manager",
logging.INFO,
retd.get("message", "No motds or info message returned from API."),
) in caplog.record_tuples


def test_list_all_active_motds_no_keys():
"""List motds without any keys returned."""
returned_dict: Dict = {
"motds": [{"MOTD ID": 1, "Message": "Test", "Created": "2022-08-05 08:31"}]
}
# Create mocker
with Mocker() as mock:

# Create mocked request - real request not executed
mock.get(DDSEndpoint.MOTD, status_code=200, json=returned_dict)

with pytest.raises(ApiResponseError) as err:
with motd_manager.MotdManager(authenticate=False, no_prompt=True) as mtdm:
mtdm.list_all_active_motds(table=True) # Run active motds listing

assert "The following information was not returned: ['keys']" in str(err.value)


def test_list_all_active_motds_table(capsys: CaptureFixture):
"""List motds without any keys returned."""
returned_dict: Dict = {
"motds": [
{"MOTD ID": 1, "Message": "Test", "Created": "2022-08-05 08:31"},
{"MOTD ID": 2, "Message": "Test 2", "Created": "2022-08-05 08:54"},
],
"keys": ["MOTD ID", "Message", "Created"],
}
# Create mocker
with Mocker() as mock:

# Create mocked request - real request not executed
mock.get(DDSEndpoint.MOTD, status_code=200, json=returned_dict)

with motd_manager.MotdManager(authenticate=False, no_prompt=True) as mtdm:
mtdm.list_all_active_motds(table=True) # Run active motds listing

captured = capsys.readouterr()
assert (
"\n".join(
[
"┏━━━━━━━━━┳━━━━━━━━━┳━━━━━━━━━━━━━━━━━━┓",
"┃ MOTD ID ┃ Message ┃ Created ┃",
"┡━━━━━━━━━╇━━━━━━━━━╇━━━━━━━━━━━━━━━━━━┩",
"│ 1 │ Test │ 2022-08-05 08:31 │",
"│ 2 │ Test 2 │ 2022-08-05 08:54 │",
"└─────────┴─────────┴──────────────────┘",
]
)
in captured.out
)


def test_list_all_active_motds_nottable(capsys: CaptureFixture):
"""List motds without any keys returned."""
returned_dict: Dict = {
"motds": [
{"MOTD ID": 1, "Message": "Test", "Created": "2022-08-05 08:31"},
{"MOTD ID": 2, "Message": "Test 2", "Created": "2022-08-05 08:54"},
],
"keys": ["MOTD ID", "Message", "Created"],
}
# Create mocker
with Mocker() as mock:

# Create mocked request - real request not executed
mock.get(DDSEndpoint.MOTD, status_code=200, json=returned_dict)

with motd_manager.MotdManager(authenticate=False, no_prompt=True) as mtdm:
motds = mtdm.list_all_active_motds(table=False) # Run active motds listing

captured = capsys.readouterr()
assert captured.out == ""

assert all(x in motds for x in returned_dict["motds"])


def test_list_all_active_motds_exceptionraised(capsys: CaptureFixture):
"""List motds when exception raised."""
returned_dict: Dict = {}
# Create mocker
with Mocker() as mock:

# Create mocked request - real request not executed
mock.get(DDSEndpoint.MOTD, status_code=500, json=returned_dict)

with motd_manager.MotdManager(authenticate=False, no_prompt=True) as mtdm:
motds = mtdm.list_all_active_motds(table=False) # Run active motds listing
assert not motds # If exception raised, nothing happens because of pass


# deactivate_motd


def test_deactivate_motd_no_response(caplog: LogCaptureFixture):
"""No response from API."""
returned_response: Dict = {}
with caplog.at_level(logging.INFO):
# Create mocker
with Mocker() as mock:
# Create mocked request - real request not executed
mock.put(DDSEndpoint.MOTD, status_code=200, json=returned_response)

with motd_manager.MotdManager(authenticate=False, no_prompt=True) as mtdm:
mtdm.token = {} # required, otherwise none
mtdm.deactivate_motd(motd_id=1) # Run deactivation

assert (
"dds_cli.motd_manager",
logging.INFO,
"No response. Cannot confirm MOTD deactivation.",
) in caplog.record_tuples


def test_deactivate_motd_ok(caplog: LogCaptureFixture):
"""No response from API."""
returned_response: Dict = {"message": "Message from API about deactivation."}
with caplog.at_level(logging.INFO):
# Create mocker
with Mocker() as mock:
# Create mocked request - real request not executed
mock.put(DDSEndpoint.MOTD, status_code=200, json=returned_response)

with motd_manager.MotdManager(authenticate=False, no_prompt=True) as mtdm:
mtdm.token = {} # required, otherwise none
mtdm.deactivate_motd(motd_id=1) # Run deactivation

assert (
"dds_cli.motd_manager",
logging.INFO,
"Message from API about deactivation.",
) in caplog.record_tuples


# add_new_motd


def test_add_new_motd_no_response(caplog: LogCaptureFixture):
"""Add new MOTD without any returned response."""
returned_response: Dict = {}
with caplog.at_level(logging.INFO):
# Create mocker
with Mocker() as mock:
# Create mocked request - real request not executed
mock.post(DDSEndpoint.MOTD, status_code=200, json=returned_response)

with motd_manager.MotdManager(authenticate=False, no_prompt=True) as mtdm:
mtdm.token = {} # required, otherwise none
mtdm.add_new_motd(message="Adding this message as a MOTD.") # Add motd

assert (
"dds_cli.motd_manager",
logging.INFO,
"No response. Cannot confirm MOTD creation.",
) in caplog.record_tuples


def test_add_new_motd_ok(caplog: LogCaptureFixture):
"""Add new MOTD without any returned response."""
returned_response: Dict = {"message": "Response from API about adding a MOTD."}
with caplog.at_level(logging.INFO):
# Create mocker
with Mocker() as mock:
# Create mocked request - real request not executed
mock.post(DDSEndpoint.MOTD, status_code=200, json=returned_response)

with motd_manager.MotdManager(authenticate=False, no_prompt=True) as mtdm:
mtdm.token = {} # required, otherwise none
mtdm.add_new_motd(message="Adding this message as a MOTD.") # Add motd

assert (
"dds_cli.motd_manager",
logging.INFO,
"Response from API about adding a MOTD.",
) in caplog.record_tuples