pythonnative/tests/test_web_transport.py at refs/heads/main · pythonnative/pythonnative · GitHub

Repository navigation

History

455 lines (364 loc) · 17.8 KB

Raw

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

170

171

172

173

174

175

176

177

178

179

180

181

182

183

184

185

186

187

188

189

190

191

192

193

194

195

196

197

198

199

200

201

202

203

204

205

206

207

208

209

210

211

212

213

214

215

216

217

218

219

220

221

222

223

224

225

226

227

228

229

230

231

232

233

234

235

236

237

238

239

240

241

242

243

244

245

246

247

248

249

250

251

252

253

254

255

256

257

258

259

260

261

262

263

264

265

266

267

268

269

270

271

272

273

274

275

276

277

278

279

280

281

282

283

284

285

286

287

288

289

290

291

292

293

294

295

296

297

298

299

300

301

302

303

304

305

306

307

308

309

310

311

312

313

314

315

316

317

318

319

320

321

322

323

324

325

326

327

328

329

330

331

332

333

334

335

336

337

338

339

340

341

342

343

344

345

346

347

348

349

350

351

352

353

354

355

356

357

358

359

360

361

362

363

364

365

366

367

368

369

370

371

372

373

374

375

376

377

378

379

380

381

382

383

384

385

386

387

388

389

390

391

392

393

394

395

396

397

398

399

400

401

402

403

404

405

406

407

408

409

410

411

412

413

414

415

416

417

418

419

420

421

422

423

424

425

426

427

428

429

430

431

432

433

434

435

436

437

438

439

440

441

442

443

444

445

446

447

448

449

450

451

452

453

454

455

"""``WebTransport``: the browser preview as the native side of the bridge.

A ``FakePage`` stands in for the JavaScript in ``devserver/static``: it

receives the frames the transport sends, answers synchronous requests

the way the page does (from another thread, since the real page answers

over the dev server's socket thread), and can raise callbacks.

"""

from __future__ import annotations

import json

import threading

import time

import types

from typing import Any, Dict, Generator, List, Tuple

import pytest

import pythonnative as pn

from pythonnative import bridge

from pythonnative.bridge import PROTOCOL_VERSION, codec

from pythonnative.bridge.web import BROWSER_MODULES, WebTransport

from pythonnative.component import component

from pythonnative.element import Element

from pythonnative.native_modules import registry as modules

from pythonnative.native_views import set_backend

from pythonnative.native_views.bridge_backend import BridgeBackend

class FakePage:

"""Records everything Python sends and answers requests like the page would."""

def __init__(self, transport: WebTransport) -> None:

self.transport = transport

self.sent: List[List[Any]] = []

self.viewport = {"width": 390.0, "height": 800.0, "insets": {"top": 0, "left": 0, "bottom": 34, "right": 0}}

self.calls: List[Any] = []

self.commands: List[Any] = []

self.animations: List[Any] = []

self.module_results: Dict[Tuple[str, str], Any] = {}

self.answer_in_thread = True

self.closed = threading.Event()

# -- what PreviewPeer offers -------------------------------------------

def send(self, text: str) -> None:

message = json.loads(text)

self.sent.append(message)

kind = message[0]

if kind in ("apply", "command", "animate", "call"):

result = self._answer(message)

reply = codec.dumps(["res", message[1], result])

if self.answer_in_thread:

threading.Thread(target=self.transport.on_preview_message, args=(self, reply), daemon=True).start()

else:

self.transport.on_preview_message(self, reply)

def close(self) -> None:

self.closed.set()

# -- page behavior ------------------------------------------------------

def _answer(self, message: List[Any]) -> Any:

kind = message[0]

if kind == "apply":

from pythonnative.bridge.commits import CommitState

self.commit = getattr(self, "commit", CommitState()).prepare(message[2]).publish()

reply = self.commit.acknowledgement()

if "layout" in message[2]:

reply["layout"] = {

"application": self.commit.application,

"surface": self.commit.surface,

"revision": self.commit.revision,

"frames": [],

}

return reply

if kind == "command":

_, _, tag, name, args = message

self.commands.append((tag, name, json.loads(args) if args else None))

return codec.dumps({"ok": True}) if name == "get_value" else None

if kind == "animate":

self.animations.append(json.loads(message[3]))

return codec.dumps({"ok": True})

if kind == "call":

_, _, module, method, envelope = message

args = json.loads(envelope)["args"]

self.calls.append((module, method, args))

if (module, method) in self.module_results:

return codec.dumps(self.module_results[(module, method)])

if module == "Layout":

