Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/pytest_plugin.py
Show First 20 Lines • Show All 133 Lines • ▼ Show 20 Lines | def db_reset(self): | ||||
dbname=self.db_name, user=self.user, host=self.host, port=self.port, | dbname=self.db_name, user=self.user, host=self.host, port=self.port, | ||||
) as cnx: | ) as cnx: | ||||
with cnx.cursor() as cur: | with cnx.cursor() as cur: | ||||
cur.execute( | cur.execute( | ||||
"SELECT table_name FROM information_schema.tables " | "SELECT table_name FROM information_schema.tables " | ||||
"WHERE table_schema = %s", | "WHERE table_schema = %s", | ||||
("public",), | ("public",), | ||||
) | ) | ||||
tables = set(table for (table,) in cur.fetchall()) | tables = set(table for (table,) in cur.fetchall()) - {"dbversion"} | ||||
for table in tables: | for table in tables: | ||||
cur.execute("truncate table %s cascade" % table) | cur.execute("truncate table %s cascade" % table) | ||||
cur.execute( | cur.execute( | ||||
"SELECT sequence_name FROM information_schema.sequences " | "SELECT sequence_name FROM information_schema.sequences " | ||||
"WHERE sequence_schema = %s", | "WHERE sequence_schema = %s", | ||||
("public",), | ("public",), | ||||
) | ) | ||||
▲ Show 20 Lines • Show All 50 Lines • Show Last 20 Lines |