126 lines
4.5 KiB
Python
126 lines
4.5 KiB
Python
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
import httpx
|
|
from textual.widgets import DataTable, TabbedContent, Tree
|
|
|
|
from rex.management.app import RexManager
|
|
from rex.management.client import ManagementClient, ManagementError
|
|
from rex.server.device_manager import DeviceManager
|
|
|
|
|
|
class ManagementClientTests(unittest.IsolatedAsyncioTestCase):
|
|
async def test_reads_keys_and_encodes_names(self) -> None:
|
|
requests: list[httpx.Request] = []
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
requests.append(request)
|
|
if request.method == "GET":
|
|
return httpx.Response(
|
|
200, json={"keys": [{"name": "home", "secret": "abc"}]}
|
|
)
|
|
return httpx.Response(200, json={"key": {}})
|
|
|
|
client = ManagementClient(
|
|
"http://rex.test", transport=httpx.MockTransport(handler)
|
|
)
|
|
self.addAsyncCleanup(client.close)
|
|
|
|
self.assertEqual((await client.keys())[0]["name"], "home")
|
|
await client.create_key("living room")
|
|
self.assertEqual(requests[1].url.raw_path, b"/key/living%20room")
|
|
|
|
async def test_reports_api_error_detail(self) -> None:
|
|
def handler(_: httpx.Request) -> httpx.Response:
|
|
return httpx.Response(409, json={"detail": "already exists"})
|
|
|
|
client = ManagementClient(
|
|
"http://rex.test", transport=httpx.MockTransport(handler)
|
|
)
|
|
self.addAsyncCleanup(client.close)
|
|
|
|
with self.assertRaisesRegex(ManagementError, "already exists"):
|
|
await client.create_device("desk")
|
|
|
|
|
|
class DeviceManagerTests(unittest.TestCase):
|
|
def test_action_changes_are_persisted(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
path = Path(directory) / "devices.json"
|
|
manager = DeviceManager(path)
|
|
manager.create_device("desk")
|
|
manager.create_action("desk", "wake")
|
|
|
|
reloaded = DeviceManager(path)
|
|
self.assertIsNotNone(reloaded.get_action("desk", "wake"))
|
|
|
|
reloaded.delete_action("desk", "wake")
|
|
self.assertIsNone(DeviceManager(path).get_action("desk", "wake"))
|
|
|
|
|
|
class FakeManagementClient:
|
|
def __init__(self) -> None:
|
|
self.created_keys: list[str] = []
|
|
|
|
async def close(self) -> None:
|
|
pass
|
|
|
|
async def keys(self) -> list[dict[str, str]]:
|
|
return [{"name": "home", "secret": "abc"}]
|
|
|
|
async def devices(self) -> list[dict[str, object]]:
|
|
return [{"name": "desk", "actions": [{"name": "wake"}]}]
|
|
|
|
async def create_key(self, name: str) -> None:
|
|
self.created_keys.append(name)
|
|
|
|
|
|
class ManagerAppTests(unittest.IsolatedAsyncioTestCase):
|
|
async def test_refresh_populates_keys_and_devices(self) -> None:
|
|
app = RexManager("http://rex.test")
|
|
app.client = FakeManagementClient() # type: ignore[assignment]
|
|
|
|
async with app.run_test() as pilot:
|
|
await pilot.pause()
|
|
keys = app.query_one("#keys", DataTable)
|
|
devices = app.query_one("#devices", Tree)
|
|
|
|
self.assertEqual(keys.get_row_at(0), ["home", "abc"])
|
|
self.assertEqual(devices.root.children[0].label.plain, "desk")
|
|
self.assertEqual(devices.root.children[0].children[0].label.plain, "wake")
|
|
|
|
async def test_vim_navigation_and_create_prompt(self) -> None:
|
|
app = RexManager("http://rex.test")
|
|
client = FakeManagementClient()
|
|
app.client = client # type: ignore[assignment]
|
|
|
|
async with app.run_test() as pilot:
|
|
await pilot.press("n")
|
|
await pilot.press(*"office", "enter")
|
|
await pilot.pause()
|
|
self.assertEqual(client.created_keys, ["office"])
|
|
|
|
await pilot.press("shift+l", "j")
|
|
self.assertEqual(app.query_one(TabbedContent).active, "devices-tab")
|
|
self.assertEqual(
|
|
app.query_one("#devices", Tree).cursor_node.data,
|
|
("device", "desk"),
|
|
)
|
|
|
|
async def test_moves_between_key_columns_and_yanks_value(self) -> None:
|
|
app = RexManager("http://rex.test")
|
|
app.client = FakeManagementClient() # type: ignore[assignment]
|
|
|
|
async with app.run_test() as pilot:
|
|
with patch.object(app, "copy_to_clipboard") as copy:
|
|
await pilot.press("l", "y")
|
|
|
|
self.assertEqual(app.query_one("#keys", DataTable).cursor_column, 1)
|
|
copy.assert_called_once_with("abc")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|