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

Repository navigation

History

286 lines (226 loc) · 10.5 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

"""Public Python values survive the native transport without type erasure."""

import asyncio

import json

from dataclasses import dataclass, field

from enum import Enum

from typing import Any, ClassVar, Iterator, Literal, Mapping, NotRequired, Protocol, Sequence, TypedDict

import pytest

import pythonnative as pn

from pythonnative.native_modules.registry import BridgeModule, PythonModule

from pythonnative.sdk.schema import COMPONENTS, MODULES, ModuleSchema

from pythonnative.sdk.types import decode_value, encode_value, type_schema

class Accuracy(Enum):

BALANCED = "balanced"

HIGH = "high"

@dataclass(frozen=True)

class Coordinate:

latitude: float

longitude: float

accuracy: Accuracy = Accuracy.BALANCED

labels: list[str] = field(default_factory=list)

class Note(TypedDict):

title: str

body: NotRequired[str]

class NullableNote(TypedDict):

required: str | None

optional: NotRequired[str | None]

class LocationAPI(Protocol):

async def locate(self, *, hint: Coordinate, timeout: float = 10) -> Coordinate: ...

@pytest.fixture

def location_contract() -> Iterator[ModuleSchema]:

old = MODULES.get("TypedLocationTest")

schema = ModuleSchema.from_protocol("TypedLocationTest", LocationAPI)

MODULES[schema.name] = schema

yield schema

if old is None:

MODULES.pop(schema.name, None)

else:

MODULES[schema.name] = old

def test_nested_record_enum_and_collection_round_trip() -> None:

value = {"home": (Coordinate(12.5, -70.25, Accuracy.HIGH, ["favorite"]), None)}

schema = type_schema(Mapping[str, tuple[Coordinate, str | None]])

wire = json.loads(json.dumps(encode_value(value, schema)))

assert wire["home"][0]["accuracy"] == "high"

assert decode_value(wire, schema) == value

assert isinstance(decode_value(wire, schema)["home"][0], Coordinate)

def test_record_defaults_and_typed_dict_absence() -> None:

result = decode_value({"latitude": 1, "longitude": 2}, type_schema(Coordinate))

assert result == Coordinate(1, 2)

assert decode_value({"title": "Draft"}, type_schema(Note)) == {"title": "Draft"}

with pytest.raises(TypeError, match="requires title"):

decode_value({}, type_schema(Note))

def test_nullable_record_fields_preserve_presence() -> None:

schema = type_schema(NullableNote)

for value in ({"required": None}, {"required": None, "optional": None}, {"required": "yes", "optional": "yes"}):

assert decode_value(encode_value(value, schema), schema) == value

with pytest.raises(TypeError, match="requires required"):

decode_value({"optional": None}, schema)

def test_record_field_names_fail_before_native_compilation() -> None:

Invalid = TypedDict("Invalid", {"user-name": str})

with pytest.raises(TypeError, match="Python identifier"):

type_schema(Invalid)

def test_sequences_and_dataclass_class_variables_keep_python_semantics() -> None:

@dataclass

class Batch:

version: ClassVar[int] = 1

values: Sequence[int]

schema = type_schema(Batch)

assert encode_value(Batch(range(3)), schema) == {"values": [0, 1, 2]}

assert decode_value({"values": [0, 1, 2]}, schema) == Batch([0, 1, 2])

with pytest.raises(TypeError, match="must be number"):

decode_value(10**1000, type_schema(float))

def test_custom_component_defaults_and_platforms_survive_factory_creation() -> None:

from pythonnative.platform import _set_platform_for_test

from pythonnative.sdk import define_component, unregister_component

@dataclass(frozen=True)

class MarkerProps:

coordinate: Coordinate = field(default_factory=lambda: Coordinate(1, 2))

try:

marker = define_component("TypedMarkerTest", MarkerProps, platforms=("ios",))

element = marker(style={"padding": 4, "opacity": 0.5})

assert element.props["coordinate"] == Coordinate(1, 2)

assert element.props["padding"] == 4

COMPONENTS["TypedMarkerTest"].validate(element.props)

assert COMPONENTS["TypedMarkerTest"].defaults["coordinate"]["accuracy"] == "balanced"

_set_platform_for_test("ios")

assert pn.Platform.supports("TypedMarkerTest")

_set_platform_for_test("android")

assert not pn.Platform.supports("TypedMarkerTest")

finally:

unregister_component("TypedMarkerTest")

_set_platform_for_test(None)

def test_literal_discriminated_records() -> None:

class Success(TypedDict):

status: Literal["ok"]

value: int

class Failure(TypedDict):

status: Literal["error"]

message: str

schema = type_schema(Success | Failure)

assert decode_value({"status": "error", "message": "offline"}, schema) == {"status": "error", "message": "offline"}

with pytest.raises(TypeError, match="annotation"):

decode_value({"status": "ok", "message": "wrong shape"}, schema)

@pytest.mark.parametrize("annotation", [object, dict[int, str], complex])

def test_unsupported_annotations_fail_during_declaration(annotation: Any) -> None:

with pytest.raises(TypeError):

type_schema(annotation)

@pytest.mark.parametrize("value", [{1: "numeric key"}, object(), float("nan"), 2**53])

