postgresql_lang: use query parameters with cursor.execute() (#65093)

* postgresql_lang: use query parameters with cursor.execute()

* add changelog fragment

changelogs/fragments/65093-postgresql_lang_use_query_params_with_cursor.yml

@@ -0,0 +1,2 @@
+bugfixes:
+- postgresql_lang - use query params with cursor.execute (https://github.com/ansible/ansible/pull/65093).

lib/ansible/modules/database/postgresql/postgresql_lang.py

@@ -191,23 +191,23 @@ executed_queries = []
 
 def lang_exists(cursor, lang):
     """Checks if language exists for db"""
-    query = "SELECT lanname FROM pg_language WHERE lanname = '%s'" % lang
-    cursor.execute(query)
+    query = "SELECT lanname FROM pg_language WHERE lanname = %(lang)s"
+    cursor.execute(query, {'lang': lang})
     return cursor.rowcount > 0
 
 
 def lang_istrusted(cursor, lang):
     """Checks if language is trusted for db"""
-    query = "SELECT lanpltrusted FROM pg_language WHERE lanname = '%s'" % lang
-    cursor.execute(query)
+    query = "SELECT lanpltrusted FROM pg_language WHERE lanname = %(lang)s"
+    cursor.execute(query, {'lang': lang})
     return cursor.fetchone()[0]
 
 
 def lang_altertrust(cursor, lang, trust):
     """Changes if language is trusted for db"""
-    query = "UPDATE pg_language SET lanpltrusted = '%s' WHERE lanname = '%s'" % (trust, lang)
-    executed_queries.append(query)
-    cursor.execute(query)
+    query = "UPDATE pg_language SET lanpltrusted = %(trust)s WHERE lanname = %(lang)s"
+    cursor.execute(query, {'trust': trust, 'lang': lang})
+    executed_queries.append(cursor.mogrify(query, {'trust': trust, 'lang': lang}))
     return True
 
 
@@ -249,8 +249,8 @@ def get_lang_owner(cursor, lang):
     """
     query = ("SELECT r.rolname FROM pg_language l "
              "JOIN pg_roles r ON l.lanowner = r.oid "
-             "WHERE l.lanname = '%s'" % lang)
-    cursor.execute(query)
+             "WHERE l.lanname = %(lang)s")
+    cursor.execute(query, {'lang': lang})
     return cursor.fetchone()[0]
 
 
@@ -262,7 +262,7 @@ def set_lang_owner(cursor, lang, owner):
         lang (str): language name.
         owner (str): name of new owner.
     """
-    query = "ALTER LANGUAGE %s OWNER TO %s" % (lang, owner)
+    query = "ALTER LANGUAGE \"%s\" OWNER TO %s" % (lang, owner)
     executed_queries.append(query)
     cursor.execute(query)
     return True

test/integration/targets/postgresql_lang/tasks/postgresql_lang_add_owner_param.yml

@@ -61,7 +61,7 @@
   - assert:
       that:
       - result is changed
-      - result.queries == ['CREATE LANGUAGE "{{ test_lang }}"', 'ALTER LANGUAGE {{ test_lang }} OWNER TO {{ test_user1 }}']
+      - result.queries == ['CREATE LANGUAGE "{{ test_lang }}"', 'ALTER LANGUAGE "{{ test_lang }}" OWNER TO {{ test_user1 }}']
 
   - name: Check
     <<: *task_parameters
@@ -88,7 +88,7 @@
   - assert:
       that:
       - result is changed
-      - result.queries == ["ALTER LANGUAGE {{ test_lang }} OWNER TO {{ test_user2 }}"]
+      - result.queries == ['ALTER LANGUAGE "{{ test_lang }}" OWNER TO {{ test_user2 }}']
 
   - name: Check that nothing was actually changed
     <<: *task_parameters
@@ -116,7 +116,7 @@
       - result is changed
       # TODO: the first elem of the returned list below
       # looks like a bug, not related with the option owner, needs to be checked
-      - result.queries == ["UPDATE pg_language SET lanpltrusted = 'False' WHERE lanname = '{{ test_lang }}'", "ALTER LANGUAGE {{ test_lang }} OWNER TO {{ test_user2 }}"]
+      - result.queries == ["UPDATE pg_language SET lanpltrusted = false WHERE lanname = '{{ test_lang }}'", 'ALTER LANGUAGE "{{ test_lang }}" OWNER TO {{ test_user2 }}']
 
   - name: Check
     <<: *task_parameters

test/integration/targets/postgresql_lang/tasks/postgresql_lang_initial.yml

@@ -183,7 +183,7 @@
 - assert:
     that:
     - result is changed
-    - result.queries == ['CREATE TRUSTED LANGUAGE "plpythonu"', "UPDATE pg_language SET lanpltrusted = 'True' WHERE lanname = 'plpythonu'"]
+    - result.queries == ['CREATE TRUSTED LANGUAGE "plpythonu"', "UPDATE pg_language SET lanpltrusted = true WHERE lanname = 'plpythonu'"]
 
 - name: postgresql_lang - check that lang exists and it's trusted after previous step
   become_user: "{{ pg_user }}"