stroblme
75b3a93094
Docs / docs (push) Successful in 33s
Playwright Tests / test-playwright (1, 2) (push) Failing after 1m37s
Playwright Tests / test-playwright (2, 2) (push) Failing after 27s
pre-commit / pre-commit (push) Failing after 2m37s
Test Backend / test-backend (3.12) (push) Successful in 5m7s
Test Backend / test-backend (3.14) (push) Successful in 5m2s
Compose Smoke Test / test-compose (push) Failing after 28s
Playwright Tests / merge-reports (push) Failing after 2m29s
130 lines
5.2 KiB
Python
130 lines
5.2 KiB
Python
"""Direct and portal browsers recover state without widening camera access."""
|
|
|
|
import asyncio
|
|
import json
|
|
from contextlib import suppress
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from fluksio.api import deps
|
|
from fluksio.api.routes import flows
|
|
from fluksio.cloud.connector import CloudConnector
|
|
from fluksio.flow import events
|
|
from fluksio.flow.events import QUEUE_SIZE, EventBus
|
|
|
|
|
|
@pytest.mark.parametrize("tunnel", [False, True])
|
|
@pytest.mark.parametrize("burst", [0, QUEUE_SIZE + 1])
|
|
def test_a_browser_catches_changes_during_its_snapshot(monkeypatch, tunnel, burst):
|
|
async def run():
|
|
bus = EventBus()
|
|
bus.bind(asyncio.get_running_loop())
|
|
monkeypatch.setattr(events, "event_bus", bus)
|
|
monkeypatch.setattr(flows, "event_bus", bus)
|
|
user = SimpleNamespace(id="viewer", is_superuser=False)
|
|
for module in (flows, deps):
|
|
monkeypatch.setattr(module, "user_from_token", lambda *a, **kw: user)
|
|
monkeypatch.setattr(
|
|
module, "hidden_from", lambda *a, **kw: frozenset({"f.camera"})
|
|
)
|
|
monkeypatch.setattr(flows, "panel_scope", lambda *a: None)
|
|
monkeypatch.setattr(flows, "socket_token", lambda *a: "test-token")
|
|
|
|
async def immediate(function, *args):
|
|
return function(*args)
|
|
|
|
monkeypatch.setattr(flows, "run_in_threadpool", immediate)
|
|
current = {"f.flag": {"value": 0}, "f.camera": {"value": "private"}}
|
|
controller = SimpleNamespace(
|
|
values=lambda: dict(current),
|
|
node_statuses=lambda: [],
|
|
issues=[],
|
|
paused_flows=lambda: [],
|
|
placer=None,
|
|
)
|
|
app = SimpleNamespace(state=SimpleNamespace(flow_controller=controller))
|
|
received = []
|
|
client_values = {}
|
|
caught_up = asyncio.Event()
|
|
drained = asyncio.Event()
|
|
|
|
class Socket:
|
|
headers = {}
|
|
|
|
async def accept(self):
|
|
pass
|
|
|
|
async def receive(self):
|
|
await asyncio.Future()
|
|
|
|
async def send(self, text):
|
|
frame = json.loads(text)
|
|
if frame["op"] == "ws_msg":
|
|
await self.send_text(frame["text"])
|
|
|
|
async def send_text(self, text):
|
|
payload = json.loads(text)
|
|
received.append(payload)
|
|
if payload["type"] == "snapshot":
|
|
assert "f.camera" not in payload["values"]
|
|
client_values.update(payload["values"])
|
|
if payload["values"]["f.flag"]["value"] == 1:
|
|
caught_up.set()
|
|
elif len(received) == 1:
|
|
# The state was read, but its frame is still sending.
|
|
# This value changes only once, then a burst loses its
|
|
# event and leaves only a snapshot able to recover it.
|
|
current["f.flag"] = {"value": 1}
|
|
bus.publish(
|
|
{"type": "message_value", "name": "f.flag", "value": 1}
|
|
)
|
|
for i in range(burst):
|
|
current["f.noise"] = {"value": i}
|
|
bus.publish(
|
|
{"type": "message_value", "name": "f.noise", "value": i}
|
|
)
|
|
# Recovery must forward retained critical events too.
|
|
bus.publish({"type": "node_error", "node": "f.example"})
|
|
await asyncio.sleep(0)
|
|
for event in payload.get("events", [payload]):
|
|
if event["type"] == "message_value":
|
|
client_values[event["name"]] = {"value": event["value"]}
|
|
if event.get("name") == "f.flag" and event.get("value") == 1:
|
|
caught_up.set()
|
|
if event["type"] == "node_error":
|
|
drained.set()
|
|
|
|
socket = Socket()
|
|
socket.app = app
|
|
if tunnel:
|
|
action = CloudConnector(app)._stream(
|
|
socket,
|
|
{"id": "review", "path": "/api/v1/flows/ws", "query": "token=test"},
|
|
)
|
|
else:
|
|
action = flows.flow_events(socket)
|
|
task = asyncio.create_task(action)
|
|
try:
|
|
await asyncio.wait_for(caught_up.wait(), timeout=2)
|
|
await asyncio.wait_for(drained.wait(), timeout=2)
|
|
finally:
|
|
task.cancel()
|
|
with suppress(asyncio.CancelledError):
|
|
await task
|
|
assert sum(item["type"] == "snapshot" for item in received) == (
|
|
2 if burst else 1
|
|
)
|
|
flattened = [
|
|
event for payload in received for event in payload.get("events", [payload])
|
|
]
|
|
assert sum(event["type"] == "node_error" for event in flattened) == 1
|
|
if burst:
|
|
# The client replaces values in arrival order. A snapshot ahead
|
|
# of the retained backlog temporarily rolls values backwards.
|
|
assert flattened[-1]["type"] == "snapshot"
|
|
assert client_values == {k: v for k, v in current.items() if k != "f.camera"}
|
|
assert not bus._subscribers
|
|
|
|
asyncio.run(run())
|