def test_values_cannot_be_silently_dropped_or_coerced(value: Any) -> None:

with pytest.raises(TypeError):

encode_value(value)

def test_typed_native_call_preserves_defaults_and_results(location_contract: ModuleSchema) -> None:

class Transport:

def call(self, module: str, method: str, envelope: str) -> str:

request = json.loads(envelope)

assert request["args"]["timeout"] == 10

assert request["args"]["hint"]["accuracy"] == "balanced"

return json.dumps({"ok": True, "value": request["args"]["hint"]})

result = asyncio.run(BridgeModule("TypedLocationTest", Transport()).call_async("locate", hint=Coordinate(1, 2)))

assert type(result) is Coordinate

assert result == Coordinate(1, 2)

def test_fallback_receives_same_typed_arguments(location_contract: ModuleSchema) -> None:

class Implementation:

async def locate(self, *, hint: Coordinate, timeout: float) -> Coordinate:

assert type(hint) is Coordinate

assert timeout == 10

return hint

result = asyncio.run(

PythonModule("TypedLocationTest", Implementation()).call_async("locate", hint={"latitude": 1, "longitude": 2})

)

assert result == Coordinate(1, 2)

def test_media_event_shapes_match_real_native_payloads() -> None:

image = COMPONENTS["Image"].decode_event("on_load", [{"width": 64, "height": 32}])

assert image == [pn.ImageLoadEvent(64, 32)]

web = COMPONENTS["WebView"].decode_event(

"on_navigation_state_change",

[

{

"url": "https://example.com",

"loading": False,

"can_go_back": True,

"can_go_forward": False,

"title": "Example",

}

)

assert web[0].can_go_back is True

assert web[0].title == "Example"

for property_name in ("max_lines", "text_transform"):

assert COMPONENTS["Text"].props[property_name]["native"]["invalidates_layout"]

def test_null_and_property_removal_have_distinct_wire_and_native_state() -> None:

from pythonnative.bridge import codec

from pythonnative.bridge.commits import PROTOCOL_VERSION, CommitError, CommitState

from pythonnative.mutations import UNSET, CreateOp, UpdateOp

operations, _ = codec.encode_transaction(

[

CreateOp(1, "Picker", {"value": "draft"}),

UpdateOp(1, {"value": None}),

]

)

state = (

CommitState()

.prepare(

{

"version": PROTOCOL_VERSION,

"application": "null-test",

"surface": 1,

"revision": 1,

"ops": json.loads(operations),

}

)

.publish()

)

assert state.views[1].props["value"] is None

operations, _ = codec.encode_transaction([UpdateOp(1, {"value": UNSET})])

assert json.loads(operations) == [["u", 1, {}, ["value"]]]

envelope = {

"version": PROTOCOL_VERSION,

"application": "null-test",

"surface": 1,

"revision": 2,

"ops": json.loads(operations),

}

state = state.prepare(envelope).publish()

assert "value" not in state.views[1].props

with pytest.raises(CommitError, match="both set and removed"):

state.prepare({**envelope, "revision": 3, "ops": [["u", 1, {"value": "conflict"}, ["value"]]]})

def test_unserializable_native_props_are_errors_unless_declared_python_only() -> None:

from pythonnative.bridge.codec import split_props

value = object()

with pytest.raises(TypeError, match="bridge-serializable"):

split_props({"value": value})

assert split_props({"value": value}, frozenset({"value"})) == ({}, {"value": value})

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

from pythonnative.native_modules import registry

schema = ModuleSchema.from_protocol("TypedLocationTest", LocationAPI, events={"changed": Coordinate})

monkeypatch.setitem(MODULES, schema.name, schema)

module = PythonModule(schema.name)

monkeypatch.setitem(registry._modules, schema.name, module)

received: list[Coordinate] = []

unsubscribe = module.add_listener("changed", received.append)

registry.emit(schema.name, "changed", {"latitude": 2, "longitude": 3})

assert received == [Coordinate(2, 3)]

with pytest.raises(TypeError):

registry.emit(schema.name, "changed", {"latitude": "bad", "longitude": 3})

assert len(received) == 1

unsubscribe()

registry.emit(schema.name, "changed", {"latitude": 4, "longitude": 5})

assert len(received) == 1

def test_cancelling_typed_native_call_releases_native_work(location_contract: ModuleSchema) -> None:

calls: list[tuple[str, dict[str, Any]]] = []

class Transport:

def call(self, module: str, method: str, envelope: str) -> str:

request = json.loads(envelope)

calls.append((method, request))

return json.dumps({"pending": True} if method == "locate" else {"ok": True})

async def exercise() -> None:

task = asyncio.create_task(

BridgeModule("TypedLocationTest", Transport()).call_async("locate", hint=Coordinate(1, 2))

)

await asyncio.sleep(0)

task.cancel()

with pytest.raises(asyncio.CancelledError):

await task

asyncio.run(exercise())

assert [method for method, _ in calls] == ["locate", "_pn_cancel"]

assert calls[1][1]["args"]["call_id"] == calls[0][1]["call_id"]

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

Repository navigation

Expand file tree

Breadcrumbs

test_native_values.py

Latest commit

History

Breadcrumbs

test_native_values.py

File metadata and controls