forked from retoor/devplacepy
retoor
d31ccba6c1
Add a new `/dbapi` router package providing a generic database API over `dataset`, restricted to admin sessions, admin API keys, and the internal gateway key. Includes: - `tables.py`: list all tables and inspect table schemas - `crud.py`: full CRUD operations (GET, POST, PATCH, DELETE) with soft-delete awareness, born-live inserts, `?include_deleted`, `.../restore`, and `?hard=true` purge - `query.py`: validated read-only SELECT execution via sqlglot parsing, classification, and EXPLAIN dry-run; async query jobs with WebSocket streaming via `DbApiJobService` - `nl.py`: natural-language-to-SQL conversion using the platform AI gateway with re-prompting until validation passes Also register `DbApiJobService` and `PubSubService` in the service manager, add `DBAPI_DIR` to config data paths, and force cleartext `http://` connections to HTTP/1.1 in `curl_transport` to fix large request failures against uvicorn's HTTP/1.1-only internal gateway.
80 lines
2.0 KiB
Python
80 lines
2.0 KiB
Python
# retoor <retoor@molodetz.nl>
|
|
|
|
from tests.conftest import run_async
|
|
|
|
from devplacepy.services.pubsub.hub import PubSubHub, topic_matches
|
|
|
|
|
|
class FakeSocket:
|
|
def __init__(self):
|
|
self.frames = []
|
|
|
|
async def send_json(self, frame):
|
|
self.frames.append(frame)
|
|
|
|
|
|
class DeadSocket:
|
|
async def send_json(self, frame):
|
|
raise RuntimeError("socket closed")
|
|
|
|
|
|
def test_topic_matches():
|
|
assert topic_matches("foo", "foo")
|
|
assert topic_matches("*", "anything.here")
|
|
assert topic_matches("foo.*", "foo.bar")
|
|
assert topic_matches("foo.*", "foo")
|
|
assert not topic_matches("foo.*", "fob")
|
|
assert not topic_matches("foo", "bar")
|
|
|
|
|
|
def test_publish_delivers_only_to_matching():
|
|
hub = PubSubHub()
|
|
a, b = FakeSocket(), FakeSocket()
|
|
hub.subscribe("public.x", a)
|
|
hub.subscribe("public.y", b)
|
|
delivered = run_async(hub.publish("public.x", {"n": 1}))
|
|
assert delivered == 1
|
|
assert a.frames == [{"n": 1}]
|
|
assert b.frames == []
|
|
|
|
|
|
def test_wildcard_delivery():
|
|
hub = PubSubHub()
|
|
a = FakeSocket()
|
|
hub.subscribe("foo.*", a)
|
|
run_async(hub.publish("foo.bar", {"k": 1}))
|
|
run_async(hub.publish("foo", {"k": 2}))
|
|
run_async(hub.publish("fob", {"k": 3}))
|
|
assert a.frames == [{"k": 1}, {"k": 2}]
|
|
|
|
|
|
def test_unsubscribe_stops_delivery():
|
|
hub = PubSubHub()
|
|
a = FakeSocket()
|
|
hub.subscribe("t", a)
|
|
hub.unsubscribe("t", a)
|
|
assert run_async(hub.publish("t", {})) == 0
|
|
|
|
|
|
def test_drop_socket_removes_all_subscriptions():
|
|
hub = PubSubHub()
|
|
a = FakeSocket()
|
|
hub.subscribe("a", a)
|
|
hub.subscribe("b", a)
|
|
hub.drop_socket(a)
|
|
assert hub.topics() == []
|
|
|
|
|
|
def test_dead_socket_is_dropped_silently():
|
|
hub = PubSubHub()
|
|
hub.subscribe("t", DeadSocket())
|
|
assert run_async(hub.publish("t", {})) == 0
|
|
|
|
|
|
def test_topics_lists_subscriber_counts():
|
|
hub = PubSubHub()
|
|
a, b = FakeSocket(), FakeSocket()
|
|
hub.subscribe("x", a)
|
|
hub.subscribe("x", b)
|
|
assert hub.topics() == [{"topic": "x", "subscribers": 2}]
|