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

Repository navigation

History

180 lines (126 loc) · 5.14 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

"""Unit tests for pn.AsyncStorage and pn.use_persisted_state."""

from __future__ import annotations

import asyncio

import json

import os

from pathlib import Path

from typing import Generator

import pytest

from pythonnative.component import component

from pythonnative.element import Element

from pythonnative.native_modules.registry import native_module

from pythonnative.reconciler import Reconciler

from pythonnative.runtime import drain, run_blocking

from pythonnative.storage import AsyncStorage, use_persisted_state

from pythonnative.testing import FakeBackend as _StubBackend

@pytest.fixture(autouse=True)

def _reset_fallback_store(tmp_path: Path) -> Generator[None, None, None]:

"""Isolate the fallback Storage module per test by pointing it at a temp dir."""

os.environ["PN_STORAGE_DIR"] = str(tmp_path)

native_module("Storage").impl._reset() # type: ignore[attr-defined]

yield

native_module("Storage").impl._reset() # type: ignore[attr-defined]

os.environ.pop("PN_STORAGE_DIR", None)

def test_set_and_get_round_trip() -> None:

async def run() -> str | None:

await AsyncStorage.set("name", "Alice")

return await AsyncStorage.get("name")

assert asyncio.run(run()) == "Alice"

def test_get_missing_key_returns_none() -> None:

async def run() -> str | None:

return await AsyncStorage.get("never-set")

assert asyncio.run(run()) is None

def test_set_rejects_non_string_values() -> None:

async def run() -> None:

with pytest.raises(TypeError):

await AsyncStorage.set("k", 42) # type: ignore[arg-type]

asyncio.run(run())

def test_set_json_and_get_json_round_trip_complex_value() -> None:

payload = {"name": "Alice", "tags": [1, 2, 3], "active": True}

async def run() -> object:

await AsyncStorage.set_json("user", payload)

return await AsyncStorage.get_json("user")

assert asyncio.run(run()) == payload

def test_get_json_returns_none_for_invalid_json() -> None:

async def run() -> object:

# Bypass set_json to write a non-JSON value.

await AsyncStorage.set("user", "not json {")

return await AsyncStorage.get_json("user")

assert asyncio.run(run()) is None

def test_delete_removes_value() -> None:

async def run() -> str | None:

await AsyncStorage.set("temp", "v")

await AsyncStorage.delete("temp")

return await AsyncStorage.get("temp")

assert asyncio.run(run()) is None

def test_all_keys_returns_persisted_keys() -> None:

async def run() -> set:

await AsyncStorage.set("a", "1")

await AsyncStorage.set("b", "2")

keys = await AsyncStorage.all_keys()

return set(keys)

assert asyncio.run(run()) == {"a", "b"}

def test_clear_removes_everything() -> None:

async def run() -> list:

await AsyncStorage.set("a", "1")

await AsyncStorage.set("b", "2")

await AsyncStorage.clear()

return await AsyncStorage.all_keys()

assert asyncio.run(run()) == []

def test_fallback_backend_persists_to_disk(tmp_path: Path) -> None:

async def write() -> None:

await AsyncStorage.set("name", "Alice")

asyncio.run(write())

# File should now exist with the value serialised.

on_disk = json.loads((tmp_path / "pn_async_storage.json").read_text())

assert on_disk == {"name": "Alice"}

# ======================================================================

# use_persisted_state

# ======================================================================

def test_use_persisted_state_starts_with_initial() -> None:

captured: list = []

@component

def screen() -> Element:

value, _set = use_persisted_state("theme", "light")

captured.append(value)

return Element("View", {}, [])

rec = Reconciler(_StubBackend())

rec.mount(screen())

assert captured[0] == "light"

def test_use_persisted_state_loads_existing_value() -> None:

async def seed() -> None:

await AsyncStorage.set_json("theme", "dark")

asyncio.run(seed())

captured: list = []

@component

def screen() -> Element:

value, _set = use_persisted_state("theme", "light")

captured.append(value)

return Element("View", {}, [])

rec = Reconciler(_StubBackend())

rec.mount(screen())

# The first render returns the initial; the async load runs as a

# task on the framework loop. Pump the loop, then flush the

# resulting local re-render.

assert captured[0] == "light"

drain()

rec.flush_dirty()

assert captured[-1] == "dark"

def test_use_persisted_state_setter_persists_writes() -> None:

setters: list = []

@component

def screen() -> Element:

value, set_value = use_persisted_state("theme", "light")

setters.append((value, set_value))

return Element("View", {}, [])

rec = Reconciler(_StubBackend())

rec.mount(screen())

# Pump the load effect so loaded=True and the setter actually writes.

drain()

rec.flush_dirty()

_, set_value = setters[-1]

set_value("dark")

rec.flush_dirty()

# Pump the fire-and-forget persistence write.

drain()

assert run_blocking(AsyncStorage.get_json("theme")) == "dark"

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

Repository navigation

Expand file tree

Breadcrumbs

test_storage.py

Latest commit

History

Breadcrumbs

test_storage.py

File metadata and controls