update how workspaces are created so tests can utilize functionality

main
Marshall Hallenbeck 2023-03-20 21:14:07 -04:00 committed by Marshall Hallenbeck
parent 64102b35db
commit 502e0b0eea
1 changed files with 30 additions and 0 deletions

View File

@ -455,6 +455,36 @@ def delete_workspace(workspace_name):
shutil.rmtree(os.path.join(WORKSPACE_DIR, workspace_name))
def create_workspace(workspace_name, p_loader, protocols):
os.mkdir(os.path.join(WORKSPACE_DIR, workspace_name))
for protocol in protocols.keys():
try:
protocol_object = p_loader.load_protocol(protocols[protocol]['dbpath'])
except KeyError:
continue
proto_db_path = os.path.join(WORKSPACE_DIR, workspace_name, protocol + '.db')
if not os.path.exists(proto_db_path):
print('[*] Initializing {} protocol database'.format(protocol.upper()))
conn = sqlite3.connect(proto_db_path)
c = conn.cursor()
# try to prevent some weird sqlite I/O errors
c.execute('PRAGMA journal_mode = OFF')
c.execute('PRAGMA foreign_keys = 1')
getattr(protocol_object, 'database').db_schema(c)
# commit the changes and close everything off
conn.commit()
conn.close()
def delete_workspace(workspace_name):
shutil.rmtree(os.path.join(WORKSPACE_DIR, workspace_name))
def initialize_db(logger):
if not os.path.exists(os.path.join(WS_PATH, 'default')):
logger.info('Creating default workspace')