tests/_helpers.py
def job_metadata_importer(*, db_conn, command, run_id, subcommand=None, status,
                          start_time='2017-08-15 01:15:39.54785+00', extra_metadata={}, **other):
    """Helper function for importing job_metadata data."""
    with db_conn, db_conn.cursor() as cursor:
        cursor.execute("""INSERT INTO job_metadata(command, run_id, subcommand, db_user, command_line,
                                                   start_time, status, extra_metadata)
                              VALUES(%s, %s, %s, 'test_user', %s, %s, %s, %s)
                           RETURNING command, run_id""",
                       [command, run_id, subcommand, ' '.join(sys.argv), start_time, status,
                        json.dumps(extra_metadata)])
        job_metadata_pk = [(x.command, x.run_id) for x in cursor.fetchall()]
    return job_metadata_pk[0]
Helper function for importing job_metadata data.
Helper function for importing job_metadata data.
Helper function for importing job_metadata data .
def job_metadata_importer(*, db_conn, command, run_id, subcommand=None, status,
                          start_time='2017-08-15 01:15:39.54785+00', extra_metadata={}, **other):
    with db_conn, db_conn.cursor() as cursor:
        cursor.execute("""INSERT INTO job_metadata(command, run_id, subcommand, db_user, command_line,
                                                   start_time, status, extra_metadata)
                              VALUES(%s, %s, %s, 'test_user', %s, %s, %s, %s)
                           RETURNING command, run_id""",
                       [command, run_id, subcommand, ' '.join(sys.argv), start_time, status,
                        json.dumps(extra_metadata)])
        job_metadata_pk = [(x.command, x.run_id) for x in cursor.fetchall()]
    return job_metadata_pk[0]
def job_metadata_importer ( * , db_conn , command , run_id , subcommand = None , status , start_time = '2017-08-15 01:15:39.54785+00' , extra_metadata = { } , ** other ) : with db_conn , db_conn . cursor ( ) as cursor : cursor . execute ( """INSERT INTO job_metadata(command, run_id, subcommand, db_user, command_line,
                                                   start_time, status, extra_metadata)
                              VALUES(%s, %s, %s, 'test_user', %s, %s, %s, %s)
                           RETURNING command, run_id""" , [ command , run_id , subcommand , ' ' . join ( sys . argv ) , start_time , status , json . dumps ( extra_metadata ) ] ) job_metadata_pk = [ ( x . command , x . run_id ) for x in cursor . fetchall ( ) ] return job_metadata_pk [ 0 ]
Helper function for importing job_metadata data.
Helper function for importing job_metadata data .
"""Helper function for importing job_metadata data."""
db_conn
command
run_id
subcommand
status
start_time
extra_metadata