return codec.dumps(

{

"ok": True,

"value": {

"application": self.commit.application,

"surface": self.commit.surface,

"revision": self.commit.revision,

"frames": [],

}

)

if method in ("attach_root", "viewport"):

return codec.dumps({"ok": True, "value": self.viewport})

return codec.dumps({"ok": True, "value": None})

return None

# -- helpers ------------------------------------------------------------

def ops(self) -> List[List[Any]]:

out: List[List[Any]] = []

for message in self.sent:

if message[0] == "apply":

out.extend(message[2]["ops"])

return out

def dev_messages(self) -> List[Dict[str, Any]]:

return [m[1] for m in self.sent if m[0] == "dev"]

def _event(self, args: Any) -> Any:

self.sequence = getattr(self, "sequence", 0) + 1

state = self.commit.acknowledgement()

state.pop("ok")

state.update(sequence=self.sequence, args=args)

return state

def callback(self, kind: str, tag: int, name: str, payload: Any) -> None:

if kind == "event":

payload = self._event(payload)

text = codec.dumps(["cb", kind, tag, name, payload if isinstance(payload, str) else codec.dumps(payload)])

self.transport.on_preview_message(self, text)

@pytest.fixture

def web() -> Generator[Any, None, None]:

from pythonnative.hosts import native as hosts

transport = WebTransport(log=lambda line: None)

bridge.set_transport(transport)

backend = BridgeBackend(transport)

set_backend(backend)

modules._reset_for_tests()

hosts._reset_for_tests()

page = FakePage(transport)

transport.on_preview_connected(page, {})

transport.drain_main()

yield types.SimpleNamespace(transport=transport, page=page, backend=backend)

transport.on_preview_disconnected(page)

transport.drain_main()

hosts._reset_for_tests()

modules._reset_for_tests()

set_backend(None)

bridge._reset_for_tests()

def _install_app(monkeypatch: pytest.MonkeyPatch, name: str, root: Any) -> str:

import sys

module = types.ModuleType(name)

module.App = root # type: ignore[attr-defined]

monkeypatch.setitem(sys.modules, name, module)

return name

# ----------------------------------------------------------------------

# Wire format

# ----------------------------------------------------------------------

def test_apply_waits_for_revision_acknowledgement(web: Any) -> None:

envelope = {

"version": PROTOCOL_VERSION,

"application": "test",

"surface": 1,

"revision": 1,

"ops": [["c", 1, "View", {}]],

}

result = json.loads(web.transport.apply(codec.dumps(envelope)))

assert result == {"ok": True, "application": "test", "surface": 1, "revision": 1}

assert web.page.sent[-1][0] == "apply"

assert web.page.sent[-1][2] == envelope

def test_requests_return_nothing_when_no_page_is_attached() -> None:

transport = WebTransport(log=lambda line: None)

assert transport.command(1, "focus", "{}") is None

def test_command_and_animate_round_trip_json_text(web: Any) -> None:

assert web.transport.command(4, "focus", "{}") is None

assert json.loads(web.transport.command(4, "get_value", '{"x": 1}') or "") == {"ok": True}

assert web.page.commands == [(4, "focus", {}), (4, "get_value", {"x": 1})]

assert json.loads(web.transport.animate(4, '{"op": "start", "id": 9}') or "") == {"ok": True}

assert web.page.animations == [{"op": "start", "id": 9}]

def test_request_timeout_is_reported_not_raised(web: Any, monkeypatch: pytest.MonkeyPatch) -> None:

from pythonnative.bridge import web as web_mod

monkeypatch.setattr(web_mod, "REQUEST_TIMEOUT_S", 0.05)

web.page.send = lambda text: web.page.sent.append(json.loads(text)) # never answers

assert web.transport.command(1, "focus", "{}") is None

def test_disconnect_fails_waiting_requests(web: Any) -> None:

results: List[Any] = []

web.page.send = lambda text: web.page.sent.append(json.loads(text)) # never answers

def _command() -> None:

results.append(web.transport.command(1, "focus", "{}"))

thread = threading.Thread(target=_command)

thread.start()

time.sleep(0.05)

web.transport.on_preview_disconnected(web.page)

thread.join(timeout=2.0)

assert results == [None]

assert not web.transport.connected

# ----------------------------------------------------------------------

# Module routing

# ----------------------------------------------------------------------

def test_browser_modules_go_to_the_page_and_others_fall_back_to_python(web: Any) -> None:

web.page.module_results[("Clipboard", "get_string")] = {"ok": True, "value": "from page"}

assert modules.native_module("Clipboard").call("get_string") == "from page"

assert ("Clipboard", "get_string", {}) in web.page.calls

# Storage has no browser implementation: the Python fallback serves it.

storage = modules.native_module("Storage")

storage.call("set", key="k", value="v")

assert storage.call("get", key="k") == "v"

storage.call("delete", key="k")

assert all(module != "Storage" for module, _, _ in web.page.calls)

assert "Host" in BROWSER_MODULES and "Alert" in BROWSER_MODULES

def test_host_post_pumps_the_main_queue_without_touching_the_page(web: Any) -> None:

ran: List[int] = []

web.transport.post_to_application(lambda: ran.append(1))

assert ran == [] # queued, not inline

assert not any(m[0] == "call" for m in web.page.sent)

web.transport.drain_main()

assert ran == [1]

def test_python_fallback_reports_unknown_modules_and_errors(web: Any) -> None:

from pythonnative.native_modules.registry import NativeModuleError

with pytest.raises(NativeModuleError):

modules.native_module("NoSuchModule").call("anything")

def test_async_python_fallbacks_settle_through_module_messages(web: Any) -> None:

from pythonnative.runtime import run_blocking

async def _go() -> Any:

return await modules.native_module("Storage").call_async("get", key="missing")

assert run_blocking(_go(), timeout=2.0) is None

# ----------------------------------------------------------------------

# Callbacks from the page

# ----------------------------------------------------------------------

def test_events_from_the_page_reach_handlers_on_the_main_thread(web: Any) -> None:

from pythonnative.events import get_event_registry

seen: List[Any] = []

from pythonnative.mutations import CreateOp

web.backend.apply_mutations([CreateOp(42, "Button", {"title": "Press"})])

get_event_registry().set_events(42, {"on_press": lambda *args: seen.append(args)})

web.page.callback("event", 42, "on_press", [])

assert seen == [] # delivered on the main loop, not on the socket thread

web.transport.drain_main()

assert seen == [()]

def test_page_callbacks_are_never_answered(web: Any) -> None:

from pythonnative.events import get_event_registry

from pythonnative.mutations import CreateOp

web.backend.apply_mutations(

[CreateOp(5, "VirtualList", {"dataset": {"base": 0, "revision": 1, "changes": [["reset", []]]}})]

)

seen: List[Any] = []

def bind_row(payload: Any) -> Dict[str, int]:

seen.append(payload)

return {"root": 77}

get_event_registry().set_events(5, {"on_bind_row": bind_row})

before = len(web.page.sent)

web.page.callback("event", 5, "on_bind_row", [{"index": 0}])

web.transport.drain_main()

assert seen == [{"index": 0}]

assert not any(message[0] == "res" for message in web.page.sent[before:])

def test_dev_messages_and_peer_changes_run_hooks_on_the_main_thread(web: Any) -> None:

devs: List[Dict[str, Any]] = []

peers: List[bool] = []

web.transport.on_dev_message = devs.append

web.transport.on_peer_changed = peers.append

web.transport.on_preview_message(web.page, json.dumps(["dev", {"type": "log", "text": "hi"}]))

web.transport.on_preview_message(web.page, "not json")

web.transport.on_preview_message(web.page, json.dumps(["mystery"]))

web.transport.drain_main()

assert devs == [{"type": "log", "text": "hi"}]

web.transport.send_dev({"type": "hello", "entry": "app.main"})

assert web.page.sent[-1] == ["dev", {"type": "hello", "entry": "app.main"}]

other = FakePage(web.transport)

web.transport.on_preview_connected(other, {})

web.transport.drain_main()

assert peers == [True]

web.transport.on_preview_disconnected(other)

web.transport.drain_main()

assert peers == [True, False]

def test_stale_peer_messages_are_ignored(web: Any) -> None:

stale = FakePage(web.transport)

seen: List[Any] = []

web.transport.on_dev_message = seen.append

web.transport.on_preview_message(stale, json.dumps(["dev", {"type": "log"}]))

web.transport.drain_main()

assert seen == []

def test_gesture_stream_drives_the_python_arbiter(web: Any) -> None:

from pythonnative.events import get_event_registry

taps: List[Any] = []

from pythonnative.mutations import CreateOp

web.backend.apply_mutations([CreateOp(9, "View", {})])

get_event_registry().set_events(9, {"gesture:0": lambda payload: taps.append(payload)})

specs = [{"kind": "tap", "n_taps": 1, "max_distance": 20.0}]

t = web.transport

def _pointer(phase: str, x: float, y: float) -> None:

# The page sends window coordinates beside the view-local point.

info = {"id": 1, "x": x, "y": y, "absolute_x": x + 100, "absolute_y": y + 200, "specs": specs}

t.on_preview_message(web.page, json.dumps(["gesture", 9, phase, info]))

_pointer("down", 10, 10)

_pointer("up", 11, 10)

t.drain_main(timeout=0.5)

assert len(taps) == 1

# The arbiter reports the window-relative point the page sent alongside the local one.

assert taps[0]["absolute_x"] - taps[0]["x"] == 100.0

assert taps[0]["absolute_y"] - taps[0]["y"] == 200.0

_pointer("clear", 0, 0)

t.drain_main()

assert 9 not in t._gestures

def test_new_device_modules_are_served_by_the_page(web: Any) -> None:

for name in ("Keyboard", "AccessibilityInfo", "Localization", "Device"):

assert name in BROWSER_MODULES

web.page.module_results[("Keyboard", "is_visible")] = {"ok": True, "value": True}

web.page.module_results[("Localization", "get_timezone")] = {"ok": True, "value": "Europe/Paris"}

assert modules.native_module("Keyboard").call("is_visible") is True

assert modules.native_module("Localization").call("get_timezone") == "Europe/Paris"

assert ("Keyboard", "is_visible", {}) in web.page.calls

assert ("Localization", "get_timezone", {}) in web.page.calls

# ----------------------------------------------------------------------

# Full mount: the page hosts a screen rendered by the reconciler

# ----------------------------------------------------------------------

def test_screen_mounts_through_the_page(web: Any, monkeypatch: pytest.MonkeyPatch) -> None:

from pythonnative import platform_metrics

from pythonnative.hosts import native as hosts

pressed: List[int] = []

@component

def Root() -> Element:

count, set_count = pn.use_state(0)

def tap() -> None:

pressed.append(1)

set_count(count + 1)

return pn.Column(

pn.Text(f"count {count}"),

pn.Button("Tap", on_press=tap),

style={"flex": 1},

)

path = _install_app(monkeypatch, "web_preview_app", Root)

payload = {"path": path, "args": None, **web.page.viewport, "color_scheme": "light"}

web.page.callback("host", 1, "create", payload)

web.transport.drain_main(timeout=1.0)

root_tag = next(

args["tag"] for module, method, args in web.page.calls if (module, method) == ("Host", "attach_root")

)

created = {op[1]: op[2] for op in web.page.ops() if op[0] == "c"}

assert created[root_tag] == "Column"

assert "Text" in created.values() and "Button" in created.values()

assert ("Host", "attach_root", {"screen": 1, "tag": root_tag}) in web.page.calls

# Children get frames (the root fills the viewport), and the Text was measured by the page.

assert any(

message[0] == "apply" and message[2].get("layout", {}).get("roots") == [root_tag] for message in web.page.sent

)

assert not any(module == "Layout" for module, _, _ in web.page.calls)

window = platform_metrics.get_window_dimensions()

assert (window.width, window.height) == (390.0, 800.0)

host = hosts.host_for_screen(1)

assert host is not None

button_tag = next(tag for tag, name in created.items() if name == "Button")

text_tag = next(tag for tag, name in created.items() if name == "Text")

web.page.callback("event", button_tag, "on_press", [])

web.transport.drain_main(timeout=1.0)

from pythonnative.runtime import drain

drain()

assert pressed == [1]

updates = [op for op in web.page.ops() if op[0] == "u" and op[1] == text_tag]

assert updates and updates[-1][2]["text"] == "count 1"

web.page.callback("host", 1, "layout", {"width": 800.0, "height": 390.0})

web.transport.drain_main(timeout=1.0)

assert host.reconciler is not None and host.reconciler.viewport_size == (800.0, 390.0)

web.page.callback("host", 1, "destroy", {})

web.transport.drain_main(timeout=1.0)

assert hosts.host_for_screen(1) is None

assert any(op[0] == "d" and op[1] == root_tag for op in web.page.ops())

platform_metrics.reset_window_dimensions()

platform_metrics.reset_safe_area_insets()

pn.appearance.reset_color_scheme()

def test_backend_reset_forgets_every_view_when_the_page_goes_away(web: Any, monkeypatch: pytest.MonkeyPatch) -> None:

@component

def Root() -> Element:

return pn.Text("bye")

path = _install_app(monkeypatch, "web_preview_reset_app", Root)

web.page.callback("host", 1, "create", {"path": path, **web.page.viewport})

web.transport.drain_main(timeout=1.0)

assert web.backend.live_view_count() > 0

from pythonnative.hosts.native import live_hosts

for host in list(live_hosts()):

host.on_destroy()

web.backend.reset()

assert web.backend.live_view_count() == 0

pythonnative/tests/test_web_transport.py at refs/heads/main · pythonnative/pythonnative · GitHub

Repository navigation

Expand file tree

Breadcrumbs

test_web_transport.py

Latest commit

History

Breadcrumbs

test_web_transport.py

File metadata and controls