IntenseWebs/pgadmin4
3
0
Fork 0
mirror of https://github.com/pgadmin-org/pgadmin4.git synced 2025-02-25 18:55:31 -06:00
Code Issues Packages Projects Releases Wiki Activity

Add the ability to import and export server definitions from a config database. Fixes #3772

Browse Source
This commit is contained in:
Dave Page
2018-11-21 16:09:05 +00:00
parent 3cfc3366d7
commit f0327f5219
8 changed files with 606 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
web/pgadmin/setup/tests/__init__.py Normal file
View File

@@ -0,0 +1,8 @@
##########################################################################
#
# pgAdmin 4 - PostgreSQL Tools
#
# Copyright (C) 2013 - 2018, The pgAdmin Development Team
# This software is released under the PostgreSQL Licence
#
##########################################################################

41
web/pgadmin/setup/tests/servers.json Normal file
View File

@@ -0,0 +1,41 @@
{
"Servers": {
"1": {
"Name": "Minimally Defined Server",
"Group": "Server Group 1",
"Port": 5432,
"Username": "postgres",
"Host": "localhost",
"SSLMode": "prefer",
"MaintenanceDB": "postgres"
},
"2": {
"Name": "Fully Defined Server",
"Group": "Server Group 2",
"Host": "host.domain.com",
"HostAddr": "192.168.1.2",
"Port": 5432,
"MaintenanceDB": "postgres",
"Username": "postgres",
"Role": "my_role_name",
"SSLMode": "require",
"Comment": "This server has every option configured in the JSON",
"DBRestriction": "live_db test_db",
"PassFile": "/path/to/pgpassfile",
"SSLCert": "/path/to/sslcert.crt",
"SSLKey": "/path/to/sslcert.key",
"SSLRootCert": "/path/to/sslroot.crt",
"SSLCrl": "/path/to/sslcrl.crl",
"SSLCompression": 1,
"BGColor": "#ff9900",
"FGColor": "#000000",
"Service": "postgresql-10",
"Timeout": 60,
"UseSSHTunnel": 1,
"TunnelHost": "192.168.1.253",
"TunnelPort": "22",
"TunnelUsername": "username",
"TunnelAuthentication": 0
}
}
}

48
web/pgadmin/setup/tests/test_export_import_servers.py Normal file
View File

@@ -0,0 +1,48 @@
##########################################################################
#
# pgAdmin 4 - PostgreSQL Tools
#
# Copyright (C) 2013 - 2018, The pgAdmin Development Team
# This software is released under the PostgreSQL Licence
#
##########################################################################
from pgadmin.utils.route import BaseTestGenerator
import os
import json
import tempfile
class ImportExportServersTestCase(BaseTestGenerator):
"""
This class validates the import/export servers functionality
"""
scenarios = [
# Fetching the default url for server group node
('Check Server Import/Export', dict())
]
def runTest(self):
path = os.path.dirname(__file__)
setup = os.path.realpath(os.path.join(path, "../../../setup.py"))
# Load the servers
os.system("python %s --load-servers %s 2> %s" %
(setup, os.path.join(path, "servers.json"), os.devnull))
# And dump them again
tf = tempfile.NamedTemporaryFile(delete=False)
os.system("python %s --dump-servers %s 2> %s" %
(setup, tf.name, os.devnull))
# Compare the JSON files, ignoring servers that exist in our
# generated file but not the test config (they are likely
# auto-discovered)
src = json.loads(open(os.path.join(path, "servers.json")).read())
tgt = json.loads(open(tf.name).read())
for server in src["Servers"]:
a = json.dumps(src["Servers"][server], sort_keys=True)
b = json.dumps(tgt["Servers"][server], sort_keys=True)
self.assertTrue(a, b)

20
web/servers.json Normal file
View File

@@ -0,0 +1,20 @@
{
"Servers": {
"1": {
"Name": "OpenStack PEM Server",
"Group": "Servers",
"Host": "172.16.253.234",
"Port": 5432,
"MaintenanceDB": "postgres",
"Username": "postgres",
"SSLMode": "prefer",
"SSLCompression": 1,
"BGColor": "#ffffff",
"FGColor": "#000000",
"Timeout": 0,
"UseSSHTunnel": 0,
"TunnelPort": "22",
"TunnelAuthentication": 0
}
}
}

358
web/setup.py
View File

