pythonnative/tests/test_devclient.py at main · pythonnative/pythonnative · GitHub

Repository navigation

History

391 lines (325 loc) · 16.1 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

"""The on-device dev client: URL handling, overlay sync, and reload wiring."""

from __future__ import annotations

import json

import os

import sys

import time

from pathlib import Path

from typing import Any, Iterator, List

import pytest

from pythonnative import devclient

from pythonnative.devserver import DevServer

from pythonnative.devserver.watcher import snapshot_sources

from pythonnative.hot_reload import configure_dev_environment

def _write(path: Path, text: str) -> None:

path.parent.mkdir(parents=True, exist_ok=True)

path.write_text(text, encoding="utf-8")

def _url(server: DevServer) -> str:

"""The device URL ``pn start`` prints for ``server``, dev token included."""

return server.info.preview_url("127.0.0.1")

def _wait(predicate: Any, timeout: float = 5.0) -> None:

deadline = time.monotonic() + timeout

while time.monotonic() < deadline:

if predicate():

return

time.sleep(0.01)

raise AssertionError("condition not met in time")

@pytest.fixture

def project(tmp_path: Path) -> Path:

root = tmp_path / "proj"

_write(root / "app" / "__init__.py", "")

_write(root / "app" / "main.py", "VALUE = 1\n")

_write(root / "app" / "assets" / "note.txt", "hi")

return root

@pytest.fixture

def server(project: Path) -> Iterator[DevServer]:

srv = DevServer(str(project), "app.main", host="127.0.0.1", port=0, watch=False, log=lambda _: None)

srv.start()

try:

yield srv

finally:

srv.stop()

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

# URLs

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

