Drop the sharing settings of tools deleted with their connection

Removing a connection with its tools now also deletes those tools' sharing
switches in the same transaction, as deleting a tool on its own does.
Team grants and chat preferences were already removed with the tools.
This commit is contained in:
arc53-machine committed 2026-09-29 14:13:35 +01:00
1 parent a45e369d0f
commit b2d26c72ed
2 files changed
+30 -1

No files matched your search

+8 -1
View File
@@ -1442,8 +1442,15 @@ def remove_connection(conn, row: dict, *, sources: str = "keep", tools: str = "d
except CredentialDecryptionError:
pass
if tools == "delete":
# Team grants go with the tools (delete trigger) and chat preferences
# by FK cascade; the owner's sharing switches have no FK.
conn.execute(
text("DELETE FROM user_tools WHERE connection_id = CAST(:id AS uuid) AND user_id = :user_id"),
text(
"WITH gone AS (DELETE FROM user_tools WHERE connection_id = CAST(:id AS uuid) "
"AND user_id = :user_id RETURNING id) "
"DELETE FROM resource_share_settings WHERE resource_type = 'tool' "
"AND resource_id IN (SELECT id FROM gone)"
),
{"id": connection_id, "user_id": row["user_id"]},
)
else:
@@ -469,6 +469,28 @@ class TestRemove:
).one()
assert kept.connection_id is None and kept.sync_frequency == "never"
def test_deleted_tools_leave_no_sharing_rows(self, pg_conn):
from docsgpt.api.user.resource_access import set_settings
from docsgpt.storage.db.repositories.user_tool_preferences import UserToolPreferencesRepository
cid = _connection(pg_conn, provider="telegram", auth_kind="api_key", secrets={"credentials": {"token": "t"}})
tool_id = str(pg_conn.execute(
text(
"INSERT INTO user_tools (user_id, name, connection_id) "
"VALUES ('alice', 'telegram', CAST(:c AS uuid)) RETURNING id"
),
{"c": cid},
).scalar())
set_settings(pg_conn, "tool", tool_id, {"editors_can_share": True}, "alice")
UserToolPreferencesRepository(pg_conn).set_in_chat("bob", tool_id, True)
with patch("docsgpt.connectors.service.revoke_at_provider"):
service.remove_connection(pg_conn, _row(pg_conn, cid))
for table, column in (("resource_share_settings", "resource_id"), ("user_tool_preferences", "tool_id")):
count = pg_conn.execute(
text(f"SELECT count(*) FROM {table} WHERE {column} = CAST(:i AS uuid)"), {"i": tool_id}
).scalar()
assert count == 0, table
def test_delete_sources_returns_them(self, pg_conn):
cid = _connection(pg_conn, secrets={})
source = _source(pg_conn, cid)