@@ -10,9 +10,12 @@
"""Perform the initial setup of the application, by creating the auth
and settings database."""
import argparse
import json
import os
import sys
from pgadmin.model import db, Version, SCHEMA_VERSION as CURRENT_SCHEMA_VERSION
from pgadmin.model import db, User, Version, ServerGroup, Server, \
SCHEMA_VERSION as CURRENT_SCHEMA_VERSION
if sys.version_info[0] >= 3:
import builtins
@@ -33,16 +36,311 @@ if sys.path[0] != root:
from pgadmin import create_app
if __name__ == '__main__':
# Configuration settings
import config
from pgadmin.model import SCHEMA_VERSION
from pgadmin.setup import db_upgrade, create_app_data_directory
config.SETTINGS_SCHEMA_VERSION = SCHEMA_VERSION
if "PGADMIN_TESTING_MODE" in os. environ and \
os.environ["PGADMIN_TESTING_MODE"] == "1":
config.SQLITE_PATH = config.TEST_SQLITE_PATH
def add_value(attr_dict, key, value):
"""Add a value to the attribute dict if non-empty.
Args:
attr_dict (dict): The dictionary to add the values to
key (str): The key for the new value
value (str): The value to add
Returns:
The updated attribute dictionary
"""
if value != "" and value is not None:
attr_dict[key] = value
return attr_dict
def dump_servers(args):
"""Dump the server groups and servers.
Args:
args (ArgParser): The parsed command line options
"""
app = create_app()
with app.app_context():
# What user?
if args.user is not None:
dump_user = args.user
else:
dump_user = config.DESKTOP_USER
user = User.query.filter_by(email=dump_user).first()
if user is None:
print("The specified user ID (%s) could not be found." %
dump_user)
sys.exit(1)
user_id = user.id
# Dict to collect the output
object_dict = {}
# Counters
servers_dumped = 0
# Dump servers
servers = Server.query.filter_by(user_id=user_id).all()
server_dict = {}
for server in servers:
if args.servers is None or str(server.id) in args.servers:
# Get the group name
group_name = ServerGroup.query.filter_by(
user_id=user_id, id=server.servergroup_id).first().name
attr_dict = {}
add_value(attr_dict, "Name", server.name)
add_value(attr_dict, "Group", group_name)
add_value(attr_dict, "Host", server.host)
add_value(attr_dict, "HostAddr", server.hostaddr)
add_value(attr_dict, "Port", server.port)
add_value(attr_dict, "MaintenanceDB", server.maintenance_db)
add_value(attr_dict, "Username", server.username)
add_value(attr_dict, "Role", server.role)
add_value(attr_dict, "SSLMode", server.ssl_mode)
add_value(attr_dict, "Comment", server.comment)
add_value(attr_dict, "DBRestriction", server.db_res)
add_value(attr_dict, "PassFile", server.passfile)
add_value(attr_dict, "SSLCert", server.sslcert)
add_value(attr_dict, "SSLKey", server.sslkey)
add_value(attr_dict, "SSLRootCert", server.sslrootcert)
add_value(attr_dict, "SSLCrl", server.sslcrl)
add_value(attr_dict, "SSLCompression", server.sslcompression)
add_value(attr_dict, "BGColor", server.bgcolor)
add_value(attr_dict, "FGColor", server.fgcolor)
add_value(attr_dict, "Service", server.service)
add_value(attr_dict, "Timeout", server.connect_timeout)
add_value(attr_dict, "UseSSHTunnel", server.use_ssh_tunnel)
add_value(attr_dict, "TunnelHost", server.tunnel_host)
add_value(attr_dict, "TunnelPort", server.tunnel_port)
add_value(attr_dict, "TunnelUsername", server.tunnel_username)
add_value(attr_dict, "TunnelAuthentication",
server.tunnel_authentication)
servers_dumped = servers_dumped + 1
server_dict[servers_dumped] = attr_dict
object_dict["Servers"] = server_dict
try:
f = open(args.dump_servers, "w")
except Exception as e:
print("Error opening output file %s: [%d] %s" %
(args.dump_servers, e.errno, e.strerror))
sys.exit(1)
try:
f.write(json.dumps(object_dict, indent=4))
except Exception as e:
print("Error writing output file %s: [%d] %s" %
(args.dump_servers, e.errno, e.strerror))
sys.exit(1)
f.close()
print("Configuration for %s servers dumped to %s." %
(servers_dumped, args.dump_servers))
def load_servers(args):
"""Load server groups and servers.
Args:
args (ArgParser): The parsed command line options
"""
try:
with open(args.load_servers) as f:
data = json.load(f)
except json.decoder.JSONDecodeError as e:
print("Error parsing input file %s: %s" %
(args.load_servers, e))
sys.exit(1)
except Exception as e:
print("Error reading input file %s: [%d] %s" %
(args.load_servers, e.errno, e.strerror))
sys.exit(1)
f.close()
app = create_app()
with app.app_context():
# What user?
if args.user is not None:
dump_user = args.user
else:
dump_user = config.DESKTOP_USER
user = User.query.filter_by(email=dump_user).first()
if user is None:
print("The specified user ID (%s) could not be found." %
dump_user)
sys.exit(1)
user_id = user.id
# Counters
groups_added = 0
servers_added = 0
# Get the server group
groups = ServerGroup.query.all()
def print_summary():
print("Added %d Server Group(s) and %d Server(s)." %
(groups_added, servers_added))
# Loop through the servers...
if "Servers" not in data:
print("'Servers' attribute not found in file '%s'" %
args.load_servers)
print_summary()
sys.exit(1)
for server in data["Servers"]:
obj = data["Servers"][server]
def check_attrib(attrib):
if attrib not in obj:
print("'%s' attribute not found for server '%s'" %
(attrib, server))
print_summary()
sys.exit(1)
check_attrib("Name")
check_attrib("Group")
check_attrib("Port")
check_attrib("Username")
check_attrib("SSLMode")
check_attrib("MaintenanceDB")
if "Host" not in obj and \
"HostAddr" not in obj and \
"Service" not in obj:
print("'Host', 'HostAddr' or 'Service' attribute not found "
"for server '%s'" % server)
print_summary()
sys.exit(1)
# Get the group. Create if necessary
group_id = -1
for g in groups:
if g.name == obj["Group"]:
group_id = g.id
break
if group_id == -1:
new_group = ServerGroup()
new_group.name = obj["Group"]
new_group.user_id = user_id
db.session.add(new_group)
try:
db.session.commit()
except Exception as e:
print("Error creating server group '%s': %s" %
(new_group.name, e))
print_summary()
sys.exit(1)
group_id = new_group.id
groups_added = groups_added + 1
groups = ServerGroup.query.all()
# Create the server
new_server = Server()
new_server.name = obj["Name"]
new_server.servergroup_id = group_id
new_server.user_id = user_id
new_server.host = obj["Host"]
if "HostAddr" in obj:
new_server.hostaddr = obj["HostAddr"]
new_server.port = obj["Port"]
new_server.maintenance_db = obj["MaintenanceDB"]
new_server.username = obj["Username"]
if "Role" in obj:
new_server.role = obj["Role"]
new_server.ssl_mode = obj["SSLMode"]
if "Comment" in obj:
new_server.comment = obj["Comment"]
if "DBRestriction" in obj:
new_server.db_res = obj["DBRestriction"]
if "PassFile" in obj:
new_server.passfile = obj["PassFile"]
if "SSLCert" in obj:
new_server.sslcert = obj["SSLCert"]
if "SSLKey" in obj:
new_server.sslkey = obj["SSLKey"]
if "SSLRootCert" in obj:
new_server.sslrootcert = obj["SSLRootCert"]
if "SSLCrl" in obj:
new_server.sslcrl = obj["SSLCrl"]
if "SSLCompression" in obj:
new_server.sslcompression = obj["SSLCompression"]
if "BGColor" in obj:
new_server.bgcolor = obj["BGColor"]
if "FGColor" in obj:
new_server.fgcolor = obj["FGColor"]
if "Service" in obj:
new_server.service = obj["Service"]
if "Timeout" in obj:
new_server.connect_timeout = obj["Timeout"]
if "UseSSHTunnel" in obj:
new_server.use_ssh_tunnel = obj["UseSSHTunnel"]
if "TunnelHost" in obj:
new_server.tunnel_host = obj["TunnelHost"]
if "TunnelPort" in obj:
new_server.tunnel_port = obj["TunnelPort"]
if "TunnelUsername" in obj:
new_server.tunnel_username = obj["TunnelUsername"]
if "TunnelAuthentication" in obj:
new_server.tunnel_authentication = obj["TunnelAuthentication"]
db.session.add(new_server)
try:
db.session.commit()
except Exception as e:
print("Error creating server '%s': %s" %
(new_server.name, e))
print_summary()
sys.exit(1)
servers_added = servers_added + 1
print_summary()
def setup_db():
"""Setup the configuration database."""
create_app_data_directory(config)
@@ -70,3 +368,43 @@ if __name__ == '__main__':
version = Version.query.filter_by(name='ConfigDB').first()
version.value = CURRENT_SCHEMA_VERSION
db.session.commit()
if __name__ == '__main__':
# Configuration settings
import config
from pgadmin.model import SCHEMA_VERSION
from pgadmin.setup import db_upgrade, create_app_data_directory
parser = argparse.ArgumentParser(description='Setup the pgAdmin config DB')
imp_exp_group = parser.add_mutually_exclusive_group(required=False)
exp_group = imp_exp_group.add_argument_group('Dump server config')
exp_group.add_argument('--dump-servers', metavar="OUTPUT_FILE",
help='Dump the servers in the DB', required=False)
exp_group.add_argument('--servers', metavar="SERVERS", nargs='*',
help='One or more servers to dump', required=False)
imp_group = imp_exp_group.add_argument_group('Load server config')
imp_group.add_argument('--load-servers', metavar="INPUT_FILE",
help='Load servers into the DB', required=False)
imp_exp_group.add_argument('--user', metavar="USER_NAME",
help='Dump/load servers for the specified '
'username', required=False)
args, extra = parser.parse_known_args()
config.SETTINGS_SCHEMA_VERSION = SCHEMA_VERSION
if "PGADMIN_TESTING_MODE" in os.environ and \
os.environ["PGADMIN_TESTING_MODE"] == "1":
config.SQLITE_PATH = config.TEST_SQLITE_PATH
# What to do?
if args.dump_servers is not None:
dump_servers(args)
elif args.load_servers is not None:
load_servers(args)
else:
setup_db()