From bf89b9d57a7fc5df5753c2af5ca2f9fa44d0d4b2 Mon Sep 17 00:00:00 2001 From: Dave Page Date: Wed, 19 Aug 2026 12:05:03 +0100 Subject: [PATCH 1/4] Editable data for simple auto-updatable views (#2363) ViewCommand.can_edit() now runs a new view_base_table.sql template to check PostgreSQL's own information_schema.views.is_updatable/ is_trigger_updatable plus a single-base-table check via view_table_usage, then reuses the existing primary_keys.sql and get_columns.sql templates to resolve the base table's primary key columns and confirm they're still exposed under their original names in the view's own output. Editability and the resolved PK info are cached on the instance. get_primary_keys(), has_oids() and save() are added to mirror TableCommand, and get_columns_types() is added because the poll() endpoint calls it whenever can_edit() is true - without it, polling results for a now-editable view raised an AttributeError. MViewCommand inherits this unchanged and correctly stays read-only, since materialized views have no information_schema.views row at all. Adds web/pgadmin/tools/sqleditor/tests/test_view_command_editable.py, covering a simple 1:1 view (including an actual UPDATE through the view landing in the base table), a view omitting the PK column, a view with a WHERE clause, a join-based view, a trigger-backed view, and a materialized view. --- web/pgadmin/tools/sqleditor/command.py | 173 ++++++++++ .../sqleditor/sql/default/view_base_table.sql | 19 + .../tests/test_view_command_editable.py | 324 ++++++++++++++++++ 3 files changed, 516 insertions(+) create mode 100644 web/pgadmin/tools/sqleditor/templates/sqleditor/sql/default/view_base_table.sql create mode 100644 web/pgadmin/tools/sqleditor/tests/test_view_command_editable.py diff --git a/web/pgadmin/tools/sqleditor/command.py b/web/pgadmin/tools/sqleditor/command.py index fb7a9d5f87b..99832bbef8d 100644 --- a/web/pgadmin/tools/sqleditor/command.py +++ b/web/pgadmin/tools/sqleditor/command.py @@ -659,6 +659,13 @@ def __init__(self, **kwargs): # call base class init to fetch the table name super().__init__(**kwargs) + # Cache for the editability check (and the primary key info it + # resolves along the way), so a view's updatability only needs to + # be worked out once per request. None means "not yet determined". + self._can_edit = None + self._pk_names = '' + self._primary_keys = OrderedDict() + def get_sql(self, default_conn=None): """ This method is used to create a proper SQL query @@ -684,8 +691,174 @@ def get_sql(self, default_conn=None): return sql def can_edit(self): + """ + A view is editable only if PostgreSQL itself classifies it as a + simple automatically updatable view (single base table, no + INSTEAD OF triggers), and the base table's primary key columns are + exposed under their original names in the view's own output. + + This never raises: any missing connection, failed query, or + unexpected error is treated as "not editable". + """ + if self._can_edit is not None: + return self._can_edit + + try: + driver = get_driver(PG_DEFAULT_DRIVER) + manager = driver.connection_manager(self.sid) + conn = manager.connection(did=self.did, conn_id=self.conn_id) + + if not conn.connected(): + return False + + # Resolve the base table backing this view (if any). + query = render_template( + "/".join([self.sql_path, 'view_base_table.sql']), + nsp_name=self.nsp_name, + object_name=self.object_name, + conn=conn, + ) + status, result = conn.execute_dict(query) + if not status: + return False + + if len(result['rows']) == 0: + # Not a simple auto-updatable view (join, trigger-backed, + # not updatable at all, etc). This is a stable catalog + # fact for this view, so cache it. + self._can_edit = False + return self._can_edit + + base_nspname = result['rows'][0]['nspname'] + base_relname = result['rows'][0]['relname'] + + # The base table's real primary key columns. + pk_query = render_template( + "/".join([self.sql_path, 'primary_keys.sql']), + table_name=base_relname, + table_nspname=base_nspname, + conn=conn, + ) + status, pk_result = conn.execute_dict(pk_query) + if not status: + return False + + # The view's own output column names. + cols_query = render_template( + "/".join([self.sql_path, 'get_columns.sql']), + obj_id=self.obj_id, + conn=conn, + ) + status, cols_result = conn.execute_dict(cols_query) + if not status: + return False + + view_column_names = { + row['attname'] for row in cols_result['rows'] + } + + # Keep only the primary-key columns that are also present, + # under their original name, in the view's own output. There + # is no reliable way to map a renamed/omitted PK column back + # to the base table through aliasing, so such views are + # deliberately left non-editable. + primary_keys = OrderedDict() + pk_names = '' + for row in pk_result['rows']: + if row['attname'] in view_column_names: + pk_names += driver.qtIdent(conn, row['attname']) + ',' + primary_keys[row['attname']] = row['typname'] + + if len(primary_keys) == 0: + self._can_edit = False + return self._can_edit + + if pk_names != '': + # Remove last character from the string + pk_names = pk_names[:-1] + + self._pk_names = pk_names + self._primary_keys = primary_keys + self._can_edit = True + except Exception: + # Fail closed - never let can_edit() raise. + return False + + return self._can_edit + + def get_primary_keys(self, default_conn=None): + """ + This function is used to fetch the primary key columns of the + view's underlying base table, filtered to the ones the view + itself still exposes under their original name. Resolved (and + cached) by can_edit(), which is run first if it hasn't been yet. + """ + if self._can_edit is None: + self.can_edit() + + return self._pk_names, self._primary_keys + + def has_oids(self, default_conn=None): + """ + Views cannot have oids in any currently supported PostgreSQL + version. + """ return False + def save(self, + changed_data, + columns_info, + client_primary_key='__temp_PK', + default_conn=None): + """ + This function is used to save the data into the database. + + Args: + changed_data: Contains data to be saved + columns_info: + client_primary_key: + default_conn: + """ + driver = get_driver(PG_DEFAULT_DRIVER) + if default_conn is None: + manager = driver.connection_manager(self.sid) + conn = manager.connection(did=self.did, conn_id=self.conn_id) + else: + conn = default_conn + + return save_changed_data(changed_data=changed_data, + columns_info=columns_info, + command_obj=self, + client_primary_key=client_primary_key, + conn=conn) + + def get_columns_types(self, conn): + """ + Fetch column type/attribute info for the view's own output + columns, the same way TableCommand does for a table. Reused + as-is: for a simple 1:1 view the driver reports every result + column's table_oid as the view's own oid (see + _check_single_table), so this resolves against the view's own + catalog entry exactly like it does for a table. + """ + columns_info = conn.get_column_info() + has_oids = self.has_oids() + table_name = None + table_nspname = None + table_oid = _check_single_table(columns_info) + if table_oid is None: + table_name = self.object_name + table_nspname = self.nsp_name + + return get_columns_types(conn=conn, + columns_info=columns_info, + has_oids=has_oids, + table_oid=table_oid, + is_query_tool=False, + table_name=table_name, + table_nspname=table_nspname, + ) + def can_filter(self): return True diff --git a/web/pgadmin/tools/sqleditor/templates/sqleditor/sql/default/view_base_table.sql b/web/pgadmin/tools/sqleditor/templates/sqleditor/sql/default/view_base_table.sql new file mode 100644 index 00000000000..fa5569d1a12 --- /dev/null +++ b/web/pgadmin/tools/sqleditor/templates/sqleditor/sql/default/view_base_table.sql @@ -0,0 +1,19 @@ +{# ============= Fetch the base table backing a simple auto-updatable view ============= #} +SELECT DISTINCT vtu.table_schema AS nspname, vtu.table_name AS relname +FROM information_schema.view_table_usage vtu +WHERE vtu.view_schema = {{nsp_name|qtLiteral(conn)}} + AND vtu.view_name = {{object_name|qtLiteral(conn)}} + AND ( + SELECT count(DISTINCT (v2.table_schema, v2.table_name)) + FROM information_schema.view_table_usage v2 + WHERE v2.view_schema = {{nsp_name|qtLiteral(conn)}} + AND v2.view_name = {{object_name|qtLiteral(conn)}} + ) = 1 + AND EXISTS ( + SELECT 1 + FROM information_schema.views v + WHERE v.table_schema = {{nsp_name|qtLiteral(conn)}} + AND v.table_name = {{object_name|qtLiteral(conn)}} + AND v.is_updatable = 'YES' + AND v.is_trigger_updatable = 'NO' + ); diff --git a/web/pgadmin/tools/sqleditor/tests/test_view_command_editable.py b/web/pgadmin/tools/sqleditor/tests/test_view_command_editable.py new file mode 100644 index 00000000000..784506c814f --- /dev/null +++ b/web/pgadmin/tools/sqleditor/tests/test_view_command_editable.py @@ -0,0 +1,324 @@ +########################################################################## +# +# pgAdmin 4 - PostgreSQL Tools +# +# Copyright (C) 2013 - 2026, The pgAdmin Development Team +# This software is released under the PostgreSQL Licence +# +########################################################################## + +""" +Tests for ViewCommand.can_edit()/get_primary_keys()/save() - the +"View/Edit Data" grid support for simple auto-updatable views +(issue #2363 / RM #3997). +""" + +import json +import secrets + +from pgadmin.browser.server_groups.servers.databases.tests import utils as \ + database_utils +from pgadmin.utils.route import BaseTestGenerator +from regression import parent_node_dict +from regression.python_test_utils import test_utils as utils +from pgadmin.tools.sqleditor.tests.execute_query_test_utils import \ + async_poll + +# cmd_type value for "All Rows", same as VIEW_ALL_ROWS in +# pgadmin.tools.sqleditor.command. +VIEW_ALL_ROWS = 3 + + +class TestViewCommandEditable(BaseTestGenerator): + """ This class tests whether ViewCommand.can_edit() correctly + classifies simple auto-updatable views as editable, and that an + actual save() through an editable view lands in the base table. """ + + scenarios = [ + ('Simple 1:1 view is editable', dict( + setup_sql=""" + CREATE TABLE {base1} ( + id SERIAL PRIMARY KEY, + name VARCHAR(50) + ); + INSERT INTO {base1} (id, name) VALUES (1, 'foo'); + CREATE VIEW {view} AS SELECT id, name FROM {base1}; + """, + teardown_sql=""" + DROP VIEW IF EXISTS {view}; + DROP TABLE IF EXISTS {base1}; + """, + view_key='view', + obj_type='view', + expected_can_edit=True, + expected_primary_keys={'id': 'int4'}, + do_save=True, + )), + ('View omitting the primary key column is not editable', dict( + setup_sql=""" + CREATE TABLE {base1} ( + id SERIAL PRIMARY KEY, + name VARCHAR(50) + ); + INSERT INTO {base1} (id, name) VALUES (1, 'foo'); + CREATE VIEW {view} AS SELECT name FROM {base1}; + """, + teardown_sql=""" + DROP VIEW IF EXISTS {view}; + DROP TABLE IF EXISTS {base1}; + """, + view_key='view', + obj_type='view', + expected_can_edit=False, + expected_primary_keys=None, + do_save=False, + )), + ('View with a WHERE clause is still editable', dict( + setup_sql=""" + CREATE TABLE {base1} ( + id SERIAL PRIMARY KEY, + name VARCHAR(50) + ); + INSERT INTO {base1} (id, name) VALUES (1, 'foo'); + CREATE VIEW {view} AS + SELECT id, name FROM {base1} WHERE id > 0; + """, + teardown_sql=""" + DROP VIEW IF EXISTS {view}; + DROP TABLE IF EXISTS {base1}; + """, + view_key='view', + obj_type='view', + expected_can_edit=True, + expected_primary_keys={'id': 'int4'}, + do_save=False, + )), + ('Join-based view is not editable', dict( + setup_sql=""" + CREATE TABLE {base1} ( + id SERIAL PRIMARY KEY, + name VARCHAR(50) + ); + CREATE TABLE {base2} ( + id SERIAL PRIMARY KEY, + base1_id INTEGER + ); + INSERT INTO {base1} (id, name) VALUES (1, 'foo'); + INSERT INTO {base2} (id, base1_id) VALUES (1, 1); + CREATE VIEW {view} AS + SELECT a.id, a.name, b.base1_id + FROM {base1} a JOIN {base2} b ON a.id = b.base1_id; + """, + teardown_sql=""" + DROP VIEW IF EXISTS {view}; + DROP TABLE IF EXISTS {base2}; + DROP TABLE IF EXISTS {base1}; + """, + view_key='view', + obj_type='view', + expected_can_edit=False, + expected_primary_keys=None, + do_save=False, + )), + ('View with an INSTEAD OF UPDATE trigger is not editable', dict( + setup_sql=""" + CREATE TABLE {base1} ( + id SERIAL PRIMARY KEY, + name VARCHAR(50) + ); + INSERT INTO {base1} (id, name) VALUES (1, 'foo'); + CREATE VIEW {view} AS SELECT id, name FROM {base1}; + CREATE FUNCTION {trig_func}() RETURNS trigger AS $$ + BEGIN + RETURN NULL; + END; + $$ LANGUAGE plpgsql; + CREATE TRIGGER {trig_name} + INSTEAD OF UPDATE ON {view} + FOR EACH ROW EXECUTE FUNCTION {trig_func}(); + """, + teardown_sql=""" + DROP TRIGGER IF EXISTS {trig_name} ON {view}; + DROP VIEW IF EXISTS {view}; + DROP FUNCTION IF EXISTS {trig_func}(); + DROP TABLE IF EXISTS {base1}; + """, + view_key='view', + obj_type='view', + expected_can_edit=False, + expected_primary_keys=None, + do_save=False, + )), + ('Materialized view is not editable', dict( + setup_sql=""" + CREATE TABLE {base1} ( + id SERIAL PRIMARY KEY, + name VARCHAR(50) + ); + INSERT INTO {base1} (id, name) VALUES (1, 'foo'); + CREATE MATERIALIZED VIEW {mview} AS + SELECT id, name FROM {base1}; + """, + teardown_sql=""" + DROP MATERIALIZED VIEW IF EXISTS {mview}; + DROP TABLE IF EXISTS {base1}; + """, + view_key='mview', + obj_type='mview', + expected_can_edit=False, + expected_primary_keys=None, + do_save=False, + )), + ] + + def setUp(self): + self._initialize_database_connection() + + def runTest(self): + self._build_names() + self._create_test_objects() + try: + self._initialize_view_data() + start_data, poll_data = self._start_view_data() + self.assertEqual( + start_data['data']['can_edit'], self.expected_can_edit) + + if self.expected_can_edit: + self.assertEqual( + poll_data['data']['primary_keys'], + self.expected_primary_keys) + + if self.do_save: + self._save_through_view() + self._check_base_table_updated() + finally: + self._close_query_tool() + + def tearDown(self): + self._drop_test_objects() + database_utils.disconnect_database(self, self.server_id, self.db_id) + + # -- setup helpers ----------------------------------------------- + + def _initialize_database_connection(self): + database_info = parent_node_dict["database"][-1] + self.db_name = database_info["db_name"] + self.server_id = database_info["server_id"] + self.db_id = database_info["db_id"] + + db_con = database_utils.connect_database( + self, utils.SERVER_GROUP, self.server_id, self.db_id) + + if not db_con["info"] == "Database connected.": + raise Exception("Could not connect to the database.") + + self.connection = utils.get_db_connection( + self.db_name, + self.server['username'], + self.server['db_password'], + self.server['host'], + self.server['port'] + ) + + def _build_names(self): + suffix = str(secrets.choice(range(100000, 999999))) + self.base1 = 'test_editview_base1_' + suffix + self.base2 = 'test_editview_base2_' + suffix + self.view = 'test_editview_v_' + suffix + self.mview = 'test_editview_mv_' + suffix + self.trig_func = 'test_editview_trig_func_' + suffix + self.trig_name = 'test_editview_trig_' + suffix + + self._names = dict( + base1=self.base1, base2=self.base2, view=self.view, + mview=self.mview, trig_func=self.trig_func, + trig_name=self.trig_name, + ) + + # The actual relation name (view or materialized view) driving + # this scenario. + self.relname = self._names[self.view_key] + self.relkind = 'm' if self.obj_type == 'mview' else 'v' + + def _create_test_objects(self): + sql = self.setup_sql.format(**self._names) + utils.create_table_with_query(self.server, self.db_name, sql) + + def _drop_test_objects(self): + try: + sql = self.teardown_sql.format(**self._names) + utils.create_table_with_query(self.server, self.db_name, sql) + except Exception: + pass + + # -- View/Edit Data flow (mirrors test_view_data.py) -------------- + + def _get_relation_oid(self): + pg_cursor = self.connection.cursor() + pg_cursor.execute( + "SELECT oid FROM pg_catalog.pg_class WHERE relname = %s " + "AND relkind = %s", (self.relname, self.relkind)) + result = pg_cursor.fetchall() + self.connection.commit() + return result[0][0] + + def _initialize_view_data(self): + obj_id = self._get_relation_oid() + self.trans_id = str(secrets.choice(range(1, 9999999))) + url = '/sqleditor/initialize/viewdata/{0}/{1}/{2}/{3}/{4}/{5}/{6}' \ + .format(self.trans_id, VIEW_ALL_ROWS, self.obj_type, + utils.SERVER_GROUP, self.server_id, self.db_id, obj_id) + response = self.tester.post(url) + self.assertEqual(response.status_code, 200) + + def _start_view_data(self): + """Kick off the view/edit data query and poll it to completion. + + Returns (start_data, poll_data): the JSON response from + `view_data/start` (which carries `can_edit`) and the final `poll` + response (which carries `primary_keys`, resolved from + `get_primary_keys()`). + """ + url = "/sqleditor/view_data/start/{0}".format(self.trans_id) + response = self.tester.get(url) + self.assertEqual(response.status_code, 200) + start_data = json.loads(response.data.decode('utf-8')) + + poll_response = async_poll( + tester=self.tester, + poll_url='/sqleditor/poll/{0}'.format(self.trans_id)) + self.assertEqual(poll_response.status_code, 200) + poll_data = json.loads(poll_response.data.decode('utf-8')) + + return start_data, poll_data + + def _save_through_view(self): + save_payload = { + "updated": { + "1": { + "err": False, + "data": {"name": "bar"}, + "primary_keys": {"id": 1} + } + }, + "added": {}, + "deleted": {}, + } + url = '/sqleditor/save/{0}'.format(self.trans_id) + response = self.tester.post( + url, data=json.dumps(save_payload), content_type='html/json') + self.assertEqual(response.status_code, 200) + response_data = json.loads(response.data.decode('utf-8')) + self.assertEqual(response_data['data']['status'], True) + + def _check_base_table_updated(self): + pg_cursor = self.connection.cursor() + pg_cursor.execute( + "SELECT name FROM {0} WHERE id = 1".format(self.base1)) + result = pg_cursor.fetchall() + self.connection.commit() + self.assertEqual(result[0][0], 'bar') + + def _close_query_tool(self): + url = '/sqleditor/close/{0}'.format(self.trans_id) + self.tester.delete(url) From d8bb510eea9925a8f607af4f66963367759fbc2f Mon Sep 17 00:00:00 2001 From: Dave Page Date: Wed, 19 Aug 2026 12:28:23 +0100 Subject: [PATCH 2/4] Fix editable-view review findings: PK aliasing, trigger scope, insert, role filtering Four issues found in review of the editable-view-data feature: - Critical: can_edit()'s name-only PK match could be fooled by a view column that merely shares a name with the base table's real PK without being it (e.g. `SELECT legacy AS id, id AS realid FROM t`), letting an UPDATE/DELETE through the view silently rewrite every base row sharing that value instead of just one. Since there's no reliable way to resolve this through aliasing (per the design spec), save() now checks the actual rows-affected count for each view UPDATE/DELETE and rolls back and rejects the change if it isn't exactly what was intended, rather than letting it stand. Scoped to ViewCommand/ MViewCommand only (matched by object_type, not isinstance, to avoid a circular import) - tables are already protected by a real PRIMARY KEY constraint. - view_base_table.sql only excluded INSTEAD OF UPDATE triggers (is_trigger_updatable); a view with only an INSTEAD OF DELETE or INSERT trigger passed through uncaught. Added is_trigger_deletable and is_trigger_insertable_into to the same check. - Row insertion through a view was reachable via the existing "Add row" UI (gated only on the shared can_edit flag) but was never designed for. save() now explicitly rejects any newly-added row when the target is a view. - information_schema.view_table_usage is filtered by pg_has_role(owner, 'USAGE'), so it returned nothing for a role with direct grants but no ownership/membership - the normal case in most server-mode deployments. Replaced with a pg_depend/pg_rewrite-based lookup of the view's _RETURN rule, which carries no such filter. Added tests for all four: an aliased-PK view whose update is rejected and confirmed unchanged in the base table, a view with only an INSTEAD OF DELETE trigger, an insert attempt against an editable view, and a non-owner role (fresh LOGIN, direct grants only) still getting can_edit()=True. --- web/pgadmin/tools/sqleditor/command.py | 19 +- .../sqleditor/sql/default/view_base_table.sql | 28 +- .../tests/test_view_command_editable.py | 357 ++++++++++++++++++ .../sqleditor/utils/save_changed_data.py | 94 ++++- 4 files changed, 481 insertions(+), 17 deletions(-) diff --git a/web/pgadmin/tools/sqleditor/command.py b/web/pgadmin/tools/sqleditor/command.py index 99832bbef8d..1b6a3d79308 100644 --- a/web/pgadmin/tools/sqleditor/command.py +++ b/web/pgadmin/tools/sqleditor/command.py @@ -694,13 +694,20 @@ def can_edit(self): """ A view is editable only if PostgreSQL itself classifies it as a simple automatically updatable view (single base table, no - INSTEAD OF triggers), and the base table's primary key columns are - exposed under their original names in the view's own output. + INSTEAD OF UPDATE/DELETE/INSERT triggers - a view with any of + these is left entirely out of scope, not just the one the + triggering operation would use), and the base table's primary key + columns are exposed under their original names in the view's own + output. This never raises: any missing connection, failed query, or unexpected error is treated as "not editable". """ - if self._can_edit is not None: + # Use getattr rather than direct attribute access: a ViewCommand + # unpickled from a session created before this attribute existed + # (see session_obj['command_obj'] in sqleditor/__init__.py) won't + # have it, and this must fail closed rather than raise. + if getattr(self, '_can_edit', None) is not None: return self._can_edit try: @@ -714,6 +721,7 @@ def can_edit(self): # Resolve the base table backing this view (if any). query = render_template( "/".join([self.sql_path, 'view_base_table.sql']), + obj_id=self.obj_id, nsp_name=self.nsp_name, object_name=self.object_name, conn=conn, @@ -793,10 +801,11 @@ def get_primary_keys(self, default_conn=None): itself still exposes under their original name. Resolved (and cached) by can_edit(), which is run first if it hasn't been yet. """ - if self._can_edit is None: + if getattr(self, '_can_edit', None) is None: self.can_edit() - return self._pk_names, self._primary_keys + return getattr(self, '_pk_names', ''), \ + getattr(self, '_primary_keys', OrderedDict()) def has_oids(self, default_conn=None): """ diff --git a/web/pgadmin/tools/sqleditor/templates/sqleditor/sql/default/view_base_table.sql b/web/pgadmin/tools/sqleditor/templates/sqleditor/sql/default/view_base_table.sql index fa5569d1a12..10369d7f900 100644 --- a/web/pgadmin/tools/sqleditor/templates/sqleditor/sql/default/view_base_table.sql +++ b/web/pgadmin/tools/sqleditor/templates/sqleditor/sql/default/view_base_table.sql @@ -1,14 +1,20 @@ {# ============= Fetch the base table backing a simple auto-updatable view ============= #} -SELECT DISTINCT vtu.table_schema AS nspname, vtu.table_name AS relname -FROM information_schema.view_table_usage vtu -WHERE vtu.view_schema = {{nsp_name|qtLiteral(conn)}} - AND vtu.view_name = {{object_name|qtLiteral(conn)}} - AND ( - SELECT count(DISTINCT (v2.table_schema, v2.table_name)) - FROM information_schema.view_table_usage v2 - WHERE v2.view_schema = {{nsp_name|qtLiteral(conn)}} - AND v2.view_name = {{object_name|qtLiteral(conn)}} - ) = 1 +{# Resolved via pg_depend/pg_rewrite (not information_schema.view_table_usage, #} +{# which is filtered by pg_has_role() on the base table's owner and so misses #} +{# roles that only have direct GRANTs on the view/table, not ownership). #} +WITH base_tables AS ( + SELECT DISTINCT cl.relname, nsp.nspname + FROM pg_catalog.pg_depend dep + JOIN pg_catalog.pg_rewrite rw ON rw.oid = dep.objid + JOIN pg_catalog.pg_class cl ON cl.oid = dep.refobjid + JOIN pg_catalog.pg_namespace nsp ON nsp.oid = cl.relnamespace + WHERE rw.ev_class = {{obj_id}}::oid + AND dep.deptype != 'i' + AND cl.relkind IN ('r', 'p') +) +SELECT nspname, relname +FROM base_tables +WHERE (SELECT count(*) FROM base_tables) = 1 AND EXISTS ( SELECT 1 FROM information_schema.views v @@ -16,4 +22,6 @@ WHERE vtu.view_schema = {{nsp_name|qtLiteral(conn)}} AND v.table_name = {{object_name|qtLiteral(conn)}} AND v.is_updatable = 'YES' AND v.is_trigger_updatable = 'NO' + AND v.is_trigger_deletable = 'NO' + AND v.is_trigger_insertable_into = 'NO' ); diff --git a/web/pgadmin/tools/sqleditor/tests/test_view_command_editable.py b/web/pgadmin/tools/sqleditor/tests/test_view_command_editable.py index 784506c814f..2b29105a6ce 100644 --- a/web/pgadmin/tools/sqleditor/tests/test_view_command_editable.py +++ b/web/pgadmin/tools/sqleditor/tests/test_view_command_editable.py @@ -149,6 +149,40 @@ class TestViewCommandEditable(BaseTestGenerator): expected_primary_keys=None, do_save=False, )), + # information_schema.views.is_trigger_updatable only reflects + # INSTEAD OF UPDATE triggers - a view with *only* an INSTEAD OF + # DELETE trigger reports is_updatable='YES' AND + # is_trigger_updatable='NO', so this must be caught by the + # is_trigger_deletable check instead. + ('View with only an INSTEAD OF DELETE trigger is not editable', dict( + setup_sql=""" + CREATE TABLE {base1} ( + id SERIAL PRIMARY KEY, + name VARCHAR(50) + ); + INSERT INTO {base1} (id, name) VALUES (1, 'foo'); + CREATE VIEW {view} AS SELECT id, name FROM {base1}; + CREATE FUNCTION {trig_func}() RETURNS trigger AS $$ + BEGIN + RETURN NULL; + END; + $$ LANGUAGE plpgsql; + CREATE TRIGGER {trig_name} + INSTEAD OF DELETE ON {view} + FOR EACH ROW EXECUTE FUNCTION {trig_func}(); + """, + teardown_sql=""" + DROP TRIGGER IF EXISTS {trig_name} ON {view}; + DROP VIEW IF EXISTS {view}; + DROP FUNCTION IF EXISTS {trig_func}(); + DROP TABLE IF EXISTS {base1}; + """, + view_key='view', + obj_type='view', + expected_can_edit=False, + expected_primary_keys=None, + do_save=False, + )), ('Materialized view is not editable', dict( setup_sql=""" CREATE TABLE {base1} ( @@ -322,3 +356,326 @@ def _check_base_table_updated(self): def _close_query_tool(self): url = '/sqleditor/close/{0}'.format(self.trans_id) self.tester.delete(url) + + +class _ViewSaveTestMixin: + """ Shared plumbing for the ad-hoc single-scenario tests below: a + fresh connection, a helper to fetch a relation's oid, and the + initialize/start/poll/save/close HTTP calls used by + TestViewCommandEditable, without the scenario-table machinery (each + of these tests needs its own bespoke setup/assertions). """ + + def _connect(self): + database_info = parent_node_dict["database"][-1] + self.db_name = database_info["db_name"] + self.server_id = database_info["server_id"] + self.db_id = database_info["db_id"] + + db_con = database_utils.connect_database( + self, utils.SERVER_GROUP, self.server_id, self.db_id) + if not db_con["info"] == "Database connected.": + raise Exception("Could not connect to the database.") + + self.connection = utils.get_db_connection( + self.db_name, + self.server['username'], + self.server['db_password'], + self.server['host'], + self.server['port'] + ) + + def _get_relation_oid(self, relname, relkind='v'): + pg_cursor = self.connection.cursor() + pg_cursor.execute( + "SELECT oid FROM pg_catalog.pg_class WHERE relname = %s " + "AND relkind = %s", (relname, relkind)) + result = pg_cursor.fetchall() + self.connection.commit() + return result[0][0] + + def _initialize_view_data(self, obj_id, obj_type='view', body=None): + trans_id = str(secrets.choice(range(1, 9999999))) + url = '/sqleditor/initialize/viewdata/{0}/{1}/{2}/{3}/{4}/{5}/{6}' \ + .format(trans_id, VIEW_ALL_ROWS, obj_type, + utils.SERVER_GROUP, self.server_id, self.db_id, obj_id) + if body is not None: + response = self.tester.post( + url, data=json.dumps(body), content_type='html/json') + else: + response = self.tester.post(url) + self.assertEqual(response.status_code, 200) + return trans_id + + def _start_and_poll(self, trans_id): + url = "/sqleditor/view_data/start/{0}".format(trans_id) + response = self.tester.get(url) + self.assertEqual(response.status_code, 200) + start_data = json.loads(response.data.decode('utf-8')) + + poll_response = async_poll( + tester=self.tester, + poll_url='/sqleditor/poll/{0}'.format(trans_id)) + self.assertEqual(poll_response.status_code, 200) + + return start_data + + def _save(self, trans_id, save_payload): + url = '/sqleditor/save/{0}'.format(trans_id) + response = self.tester.post( + url, data=json.dumps(save_payload), content_type='html/json') + self.assertEqual(response.status_code, 200) + return json.loads(response.data.decode('utf-8')) + + def _close_query_tool(self, trans_id): + url = '/sqleditor/close/{0}'.format(trans_id) + self.tester.delete(url) + + +class TestViewSaveRejectsAmbiguousPrimaryKey( + _ViewSaveTestMixin, BaseTestGenerator): + """ Regression test for the aliasing gap can_edit() can't close. + + can_edit()'s primary-key check is name-only (deliberately - see the + design spec's pg_depend note on why per-column alias resolution + isn't reliable): it only confirms the base table's real PK column + name is *also* present, under the same name, in the view's own + output. It has no way to tell that a differently-derived column + happens to share that name. + + `CREATE VIEW v AS SELECT legacy AS id, id AS realid, name FROM t` + passes that check (t's real PK is `id`, and the view exposes an + output column literally called `id`), but the view's `id` is + actually `t.legacy`, not `t.id`. An UPDATE through the view with + `WHERE id = ` would then rewrite every base row sharing that + `legacy` value, not just one - so the save path's rows-affected + safety net must catch and reject this rather than silently + corrupting more rows than intended. """ + + scenarios = [('default', dict())] + + def setUp(self): + self._connect() + suffix = str(secrets.choice(range(100000, 999999))) + self.base1 = 'test_editview_alias_base_' + suffix + self.view = 'test_editview_alias_v_' + suffix + + def runTest(self): + setup_sql = """ + CREATE TABLE {base1} ( + id SERIAL PRIMARY KEY, + legacy INTEGER, + name VARCHAR(50) + ); + INSERT INTO {base1} (id, legacy, name) VALUES + (1, 5, 'foo'), + (2, 5, 'bar'); + CREATE VIEW {view} AS + SELECT legacy AS id, id AS realid, name FROM {base1}; + """.format(base1=self.base1, view=self.view) + utils.create_table_with_query(self.server, self.db_name, setup_sql) + + try: + obj_id = self._get_relation_oid(self.view) + trans_id = self._initialize_view_data(obj_id) + + # Confirm the exploit precondition: can_edit() is fooled by + # the name-only match. + start_data = self._start_and_poll(trans_id) + self.assertTrue(start_data['data']['can_edit']) + + save_payload = { + "updated": { + "1": { + "err": False, + "data": {"name": "CHANGED"}, + "primary_keys": {"id": 5} + } + }, + "added": {}, + "deleted": {}, + } + response_data = self._save(trans_id, save_payload) + + # Rejected, not silently applied to both rows. + self.assertEqual(response_data['data']['status'], False) + + self._close_query_tool(trans_id) + + pg_cursor = self.connection.cursor() + pg_cursor.execute( + "SELECT name FROM {0} ORDER BY id".format(self.base1)) + result = pg_cursor.fetchall() + self.connection.commit() + # Neither base row was changed. + self.assertEqual([r[0] for r in result], ['foo', 'bar']) + finally: + self._drop_test_objects() + + def _drop_test_objects(self): + try: + utils.create_table_with_query( + self.server, self.db_name, + "DROP VIEW IF EXISTS {view}; " + "DROP TABLE IF EXISTS {base1};".format( + view=self.view, base1=self.base1)) + except Exception: + pass + + def tearDown(self): + database_utils.disconnect_database(self, self.server_id, self.db_id) + + +class TestViewSaveRejectsInsertedRow(_ViewSaveTestMixin, BaseTestGenerator): + """ Regression test: the frontend's "Add row" is gated only on the + backend's can_edit flag (shared, unchanged behaviour also used for + tables), so once can_edit() can return True for a view, a user can + reach the INSERT path through the existing UI even though row + insertion into a view was never designed for or tested (out of + scope per the design spec). save() must reject any row marked as + newly-inserted when the target is a view. """ + + scenarios = [('default', dict())] + + def setUp(self): + self._connect() + suffix = str(secrets.choice(range(100000, 999999))) + self.base1 = 'test_editview_insert_base_' + suffix + self.view = 'test_editview_insert_v_' + suffix + + def runTest(self): + setup_sql = """ + CREATE TABLE {base1} ( + id SERIAL PRIMARY KEY, + name VARCHAR(50) + ); + INSERT INTO {base1} (id, name) VALUES (1, 'foo'); + CREATE VIEW {view} AS SELECT id, name FROM {base1}; + """.format(base1=self.base1, view=self.view) + utils.create_table_with_query(self.server, self.db_name, setup_sql) + + try: + obj_id = self._get_relation_oid(self.view) + trans_id = self._initialize_view_data(obj_id) + + start_data = self._start_and_poll(trans_id) + self.assertTrue(start_data['data']['can_edit']) + + save_payload = { + "updated": {}, + "added": { + "2": { + "err": False, + "data": { + "id": "99", + "__temp_PK": "2", + "name": "new row" + } + } + }, + "deleted": {}, + "added_index": {"2": "2"}, + } + response_data = self._save(trans_id, save_payload) + + self.assertEqual(response_data['data']['status'], False) + + self._close_query_tool(trans_id) + + pg_cursor = self.connection.cursor() + pg_cursor.execute( + "SELECT count(*) FROM {0}".format(self.base1)) + result = pg_cursor.fetchall() + self.connection.commit() + # No row was actually inserted into the base table. + self.assertEqual(int(result[0][0]), 1) + finally: + self._drop_test_objects() + + def _drop_test_objects(self): + try: + utils.create_table_with_query( + self.server, self.db_name, + "DROP VIEW IF EXISTS {view}; " + "DROP TABLE IF EXISTS {base1};".format( + view=self.view, base1=self.base1)) + except Exception: + pass + + def tearDown(self): + database_utils.disconnect_database(self, self.server_id, self.db_id) + + +class TestViewCommandEditableForNonOwnerRole( + _ViewSaveTestMixin, BaseTestGenerator): + """ Regression test: information_schema.view_table_usage (previously + used to resolve a view's base table) is filtered by + pg_has_role(owner, 'USAGE'), so it returns nothing for a role that + has direct GRANTs on the view/table but isn't a member of the + owning role - which is the normal case in most real server-mode + deployments (the connecting role is rarely the table owner). The + base-table lookup is now resolved via pg_depend/pg_rewrite instead, + which carries no such role filter. This test authenticates as a + fresh, non-superuser LOGIN role with only direct grants (no + ownership, no role membership) and confirms can_edit() is still + True. """ + + scenarios = [('default', dict())] + + ROLE_PASSWORD = 'Editview_probe_pw1!' + + def setUp(self): + self._connect() + suffix = str(secrets.choice(range(100000, 999999))) + self.base1 = 'test_editview_role_base_' + suffix + self.view = 'test_editview_role_v_' + suffix + self.role_name = 'test_editview_role_' + suffix + + def runTest(self): + setup_sql = """ + CREATE TABLE {base1} ( + id SERIAL PRIMARY KEY, + name VARCHAR(50) + ); + INSERT INTO {base1} (id, name) VALUES (1, 'foo'); + CREATE VIEW {view} AS SELECT id, name FROM {base1}; + CREATE ROLE {role} LOGIN PASSWORD '{password}'; + GRANT SELECT, UPDATE ON {base1} TO {role}; + GRANT SELECT, UPDATE ON {view} TO {role}; + """.format(base1=self.base1, view=self.view, role=self.role_name, + password=self.ROLE_PASSWORD) + utils.create_table_with_query(self.server, self.db_name, setup_sql) + + try: + obj_id = self._get_relation_oid(self.view) + # Log the async connection in as the restricted role - a + # fresh, password-authenticated connection (see + # Connection.connect()), not a superuser SET ROLE - so this + # genuinely exercises a non-owner, non-superuser session. + trans_id = self._initialize_view_data( + obj_id, + body={ + "user": self.role_name, + "password": self.ROLE_PASSWORD, + } + ) + + start_data = self._start_and_poll(trans_id) + self.assertTrue(start_data['data']['can_edit']) + + self._close_query_tool(trans_id) + finally: + self._drop_test_objects() + + def _drop_test_objects(self): + try: + utils.create_table_with_query( + self.server, self.db_name, + "DROP VIEW IF EXISTS {view}; " + "DROP TABLE IF EXISTS {base1}; " + "DROP ROLE IF EXISTS {role};".format( + view=self.view, base1=self.base1, role=self.role_name)) + except Exception: + pass + + def tearDown(self): + database_utils.disconnect_database(self, self.server_id, self.db_id) diff --git a/web/pgadmin/tools/sqleditor/utils/save_changed_data.py b/web/pgadmin/tools/sqleditor/utils/save_changed_data.py index 78776697ff3..b144f24e0c7 100644 --- a/web/pgadmin/tools/sqleditor/utils/save_changed_data.py +++ b/web/pgadmin/tools/sqleditor/utils/save_changed_data.py @@ -8,6 +8,7 @@ ########################################################################## from flask import render_template +from flask_babel import gettext from collections import OrderedDict from pgadmin.tools.sqleditor.utils.constant_definition import TX_STATUS_IDLE @@ -16,6 +17,18 @@ ignore_type_cast_list = ['character', 'character[]', 'bit', 'bit[]'] +def _is_view_command(command_obj): + """ + True for a ViewCommand/MViewCommand target, matched by object_type + rather than isinstance() to avoid a circular import with + pgadmin.tools.sqleditor.command (which imports this module at load + time). A table's own real PRIMARY KEY constraint already guarantees + the safety nets below can't trigger for it, so they're scoped to + views only. + """ + return getattr(command_obj, 'object_type', None) in ('view', 'mview') + + def save_changed_data(changed_data, columns_info, conn, command_obj, client_primary_key, auto_commit=True): """ @@ -38,6 +51,21 @@ def save_changed_data(changed_data, columns_info, conn, command_obj, list_of_sql = {} _rowid = None + # Row insertion through a view was never designed for or tested (the + # frontend's "Add row" is gated only on can_edit, shared with tables, + # so this has to be enforced here). Reject the whole save rather than + # silently dropping just the added rows. + if _is_view_command(command_obj) and changed_data.get('added'): + return ( + False, + gettext( + 'Inserting new rows into a view is not currently ' + 'supported.' + ), + [], + None + ) + pgadmin_alias = { col_name: col_info['pgadmin_alias'] for col_name, col_info in columns_info.items() @@ -197,7 +225,12 @@ def save_changed_data(changed_data, columns_info, conn, command_obj, list_of_sql[of_type].append({'sql': sql, 'data': data, 'row_id': - data.get(client_primary_key)}) + data.get(client_primary_key), + # A single UPDATE statement is + # rendered per row here, so + # exactly one base row should + # ever be affected. + 'expected_rows': 1}) # For deleted rows elif of_type == 'deleted': @@ -239,7 +272,13 @@ def save_changed_data(changed_data, columns_info, conn, command_obj, nsp_name=command_obj.nsp_name, conn=conn ) - list_of_sql[of_type].append({'sql': sql, 'data': {}}) + # A single DELETE statement is rendered covering every row in + # rows_to_delete, so that many base rows (no more, no fewer) + # should ever be affected. + list_of_sql[of_type].append({ + 'sql': sql, 'data': {}, + 'expected_rows': len(rows_to_delete) + }) def failure_handle(res, row_id): mogrified_sql = conn.mogrify(item['sql'], item['data']) @@ -284,11 +323,27 @@ def failure_handle(res, row_id): row_added = None + # execute_void() never updates the connection's tracked + # row count (see Connection.execute_void/row_count), so + # the rows-affected safety net below - which needs the + # true count for this exact UPDATE/DELETE - has to go + # through execute_dict() instead for a view's save. It's + # otherwise equivalent for a statement with no RETURNING + # clause: no columns come back, so no fetchall() is + # attempted, only cur.rowcount is captured. + needs_rows_affected = ( + _is_view_command(command_obj) and + opr in ('updated', 'deleted') + ) + try: # Fetch oids/primary keys if 'select_sql' in item and item['select_sql']: status, res = conn.execute_dict( item['sql'], item['data']) + elif needs_rows_affected: + status, res = conn.execute_dict( + item['sql'], item['data']) else: status, res = conn.execute_void( item['sql'], item['data']) @@ -316,6 +371,41 @@ def failure_handle(res, row_id): item['client_row']: sel_res['rows'][0]} rows_affected = conn.rows_affected() + + # Safety net for views only (tables are already protected + # by a real PRIMARY KEY constraint, so this can never + # trigger for them): a view's apparent primary key is + # only as reliable as its column names, and a renamed/ + # aliased column that happens to share a name with the + # base table's real PK (e.g. `SELECT legacy AS id, id AS + # realid FROM t`) can pass can_edit()'s name-only check + # while not actually identifying a single base row. Refuse + # to let such a change stand rather than silently + # rewriting more rows than intended. + if needs_rows_affected and \ + rows_affected != item.get('expected_rows', 1): + return failure_handle( + gettext( + 'This change was not applied: it would have ' + 'affected %(actual)s row(s) in the ' + 'underlying table instead of the %(expected)s ' + 'expected. The view\'s apparent primary key ' + 'does not uniquely identify the affected ' + 'row(s).' + ) % { + 'actual': rows_affected, + 'expected': item.get('expected_rows', 1) + }, + item.get('row_id', 0) + ) + + if needs_rows_affected: + # execute_dict() was only used here to get an accurate + # rows_affected; restore the plain execute_void()-style + # result shape (None) so downstream reporting of a + # successful update/delete is unchanged. + res = None + mogrified_sql = conn.mogrify(item['sql'], item['data']) mogrified_sql = mogrified_sql if mogrified_sql is not None \ else item['sql'] From c2d1d27be3661e11d12ece173a758e0bb908afc1 Mon Sep 17 00:00:00 2001 From: Dave Page Date: Wed, 19 Aug 2026 12:55:17 +0100 Subject: [PATCH 3/4] Fix final-review findings: docs, default_conn threading, save() guard Three issues from the whole-branch review, all "ready to merge, with fixes": - docs/en_US/editgrid.rst still said views cannot be edited and updatable views (using rules) are not supported. Corrected to describe what the code now does: simple auto-updatable views (single base table, no INSTEAD OF triggers, PK exposed under its own name) support UPDATE/DELETE but not row insertion; materialized, join-based and trigger-backed views stay read only. - ViewCommand.can_edit()/get_primary_keys() ignored the default_conn they were given (get_primary_keys() already accepted it but discarded it; can_edit() didn't even take it), resolving a second connection on the same conn_id instead - risking disturbance of an in-flight async cursor's results, per __init__.py's own comment on why start_view_data() resolves a separate default_conn in the first place. can_edit() now takes default_conn=None and uses it when supplied; get_primary_keys() forwards whatever it was given. - ViewCommand.save() (and MViewCommand, which inherits it) had no can_edit() guard, so a non-editable instance would reach save_changed_data() with an incomplete columns_info and fail with a KeyError after a BEGIN had already been issued - a dangling transaction and a 500, not an intentional guard. Added an explicit check at the top of save(). Deliberately does not call forbidden() the way GridCommand.save() does: forbidden() returns a raw HTTP Response, and the one real caller of ViewCommand.save() always unpacks a 4-tuple from it, which raises TypeError on a Response (verified) - a 500 instead of a clean refusal. Returns the same message in the 4-tuple shape save_changed_data() itself already uses for its own early refusals. Added tests: TestViewSaveGuardsNonEditable (a view missing its PK column, and a materialized view) confirms save() refuses cleanly, with no dangling transaction and no change to the base table. --- docs/en_US/editgrid.rst | 8 +- web/pgadmin/tools/sqleditor/command.py | 49 +++++++- .../tests/test_view_command_editable.py | 113 ++++++++++++++++++ 3 files changed, 163 insertions(+), 7 deletions(-) diff --git a/docs/en_US/editgrid.rst b/docs/en_US/editgrid.rst index 65ddfc39f43..59f2d906d1c 100644 --- a/docs/en_US/editgrid.rst +++ b/docs/en_US/editgrid.rst @@ -15,8 +15,12 @@ to specify the number of rows you would like to display in the editor panel. To modify the content of a table, each row in the table must be uniquely identifiable. If the table definition does not include an OID or a primary key, -the displayed data is read only. Note that views cannot be edited; updatable -views (using rules) are not supported. +the displayed data is read only. Simple, automatically-updatable views (a +single base table, with no ``INSTEAD OF`` triggers, whose base table's +primary key columns are exposed in the view under their original names) can +also be edited; updating and deleting rows is supported, but inserting new +rows through a view is not. Materialized views, views based on more than one +table, and views relying on ``INSTEAD OF`` triggers remain read only. The editor features a toolbar that allows quick access to frequently used options, and a work environment divided into two panels: diff --git a/web/pgadmin/tools/sqleditor/command.py b/web/pgadmin/tools/sqleditor/command.py index 1b6a3d79308..1a351020e1f 100644 --- a/web/pgadmin/tools/sqleditor/command.py +++ b/web/pgadmin/tools/sqleditor/command.py @@ -690,7 +690,7 @@ def get_sql(self, default_conn=None): return sql - def can_edit(self): + def can_edit(self, default_conn=None): """ A view is editable only if PostgreSQL itself classifies it as a simple automatically updatable view (single base table, no @@ -700,6 +700,14 @@ def can_edit(self): columns are exposed under their original names in the view's own output. + Args: + default_conn: an already-resolved connection to reuse (see + sqleditor/__init__.py's start_view_data(), which resolves + one specifically so metadata calls like this one don't + run on the same conn_id-keyed connection as an in-flight + async query and its cursor). Only resolved independently + when the caller doesn't supply one. + This never raises: any missing connection, failed query, or unexpected error is treated as "not editable". """ @@ -712,8 +720,12 @@ def can_edit(self): try: driver = get_driver(PG_DEFAULT_DRIVER) - manager = driver.connection_manager(self.sid) - conn = manager.connection(did=self.did, conn_id=self.conn_id) + if default_conn is None: + manager = driver.connection_manager(self.sid) + conn = manager.connection(did=self.did, + conn_id=self.conn_id) + else: + conn = default_conn if not conn.connected(): return False @@ -799,10 +811,13 @@ def get_primary_keys(self, default_conn=None): This function is used to fetch the primary key columns of the view's underlying base table, filtered to the ones the view itself still exposes under their original name. Resolved (and - cached) by can_edit(), which is run first if it hasn't been yet. + cached) by can_edit(), which is run first if it hasn't been yet - + forwarding whatever connection this was given, rather than + letting can_edit() resolve one of its own independently of the + caller (see can_edit()'s default_conn docstring). """ if getattr(self, '_can_edit', None) is None: - self.can_edit() + self.can_edit(default_conn) return getattr(self, '_pk_names', ''), \ getattr(self, '_primary_keys', OrderedDict()) @@ -828,6 +843,30 @@ def save(self, client_primary_key: default_conn: """ + # Before this class existed, every view fell through to + # GridCommand.save() below, which always refuses. Match that: a + # non-editable view (join-based, trigger-backed, matview, PK not + # exposed under its own name, etc.) must still be refused here, + # not attempted - can_edit() being false isn't otherwise checked + # anywhere on this path before a transaction gets started. + # + # This deliberately does NOT call forbidden() the way + # GridCommand.save() does: forbidden() returns a raw HTTP + # Response, but the one real caller of this method - the + # /sqleditor/save/ endpoint - always does + # `status, res, query_results, _rowid = trans_obj.save(...)`, + # and unpacking a Response that way raises TypeError (verified), + # turning a clean refusal into a 500. Same message/intent as + # forbidden(), in the 4-tuple shape save_changed_data() itself + # already returns for its own early refusals. + if not self.can_edit(default_conn): + return ( + False, + gettext("Data cannot be saved for the current object."), + [], + None + ) + driver = get_driver(PG_DEFAULT_DRIVER) if default_conn is None: manager = driver.connection_manager(self.sid) diff --git a/web/pgadmin/tools/sqleditor/tests/test_view_command_editable.py b/web/pgadmin/tools/sqleditor/tests/test_view_command_editable.py index 2b29105a6ce..451315adc68 100644 --- a/web/pgadmin/tools/sqleditor/tests/test_view_command_editable.py +++ b/web/pgadmin/tools/sqleditor/tests/test_view_command_editable.py @@ -679,3 +679,116 @@ def _drop_test_objects(self): def tearDown(self): database_utils.disconnect_database(self, self.server_id, self.db_id) + + +class TestViewSaveGuardsNonEditable(_ViewSaveTestMixin, BaseTestGenerator): + """ Regression test: before ViewCommand/MViewCommand implemented + their own save(), every view fell through to GridCommand.save(), + which always refuses. Now that they implement save() themselves, a + non-editable instance (can_edit() False - a view missing its PK + column, a materialized view, a join-based view, etc.) must still be + refused up front, not attempted: without this guard, save() would + reach save_changed_data() with an incomplete columns_info (built for + a non-editable object, missing keys like not_null) and fail with a + KeyError - after a BEGIN had already been issued, leaving a dangling + transaction and a 500 instead of a clean refusal. """ + + scenarios = [ + ('View missing its primary key column', dict( + setup_sql=""" + CREATE TABLE {base1} ( + id SERIAL PRIMARY KEY, + name VARCHAR(50) + ); + INSERT INTO {base1} (id, name) VALUES (1, 'foo'); + CREATE VIEW {view} AS SELECT name FROM {base1}; + """, + teardown_sql=""" + DROP VIEW IF EXISTS {view}; + DROP TABLE IF EXISTS {base1}; + """, + relname_key='view', + obj_type='view', + )), + ('Materialized view', dict( + setup_sql=""" + CREATE TABLE {base1} ( + id SERIAL PRIMARY KEY, + name VARCHAR(50) + ); + INSERT INTO {base1} (id, name) VALUES (1, 'foo'); + CREATE MATERIALIZED VIEW {mview} AS + SELECT id, name FROM {base1}; + """, + teardown_sql=""" + DROP MATERIALIZED VIEW IF EXISTS {mview}; + DROP TABLE IF EXISTS {base1}; + """, + relname_key='mview', + obj_type='mview', + )), + ] + + def setUp(self): + self._connect() + suffix = str(secrets.choice(range(100000, 999999))) + self.base1 = 'test_editview_guard_base_' + suffix + self.view = 'test_editview_guard_v_' + suffix + self.mview = 'test_editview_guard_mv_' + suffix + self._names = dict( + base1=self.base1, view=self.view, mview=self.mview) + self.relname = self._names[self.relname_key] + + def runTest(self): + sql = self.setup_sql.format(**self._names) + utils.create_table_with_query(self.server, self.db_name, sql) + + try: + relkind = 'm' if self.obj_type == 'mview' else 'v' + obj_id = self._get_relation_oid(self.relname, relkind) + trans_id = self._initialize_view_data(obj_id, self.obj_type) + + start_data = self._start_and_poll(trans_id) + # Precondition: this instance really is non-editable. + self.assertFalse(start_data['data']['can_edit']) + + save_payload = { + "updated": { + "1": { + "err": False, + "data": {"name": "should-not-apply"}, + "primary_keys": {"id": 1} + } + }, + "added": {}, + "deleted": {}, + } + response_data = self._save(trans_id, save_payload) + + # A clean refusal, not a 500 from an unhandled KeyError. + self.assertEqual(response_data['data']['status'], False) + self.assertIn( + 'cannot be saved', + response_data['data']['result'].lower()) + # No transaction was left dangling by the refused save. + self.assertEqual(response_data['data']['transaction_status'], 0) + + self._close_query_tool(trans_id) + + pg_cursor = self.connection.cursor() + pg_cursor.execute( + "SELECT name FROM {0} WHERE id = 1".format(self.base1)) + result = pg_cursor.fetchall() + self.connection.commit() + # Nothing was actually changed in the base table either. + self.assertEqual(result[0][0], 'foo') + finally: + try: + teardown_sql = self.teardown_sql.format(**self._names) + utils.create_table_with_query( + self.server, self.db_name, teardown_sql) + except Exception: + pass + + def tearDown(self): + database_utils.disconnect_database(self, self.server_id, self.db_id) From 8a27b08328a81fd9ed90fc84203b6262202d5edb Mon Sep 17 00:00:00 2001 From: Dave Page Date: Thu, 20 Aug 2026 09:09:41 +0100 Subject: [PATCH 4/4] Guard self.trans_id in test teardown to avoid masking failures CodeRabbit review on #10322: if _get_relation_oid() raises IndexError before self.trans_id is assigned, the finally block's unconditional _close_query_tool() call raised AttributeError, masking the original failure. Initialize self.trans_id to None in setUp and only close the query tool when it was actually assigned. --- .../tools/sqleditor/tests/test_view_command_editable.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/web/pgadmin/tools/sqleditor/tests/test_view_command_editable.py b/web/pgadmin/tools/sqleditor/tests/test_view_command_editable.py index 451315adc68..06646b17ff3 100644 --- a/web/pgadmin/tools/sqleditor/tests/test_view_command_editable.py +++ b/web/pgadmin/tools/sqleditor/tests/test_view_command_editable.py @@ -206,6 +206,7 @@ class TestViewCommandEditable(BaseTestGenerator): ] def setUp(self): + self.trans_id = None self._initialize_database_connection() def runTest(self): @@ -226,7 +227,8 @@ def runTest(self): self._save_through_view() self._check_base_table_updated() finally: - self._close_query_tool() + if self.trans_id is not None: + self._close_query_tool() def tearDown(self): self._drop_test_objects()