tests/core/test_session.py¶
Source from this local checkout, regenerated when the reader rebuilds.
Line links use #L<number>; a GitHub line range opens its first line.
1 # Copyright 2025 Softwell S.r.l.2 #3 # Licensed under the Apache License, Version 2.0 (the "License");4 # you may not use this file except in compliance with the License.5 # You may obtain a copy of the License at6 #7 # https://www.apache.org/licenses/LICENSE-2.08 #9 # Unless required by applicable law or agreed to in writing, software10 # distributed under the License is distributed on an "AS IS" BASIS,11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.12 # See the License for the specific language governing permissions and13 # limitations under the License.14 15 """Session tests (Macro 2 Phase 4 + core 1b Phase 3): store contract + cookie flow.16 17 The store contract suite runs over a FACTORY (invariant §5.9): a callable18 returning a fresh configured store, so a future backend can plug into the SAME19 suite as ``MemorySessionStore`` (the only shipped store). The cookie flow is20 driven directly through a ``SessionMixin/MiddlewareMixin/BaseServer``21 composition at the ASGI level (no uvicorn), the same driving style as22 ``test_middleware.py``.23 """24 25 from __future__ import annotations26 27 import time28 29 import pytest30 31 from genro_asgi import (32 Avatar,33 BaseApplication,34 BaseServer,35 MemorySessionStore,36 Session,37 SessionMixin,38 SessionStore,39 )40 from genro_asgi.middleware import MiddlewareMixin41 from genro_asgi.middleware.session import COOKIE_LIFETIME_FACTOR, SessionMiddleware42 from genro_asgi.types import Message, Receive, Scope, Send43 44 # --- store contract suite (over a FACTORY, §5.9) ---45 46 47 @pytest.fixture48 def store_factory():49 """A callable returning a fresh configured ``MemorySessionStore``."""50 51 def make(**kwargs):52 return MemorySessionStore(**kwargs)53 54 return make55 56 57 class TestSessionStoreContract:58 def test_is_a_session_store(self, store_factory) -> None:59 assert isinstance(store_factory(), SessionStore)60 61 def test_create_default_is_anonymous(self, store_factory) -> None:62 assert store_factory().create().avatar() is None63 64 def test_create_get_roundtrip(self, store_factory) -> None:65 store = store_factory()66 created = store.create(avatar=Avatar("alice"))67 fetched = store.get(created.id)68 assert fetched is created69 assert fetched.avatar().identity == "alice"70 71 def test_get_unknown_returns_none(self, store_factory) -> None:72 assert store_factory().get("nope") is None73 74 def test_ttl_expiry(self, store_factory) -> None:75 store = store_factory(default_ttl=3600)76 session = store.create()77 session.meta["last_access"] = time.time() - 10_00078 assert store.get(session.id) is None79 80 def test_delete(self, store_factory) -> None:81 store = store_factory()82 session = store.create()83 store.delete(session.id)84 assert store.get(session.id) is None85 86 def test_purge_expired_removes_only_expired(self, store_factory) -> None:87 store = store_factory(default_ttl=3600)88 live = store.create()89 expired = store.create()90 expired.meta["last_access"] = time.time() - 10_00091 assert store.purge_expired() == 192 assert store.get(live.id) is not None93 assert store.get(expired.id) is None94 95 def test_dump_restore_keeps_avatar_drops_data(self, store_factory) -> None:96 store = store_factory()97 session = store.create(avatar=Avatar("bob", ["user"]))98 session.data["k"] = "v"99 dumped = store.dump()100 fresh = store_factory()101 fresh.restore(dumped)102 restored = fresh.get(session.id)103 assert restored is not None104 assert restored.avatar().identity == "bob"105 assert restored.avatar().tags == ["user"]106 assert len(restored.data) == 0107 108 def test_save_is_in_the_contract(self, store_factory) -> None:109 # save() is part of the SessionStore Protocol on every backend; calling it110 # on a live session must succeed (the write-back seam).111 store = store_factory()112 session = store.create()113 session.attach_avatar(Avatar("carol", ["ops"]))114 store.save(session)115 again = store.get(session.id)116 assert again is not None117 assert again.avatar() is not None and again.avatar().identity == "carol"118 119 120 # --- delta-checked mass reap at create time (session-store minimization) ---121 122 123 class TestPurgeDeltaCheck:124 def test_create_within_interval_skips_the_mass_reap(self) -> None:125 store = MemorySessionStore(default_ttl=3600)126 expired = store.create()127 expired.meta["last_access"] = time.time() - 10_000128 store.create() # _last_purge is fresh: no reap happens here129 assert expired.id in store.dump()130 131 def test_create_past_interval_runs_the_mass_reap(self) -> None:132 store = MemorySessionStore(default_ttl=3600)133 expired = store.create()134 expired.meta["last_access"] = time.time() - 10_000135 store._last_purge = time.time() - 10_000 # the interval has elapsed136 store.create()137 assert expired.id not in store.dump()138 139 def test_purge_expired_resets_the_delta_clock(self) -> None:140 store = MemorySessionStore(default_ttl=3600)141 store._last_purge = time.time() - 10_000142 store.purge_expired()143 assert time.time() - store._last_purge < 60144 145 146 # --- pickle snapshot: the whole store crosses a restart (data Bag INCLUDED) ---147 148 149 class TestSnapshot:150 def test_roundtrip_keeps_avatar_and_data(self, tmp_path) -> None:151 store = MemorySessionStore(default_ttl=3600)152 session = store.create(avatar=Avatar("alice", ["admin"]))153 session.data["cart"] = "kept"154 session.attach_avatar(Avatar("alice@erp", ["operator"]), "erp")155 path = tmp_path / "sessions.pickle"156 assert store.save_snapshot(path) == 1157 fresh = MemorySessionStore(default_ttl=3600)158 assert fresh.load_snapshot(path) == 1159 restored = fresh.get(session.id)160 assert restored is not None161 assert restored.avatar().identity == "alice"162 assert restored.avatar("erp").identity == "alice@erp"163 assert restored.data["cart"] == "kept" # unlike dump(), the Bag survives164 165 def test_save_reaps_expired_sessions_first(self, tmp_path) -> None:166 store = MemorySessionStore(default_ttl=3600)167 live = store.create()168 expired = store.create()169 expired.meta["last_access"] = time.time() - 10_000170 path = tmp_path / "sessions.pickle"171 assert store.save_snapshot(path) == 1172 fresh = MemorySessionStore()173 fresh.load_snapshot(path)174 assert fresh.get(live.id) is not None175 assert fresh.get(expired.id) is None176 177 def test_load_drops_a_session_expired_on_disk(self, tmp_path) -> None:178 # the session is live at save time and past its TTL at load time:179 # the TTL is the only filter of the restore side180 store = MemorySessionStore(default_ttl=1)181 aged = store.create()182 path = tmp_path / "sessions.pickle"183 store.save_snapshot(path)184 time.sleep(1.1)185 fresh = MemorySessionStore()186 assert fresh.load_snapshot(path) == 0187 assert fresh.get(aged.id) is None188 189 def test_save_creates_parent_directories(self, tmp_path) -> None:190 store = MemorySessionStore()191 store.create()192 path = tmp_path / "deep" / "nested" / "sessions.pickle"193 store.save_snapshot(path)194 assert path.is_file()195 196 197 class SnapshotServer(SessionMixin, MiddlewareMixin, BaseServer):198 """The session composition with the snapshot armed via ``save_session=``."""199 200 201 async def lifespan_cycle(server: BaseServer) -> list[Message]:202 """Drive one full lifespan protocol (startup then shutdown); return the acks."""203 messages = [{"type": "lifespan.startup"}, {"type": "lifespan.shutdown"}]204 sent: list[Message] = []205 206 async def receive() -> Message:207 return messages.pop(0)208 209 async def send(message: Message) -> None:210 sent.append(message)211 212 await server({"type": "lifespan"}, receive, send)213 return sent214 215 216 class TestSnapshotLifespan:217 async def test_sessions_survive_a_server_restart(self, tmp_path) -> None:218 path = tmp_path / "demo.pickle"219 server = SnapshotServer(applications=[EchoApp(mount="")], save_session=str(path))220 session = server.session_store.create(avatar=Avatar("alice", ["admin"]))221 session.data["cart"] = "kept"222 acks = await lifespan_cycle(server) # shutdown writes the snapshot223 assert {m["type"] for m in acks} == {"lifespan.startup.complete", "lifespan.shutdown.complete"}224 assert path.is_file()225 reborn = SnapshotServer(applications=[EchoApp(mount="")], save_session=str(path))226 await lifespan_cycle(reborn) # startup loads the snapshot227 restored = reborn.session_store.get(session.id)228 assert restored is not None229 assert restored.avatar().identity == "alice"230 assert restored.data["cart"] == "kept"231 232 async def test_absent_snapshot_file_starts_empty(self, tmp_path) -> None:233 path = tmp_path / "never-written.pickle"234 server = SnapshotServer(applications=[EchoApp(mount="")], save_session=str(path))235 acks = await lifespan_cycle(server)236 assert {m["type"] for m in acks} == {"lifespan.startup.complete", "lifespan.shutdown.complete"}237 assert path.is_file() # the shutdown still writes one238 239 async def test_disarmed_server_writes_nothing(self, tmp_path) -> None:240 server = SnapshotServer(applications=[EchoApp(mount="")])241 assert server.save_session is None242 await lifespan_cycle(server)243 assert list(tmp_path.rglob("*.pickle")) == []244 245 246 # --- MemorySessionStore.restore drops an expired dumped session (Macro 2 item 8) ---247 248 249 class TestMemoryStoreRestore:250 def test_restore_drops_expired_session(self) -> None:251 store = MemorySessionStore(default_ttl=3600)252 session = store.create(avatar=Avatar("bob"))253 dumped = store.dump()254 dumped[session.id]["meta"]["last_access"] = time.time() - 10_000255 fresh = MemorySessionStore()256 fresh.restore(dumped)257 assert fresh.get(session.id) is None258 259 260 # --- Session / Avatar units ---261 262 263 class TestSessionUnit:264 def test_session_holds_avatar_and_bag(self) -> None:265 session = Session("tok", avatar=Avatar("alice"), ttl=3600)266 assert session.id == "tok"267 assert isinstance(session.avatar(), Avatar)268 assert session.avatar().identity == "alice"269 assert not session.is_expired()270 271 def test_anonymous_session_avatar_is_none(self) -> None:272 assert Session("tok", avatar=None, ttl=3600).avatar() is None273 274 def test_avatar_normalizes_none_tags(self) -> None:275 assert Avatar("alice", None).tags == []276 277 def test_touch_updates_last_access(self) -> None:278 session = Session("tok", avatar=None, ttl=3600)279 session.meta["last_access"] = 0.0280 session.touch()281 assert session.meta["last_access"] > 0.0282 283 def test_zero_ttl_is_expired(self) -> None:284 assert Session("tok", avatar=None, ttl=0).is_expired()285 286 def test_new_session_is_not_dirty(self) -> None:287 assert Session("tok", avatar=None, ttl=3600).dirty is False288 289 def test_touch_does_not_mark_dirty(self) -> None:290 # a read-only request only touches last_access; it must stay non-dirty291 session = Session("tok", avatar=None, ttl=3600)292 session.touch()293 assert session.dirty is False294 295 def test_mark_dirty_and_clear(self) -> None:296 session = Session("tok", avatar=None, ttl=3600)297 session.mark_dirty()298 assert session.dirty is True299 session.clear_dirty()300 assert session.dirty is False301 302 def test_attach_avatar_marks_dirty(self) -> None:303 session = Session("tok", avatar=None, ttl=3600)304 session.attach_avatar(Avatar("alice"))305 assert session.dirty is True306 307 308 # --- the dressing model: keyed avatars on one session ---309 310 311 class TestSessionAvatars:312 def test_constructor_avatar_dresses_the_root_slot(self) -> None:313 session = Session("tok", avatar=Avatar("alice"), ttl=3600)314 assert session.avatar() is session.avatar(Session.ROOT_AVATAR_KEY)315 assert list(session.avatars) == ["root"]316 317 def test_attach_under_explicit_key_and_read_back(self) -> None:318 session = Session("tok", avatar=Avatar("alice"), ttl=3600)319 session.attach_avatar(Avatar("alice@erp", ["operator"]), "erp")320 assert session.avatar("erp").identity == "alice@erp"321 assert session.avatar().identity == "alice" # root untouched by a sub-login322 323 def test_unclaimed_slot_is_none(self) -> None:324 session = Session("tok", avatar=Avatar("alice"), ttl=3600)325 assert session.avatar("erp") is None326 327 def test_keyed_attach_marks_dirty(self) -> None:328 session = Session("tok", avatar=Avatar("alice"), ttl=3600)329 session.clear_dirty()330 session.attach_avatar(Avatar("alice@erp"), "erp")331 assert session.dirty is True332 333 def test_avatars_view_is_enumerable(self) -> None:334 session = Session("tok", avatar=Avatar("alice"), ttl=3600)335 session.attach_avatar(Avatar("alice@erp"), "erp")336 assert sorted(session.avatars) == ["erp", "root"]337 assert len(session.avatars) == 2338 assert "erp" in session.avatars339 340 def test_avatars_view_is_read_only(self) -> None:341 session = Session("tok", avatar=Avatar("alice"), ttl=3600)342 with pytest.raises(TypeError):343 session.avatars["erp"] = Avatar("intruder") # type: ignore[index]344 345 def test_memory_store_roundtrips_every_keyed_avatar(self) -> None:346 store = MemorySessionStore(default_ttl=3600)347 session = store.create(avatar=Avatar("alice", ["admin"]))348 session.attach_avatar(Avatar("alice@erp", ["operator"]), "erp")349 fresh = MemorySessionStore(default_ttl=3600)350 fresh.restore(store.dump())351 restored = fresh.get(session.id)352 assert restored.avatar().identity == "alice"353 assert restored.avatar().tags == ["admin"]354 assert restored.avatar("erp").identity == "alice@erp"355 assert restored.avatar("erp").tags == ["operator"]356 assert restored.dirty is False357 358 def test_anonymous_session_serializes_an_empty_wardrobe(self) -> None:359 store = MemorySessionStore(default_ttl=3600)360 session = store.create()361 assert store.dump()[session.id]["avatars"] == {}362 fresh = MemorySessionStore(default_ttl=3600)363 fresh.restore(store.dump())364 assert fresh.get(session.id).avatar() is None365 366 367 # --- ASGI cookie flow ---368 369 370 class SessionServer(SessionMixin, MiddlewareMixin, BaseServer):371 """The Phase 4 composition: sessions armed over the middleware chain."""372 373 374 class EchoApp(BaseApplication):375 """Echoes the id of the session attached to the scope, or a marker."""376 377 async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:378 session = scope.get("session")379 body = session.id.encode() if session is not None else b"no-session"380 await send({"type": "http.response.start", "status": 200, "headers": []})381 await send({"type": "http.response.body", "body": body})382 383 384 async def http_get(385 server: BaseServer, path: str = "/", cookie: str | None = None386 ) -> tuple[Scope, list[Message]]:387 """Drive one GET through ``server`` at the ASGI level; return the scope and what it sent."""388 headers = [(b"cookie", cookie.encode())] if cookie is not None else []389 scope: Scope = {"type": "http", "method": "GET", "path": path, "headers": headers}390 sent: list[Message] = []391 392 async def receive() -> Message:393 return {"type": "http.request"}394 395 async def send(message: Message) -> None:396 sent.append(message)397 398 await server(scope, receive, send)399 return scope, sent400 401 402 def set_cookie_value(sent: list[Message]) -> str | None:403 start = next(m for m in sent if m["type"] == "http.response.start")404 for name, value in start["headers"]:405 if name == b"set-cookie":406 return value.decode()407 return None408 409 410 def response_body(sent: list[Message]) -> bytes:411 return b"".join(m.get("body", b"") for m in sent if m["type"] == "http.response.body")412 413 414 def cookie_token(sent: list[Message]) -> str:415 cookie = set_cookie_value(sent)416 assert cookie is not None417 return cookie.split(";")[0].split("=", 1)[1]418 419 420 class TestSessionCookieFlow:421 async def test_first_request_sets_cookie(self) -> None:422 server = SessionServer(applications=[EchoApp(mount="")])423 scope, sent = await http_get(server)424 cookie = set_cookie_value(sent)425 assert cookie is not None426 assert cookie.startswith("session_id=")427 assert "HttpOnly" in cookie428 assert response_body(sent) == scope["session"].id.encode()429 430 async def test_returning_cookie_reuses_the_session(self) -> None:431 server = SessionServer(applications=[EchoApp(mount="")])432 _, first = await http_get(server)433 token = cookie_token(first)434 scope2, second = await http_get(server, cookie=f"session_id={token}")435 assert scope2["session"].id == token436 assert set_cookie_value(second) is None437 assert server.session(scope2) is scope2["session"]438 439 async def test_expired_cookie_issues_new_session(self) -> None:440 server = SessionServer(applications=[EchoApp(mount="")])441 _, first = await http_get(server)442 token = cookie_token(first)443 stored = server.session_store.get(token)444 assert stored is not None445 stored.meta["last_access"] = time.time() - 10_000446 scope2, second = await http_get(server, cookie=f"session_id={token}")447 assert scope2["session"].id != token448 assert set_cookie_value(second) is not None449 450 async def test_composition_without_session_mixin_returns_none(self) -> None:451 class Plain(MiddlewareMixin, BaseServer):452 pass453 454 server = Plain(applications=[EchoApp(mount="")])455 scope, sent = await http_get(server)456 assert server.session(scope) is None457 assert response_body(sent) == b"no-session"458 assert set_cookie_value(sent) is None459 460 async def test_explicit_session_false_disarms_the_middleware(self) -> None:461 server = SessionServer(applications=[EchoApp(mount="")], middleware={"session": False})462 scope, sent = await http_get(server)463 assert scope.get("session") is None464 assert set_cookie_value(sent) is None465 466 async def test_secure_option_adds_secure_cookie_attribute(self) -> None:467 server = SessionServer(applications=[EchoApp(mount="")], middleware={"session": {"secure": True}})468 _, sent = await http_get(server)469 cookie = set_cookie_value(sent)470 assert cookie is not None471 assert "Secure" in cookie472 473 async def test_default_cookie_has_no_secure_attribute(self) -> None:474 server = SessionServer(applications=[EchoApp(mount="")])475 _, sent = await http_get(server)476 cookie = set_cookie_value(sent)477 assert cookie is not None478 assert "Secure" not in cookie479 480 async def test_cookie_max_age_outlives_the_sliding_ttl(self) -> None:481 # Max-Age is fixed from issue time while the server TTL slides:482 # the cookie is issued COOKIE_LIFETIME_FACTOR times wider (legacy-style)483 server = SessionServer(applications=[EchoApp(mount="")], session_ttl=3600)484 _, sent = await http_get(server)485 cookie = set_cookie_value(sent)486 assert cookie is not None487 assert f"Max-Age={3600 * COOKIE_LIFETIME_FACTOR}" in cookie488 489 490 # --- attach_avatar (login seam): identity attached in place, id unchanged ---491 492 493 class TestAttachAvatar:494 def test_attach_sets_the_avatar_on_the_existing_session(self) -> None:495 session = MemorySessionStore().create()496 session.attach_avatar(Avatar("alice", ["admin"]))497 assert session.avatar() is not None498 assert session.avatar().identity == "alice"499 assert session.avatar().tags == ["admin"]500 501 def test_attach_preserves_session_data_and_id(self) -> None:502 store = MemorySessionStore()503 session = store.create()504 session_id = session.id505 session.data["cart"] = "kept"506 session.attach_avatar(Avatar("bob"))507 assert session.id == session_id # the id never changes at login508 assert session.data["cart"] == "kept" # the cart survives the login509 assert store.get(session_id) is session510 511 512 class TestTheSessionOfAScopeWithNoChain:513 """``get_session`` (#68 N24): what reads a session where no middleware ran.514 515 A websocket handshake never passes through the chain, so the motor asks the516 layer itself. The reading is pure — it hands back what the store has and517 creates nothing, because a handshake has no moment to issue a cookie in.518 """519 520 def session_layer(self, server: SessionServer) -> SessionMiddleware:521 layer = server.get_middleware(SessionMiddleware)522 assert layer is not None523 return layer524 525 async def test_it_hands_back_the_session_the_cookie_names(self) -> None:526 server = SessionServer(applications=[EchoApp(mount="")])527 _, first = await http_get(server)528 token = cookie_token(first)529 scope: Scope = {"type": "websocket", "headers": [(b"cookie", f"session_id={token}".encode())]}530 assert self.session_layer(server).get_session(scope).id == token531 532 async def test_a_scope_with_no_cookie_reads_none(self) -> None:533 server = SessionServer(applications=[EchoApp(mount="")])534 assert self.session_layer(server).get_session({"type": "websocket", "headers": []}) is None535 536 async def test_an_unknown_token_reads_none(self) -> None:537 server = SessionServer(applications=[EchoApp(mount="")])538 scope: Scope = {"type": "websocket", "headers": [(b"cookie", b"session_id=nobody")]}539 assert self.session_layer(server).get_session(scope) is None540 541 async def test_it_creates_nothing(self) -> None:542 # ``dump`` is the store's own census: what it would persist.543 server = SessionServer(applications=[EchoApp(mount="")])544 before = server.session_store.dump()545 self.session_layer(server).get_session({"type": "websocket", "headers": []})546 assert server.session_store.dump() == before547 548 def test_the_layer_is_none_when_that_middleware_is_off(self) -> None:549 server = SessionServer(applications=[EchoApp(mount="")], middleware={"session": False})550 assert server.get_middleware(SessionMiddleware) is None551 552 553 # --- SessionMiddleware and login: the cookie is issued for a NEW session only ---554 555 556 class PromotingApp(BaseApplication):557 """A login-like handler: attaches the avatar via the request facade in place."""558 559 def __init__(self, avatar: Avatar, **kwargs) -> None:560 super().__init__(**kwargs)561 self._avatar = avatar562 563 async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:564 scope["session"].attach_avatar(self._avatar)565 await send({"type": "http.response.start", "status": 200, "headers": []})566 await send({"type": "http.response.body", "body": scope["session"].id.encode()})567 568 569 class TestPromotedSessionCookie:570 async def test_promotion_in_a_first_request_rides_the_new_session_cookie(self) -> None:571 app = PromotingApp(Avatar("alice", ["admin"]), mount="")572 server = SessionServer(applications=[app])573 scope, sent = await http_get(server)574 session_id = scope["session"].id575 assert scope["session"].avatar() is not None576 assert scope["session"].avatar().identity == "alice"577 cookie = set_cookie_value(sent) # issued for the NEW session, not for the login578 assert cookie is not None579 assert cookie.startswith(f"session_id={session_id}")580 assert "HttpOnly" in cookie581 assert response_body(sent) == session_id.encode()582 583 async def test_promotion_on_a_returning_session_sends_no_cookie(self) -> None:584 app = PromotingApp(Avatar("bob", ["user"]), mount="")585 server = SessionServer(applications=[app])586 anonymous = server.session_store.create()587 anonymous.data["cart"] = "kept"588 scope, sent = await http_get(server, cookie=f"session_id={anonymous.id}")589 assert scope["session"] is anonymous # same session, same id590 assert scope["session"].avatar() is not None591 assert scope["session"].data["cart"] == "kept" # the cart survives the login592 assert set_cookie_value(sent) is None # the client's cookie is still valid593 594 async def test_unchanged_returning_session_still_sends_no_cookie(self) -> None:595 server = SessionServer(applications=[EchoApp(mount="")])596 _, first = await http_get(server)597 token = cookie_token(first)598 _, second = await http_get(server, cookie=f"session_id={token}")599 assert set_cookie_value(second) is None