95 lines
3.3 KiB
Python
95 lines
3.3 KiB
Python
from pathlib import Path
|
|
from unittest.mock import Mock, patch
|
|
|
|
import pytest
|
|
|
|
from pigway_frame_flow.errors import ErrorCode, FlowError
|
|
from pigway_frame_flow.integrations.kokonna_official.capabilities import capabilities
|
|
from pigway_frame_flow.integrations.kokonna_official.rpc import OfficialRPC
|
|
|
|
|
|
def test_login_session_not_password_used_for_reads():
|
|
login = Mock(status_code=200)
|
|
login.json.return_value = {"result": {"token": "session-secret"}}
|
|
devices = Mock(status_code=200)
|
|
devices.json.return_value = {"result": {"data": []}}
|
|
rpc = OfficialRPC()
|
|
with patch("requests.post", side_effect=[login, devices]) as post:
|
|
rpc.login("account", "password-secret")
|
|
assert rpc.devices() == {"data": []}
|
|
assert post.call_args.kwargs["headers"] == {
|
|
"Authorization": "Bearer session-secret"
|
|
}
|
|
assert "password-secret" not in str(post.call_args.kwargs)
|
|
rpc.disconnect()
|
|
assert not rpc.authorized
|
|
with patch("requests.post") as post, pytest.raises(FlowError):
|
|
rpc.devices()
|
|
post.assert_not_called()
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"method",
|
|
[
|
|
"appFrame.setFrameCover",
|
|
"appFrame.reset",
|
|
"appFrame.uploadImage",
|
|
"arbitrary.method",
|
|
],
|
|
)
|
|
def test_all_unenabled_operations_blocked_without_network(method):
|
|
with patch("requests.post") as post, pytest.raises(FlowError) as error:
|
|
OfficialRPC("session")._call(method, {})
|
|
assert error.value.code == ErrorCode.UNSUPPORTED_CAPABILITY
|
|
post.assert_not_called()
|
|
|
|
|
|
def test_rpc_http_500_relogin_is_auth_error_not_cloud_failure():
|
|
response = Mock(status_code=500)
|
|
response.json.return_value = {"error": {"code": -1, "message": "请重新登陆"}}
|
|
with (
|
|
patch("requests.post", return_value=response),
|
|
pytest.raises(FlowError) as error,
|
|
):
|
|
OfficialRPC("secret").devices()
|
|
assert error.value.code == ErrorCode.OFFICIAL_AUTH_REQUIRED
|
|
assert "secret" not in str(error.value)
|
|
|
|
|
|
def test_inventory_does_not_claim_unverified_writes_implemented():
|
|
entries = {entry.method: entry for entry in capabilities()}
|
|
assert entries["appFrame.setRotate"].implemented is False
|
|
assert entries["appFrame.getFrameCovers"].evidence == "live_verified"
|
|
assert all(
|
|
not entry.implemented
|
|
for entry in entries.values()
|
|
if entry.evidence == "apk_discovered"
|
|
)
|
|
|
|
|
|
def test_api_key_driver_has_no_official_dependency():
|
|
for name in ["client.py", "driver.py"]:
|
|
source = (Path("pigway_frame_flow/drivers/kokonna") / name).read_text()
|
|
assert "kokonna_official" not in source
|
|
assert "WidgetReader" not in source
|
|
|
|
|
|
def test_widget_disconnect_does_not_return_authenticated_cache():
|
|
from pigway_frame_flow.integrations.kokonna_official.widgets import WidgetReader
|
|
|
|
response = Mock(status_code=200)
|
|
response.json.return_value = {
|
|
"result": {
|
|
"rows": [],
|
|
"robot": {"frameCoverRobotId": None},
|
|
"frameCoverRobot": None,
|
|
}
|
|
}
|
|
reader = WidgetReader("session", 123)
|
|
with patch("requests.post", return_value=response):
|
|
assert reader.read().status == "NO_WIDGET"
|
|
reader.rpc.disconnect()
|
|
with patch("requests.post") as post:
|
|
assert reader.read().status == "AUTH_REQUIRED"
|
|
post.assert_not_called()
|