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.
46 lines
1.5 KiB
Python
46 lines
1.5 KiB
Python
# retoor <retoor@molodetz.nl>
|
|
|
|
from devplacepy.database import set_setting
|
|
from devplacepy.services.pubsub import policy
|
|
from devplacepy.services.pubsub.policy import Actor
|
|
|
|
ADMIN = Actor("admin", "a", "adm")
|
|
INTERNAL = Actor("internal", "internal", "internal")
|
|
USER = Actor("user", "u1", "bob")
|
|
GUEST = Actor("guest", "", "guest")
|
|
|
|
|
|
def test_valid_topic():
|
|
assert policy.valid_topic("public.demo")
|
|
assert policy.valid_topic("user.u1.inbox")
|
|
assert not policy.valid_topic("bad topic")
|
|
assert not policy.valid_topic("")
|
|
|
|
|
|
def test_privileged_can_use_any_topic():
|
|
for actor in (ADMIN, INTERNAL):
|
|
assert policy.can_subscribe(actor, "anything.here")
|
|
assert policy.can_publish(actor, "public.x")
|
|
assert policy.can_subscribe(actor, "*")
|
|
|
|
|
|
def test_user_namespace_rules():
|
|
assert policy.can_publish(USER, "user.u1.x")
|
|
assert not policy.can_publish(USER, "user.u2.x")
|
|
assert not policy.can_publish(USER, "public.x")
|
|
assert policy.can_subscribe(USER, "public.x")
|
|
assert policy.can_subscribe(USER, "user.u1.feed")
|
|
assert not policy.can_subscribe(USER, "user.u2.feed")
|
|
assert not policy.can_subscribe(USER, "*")
|
|
|
|
|
|
def test_guest_gated_by_setting(local_db):
|
|
set_setting("pubsub_allow_guests", "0")
|
|
try:
|
|
assert not policy.can_subscribe(GUEST, "public.x")
|
|
set_setting("pubsub_allow_guests", "1")
|
|
assert policy.can_subscribe(GUEST, "public.x")
|
|
assert not policy.can_publish(GUEST, "public.x")
|
|
finally:
|
|
set_setting("pubsub_allow_guests", "0")
|