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_mysql(self):
self.assertEqual(
'select "c1","c2" from testSchema.t1 limit 10',
MySQLExplorer._get_sample_query(
self.schema,
self.schema.get_children()[0],
self.schema.get_children()[0].get_children(),
),
def explorer(self):
return MySQLExplorer(self.namespace)
def setUp(self):
self.explorer = MySQLExplorer(
Namespace(
host="127.0.0.1",
user="piiuser",
password="p11secret",
database="piidb",
include_schema=(),
exclude_schema=(),
include_table=(),
exclude_table=(),
catalog=None,
)
def test_mysql(self):
self.assertEqual(
'select "c1","c2" from testSchema.t1',
MySQLExplorer._get_select_query(
self.schema,
self.schema.get_children()[0],
self.schema.get_children()[0].get_children(),
),
exclude_table=(),
catalog=None,
)
@classmethod
def get_connection(cls):
return pymysql.connect(
host="127.0.0.1", user="piiuser", password="p11secret", database="piidb"
)
@property
def explorer(self):
return MySQLExplorer(self.namespace)
class SmallSampleMysqlExplorer(MySQLExplorer):
@property
def small_table_max(self):
return 5
class SmallSampleMySqlExplorerTest(VanillaMySqlExplorerTest):
@property
def explorer(self):
return SmallSampleMysqlExplorer(self.namespace)
class VanillaPGExplorerTest(CommonSampleDataTestCases.CommonSampleDataTests):
@property
def deep_scan_result(self):
return [
["public", "sample", "address", True],
def setUp(self):
self.explorer = MySQLExplorer(
Namespace(
host="127.0.0.1",
user="piiuser",
password="p11secret",
database="piidb",
include_schema=(),
exclude_schema=(),
include_table=(),
exclude_table=(),
catalog=None,
)
def __init__(self, ns):
super(MySQLExplorer, self).__init__(ns)
self._mysql_database = (
ns.database if "database" in vars(ns) and ns.database is not None else None
)
def factory(cls, ns):
logging.debug("Relational Db Factory entered")
explorer = None
if ns.connection_type == "mysql":
explorer = MySQLExplorer(ns)
elif ns.connection_type == "postgres" or ns.connection_type == "redshift":
explorer = PostgreSQLExplorer(ns)
elif ns.connection_type == "oracle":
explorer = OracleExplorer(ns)
assert explorer is not None
return explorer