Refuse a disabled connector before falling back to a legacy tool secret

tests/connectors/test_service_credentials.py

@@ -300,6 +300,21 @@ class TestApiKeyConnections:
                 pg_conn, "alice", catalog.get_definition("brave"), {"token": "secret-key-1234"},
             )
 
+    def test_disabled_connector_refused_before_other_errors(self, pg_conn, monkeypatch):
+        from docsgpt.connectors import catalog
+        from docsgpt.core.settings import settings
+        from docsgpt.security.encryption import DEFAULT_ENCRYPTION_KEY
+        from docsgpt.storage.db.repositories.connector_policies import ConnectorPoliciesRepository
+
+        monkeypatch.setattr(settings, "ENCRYPTION_SECRET_KEY", DEFAULT_ENCRYPTION_KEY)
+        monkeypatch.setattr(settings, "AUTH_TYPE", "session_jwt")
+        ConnectorPoliciesRepository(pg_conn).upsert("telegram", enabled=False)
+        telegram = catalog.get_definition("telegram")
+        # Callers fall back to a legacy per-tool secret on these other errors.
+        for credentials in ({"token": "123456:ABCDEFG"}, {}):
+            with pytest.raises(service.ConnectorDisabled):
+                service.create_api_key_connection(pg_conn, "alice", telegram, credentials)
+
     def test_default_key_allowed_single_user(self, pg_conn, monkeypatch):
         from docsgpt.connectors import catalog
         from docsgpt.core.settings import settings

tests/connectors/test_tool_routes.py

@@ -136,6 +136,17 @@ class TestCreateConnectedTool:
             resp = _call(app, CreateTool, _telegram())
         assert resp.status_code == 403
 
+    def test_disabled_connector_is_refused_on_the_default_key(self, app, pg_conn):
+        from docsgpt.api.user.tools.routes import CreateTool
+        from docsgpt.storage.db.repositories.connector_policies import ConnectorPoliciesRepository
+
+        ConnectorPoliciesRepository(pg_conn).upsert("telegram", enabled=False)
+        with _db(pg_conn), patch.object(service, "ensure_can_store_credentials",
+                                        side_effect=service.EncryptionKeyNotConfigured("set a key")):
+            resp = _call(app, CreateTool, _telegram())
+        assert resp.status_code == 403
+        assert pg_conn.execute(text("SELECT count(*) FROM user_tools")).scalar() == 0
+
     def test_default_key_on_multi_user_falls_back_to_the_tool(self, app, pg_conn):
         from docsgpt.api.user.tools.routes import CreateTool