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_pymssql_Connection_with(self):
with pymssqlconn() as conn:
cursor = conn.cursor()
cursor.execute("SELECT @@version AS version")
self.assertIsNotNone(conn._conn)
with self.assertRaises(InterfaceError) as context:
self.assertIsNotNone(conn._conn)
self.assertEqual(str(context.exception), "Connection is closed.")
def test_pymssql_Cursor_with(self):
conn = pymssqlconn()
with conn.cursor() as cursor:
cursor.execute("SELECT @@version AS version")
self.assertIsNotNone(conn._conn)
self.assertIsNotNone(cursor)
with self.assertRaises(InterfaceError) as context:
cursor.execute("SELECT @@version AS version")
self.assertEqual(str(context.exception), "Cursor is closed.")
try:
execute_query(mssql, options, host)
except NagiosReturn:
print("%s passed!" % mode)
except Exception as e:
failed += 1
print("%s failed with: %s" % (mode, e))
print('%d/%d tests failed.' % (failed, total))
if __name__ == '__main__':
try:
main()
except pymssql.OperationalError as e:
print(e)
sys.exit(3)
except pymssql.InterfaceError as e:
print(e)
sys.exit(3)
except IOError as e:
print(e)
sys.exit(3)
except NagiosReturn as e:
print(e.message)
sys.exit(e.code)
except Exception as e:
print(type(e))
print("Caught unexpected error. This could be caused by your sysperfinfo not containing the proper entries for this query, and you may delete this service check.")
sys.exit(3)
except CloudError:
# Create sql server
if administrator_login_password is None:
administrator_login_password = str(uuid.uuid4())
server_result = _create_sql_server(
db_context, cmd, resource_group_name, server_name, location, administrator_login,
administrator_login_password, version, tags)
# Create firewall rule to allow for Azure IPs
_create_azure_firewall_rule(db_context, cmd, resource_group_name, server_name)
# Create sql database if it does not exist
_create_sql_database(db_context, cmd, resource_group_name, server_name, database_name, location)
# check ip address(es) of the user and configure firewall rules
sql_errors = (pymssql.InterfaceError, pymssql.OperationalError)
host, user = _configure_firewall_rules(
db_context, sql_errors, cmd, server_result, resource_group_name, server_name, administrator_login,
administrator_login_password, database_name, {'tds_version': '7.0'})
user = '{}@{}'.format(administrator_login, server_name)
host = server_result.fully_qualified_domain_name
# connect to sql server and run some commands
if administrator_login_password is not None:
_run_sql_commands(host, user, administrator_login_password, database_name)
return _form_response(
_create_sql_connection_string(host, user, administrator_login_password, database_name),
host, user,
administrator_login_password if administrator_login_password is not None else '*****'
)
def connect(self):
self.initConnection()
try:
self.connector = pymssql.connect(host="%s:%d" % (self.hostname, self.port), user=self.user, password=self.password, database=self.db, login_timeout=conf.timeout, timeout=conf.timeout)
except (pymssql.InterfaceError, pymssql.OperationalError), msg:
raise SqlmapConnectionException(msg)
self.initCursor()
self.printConnected()
Fetch all remaining rows of a query result, returning them
as a list of dictionaries. An empty list is returned if
no more rows are available. Data can be accessed by 0-based
numeric column index, or by column name.
Raises OperationalError if previous call to execute*() did not
produce any result set or no call was issued yet.
"""
if self._source.get_header() == None:
raise OperationalError, "No data available."
try:
return [ row for row in self._source ]
except _mssql.MssqlDatabaseException, e:
raise OperationalError, e[0]
except _mssql.MssqlDriverException, e:
raise InterfaceError, e[0]