72 lines
2.6 KiB
Python
72 lines
2.6 KiB
Python
|
import json
|
||
|
|
import stat
|
||
|
|
from unittest.mock import Mock, patch
|
||
|
|
|
||
|
|
from pigway_frame_flow.api.app import create_app
|
||
|
|
|
||
|
|
PATH = "/api/v1/devices/living-room/official/connection"
|
||
|
|
|
||
|
|
|
||
|
|
def test_connect_select_persist_and_disconnect(client, settings):
|
||
|
|
response = client.post(
|
||
|
|
PATH, json={"username": "official-user", "password": "not-saved"}
|
||
|
|
)
|
||
|
|
assert response.status_code == 200
|
||
|
|
assert response.json()["robot_id"] is None
|
||
|
|
assert response.json()["devices"][0]["id"] == 1
|
||
|
|
assert "token" not in response.text and "not-saved" not in response.text
|
||
|
|
path = settings.config_dir / "official-living-room.secret"
|
||
|
|
assert stat.S_IMODE(path.stat().st_mode) == 0o600
|
||
|
|
assert "not-saved" not in path.read_text()
|
||
|
|
assert client.put(PATH, json={"robot_id": 1}).status_code == 200
|
||
|
|
assert client.put(PATH, json={"robot_id": 999}).status_code == 404
|
||
|
|
restored = create_app(settings, start_worker=False).state.service.registry
|
||
|
|
assert restored.official["living-room"].state().robot_id == 1
|
||
|
|
assert client.delete(PATH).status_code == 200
|
||
|
|
assert json.loads(path.read_text())["token"] == ""
|
||
|
|
assert not client.get(PATH).json()["authorized"]
|
||
|
|
|
||
|
|
|
||
|
|
def test_auth_and_validation_hide_password(client):
|
||
|
|
response = client.post(PATH, json={"username": "", "password": "secret-password"})
|
||
|
|
assert response.status_code == 422 and "secret-password" not in response.text
|
||
|
|
client.headers.clear()
|
||
|
|
assert (
|
||
|
|
client.post(PATH, json={"username": "user", "password": "secret"}).status_code
|
||
|
|
== 401
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def test_live_connection_only_login_and_read_methods(settings):
|
||
|
|
from pigway_frame_flow.integrations.kokonna_official.connection import Connection
|
||
|
|
from pigway_frame_flow.integrations.kokonna_official.widgets import WidgetReader
|
||
|
|
|
||
|
|
login = Mock(status_code=200)
|
||
|
|
login.json.return_value = {"result": {"token": "official-secret"}}
|
||
|
|
devices = Mock(status_code=200)
|
||
|
|
devices.json.return_value = {
|
||
|
|
"result": {
|
||
|
|
"data": [
|
||
|
|
{
|
||
|
|
"id": 10,
|
||
|
|
"robotNickname": "Frame",
|
||
|
|
"model": {"code": "B101", "name": "13.3-inch"},
|
||
|
|
}
|
||
|
|
]
|
||
|
|
}
|
||
|
|
}
|
||
|
|
conn = Connection(
|
||
|
|
WidgetReader(),
|
||
|
|
settings.config_dir / "official-test.secret",
|
||
|
|
settings.additional_secrets,
|
||
|
|
)
|
||
|
|
with patch("requests.post", side_effect=[login, devices, devices]) as post:
|
||
|
|
conn.login("user", "password")
|
||
|
|
conn.select(10)
|
||
|
|
assert [c.kwargs["json"]["method"] for c in post.call_args_list] == [
|
||
|
|
"app.login",
|
||
|
|
"app.getRobots",
|
||
|
|
"app.getRobots",
|
||
|
|
]
|
||
|
|
assert "official-secret" not in conn.state().model_dump_json()
|