104 lines
3.9 KiB
Python
104 lines
3.9 KiB
Python
import json
|
|
from unittest.mock import Mock, patch
|
|
|
|
import pytest
|
|
|
|
from pigway_frame_flow.api.app import create_app
|
|
from pigway_frame_flow.errors import ErrorCode, FlowError
|
|
from pigway_frame_flow.integrations.kokonna_official.connection import Connection
|
|
from pigway_frame_flow.integrations.kokonna_official.widgets import WidgetReader
|
|
from pigway_frame_flow.models import DisplayRequest
|
|
|
|
|
|
def enqueue(service, priority):
|
|
payload = DisplayRequest(
|
|
content={"type": "template", "template": "quote", "data": {}},
|
|
options={"priority": priority, "duration": 3600, "restore_previous": True},
|
|
)
|
|
return service.submit(
|
|
"display", payload.model_dump(mode="json"), "living-room", "admin", "audit"
|
|
)
|
|
|
|
|
|
def test_later_alert_is_not_starved_by_first_fifty_blocked_jobs(service):
|
|
enqueue(service, 50)
|
|
service.process_one()
|
|
for _ in range(51):
|
|
enqueue(service, 10)
|
|
high = enqueue(service, 90)
|
|
assert service.process_one()
|
|
assert service.operation(high)["status"] == "SYNCING"
|
|
assert service.registry.get("living-room").client.uploads == 2
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"content",
|
|
[
|
|
"{broken",
|
|
"{}",
|
|
'{"token":123,"username":"user","robot_id":1}',
|
|
'{"token":"secret","username":"user","robot_id":true}',
|
|
],
|
|
)
|
|
def test_broken_optional_secret_does_not_block_core_start(settings, content, caplog):
|
|
settings.config_dir.mkdir(parents=True)
|
|
(settings.config_dir / "official-living-room.secret").write_text(content)
|
|
app = create_app(settings, start_worker=False)
|
|
registry = app.state.service.registry
|
|
assert not registry.official["living-room"].state().authorized
|
|
assert registry.get("living-room").device().status == "online"
|
|
assert "secret" not in caplog.text
|
|
|
|
|
|
@pytest.mark.parametrize("action", ["disconnect", "select", "login"])
|
|
def test_disk_failure_preserves_existing_connection(settings, action):
|
|
path = settings.config_dir / "official-test.secret"
|
|
conn = Connection(WidgetReader(mock=True), path, [])
|
|
conn.login("old-user", "not-saved")
|
|
if action != "select":
|
|
conn.select(1)
|
|
before = path.read_text()
|
|
with patch.object(conn, "_save", side_effect=OSError("fixture disk failure")):
|
|
with pytest.raises(OSError):
|
|
if action == "login":
|
|
conn.login("new-user", "not-saved")
|
|
elif action == "select":
|
|
conn.select(1)
|
|
else:
|
|
conn.disconnect()
|
|
assert conn.state().authorized
|
|
assert conn.state().username == "old-user"
|
|
assert conn.state().robot_id == (None if action == "select" else 1)
|
|
assert path.read_text() == before
|
|
assert Connection(WidgetReader(mock=True), path, []).state().authorized
|
|
|
|
|
|
def test_expired_session_invalidated_in_memory_and_on_disk(settings):
|
|
path = settings.config_dir / "official-test.secret"
|
|
conn = Connection(WidgetReader(), path, [])
|
|
conn._commit(conn.reader.rpc.__class__("expired-session"), "user", 1)
|
|
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,
|
|
):
|
|
conn.widgets()
|
|
assert error.value.code == ErrorCode.OFFICIAL_AUTH_REQUIRED
|
|
assert not conn.state().authorized
|
|
assert json.loads(path.read_text())["token"] == ""
|
|
assert not Connection(WidgetReader(), path, []).state().authorized
|
|
|
|
|
|
def test_official_session_not_reflected_in_request_id(client):
|
|
client.post(
|
|
"/api/v1/devices/living-room/official/connection",
|
|
json={"username": "fixture", "password": "fixture"},
|
|
)
|
|
response = client.get("/health", headers={"X-Request-ID": "mock-session"})
|
|
assert response.headers["X-Request-ID"] != "mock-session"
|
|
error = client.get(
|
|
"/api/v1/devices/missing", headers={"X-Request-ID": "mock-session"}
|
|
)
|
|
assert error.json()["error"]["request_id"] != "mock-session"
|