@pytest.mark.parametrize(

"typed, expected",

[

("192.168.1.20", "ws://192.168.1.20:8765/ws?role=client"),

("192.168.1.20:9000", "ws://192.168.1.20:9000/ws?role=client"),

("http://mac.local:8765", "ws://mac.local:8765/ws?role=client"),

("http://mac.local:8765/", "ws://mac.local:8765/ws?role=client"),

("ws://10.0.2.2:8765/ws?role=client", "ws://10.0.2.2:8765/ws?role=client"),

("ws://10.0.2.2:8765/ws", "ws://10.0.2.2:8765/ws?role=client"),

(" localhost:8765 ", "ws://localhost:8765/ws?role=client"),

# The device URL `pn start` prints, and the forms a developer might type from it.

("http://192.168.1.20:8765/?token=abc-_1", "ws://192.168.1.20:8765/ws?role=client&token=abc-_1"),

("192.168.1.20:8765/?token=abc", "ws://192.168.1.20:8765/ws?role=client&token=abc"),

("192.168.1.20?token=abc", "ws://192.168.1.20:8765/ws?role=client&token=abc"),

("ws://10.0.2.2:8765/ws?role=client&token=abc", "ws://10.0.2.2:8765/ws?role=client&token=abc"),

("ws://10.0.2.2:8765/ws?token=abc&role=preview", "ws://10.0.2.2:8765/ws?role=client&token=abc"),

)

def test_normalize_server_url(typed: str, expected: str) -> None:

assert devclient.normalize_server_url(typed) == expected

def test_normalize_server_url_rejects_empty() -> None:

with pytest.raises(ValueError):

devclient.normalize_server_url(" ")

def test_normalize_server_url_rejects_other_schemes() -> None:

with pytest.raises(ValueError):

devclient.normalize_server_url("ftp://192.168.1.20:8765")

@pytest.mark.parametrize(

"url, shown",

[

("ws://192.168.1.20:8765/ws?role=client&token=abc", "192.168.1.20:8765/?token=abc"),

("ws://192.168.1.20:8765/ws?role=client", "192.168.1.20:8765"),

)

def test_display_server_url_round_trips_through_normalize(url: str, shown: str) -> None:

assert devclient.display_server_url(url) == shown

assert devclient.normalize_server_url(shown) == url

def test_client_without_the_token_is_refused_with_a_clear_message(server: DevServer, tmp_path: Path) -> None:

logs: List[str] = []

client = devclient.DevClient(

server.info.url("127.0.0.1"), str(tmp_path / "overlay"), forward_logs=False, log=logs.append

)

client.start()

try:

_wait(lambda: any("refused the connection" in line for line in logs))

assert client.state == "disconnected"

assert not client.synced_once.is_set()

finally:

client.stop()

def test_client_logs_never_include_the_token(server: DevServer, tmp_path: Path) -> None:

logs: List[str] = []

details: List[str] = []

client = devclient.DevClient(_url(server), str(tmp_path / "overlay"), forward_logs=False, log=logs.append)

client.add_listener(lambda state, detail: details.append(detail))

client.start()

try:

assert client.synced_once.wait(5.0)

finally:

client.stop()

assert client.server_label == f"127.0.0.1:{server.info.port}"

assert not any(server.token in line for line in logs + details)

def test_saved_server_url_round_trips_through_the_overlay(tmp_path: Path) -> None:

overlay = tmp_path / "overlay"

assert devclient.saved_server_url(str(overlay)) is None

devclient._save_server_url(str(overlay), "ws://1.2.3.4:8765/ws?role=client")

assert devclient.saved_server_url(str(overlay)) == "ws://1.2.3.4:8765/ws?role=client"

(overlay / "server.json").write_text("not json", encoding="utf-8")

assert devclient.saved_server_url(str(overlay)) is None

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

# Sync

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

def test_client_syncs_the_tree_into_the_overlay_and_reports_state(

server: DevServer, tmp_path: Path, monkeypatch: pytest.MonkeyPatch

) -> None:

overlay = tmp_path / "overlay"

states: List[str] = []

reloads: List[List[str]] = []

monkeypatch.setattr(

devclient.DevClient, "_schedule_reload", lambda self, modules, version, **kw: reloads.append(modules)

)

client = devclient.DevClient(

_url(server), str(overlay), entry_module="app.main", forward_logs=False, log=lambda _: None

)

client.add_listener(lambda state, detail: states.append(state))

client.start()

try:

assert client.synced_once.wait(5.0)

_wait(lambda: client.state == "connected")

assert (overlay / "app" / "main.py").read_text(encoding="utf-8") == "VALUE = 1\n"

assert (overlay / "app" / "assets" / "note.txt").read_text(encoding="utf-8") == "hi"

assert client.synced_version == server.snapshot.version

assert states[:3] == ["connecting", "connected", "syncing"]

# The URL is remembered for --dev-client builds.

assert devclient.saved_server_url(str(overlay)) == client.url

# Everything was new to this overlay, so the synced modules are reloaded (a

# --dev-client shell's placeholder entry is replaced the same way).

_wait(lambda: len(reloads) == 1)

assert reloads[0] == ["app", "app.main"]

_wait(lambda: len(server.clients) == 1)

assert server.clients[0].app == "app.main"

finally:

client.stop()

assert client.state == "idle"

def test_client_hello_advertises_existing_overlay_so_only_changes_flow(server: DevServer, tmp_path: Path) -> None:

overlay = tmp_path / "overlay"

# Pre-seed the overlay with the current main.py so the server has nothing new to send for it.

_write(overlay / "app" / "main.py", "VALUE = 1\n")

client = devclient.DevClient(_url(server), str(overlay), forward_logs=False, log=lambda _: None)

manifest = client._overlay_manifest()

assert set(manifest) == {"app/main.py"}

assert manifest["app/main.py"] == server.snapshot.files["app/main.py"]

def test_first_hello_seeds_the_overlay_from_the_bundled_sources(

server: DevServer, project: Path, tmp_path: Path, monkeypatch: pytest.MonkeyPatch

) -> None:

"""A build made from the server's sources connects without downloading or reloading anything."""

bundle = tmp_path / "bundle"

for rel in ("app/__init__.py", "app/main.py", "app/assets/note.txt"):

_write(bundle / rel, (project / rel).read_text(encoding="utf-8"))

_write(bundle / "app" / "__pycache__" / "main.cpython-313.pyc", "junk")

monkeypatch.syspath_prepend(str(bundle))

# Prepared the way the native templates do it: the overlay already contains an empty app/ dir.

overlay = Path(configure_dev_environment(str(tmp_path / "data")))

monkeypatch.setattr(sys, "path", [p for p in sys.path if p != str(overlay)])

assert (overlay / "app").is_dir() and not any((overlay / "app").iterdir())

reloads: List[List[str]] = []

monkeypatch.setattr(

devclient.DevClient, "_schedule_reload", lambda self, modules, version, **kw: reloads.append(modules)

)

client = devclient.DevClient(

_url(server), str(overlay), entry_module="app.main", forward_logs=False, log=lambda _: None

)

hello = client._hello()

# Seeded from the bundle: the whole tree, minus caches, hashed like the server does.

assert hello["files"] == server.snapshot.files

assert not (overlay / "app" / "__pycache__").exists()

client.start()

try:

assert client.synced_once.wait(5.0)

_wait(lambda: client.state == "connected")

# Nothing differed, so the first sync wrote nothing and reloaded nothing.

time.sleep(0.2)

assert reloads == []

finally:

client.stop()

# Seeding happens once: an existing overlay is left alone even if the bundle changes.

_write(bundle / "app" / "main.py", "VALUE = 2\n")

assert client._hello()["files"]["app/main.py"] == server.snapshot.files["app/main.py"]

def test_bundled_package_root_ignores_the_overlay_and_does_not_import(

tmp_path: Path, monkeypatch: pytest.MonkeyPatch

) -> None:

overlay = tmp_path / "overlay"

_write(overlay / "pkg_seed_test" / "__init__.py", "")

bundle = tmp_path / "bundle"

_write(bundle / "pkg_seed_test" / "__init__.py", "raise RuntimeError('must not be imported')\n")

_write(bundle / "pkg_seed_test" / "sub" / "leaf.py", "")

monkeypatch.syspath_prepend(str(bundle))

monkeypatch.syspath_prepend(str(overlay))

root = devclient._bundled_package_root("pkg_seed_test", exclude=str(overlay))

assert root is not None and Path(str(root)) == bundle / "pkg_seed_test"

assert sorted(child.name for child in root.iterdir()) == ["__init__.py", "sub"]

assert "pkg_seed_test" not in sys.modules

assert devclient._bundled_package_root("no_such_pkg_anywhere", exclude=str(overlay)) is None

def test_updates_are_applied_and_reloads_scheduled(

server: DevServer, project: Path, tmp_path: Path, monkeypatch: pytest.MonkeyPatch

) -> None:

overlay = tmp_path / "overlay"

reloads: List[List[str]] = []

monkeypatch.setattr(

devclient.DevClient, "_schedule_reload", lambda self, modules, version, **kw: reloads.append(modules)

)

client = devclient.DevClient(_url(server), str(overlay), forward_logs=False, log=lambda _: None)

client.start()

try:

assert client.synced_once.wait(5.0)

_wait(lambda: len(server.clients) == 1)

before = server.snapshot

_write(project / "app" / "main.py", "VALUE = 2\n")

_write(project / "app" / "screens.py", "X = 1\n")

(project / "app" / "assets" / "note.txt").unlink()

after = snapshot_sources(str(project), previous=before)

server._on_files_changed(before.diff(after), after)

_wait(lambda: len(reloads) >= 2)

assert (overlay / "app" / "main.py").read_text(encoding="utf-8") == "VALUE = 2\n"

assert (overlay / "app" / "screens.py").exists()

_wait(lambda: not (overlay / "app" / "assets" / "note.txt").exists())

assert reloads[-1] == ["app.main", "app.screens"]

assert client.synced_version == after.version

finally:

client.stop()

def test_client_refuses_paths_that_escape_the_overlay(tmp_path: Path) -> None:

client = devclient.DevClient("ws://127.0.0.1:1/ws", str(tmp_path / "overlay"), forward_logs=False)

client._handle(

{

"type": "update",

"version": "v",

"files": [

{"path": "../evil.py", "content": "x", "encoding": "utf-8"},

{"path": "app/ok.py", "content": "y", "encoding": "utf-8"},

"removed": ["../../etc/passwd"],

}

)

assert not (tmp_path / "evil.py").exists()

assert (tmp_path / "overlay" / "app" / "ok.py").read_text(encoding="utf-8") == "y"

def test_client_retries_when_the_server_is_down(tmp_path: Path) -> None:

logs: List[str] = []

client = devclient.DevClient("ws://127.0.0.1:1/ws", str(tmp_path / "overlay"), forward_logs=False, log=logs.append)

client.start()

try:

_wait(lambda: client.state == "disconnected")

_wait(lambda: any("cannot reach" in line for line in logs))

finally:

client.stop()

def test_log_forwarding_tees_stdout_to_the_server(server: DevServer, tmp_path: Path, capsys: Any) -> None:

client = devclient.DevClient(_url(server), str(tmp_path / "overlay"), forward_logs=True)

logs: List[str] = []

server.log = logs.append

client.start()

try:

assert client.synced_once.wait(5.0)

print("hello from the app")

_wait(lambda: any("hello from the app" in line for line in logs))

finally:

client.stop()

assert not isinstance(sys.stdout, devclient._Tee)

captured = capsys.readouterr()

assert "hello from the app" in captured.out # still printed locally

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

# Bootstrap wiring

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

def test_start_if_configured_needs_an_overlay_and_a_url(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:

monkeypatch.delenv(devclient.SERVER_URL_ENV, raising=False)

monkeypatch.setattr(devclient, "overlay_root", lambda: None)

assert devclient.start_if_configured() is None

monkeypatch.setattr(devclient, "overlay_root", lambda: str(tmp_path))

assert devclient.start_if_configured() is None # overlay but no URL anywhere

started: List[Any] = []

def fake_start(url: str, root: str, **kw: Any) -> str:

started.append((url, root, kw))

return "client"

monkeypatch.setattr(devclient, "start", fake_start)

monkeypatch.setenv(devclient.SERVER_URL_ENV, "ws://10.0.2.2:8765/ws?role=client")

monkeypatch.setenv("PN_ENTRY_MODULE", "app.entry")

assert devclient.start_if_configured() == "client"

assert started == [("ws://10.0.2.2:8765/ws?role=client", str(tmp_path), {"entry_module": "app.entry"})]

def test_start_if_configured_falls_back_to_the_saved_url(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:

monkeypatch.delenv(devclient.SERVER_URL_ENV, raising=False)

monkeypatch.delenv("PN_ENTRY_MODULE", raising=False)

monkeypatch.setattr(devclient, "overlay_root", lambda: str(tmp_path))

devclient._save_server_url(str(tmp_path), "ws://192.168.1.9:8765/ws?role=client")

started: List[Any] = []

def fake_start(url: str, root: str, **kw: Any) -> str:

started.append(url)

return "client"

monkeypatch.setattr(devclient, "start", fake_start)

assert devclient.start_if_configured() == "client"

assert started == ["ws://192.168.1.9:8765/ws?role=client"]

def test_placeholder_main_source_exports_the_connect_screen() -> None:

source = devclient.placeholder_main_source()

namespace: dict = {}

exec(compile(source, "main.py", "exec"), namespace)

assert namespace["App"] is devclient.ConnectScreen

assert namespace["__all__"] == ["App"]

def test_connect_screen_renders_without_a_client(monkeypatch: pytest.MonkeyPatch) -> None:

from pythonnative.testing import render

monkeypatch.setattr(devclient, "_current", None)

result = render(devclient.ConnectScreen())

try:

text = result.text()

assert "Dev server address" in text

assert "idle" in text

result.get_by_text("Connect")

finally:

result.unmount()

def test_overlay_manifest_skips_caches_and_the_saved_url(tmp_path: Path) -> None:

overlay = tmp_path / "overlay"

_write(overlay / "app" / "main.py", "x")

_write(overlay / "app" / "__pycache__" / "main.pyc", "junk")

_write(overlay / "server.json", json.dumps({"url": "ws://x"}))

client = devclient.DevClient("ws://127.0.0.1:1/ws", str(overlay), forward_logs=False)

assert set(client._overlay_manifest()) == {"app/main.py"}

assert os.path.exists(overlay / "server.json")

pythonnative/tests/test_devclient.py at main · pythonnative/pythonnative · GitHub

Repository navigation

Expand file tree

Breadcrumbs

test_devclient.py

Latest commit

History

Breadcrumbs

test_devclient.py

File metadata and controls