Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 23 additions & 1 deletion src/binarylane/console/printers/formatter.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@
def format_response(response: Any, show_header: bool, fields: Optional[List[str]] = None) -> List[List[str]]:
"""Convert structured response object into a 'table' (where the length of each inner list is the same)"""

action_id: Optional[int] = _get_action_id(response)
response = _extract_primary(response)

if isinstance(response, list):
Expand All @@ -33,11 +34,17 @@ def object_to_list(row: Dict[str, Any], columns: List[str]) -> List[Any]:
if isinstance(response, str):
data = [[DEFAULT_HEADING]] if show_header else []
data += [[response]]

elif isinstance(response, int):
data = [[DEFAULT_HEADING]] if show_header else []
data += [[str(response)]]
else:
data = [["name", "value"]] if show_header else []
data += [_flatten(item, True) for item in response.to_dict().items()]

# If response contained an action ID, prepend it to the formatted data
if action_id:
data.insert(1 if show_header else 0, ["action_id", str(action_id)])

return data


Expand Down Expand Up @@ -125,3 +132,18 @@ def _flatten_dict(item: Dict[str, Any], single_object: bool) -> str:

# Generic handler
return "<object>" if not single_object else "\n".join([f"{key}: {value}" for key, value in item.items()])


def _get_action_id(response: Any) -> Optional[int]:
"""Return response.links.actions[0].id or None"""

if not (links := getattr(response, "links", None)):
return None

# Most responses do not contain action links, so import is delayed
from binarylane.models.actions_links import ActionsLinks

if not isinstance(links, ActionsLinks) or not links.actions:
return None

return links.actions[0].id if links.actions[0] else None
5 changes: 4 additions & 1 deletion src/binarylane/console/printers/json_printer.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,4 +10,7 @@ class JsonPrinter(Printer):
"""Output an API response as 'raw' JSON"""

def print(self, response: Any, fields: Optional[List[str]] = None) -> None:
print(json.dumps(response.to_dict()))
print(self.format_response(response))

def format_response(self, response: Any) -> str:
return json.dumps(response.to_dict() if hasattr(response, "to_dict") else response)
6 changes: 4 additions & 2 deletions src/binarylane/console/runners/actionlink.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,8 +16,10 @@ def response(self, status_code: int, received: Any) -> None:
super().response(status_code, received)
return

action_id = links.actions[0].id
super().response(status_code, action_id)
# Show action progress on stdout
if not self._async:
action_id = links.actions[0].id
super().response(status_code, action_id)

# Print the 'other' object (e.g. server) from the response
self._printer.print(received)
32 changes: 32 additions & 0 deletions tests/models/create_server_response.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,32 @@
from __future__ import annotations

from typing import Any, Dict

import attr

# We have to use the real ActionsLinks, because handler checks for it by type
from binarylane.models.actions_links import ActionsLinks
from tests.models.server import Server


@attr.s(auto_attribs=True)
class CreateServerResponse:
server: Server
links: ActionsLinks
additional_properties: Dict[str, Any] = attr.ib(init=False, factory=dict)

def to_dict(self) -> Dict[str, Any]:
server = self.server.to_dict()

links = self.links.to_dict()

field_dict: Dict[str, Any] = {}
field_dict.update(self.additional_properties)
field_dict.update(
{
"server": server,
"links": links,
}
)

return field_dict
20 changes: 20 additions & 0 deletions tests/printers/test_formatter.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,9 @@

from typing import TYPE_CHECKING, Any, Dict, List

from binarylane.models.action_link import ActionLink
from binarylane.models.actions_links import ActionsLinks
from tests.models.create_server_response import CreateServerResponse
from tests.models.network import Network
from tests.models.network_type import NetworkType

Expand Down Expand Up @@ -93,3 +96,20 @@ def test_format_networks_v4_and_v6(servers_response: ServersResponse) -> None:
Network(ip_address="value4", type=NetworkType.PUBLIC),
]
assert formatter.format_response(servers_response, False, ["networks"]) == [["ipv4\nipv6"]]


# ActionLinkRunner when used with --async will print the action ID
def test_format_int() -> None:
assert formatter.format_response(12345, True) == [[formatter.DEFAULT_HEADING], ["12345"]]
assert formatter.format_response(12345, False) == [["12345"]]


def test_format_action_link(servers_response: ServersResponse) -> None:
action_link = ActionLink(12345, "create", "https://api.example.com/v2/actions/12345")
response = CreateServerResponse(servers_response.servers[0], ActionsLinks([action_link]))
assert formatter.format_response(response, True)[:4] == [
["name", "value"],
["action_id", "12345"],
["id", "1"],
["name", "test"],
]
40 changes: 40 additions & 0 deletions tests/printers/test_json.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,40 @@
from __future__ import annotations

from typing import Any, Dict, List

from binarylane.console.printers.json_printer import JsonPrinter

printer = JsonPrinter()


def test_format_str() -> None:
assert printer.format_response("test") == '"test"'


def test_format_list() -> None:
ns1 = "ns1.binarylane.com.au"
ns2 = "ns2.binarylane.com.au"
dns = [ns1, ns2]

assert printer.format_response(dns) == '["ns1.binarylane.com.au", "ns2.binarylane.com.au"]'


def test_format_dict() -> None:
class DnsList:
dns: List[str]
meta: Dict[str, Any]
links: List[str]

def __init__(self) -> None:
self.dns = ["ns1.binarylane.com.au", "ns2.binarylane.com.au"]

def to_dict(self) -> Dict[str, Any]:
return {"dns": self.dns}

response = DnsList()
assert printer.format_response(response) == '{"dns": ["ns1.binarylane.com.au", "ns2.binarylane.com.au"]}'


# ActionLinkRunner when used with --async will print the action ID
def test_format_int() -> None:
assert printer.format_response(12345) == "12345"
Loading