From f995344f547770f57eb3c10a582eaa0b9a6d2716 Mon Sep 17 00:00:00 2001 From: Zijian Zhang <35801754+futrime@users.noreply.github.com> Date: Mon, 12 Oct 2026 22:56:43 +0800 Subject: [PATCH] feat(web): one notification feed of every daemon's notices Serve /api/notifications, its stream, read, clear and settings on the local interface. Notices are heard from site.notices and followed off each reached SSH daemon's /api/notices/stream, tagged with the daemon, filtered by the notifications kinds in settings.yaml, and the last 200 kept. Co-Authored-By: Claude Opus 5.5 --- docs/reference/settings.md | 4 + docs/reference/web.md | 21 ++ src/hmz/web/__init__.py | 4 + src/hmz/web/notifications.py | 391 ++++++++++++++++++++ src/hmz/web/site.py | 5 + tests/integration/test_web_notifications.py | 267 +++++++++++++ tests/unit/web/test_notifications.py | 194 ++++++++++ 7 files changed, 886 insertions(+) create mode 100644 src/hmz/web/notifications.py create mode 100644 tests/integration/test_web_notifications.py create mode 100644 tests/unit/web/test_notifications.py diff --git a/docs/reference/settings.md b/docs/reference/settings.md index 2c7b1d6c..af163419 100644 --- a/docs/reference/settings.md +++ b/docs/reference/settings.md @@ -46,6 +46,9 @@ enable_sentry: false # this machine details: true # this machine btw: claude/claude-opus-5:high # this machine every_epic: true # this machine +notifications: # this machine; Settings › Notifications in the web interface + kinds: {done: true, failed: true, stopped: true, asked: true, listening: true} + system: true spelling: 2 # how envs are spelled; written with every change fallbacks: # this machine; the Fallback page - spec: claude@work/claude-opus-5 @@ -76,6 +79,7 @@ backup: # this machine | `details` | `bool` (only `true` is on) | `false` | `/settings` › General › **Details** | the TUI | immediately: turns show every tool call and all thinking instead of responses only | | `btw` | `str`: `cli[@account]/model:effort` | `""`: the running flow's first agent | `/settings` › General › **/btw agent** | the TUI | the next time `/btw` is entered | | `every_epic` | `bool` (only `true` is on) | `false`: this directory's | `ctrl+t` on [`/epics`](/reference/tui#the-runs-that-have-already-happened) | the TUI | the next time `/epics` is opened | +| `notifications` | `{kinds: {done, failed, stopped, asked, listening}: bool, system: bool}` | every kind told, and shown by the system too; only `false` turns one off | the web interface's Settings › Notifications, [`POST /api/notifications/settings`](/reference/web#api-notifications) | `hmz web` | immediately: the next notice of a kind is told or not | | `spelling` | `int` | absent: written before `-e` wrote `@` only before a provider | every write of the file, as `2` | every process, as it reads the file | absent, `envs` are read as their old spelling meant and rewritten once; `2`, as written | What error reports contain and exclude is listed on `/settings` › General › **What is sent**. diff --git a/docs/reference/web.md b/docs/reference/web.md index e4311393..db73c0f4 100644 --- a/docs/reference/web.md +++ b/docs/reference/web.md @@ -277,6 +277,27 @@ Every `error` but the last is the runtime's own sentence, e.g. An account is listed by the names of the variables it sets (`sets`), never by a value. +### Notifications {#api-notifications} + +What every daemon's notices come to, served by the interface a page is open on alone: its own +machine's as they are heard, and each SSH daemon's while it is reached, off the +`/api/notices/stream` that daemon's own interface serves -- from what it says once first +reached, and from where it was last heard after that. A notification is the notice, with its own +`id` (counted from `1` each time `hmz web` starts), `daemon` (`local` or the daemon's name) and +`read`. The last **200** are kept, in memory. + +| Method and path | Takes | Answers | +| --- | --- | --- | +| `GET /api/notifications` | | `items` (newest first), `unread` (how many are not read) | +| `GET /api/notifications/stream` | `Last-Event-ID` header, or `?last=` | `text/event-stream`: a `notification` event, `id` its `id`, for each kept after the one named -- every one kept where it names one later than any here -- then each as it comes; with none, from now. `: quiet` after **15 s** of nothing | +| `POST /api/notifications/read` | `ids` (a list), or `all: true` | As `GET` | +| `POST /api/notifications/clear` | | As `GET`, with nothing kept | +| `GET /api/notifications/settings` | | `kinds` (`{done, failed, stopped, asked, listening}`: whether each is told) and `system` (whether the page asks the system to show them too), as [`notifications`](/reference/settings#machine-settings) holds them | +| `POST /api/notifications/settings` | any of `kinds` (any of those five, each a bool), `system` (bool) | As `GET`; `400`, and nothing written, for anything else | + +A notice of a kind turned off is never kept nor sent. Which kinds are told is read afresh for +every notice, so a change takes effect at once. + ## The stream {#stream} `GET /api/held/stream` answers `text/event-stream`, starting with `retry: 2000`. Each event is diff --git a/src/hmz/web/__init__.py b/src/hmz/web/__init__.py index 5c6d7197..c24f1bbd 100644 --- a/src/hmz/web/__init__.py +++ b/src/hmz/web/__init__.py @@ -59,6 +59,7 @@ def listening(*, port: int = 0, apart: bool = True) -> Generator[Site]: from hmz.coganchor import releases + from .notifications import Notifications from .places import Places from .runs import Runs from .site import Site @@ -73,6 +74,9 @@ def listening(*, port: int = 0, apart: bool = True) -> Generator[Site]: # And which release of humanize is the newest, where what is kept has got old. releases.refresh() held_open.callback(site.server_close) + # Every daemon's notices, as this machine's interface tells them. + site.notifications = Notifications(site) + held_open.callback(site.notifications.close) # Set first as the block ends, which is what ends every stream being sent. held_open.callback(site.going.set) yield site diff --git a/src/hmz/web/notifications.py b/src/hmz/web/notifications.py new file mode 100644 index 00000000..ceb19398 --- /dev/null +++ b/src/hmz/web/notifications.py @@ -0,0 +1,391 @@ +"""Notifications: what every daemon's notices come to, on this machine's interface alone. + +Each `hmz web` serves its own machine's notices -- an epic ended, a flow asking a person, a +chat waiting for its next turn. The interface a person has open is told them of every daemon at +once: this machine's as they are heard, and each SSH daemon's, while it is reached, off the +stream its own interface serves. Each is told by which daemon said it, kept only where the kinds +a person wants to be told of say so, and kept a while -- the last few hundred -- with whether it +was read, for every page this interface serves to follow from where it last heard. + +Which kinds are told, and whether the system is asked to show them too, are the app's own +settings, kept under `notifications` in this machine's `settings.yaml` and never a daemon's. +""" + +from __future__ import annotations + +import collections +import contextlib +import http.client +import json +import threading +import time +from http import HTTPStatus +from typing import TYPE_CHECKING, Any, cast + +from .routing import Refusal, Stream, routes + +if TYPE_CHECKING: + from collections.abc import Callable, Iterator + + from .routing import Asked + +__all__ = ["KINDS", "ROUTES", "Notifications", "daemons", "preferences"] + +#: Every kind of notice there is: an epic done, failed or stopped, a person asked, and a chat +#: listening for its next turn. +KINDS = ("done", "failed", "stopped", "asked", "listening") + +#: How many notifications are kept. +_KEPT = 200 + +#: How often the daemons there are, and which are reached, are looked at again. +_LOOKING = 2.0 + +#: How long a stream says nothing before it says it is still there. +_QUIET = 15.0 + +#: How long a daemon's interface may say nothing before its stream is given up on: it says it is +#: still there more often than this. +_SILENT = 60.0 + + +def preferences(held: dict[str, Any] | None = None) -> dict[str, Any]: + """Which notifications are wanted, as `settings.yaml` holds them: every one, where it is silent. + + Args: + held: Everything `settings.yaml` holds; read afresh where None. + + Returns: + `kinds`, whether each kind is told, and `system`, whether the system is asked to show + them too. + """ + if held is None: + from hmz.coganchor import settings + + held = settings.read() + said: object = held.get("notifications") + wanted = cast("dict[str, Any]", said) if isinstance(said, dict) else {} + kinds: object = wanted.get("kinds") + each = cast("dict[str, Any]", kinds) if isinstance(kinds, dict) else {} + return { + "kinds": {kind: each.get(kind) is not False for kind in KINDS}, + "system": wanted.get("system") is not False, + } + + +def daemons(held: dict[str, Any] | None = None) -> list[str]: + """The name of every SSH daemon `settings.yaml` holds, as it holds them. + + Args: + held: Everything `settings.yaml` holds; read afresh where None. + """ + if held is None: + from hmz.coganchor import settings + + held = settings.read() + said: object = held.get("daemons") + entries = cast("list[object]", said) if isinstance(said, list) else [] + named = [ + cast("dict[str, Any]", one).get("name") + for one in entries + if isinstance(one, dict) + ] + return [str(one) for one in named if one and one != "local"] + + +class Notifications: + """Every daemon's notices, as this machine's interface tells them. + + Hears this machine's own as they come, and follows each SSH daemon's while it is reached: + the notices held as `site.notices` and the daemons reached through `site.remotes`, each + looked for again as long as it is missing. + + Args: + site: What holds this machine's notices and the daemons reached, where it has them. + wanted: Which notifications are wanted, asked afresh of every notice. + named: The SSH daemons there are, asked afresh every little while. + """ + + def __init__( + self, + site: Any, + *, + wanted: Callable[[], dict[str, Any]] = preferences, + named: Callable[[], list[str]] = daemons, + ) -> None: + self._site = site + self._wanted = wanted + self._named = named + self._changed = threading.Condition() + self._kept: collections.deque[dict[str, Any]] = collections.deque(maxlen=_KEPT) + self._told = 0 + self._closed = threading.Event() + self._hearing: object = None + #: Each daemon followed, by name: the connection its stream comes over, once there is one. + self._following: dict[str, list[http.client.HTTPConnection]] = {} + #: The last notice heard of each daemon, by name: where its stream is taken up again. + self._last: dict[str, str] = {} + self._hears() + threading.Thread( + target=self._looks, daemon=True, name="hmz-web-notifications" + ).start() + + def told(self, daemon: str, notice: dict[str, Any]) -> dict[str, Any] | None: + """Keeps one daemon's notice as a notification, where its kind is wanted. + + Args: + daemon: Which said it: `local`, or an SSH daemon's name. + notice: The notice, as its daemon serves it. + + Returns: + The notification, or None where its kind is not wanted. + """ + if not self._wanted()["kinds"].get(str(notice.get("kind")), True): + return None + with self._changed: + self._told += 1 + made = notice | {"id": str(self._told), "daemon": daemon, "read": False} + self._kept.append(made) + self._changed.notify_all() + return dict(made) + + def listed(self) -> dict[str, Any]: + """Every notification kept, newest first, and how many are unread.""" + with self._changed: + items = [dict(one) for one in reversed(self._kept)] + return {"items": items, "unread": sum(not one["read"] for one in items)} + + def read(self, ids: list[str] | None = None) -> dict[str, Any]: + """Marks notifications read: those named, or every one where none are. + + Returns: + As `listed`. + """ + with self._changed: + for one in self._kept: + if ids is None or one["id"] in ids: + one["read"] = True + return self.listed() + + def clear(self) -> dict[str, Any]: + """Forgets every notification kept. + + Returns: + As `listed`. + """ + with self._changed: + self._kept.clear() + return self.listed() + + def stream(self) -> Stream: + """Every notification after the one a page last heard, then each as it comes.""" + return Stream(self._events) + + def close(self) -> None: + """Stops following every daemon, and ends every stream.""" + self._closed.set() + with self._changed: + following = [one for held in self._following.values() for one in held] + self._changed.notify_all() + for connection in following: + with contextlib.suppress(OSError): + connection.close() + + def _events( + self, last: str, going: threading.Event + ) -> Iterator[tuple[str, str, Any] | None]: + with self._changed: + told = self._told + # One heard of an interface before this, which counted from the start again, has + # heard none of this one's. + heard = int(last) if last.isdigit() else told + after = heard if heard <= told else 0 + while not (going.is_set() or self._closed.is_set()): + with self._changed: + fresh = [dict(one) for one in self._kept if int(one["id"]) > after] + if not fresh: + self._changed.wait(_QUIET) + fresh = [dict(one) for one in self._kept if int(one["id"]) > after] + if not fresh: + yield None + continue + for one in fresh: + after = int(one["id"]) + yield one["id"], "notification", one + + def _hears(self) -> None: + """Hears this machine's own notices, once there are any to hear.""" + notices = getattr(self._site, "notices", None) + if notices is None or self._hearing is notices: + return + self._hearing = notices + notices.heard(self._heard) + + def _heard(self, said: dict[str, Any] | list[dict[str, Any]]) -> None: + for notice in said if isinstance(said, list) else [said]: + self.told("local", notice) + + def _looks(self) -> None: + """Every little while: this machine's notices, and every SSH daemon reached, followed.""" + while not self._closed.wait(_LOOKING): + # Whatever went wrong is looked at again in a while. + with contextlib.suppress(Exception): + self._hears() + self._reaches(getattr(self._site, "remotes", None)) + + def _reaches(self, remotes: Any) -> None: + """Follows every SSH daemon reached that is not followed yet.""" + if remotes is None: + return + for name in self._named(): + if name in self._following or remotes.state(name).get("state") != "up": + continue + with self._changed: + self._following[name] = [] + threading.Thread( + target=self._follows, + args=(remotes, name), + daemon=True, + name=f"hmz-web-notices-{name}", + ).start() + + def _follows(self, remotes: Any, name: str) -> None: + """Follows one daemon's notices while it is reached, from where it was last heard.""" + try: + port, key = remotes.reached(name) + connection = http.client.HTTPConnection("127.0.0.1", port, timeout=_SILENT) + with self._changed: + if self._closed.is_set(): + return + self._following[name].append(connection) + cookie = _signs_in(connection, key) + # What it said before it was first reached was said to nobody here, and is not + # told; one started again since it was last heard counts from the start again. + now = str(_asks(connection, cookie, "/api/notices")["last"]) + heard = self._last.get(name) + if heard is None or ( + heard.isdigit() and now.isdigit() and int(heard) > int(now) + ): + self._last[name] = now if heard is None else "0" + connection.request( + "GET", + "/api/notices/stream", + headers={"Cookie": cookie, "Last-Event-ID": self._last[name]}, + ) + answer = connection.getresponse() + if answer.status != HTTPStatus.OK: + return + event, ident, data = "", "", list[str]() + while not self._closed.is_set(): + line = answer.readline() + if not line: + return + said = line.decode().rstrip("\r\n") + if said.startswith("id:"): + ident = said[3:].strip() + elif said.startswith("event:"): + event = said[6:].strip() + elif said.startswith("data:"): + data.append(said[5:].removeprefix(" ")) + elif not said: + if event == "notice" and data: + notice = json.loads("\n".join(data)) + self._last[name] = ident or str(notice.get("id", "")) + self.told(name, notice) + event, ident, data = "", "", list[str]() + except Exception: # noqa: BLE001 -- followed again once it is reached again + time.sleep(_LOOKING) + finally: + with self._changed: + for connection in self._following.pop(name, []): + with contextlib.suppress(OSError): + connection.close() + + +def _signs_in(connection: http.client.HTTPConnection, key: str) -> str: + """Lets this in to a daemon's interface with its key, as a browser is let in. + + Returns: + The cookie it is handed, named for the port that interface listens on there. + + Raises: + OSError: If it would not let this in. + """ + connection.request("GET", f"/?key={key}") + answer = connection.getresponse() + answer.read() + cookie = (answer.getheader("Set-Cookie") or "").partition(";")[0] + if not cookie: + raise OSError(f"the daemon's interface would not let this in: {answer.status}") + return cookie + + +def _asks( + connection: http.client.HTTPConnection, cookie: str, path: str +) -> dict[str, Any]: + """What a daemon's interface answers one read with.""" + connection.request("GET", path, headers={"Cookie": cookie}) + answer = connection.getresponse() + body = answer.read() + if answer.status != HTTPStatus.OK: + raise OSError(f"the daemon's interface answered {answer.status}") + return cast("dict[str, Any]", json.loads(body)) + + +def _held(asked: Asked) -> Notifications: + held = getattr(asked.site, "notifications", None) + if held is None: + raise Refusal("Notifications are not kept here.", 404) + return cast("Notifications", held) + + +def _read(asked: Asked) -> dict[str, Any]: + if asked.body.get("all") is True: + return _held(asked).read() + said: object = asked.body.get("ids") + if not isinstance(said, list) or not all( + isinstance(one, str | int) and not isinstance(one, bool) + for one in cast("list[object]", said) + ): + raise Refusal("Say ids as a list, or all as true.") + return _held(asked).read([str(one) for one in cast("list[object]", said)]) + + +def _prefers(asked: Asked) -> dict[str, Any]: + said = asked.body + if extra := sorted(set(said) - {"kinds", "system"}): + raise Refusal(f"Notifications take no {', '.join(extra)}.") + if not isinstance(said.get("kinds", {}), dict): + raise Refusal("kinds is whether each kind is told, by name.") + kinds = asked.named("kinds") + if extra := sorted(set(kinds) - set(KINDS)): + raise Refusal(f"There is no kind {', '.join(extra)}.") + if not all(isinstance(one, bool) for one in kinds.values()): + raise Refusal("Each kind is true or false.") + if "system" in said and not isinstance(said["system"], bool): + raise Refusal("system is true or false.") + + def change(held: dict[str, Any]) -> None: + now = preferences(held) + now["kinds"] |= kinds + if "system" in said: + now["system"] = said["system"] + held["notifications"] = now + + from hmz.coganchor import settings + + try: + return preferences(settings.changes(change)) + except OSError as why: + raise Refusal(f"The settings cannot be written: {why}", 409) from why + + +#: What a page reads of the notifications, and does to them. +ROUTES = routes( + ("GET", "/api/notifications", lambda asked: _held(asked).listed()), + ("GET", "/api/notifications/stream", lambda asked: _held(asked).stream()), + ("POST", "/api/notifications/read", _read), + ("POST", "/api/notifications/clear", lambda asked: _held(asked).clear()), + ("GET", "/api/notifications/settings", lambda _: preferences()), + ("POST", "/api/notifications/settings", _prefers), +) diff --git a/src/hmz/web/site.py b/src/hmz/web/site.py index e6204457..39292480 100644 --- a/src/hmz/web/site.py +++ b/src/hmz/web/site.py @@ -42,6 +42,7 @@ fallbacks, flows, held, + notifications, paths, places, runs, @@ -54,6 +55,7 @@ if TYPE_CHECKING: from collections.abc import Generator + from .notifications import Notifications from .places import Places from .runs import Runs @@ -71,6 +73,7 @@ *paths.ROUTES, *places.ROUTES, *verses.ROUTES, + *notifications.ROUTES, ) #: How many ports any would do for are tried before one free on both loopbacks is given up on. @@ -152,6 +155,8 @@ def __init__(self, port: int, *, places: Places, runs: Runs) -> None: self.places = places self.hmz: Any = places.hmz self.runs = runs + #: Every daemon's notices, as this interface tells them, once it is listening. + self.notifications: Notifications | None = None #: Set as the server stops, which is what ends every stream it is sending. self.going = threading.Event() self._key = secrets.token_urlsafe(32) diff --git a/tests/integration/test_web_notifications.py b/tests/integration/test_web_notifications.py new file mode 100644 index 00000000..a1f86cb7 --- /dev/null +++ b/tests/integration/test_web_notifications.py @@ -0,0 +1,267 @@ +"""The interface's notifications: this machine's notices and an SSH daemon's, told as one. + +The interface is served in this process; this machine's notices are a stand-in told what to +say, and an SSH daemon is a stand-in for its interface, reached on a port of this machine as +a forwarded one is: it lets in what comes with its key, says where its notices stand, and +streams what it says after that. +""" + +from __future__ import annotations + +import contextlib +import http.client +import json +import threading +import time +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from typing import TYPE_CHECKING, Any, cast + +import yaml + +from hmz import home +from tests.integration.doubles_web import PATIENCE, Browser, served + +if TYPE_CHECKING: + from collections.abc import Callable, Generator + + from hmz.web.site import Site + + +class _Notices: + """This machine's notices: what hears them, told each as it is said.""" + + def __init__(self) -> None: + self.hearing: list[Callable[[Any], object]] = [] + + def heard(self, callback: Callable[[Any], object]) -> None: + self.hearing.append(callback) + + def since(self, after: str) -> list[dict[str, Any]]: + return [] + + def says(self, notice: dict[str, Any]) -> None: + for one in self.hearing: + one(notice) + + +class _Daemon(ThreadingHTTPServer): + """An SSH daemon's interface, as its forwarded port reaches it.""" + + daemon_threads = True + + def __init__(self) -> None: + super().__init__(("127.0.0.1", 0), _Answers) + #: What its stream says next, a notice at a time. + self.saying: list[dict[str, Any]] = [] + #: The `Last-Event-ID` each stream was asked from. + self.asked: list[str] = [] + + +class _Answers(BaseHTTPRequestHandler): + protocol_version = "HTTP/1.1" + + @property + def _daemon(self) -> _Daemon: + return cast("_Daemon", self.server) + + def log_message(self, format: str, *args: Any) -> None: # noqa: A002 + pass + + def do_GET(self) -> None: + if self.path == "/?key=sesame": + self.send_response(303) + self.send_header("Set-Cookie", "hmz-4242=sesame; HttpOnly; Path=/") + self.send_header("Content-Length", "0") + self.end_headers() + return + if "hmz-4242=sesame" not in self.headers.get("Cookie", ""): + self.send_error(401) + return + if self.path == "/api/notices": + body = json.dumps({"notices": [], "last": "5"}).encode() + self.send_response(200) + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + return + self._daemon.asked.append(self.headers.get("Last-Event-ID", "")) + self.send_response(200) + self.send_header("Content-Type", "text/event-stream") + self.end_headers() + self.close_connection = True + # Until the interface following it lets it go. + with contextlib.suppress(OSError): + self._says() + + def _says(self) -> None: + while True: + while self._daemon.saying: + notice = self._daemon.saying.pop(0) + said = ( + f"id: {notice['id']}\nevent: notice\ndata: {json.dumps(notice)}\n\n" + ) + self.wfile.write(said.encode()) + self.wfile.write(b": quiet\n\n") + self.wfile.flush() + time.sleep(0.05) + + +class _Remotes: + """The SSH daemons reached: `gpu`, up on the stand-in's port.""" + + def __init__(self, daemon: _Daemon) -> None: + self._daemon = daemon + + def state(self, name: str) -> dict[str, Any]: + return {"state": "up" if name == "gpu" else "idle", "detail": "", "version": ""} + + def reached(self, name: str) -> tuple[int, str]: + return int(self._daemon.server_address[1]), "sesame" + + +@contextlib.contextmanager +def _told() -> Generator[tuple[Site, Browser, _Notices, _Daemon]]: + home().mkdir(parents=True, exist_ok=True) + (home() / "settings.yaml").write_text( + yaml.safe_dump({"daemons": [{"name": "gpu", "host": "gpu.example"}]}) + ) + daemon = _Daemon() + threading.Thread(target=daemon.serve_forever, daemon=True).start() + try: + with served() as site: + notices = _Notices() + site.notices = notices # pyright: ignore[reportAttributeAccessIssue] + site.remotes = _Remotes(daemon) # pyright: ignore[reportAttributeAccessIssue] + browser = Browser(site) + browser.signs_in() + _waits(lambda: bool(notices.hearing) and bool(daemon.asked), "followed") + yield site, browser, notices, daemon + finally: + daemon.shutdown() + daemon.server_close() + + +def _waits(done: Callable[[], bool], what: str) -> None: + ending = time.monotonic() + PATIENCE + while not done(): + assert time.monotonic() < ending, f"never {what}" + time.sleep(0.05) + + +def _items(browser: Browser) -> list[dict[str, Any]]: + answer = browser.get("/api/notifications") + assert answer.status == 200, answer.body + return answer.json()["items"] + + +def _notice(kind: str, epic: str, ident: str = "1") -> dict[str, Any]: + return {"id": ident, "kind": kind, "workspace": "/w", "epic": epic, "text": ""} + + +def test_every_daemons_notices_are_told_as_one_by_who_said_them() -> None: + with _told() as (_, browser, notices, daemon): + notices.says(_notice("done", "here")) + daemon.saying.append(_notice("asked", "there", "6")) + _waits(lambda: len(_items(browser)) == 2, "told both") + + items = _items(browser) + + assert [(one["daemon"], one["epic"], one["kind"]) for one in items] == [ + ("gpu", "there", "asked"), + ("local", "here", "done"), + ] + # What the daemon said before it was first reached is not told. + assert daemon.asked == ["5"] + + +def test_a_kind_turned_off_is_never_told_and_written_down_as_off() -> None: + with _told() as (_, browser, notices, _): + answer = browser.post( + "/api/notifications/settings", {"kinds": {"listening": False}} + ) + + assert answer.status == 200, answer.body + assert answer.json() == browser.get("/api/notifications/settings").json() + written = yaml.safe_load((home() / "settings.yaml").read_text()) + assert written["notifications"]["kinds"]["listening"] is False + assert written["notifications"]["system"] is True + assert written["daemons"] == [{"name": "gpu", "host": "gpu.example"}] + notices.says(_notice("listening", "chat")) + notices.says(_notice("failed", "broke")) + assert [one["kind"] for one in _items(browser)] == ["failed"] + + +def test_settings_that_are_not_notifications_are_refused() -> None: + with _told() as (_, browser, _, _): + for wrong in ( + {"kinds": {"x": True}}, + {"kinds": {"done": "no"}}, + {"kinds": False}, + {"system": "yes"}, + {"sound": True}, + ): + answer = browser.post("/api/notifications/settings", wrong) + assert answer.status == 400, wrong + assert "notifications" not in (home() / "settings.yaml").read_text() + + +def test_notifications_are_read_and_cleared() -> None: + with _told() as (_, browser, notices, _): + notices.says(_notice("done", "a")) + notices.says(_notice("stopped", "b")) + first = _items(browser)[-1]["id"] + + answer = browser.post("/api/notifications/read", {"ids": [first]}) + assert answer.json()["unread"] == 1 + assert ( + browser.post("/api/notifications/read", {"all": True}).json()["unread"] == 0 + ) + assert browser.post("/api/notifications/clear").json() == { + "items": [], + "unread": 0, + } + + +def test_the_stream_tells_each_as_it_comes_and_takes_up_where_it_was_left() -> None: + with _told() as (site, browser, notices, _): + notices.says(_notice("done", "before")) + with _following(site, browser, "") as next_event: + notices.says(_notice("asked", "after")) + ident, event, data = next_event() + assert (event, data["epic"], data["daemon"], data["read"]) == ( + "notification", + "after", + "local", + False, + ) + with _following(site, browser, "1") as next_event: + assert next_event()[0] == ident + + +@contextlib.contextmanager +def _following( + site: Site, browser: Browser, last: str +) -> Generator[Callable[[], tuple[str, str, Any]]]: + connection = http.client.HTTPConnection("127.0.0.1", site.port, timeout=PATIENCE) + headers = {"Cookie": browser.cookie} | ({"Last-Event-ID": last} if last else {}) + connection.request("GET", "/api/notifications/stream", headers=headers) + stream = connection.getresponse() + assert stream.status == 200 + + def next_event() -> tuple[str, str, Any]: + ident = event = data = "" + while True: + line = stream.readline().decode().rstrip("\n") + if line.startswith("id: "): + ident = line[4:] + elif line.startswith("event: "): + event = line[7:] + elif line.startswith("data: "): + data = line[6:] + elif not line and event: + return ident, event, json.loads(data) + + try: + yield next_event + finally: + connection.close() diff --git a/tests/unit/web/test_notifications.py b/tests/unit/web/test_notifications.py new file mode 100644 index 00000000..cf2ce631 --- /dev/null +++ b/tests/unit/web/test_notifications.py @@ -0,0 +1,194 @@ +"""`hmz.web.notifications`: every daemon's notices, kept, filtered, read and streamed.""" + +from __future__ import annotations + +import threading +from types import SimpleNamespace +from typing import TYPE_CHECKING, Any + +import pytest + +from hmz.web.notifications import KINDS, ROUTES, Notifications, daemons, preferences +from hmz.web.routing import Asked, Refusal + +if TYPE_CHECKING: + from collections.abc import Callable, Generator + + +class _Notices: + """A stand-in for this machine's notices: what hears them, told each as it is said.""" + + def __init__(self) -> None: + self.hearing: list[Callable[[Any], object]] = [] + + def heard(self, callback: Callable[[Any], object]) -> None: + self.hearing.append(callback) + + def says(self, said: Any) -> None: + for one in self.hearing: + one(said) + + +def _notice(kind: str = "done", **more: Any) -> dict[str, Any]: + return {"id": "7", "kind": kind, "workspace": "/w", "epic": "e", "text": ""} | more + + +@pytest.fixture +def notices() -> _Notices: + return _Notices() + + +@pytest.fixture +def wanted() -> dict[str, Any]: + return preferences({}) + + +@pytest.fixture +def held(notices: _Notices, wanted: dict[str, Any]) -> Generator[Notifications]: + made = Notifications( + SimpleNamespace(notices=notices), wanted=lambda: wanted, named=list + ) + yield made + made.close() + + +def test_every_kind_is_told_and_shown_where_nothing_is_set() -> None: + assert preferences({}) == { + "kinds": dict.fromkeys(KINDS, True), + "system": True, + } + + +def test_a_kind_turned_off_is_off_and_anything_else_is_on() -> None: + said = {"notifications": {"kinds": {"done": False, "asked": "no"}, "system": False}} + + assert preferences(said) == { + "kinds": dict.fromkeys(KINDS, True) | {"done": False}, + "system": False, + } + + +def test_settings_not_written_as_notifications_are_every_one_wanted() -> None: + assert preferences({"notifications": ["done"]}) == preferences({}) + + +def test_ssh_daemons_are_named_as_written_and_local_is_never_one() -> None: + held = { + "daemons": [ + {"name": "gpu"}, + {"name": "local"}, + "box", + {"host": "x"}, + {"name": "b"}, + ] + } + + assert daemons(held) == ["gpu", "b"] + assert daemons({}) == [] + + +def test_this_machines_notices_are_kept_as_local_newest_first( + held: Notifications, notices: _Notices +) -> None: + notices.says(_notice("done", epic="a")) + notices.says([_notice("asked", epic="b")]) + + listed = held.listed() + + assert [(one["daemon"], one["epic"], one["kind"]) for one in listed["items"]] == [ + ("local", "b", "asked"), + ("local", "a", "done"), + ] + assert [one["id"] for one in listed["items"]] == ["2", "1"] + assert listed["unread"] == 2 + assert not any(one["read"] for one in listed["items"]) + + +def test_a_kind_not_wanted_is_never_kept( + held: Notifications, wanted: dict[str, Any] +) -> None: + wanted["kinds"]["listening"] = False + + assert held.told("gpu", _notice("listening")) is None + assert held.told("gpu", _notice("failed")) is not None + assert [one["kind"] for one in held.listed()["items"]] == ["failed"] + + +def test_only_the_last_few_hundred_are_kept(held: Notifications) -> None: + for _ in range(250): + held.told("local", _notice()) + + items = held.listed()["items"] + + assert len(items) == 200 + assert items[0]["id"] == "250" + + +def test_notifications_are_read_by_id_or_all_at_once(held: Notifications) -> None: + first = held.told("local", _notice()) + held.told("gpu", _notice()) + assert first is not None + + assert held.read([first["id"]])["unread"] == 1 + assert held.read()["unread"] == 0 + + +def test_clearing_forgets_every_one_and_the_next_counts_on( + held: Notifications, +) -> None: + held.told("local", _notice()) + + assert held.clear() == {"items": [], "unread": 0} + told = held.told("local", _notice()) + assert told is not None + assert told["id"] == "2" + + +def test_the_stream_takes_up_after_the_last_heard_then_tells_each_as_it_comes( + held: Notifications, +) -> None: + held.told("local", _notice(epic="a")) + held.told("gpu", _notice(epic="b")) + going = threading.Event() + events = held.stream().events("1", going) + + assert next(events) == ("2", "notification", held.listed()["items"][0]) + threading.Timer(0.05, held.told, args=("gpu", _notice(epic="c"))).start() + said = next(events) + assert said is not None + ident, name, data = said + + assert (ident, name, data["epic"], data["daemon"]) == ( + "3", + "notification", + "c", + "gpu", + ) + + +def test_a_stream_heard_of_an_earlier_interface_starts_over( + held: Notifications, +) -> None: + held.told("local", _notice()) + + events = held.stream().events("90", threading.Event()) + + said = next(events) + assert said is not None + assert said[0] == "1" + + +def test_a_page_marks_read_by_ids_or_all_and_nothing_else( + held: Notifications, +) -> None: + (read,) = [one for one in ROUTES if one.path == "/api/notifications/read"] + site = SimpleNamespace(notifications=held) + held.told("local", _notice()) + held.told("local", _notice()) + + assert read.answers(Asked(site, body={"ids": [1]}))["unread"] == 1 + assert read.answers(Asked(site, body={"all": True}))["unread"] == 0 + with pytest.raises(Refusal): + read.answers(Asked(site, body={"ids": "1"})) + with pytest.raises(Refusal, match="not kept"): + read.answers(Asked(SimpleNamespace(), body={"all": True}))