Add Reverse Engineered SQL tests for Collations. Fixes #4464
This also adds the ability to test the msql output in ALTER steps.pull/25/head
parent
6b5ca07715
commit
79e6f4c008
|
@ -23,6 +23,7 @@ Housekeeping
|
|||
| `Issue #4454 <https://redmine.postgresql.org/issues/4454>`_ - Add Reverse Engineered SQL tests for FTS Configurations.
|
||||
| `Issue #4456 <https://redmine.postgresql.org/issues/4456>`_ - Add Reverse Engineered SQL tests for Packages.
|
||||
| `Issue #4460 <https://redmine.postgresql.org/issues/4460>`_ - Add Reverse Engineered SQL tests for FTS Dictionaries.
|
||||
| `Issue #4464 <https://redmine.postgresql.org/issues/4464>`_ - Add Reverse Engineered SQL tests for Collations.
|
||||
|
||||
Bug fixes
|
||||
*********
|
||||
|
|
|
@ -511,24 +511,22 @@ class CollationView(PGChildNodeView):
|
|||
SQL = render_template("/".join([self.template_path,
|
||||
'get_name.sql']),
|
||||
scid=scid, coid=coid)
|
||||
status, name = self.conn.execute_scalar(SQL)
|
||||
status, res = self.conn.execute_dict(SQL)
|
||||
if not status:
|
||||
return internal_server_error(errormsg=name)
|
||||
return internal_server_error(errormsg=res)
|
||||
|
||||
if name is None:
|
||||
return make_json_response(
|
||||
success=0,
|
||||
errormsg=gettext(
|
||||
'Error: Object not found.'
|
||||
),
|
||||
info=gettext(
|
||||
'The specified collation could not be found.\n'
|
||||
)
|
||||
)
|
||||
if len(res['rows']) == 0:
|
||||
return gone(gettext(
|
||||
"Could not find the collation object in the database."
|
||||
))
|
||||
|
||||
data = res['rows'][0]
|
||||
|
||||
SQL = render_template("/".join([self.template_path,
|
||||
'delete.sql']),
|
||||
name=name, cascade=cascade,
|
||||
name=data['name'],
|
||||
nspname=data['schema'],
|
||||
cascade=cascade,
|
||||
conn=self.conn)
|
||||
status, res = self.conn.execute_scalar(SQL)
|
||||
if not status:
|
||||
|
@ -700,7 +698,8 @@ class CollationView(PGChildNodeView):
|
|||
|
||||
sql_header += render_template("/".join([self.template_path,
|
||||
'delete.sql']),
|
||||
name=data['name'])
|
||||
name=data['name'],
|
||||
nspname=data['schema'])
|
||||
SQL = sql_header + '\n\n' + SQL.strip('\n')
|
||||
|
||||
return ajax_response(response=SQL)
|
||||
|
|
|
@ -1 +1 @@
|
|||
DROP COLLATION {{name}}{% if cascade%} CASCADE{% endif %};
|
||||
DROP COLLATION {{ conn|qtIdent(nspname, name) }}{% if cascade%} CASCADE{% endif %};
|
||||
|
|
|
@ -1,4 +1,4 @@
|
|||
SELECT concat(quote_ident(nspname), '.', quote_ident(collname)) AS name
|
||||
SELECT nspname AS schema, collname AS name
|
||||
FROM pg_collation c, pg_namespace n
|
||||
WHERE c.collnamespace = n.oid AND
|
||||
n.oid = {{ scid }}::oid AND
|
||||
|
|
|
@ -0,0 +1,12 @@
|
|||
-- Collation: Cl1_$%{}[]()&*^!@"'`\/#a;
|
||||
|
||||
-- DROP COLLATION testschema."Cl1_$%{}[]()&*^!@""'`\/#a";
|
||||
|
||||
CREATE COLLATION testschema."Cl1_$%{}[]()&*^!@""'`\/#a"
|
||||
(LC_COLLATE = 'C', LC_CTYPE = 'C');
|
||||
|
||||
ALTER COLLATION testschema."Cl1_$%{}[]()&*^!@""'`\/#a"
|
||||
OWNER TO <OWNER>;
|
||||
|
||||
COMMENT ON COLLATION testschema."Cl1_$%{}[]()&*^!@""'`\/#a"
|
||||
IS 'Description for alter';
|
|
@ -0,0 +1,12 @@
|
|||
-- Collation: Cl1_$%{}[]()&*^!@"'`\/#;
|
||||
|
||||
-- DROP COLLATION testschema."Cl1_$%{}[]()&*^!@""'`\/#";
|
||||
|
||||
CREATE COLLATION testschema."Cl1_$%{}[]()&*^!@""'`\/#"
|
||||
(LC_COLLATE = 'C', LC_CTYPE = 'C');
|
||||
|
||||
ALTER COLLATION testschema."Cl1_$%{}[]()&*^!@""'`\/#"
|
||||
OWNER TO <OWNER>;
|
||||
|
||||
COMMENT ON COLLATION testschema."Cl1_$%{}[]()&*^!@""'`\/#"
|
||||
IS 'Description';
|
|
@ -0,0 +1,5 @@
|
|||
COMMENT ON COLLATION testschema."Cl1_$%{}[]()&*^!@""'`\/#"
|
||||
IS 'Description for alter';
|
||||
|
||||
ALTER COLLATION testschema."Cl1_$%{}[]()&*^!@""'`\/#"
|
||||
RENAME TO "Cl1_$%{}[]()&*^!@""'`\/#a";
|
|
@ -0,0 +1,37 @@
|
|||
{
|
||||
"scenarios": [
|
||||
{
|
||||
"type": "create",
|
||||
"name": "Create Collation",
|
||||
"endpoint": "NODE-collation.obj",
|
||||
"sql_endpoint": "NODE-collation.sql_id",
|
||||
"data": {
|
||||
"name": "Cl1_$%{}[]()&*^!@\"'`\\/#",
|
||||
"schema": "testschema",
|
||||
"copy_collation": "pg_catalog.\"C\"",
|
||||
"description": "Description"
|
||||
},
|
||||
"expected_sql_file": "create_collation.sql"
|
||||
}, {
|
||||
"type": "alter",
|
||||
"name": "Alter Collation",
|
||||
"endpoint": "NODE-collation.obj_id",
|
||||
"sql_endpoint": "NODE-collation.sql_id",
|
||||
"msql_endpoint": "NODE-collation.msql_id",
|
||||
"data": {
|
||||
"name": "Cl1_$%{}[]()&*^!@\"'`\\/#a",
|
||||
"schema": "testschema",
|
||||
"description": "Description for alter"
|
||||
},
|
||||
"expected_sql_file": "alter_collation.sql",
|
||||
"expected_msql_file": "msql_collation.sql"
|
||||
}, {
|
||||
"type": "delete",
|
||||
"name": "Drop Collation",
|
||||
"endpoint": "NODE-collation.delete_id",
|
||||
"data": {
|
||||
"name": "Cl1_$%{}[]()&*^!@\"'`\\/#a"
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
|
@ -9,6 +9,7 @@
|
|||
from __future__ import print_function
|
||||
import json
|
||||
import os
|
||||
import urllib
|
||||
import traceback
|
||||
from flask import url_for
|
||||
import regression
|
||||
|
@ -222,6 +223,19 @@ class ReverseEngineeredSQLTestCases(BaseTestGenerator):
|
|||
continue
|
||||
elif 'type' in scenario and scenario['type'] == 'alter':
|
||||
# Get the url and create the specific node.
|
||||
|
||||
# If msql_endpoint exists then validate the modified sql
|
||||
if 'msql_endpoint' in scenario\
|
||||
and scenario['msql_endpoint']:
|
||||
if not self.check_msql(scenario, object_id):
|
||||
print_msg = scenario['name']
|
||||
if 'expected_msql_file' in scenario:
|
||||
print_msg += " Expected MSQL File:" + scenario[
|
||||
'expected_msql_file']
|
||||
print_msg = print_msg + "... FAIL"
|
||||
print(print_msg)
|
||||
continue
|
||||
|
||||
alter_url = self.get_url(scenario['endpoint'], object_id)
|
||||
response = self.tester.put(alter_url,
|
||||
data=json.dumps(scenario['data']),
|
||||
|
@ -294,6 +308,66 @@ class ReverseEngineeredSQLTestCases(BaseTestGenerator):
|
|||
|
||||
return False, None
|
||||
|
||||
def check_msql(self, scenario, object_id):
|
||||
"""
|
||||
This function is used to check the modified SQL.
|
||||
:param scenario:
|
||||
:param object_id:
|
||||
:return:
|
||||
"""
|
||||
|
||||
msql_url = self.get_url(scenario['msql_endpoint'],
|
||||
object_id)
|
||||
|
||||
params = urllib.parse.urlencode(scenario['data'])
|
||||
url = msql_url + "?%s" % params
|
||||
response = self.tester.get(url,
|
||||
follow_redirects=True)
|
||||
try:
|
||||
self.assertEquals(response.status_code, 200)
|
||||
except Exception as e:
|
||||
self.final_test_status = False
|
||||
print(scenario['name'] + "... FAIL")
|
||||
traceback.print_exc()
|
||||
|
||||
resp = json.loads(response.data)
|
||||
resp_sql = resp['data']
|
||||
|
||||
# Remove first and last double quotes
|
||||
if resp_sql.startswith('"') and resp_sql.endswith('"'):
|
||||
resp_sql = resp_sql[1:-1]
|
||||
resp_sql = resp_sql.rstrip()
|
||||
|
||||
# Check if expected sql is given in JSON file or path of the output
|
||||
# file is given
|
||||
if 'expected_msql_file' in scenario:
|
||||
output_file = os.path.join(self.test_folder,
|
||||
scenario['expected_msql_file'])
|
||||
|
||||
if os.path.exists(output_file):
|
||||
fp = open(output_file, "r")
|
||||
# Used rstrip to remove trailing \n
|
||||
sql = fp.read().rstrip()
|
||||
# Replace place holder <owner> with the current username
|
||||
# used to connect to the database
|
||||
if 'username' in self.server:
|
||||
sql = sql.replace(self.JSON_PLACEHOLDERS['owner'],
|
||||
self.server['username'])
|
||||
try:
|
||||
self.assertEquals(sql, resp_sql)
|
||||
except Exception as e:
|
||||
self.final_test_status = False
|
||||
traceback.print_exc()
|
||||
return False
|
||||
else:
|
||||
try:
|
||||
self.assertFalse("Expected SQL File not found")
|
||||
except Exception as e:
|
||||
self.final_test_status = False
|
||||
traceback.print_exc()
|
||||
return False
|
||||
return True
|
||||
|
||||
def check_re_sql(self, scenario, object_id):
|
||||
"""
|
||||
This function is used to get the reverse engineering SQL.
|
||||
|
@ -301,11 +375,14 @@ class ReverseEngineeredSQLTestCases(BaseTestGenerator):
|
|||
:param object_id:
|
||||
:return:
|
||||
"""
|
||||
|
||||
sql_url = self.get_url(scenario['sql_endpoint'], object_id)
|
||||
response = self.tester.get(sql_url)
|
||||
|
||||
try:
|
||||
self.assertEquals(response.status_code, 200)
|
||||
except Exception as e:
|
||||
|
||||
self.final_test_status = False
|
||||
traceback.print_exc()
|
||||
return False
|
||||
|
|
Loading…
Reference in New Issue