diff --git a/priompt_sdk/client.py b/priompt_sdk/client.py index ca868e6..bccca3e 100644 --- a/priompt_sdk/client.py +++ b/priompt_sdk/client.py @@ -1,5 +1,6 @@ """Priompt Python adapter. Works with LangChain, LangGraph, or raw Python.""" +import logging import os import threading import time @@ -140,7 +141,9 @@ def subscribe(self, uri, on_change): change. TTL polling via cache_ttl is the pull side. Returns the daemon thread running the NATS subscription. - Requires `pip install nats-py` and nats_url set on the client. + Requires `pip install nats-py` and nats_url set on the client. Raises + ConnectionError if the broker can't be reached or refuses the + connection, e.g. a wrong NATS token ('Authorization Violation'). """ if not self._nats_url: raise ValueError("set nats_url on PromptClient to subscribe") @@ -151,10 +154,27 @@ def subscribe(self, uri, on_change): import nats subject = _subject(uri) + ready = threading.Event() + failed = [] # the initial connect error, if any + log = logging.getLogger("priompt_sdk") + + async def on_error(e): + # Before the first connect succeeds, raising here aborts nats.connect + # (it awaits this callback), so a wrong token fails subscribe() at + # once instead of retrying for minutes. Afterwards, nats-py + # reconnects by itself; log one line, not a traceback. + if not ready.is_set(): + raise e + log.warning("priompt subscribe %s: %s", uri, e) def run(): async def main(): - nc = await nats.connect(self._nats_url) + try: + nc = await nats.connect(self._nats_url, error_cb=on_error) + except Exception as e: + failed.append(e) + ready.set() + return async def handler(msg): self._cache.pop(uri, None) # drop stale cache @@ -166,6 +186,7 @@ async def handler(msg): on_change(msg.data.decode(), "") # pre-0.7 bare-hash body await nc.subscribe(subject, cb=handler) + ready.set() while True: await asyncio.sleep(3600) @@ -173,6 +194,14 @@ async def handler(msg): t = threading.Thread(target=run, daemon=True) t.start() + ready.wait() + if failed: + e = failed[0] + u = urlparse(self._nats_url) # don't echo the token + raise ConnectionError( + f"subscribe: cannot connect to NATS at {u.hostname}:{u.port}: " + f"{str(e) or type(e).__name__}" + ) from e return t def close(self): diff --git a/tests/test_client.py b/tests/test_client.py index e4fc95a..7d7e2be 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -183,3 +183,28 @@ def worker(): assert not errors, errors assert len(c._cache) == len(keys) # one entry per key, no garbage + + +def test_subscribe_raises_when_broker_refuses(monkeypatch): + """#4: with a wrong NATS token the broker answers 'Authorization + Violation'. subscribe() must raise that to the caller, not log tracebacks + from a background thread and return as if it were subscribed.""" + import sys + import types + + class AuthError(Exception): + pass + + async def connect(url, error_cb=None, **kw): + # Like nats-py: report each failed attempt to error_cb, then give up. + err = AuthError("nats: 'Authorization Violation'") + for _ in range(3): + if error_cb: + await error_cb(err) + raise err + + monkeypatch.setitem(sys.modules, "nats", types.SimpleNamespace(connect=connect)) + c = PromptClient("localhost:1", nats_url="nats://WRONG@127.0.0.1:4222") + with pytest.raises(ConnectionError, match="Authorization Violation") as exc: + c.subscribe("priompt://x", lambda *_: None) + assert "WRONG" not in str(exc.value) # the token is not echoed