Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_exit_without_active_connection(executor):
quit_handler = MagicMock()
pgspecial = PGSpecial()
pgspecial.register(
quit_handler,
"\\q",
"\\q",
"Quit pgcli.",
arg_type=NO_QUERY,
case_sensitive=True,
aliases=(":q",),
)
with patch.object(executor, "conn", BrokenConnection()):
# we should be able to quit the app, even without active connection
run(executor, "\\q", pgspecial=pgspecial)
quit_handler.assert_called_once()
# an exception should be raised when running a query without active connection
with pytest.raises(psycopg2.InterfaceError):
run(executor, "select 1", pgspecial=pgspecial)
self.pgspecial.register(
self.change_db,
"\\c",
"\\c[onnect] database_name",
"Change to a new database.",
aliases=("use", "\\connect", "USE"),
)
refresh_callback = lambda: self.refresh_completions(persist_priorities="all")
self.pgspecial.register(
self.quit,
"\\q",
"\\q",
"Quit pgcli.",
arg_type=NO_QUERY,
case_sensitive=True,
aliases=(":q",),
)
self.pgspecial.register(
self.quit,
"quit",
"quit",
"Quit pgcli.",
arg_type=NO_QUERY,
case_sensitive=False,
aliases=("exit",),
)
self.pgspecial.register(
refresh_callback,
"\\#",
"\\#",
def register_special_commands(self):
self.pgspecial.register(
self.change_db, '\\c', '\\c[onnect] database_name',
'Change to a new database.', aliases=('use', '\\connect', 'USE'))
refresh_callback = lambda: self.refresh_completions(
persist_priorities='all')
self.pgspecial.register(refresh_callback, '\\#', '\\#',
'Refresh auto-completions.', arg_type=NO_QUERY)
self.pgspecial.register(refresh_callback, '\\refresh', '\\refresh',
'Refresh auto-completions.', arg_type=NO_QUERY)
self.pgspecial.register(self.execute_from_file, '\\i', '\\i filename',
'Execute commands from file.')
self.pgspecial.register(self.write_to_file, '\\o', '\\o [filename]',
'Send all query results to file.')
self.pgspecial.register(self.info_connection, '\\conninfo',
'\\conninfo', 'Get connection details')
self.pgspecial.register(self.change_table_format, '\\T', '\\T [format]',
'Change the table format used to output results')
def register_special_commands(self):
self.pgspecial.register(
self.change_db, '\\c', '\\c[onnect] database_name',
'Change to a new database.', aliases=('use', '\\connect', 'USE'))
refresh_callback = lambda: self.refresh_completions(
persist_priorities='all')
self.pgspecial.register(refresh_callback, '\\#', '\\#',
'Refresh auto-completions.', arg_type=NO_QUERY)
self.pgspecial.register(refresh_callback, '\\refresh', '\\refresh',
'Refresh auto-completions.', arg_type=NO_QUERY)
self.pgspecial.register(self.execute_from_file, '\\i', '\\i filename',
'Execute commands from file.')
self.pgspecial.register(self.write_to_file, '\\o', '\\o [filename]',
'Send all query results to file.')
self.pgspecial.register(self.info_connection, '\\conninfo',
'\\conninfo', 'Get connection details')
self.pgspecial.register(self.change_table_format, '\\T', '\\T [format]',
'Change the table format used to output results')
case_sensitive=False,
aliases=("exit",),
)
self.pgspecial.register(
refresh_callback,
"\\#",
"\\#",
"Refresh auto-completions.",
arg_type=NO_QUERY,
)
self.pgspecial.register(
refresh_callback,
"\\refresh",
"\\refresh",
"Refresh auto-completions.",
arg_type=NO_QUERY,
)
self.pgspecial.register(
self.execute_from_file, "\\i", "\\i filename", "Execute commands from file."
)
self.pgspecial.register(
self.write_to_file,
"\\o",
"\\o [filename]",
"Send all query results to file.",
)
self.pgspecial.register(
self.info_connection, "\\conninfo", "\\conninfo", "Get connection details"
)
self.pgspecial.register(
self.change_table_format,
"\\T",