tests/_fixtures.py
def classification_data(db_conn, request):
    """Fixture to bootstrap DB with known classification_state data."""
    req_class_state_file = request.param
    assert req_class_state_file is not None

    class_state_import_sql = """
        CREATE TEMP TABLE classification_state_temp (LIKE classification_state INCLUDING DEFAULTS);
        ALTER TABLE classification_state_temp ALTER COLUMN run_id SET DEFAULT 1;
        ALTER TABLE classification_state_temp ALTER COLUMN virt_imei_shard DROP NOT NULL;
        COPY classification_state_temp(
            imei_norm,
            cond_name,
            start_date,
            end_date,
            block_date
        )
        FROM
            STDIN WITH CSV HEADER DELIMITER AS ','
    """
    with db_conn, db_conn.cursor() as cur:
        fn = req_class_state_file
        with open(path.join(path.abspath(path.dirname(__file__) + '/unittest_data'), fn)) as f:
            cur.copy_expert(sql=class_state_import_sql, file=f)
            cur.execute("""INSERT INTO classification_state(run_id,
                                                            imei_norm,
                                                            cond_name,
                                                            start_date,
                                                            end_date,
                                                            block_date,
                                                            virt_imei_shard)
                                SELECT run_id,
                                       imei_norm,
                                       cond_name,
                                       start_date,
                                       end_date,
                                       block_date,
                                       calc_virt_imei_shard(imei_norm)
                                  FROM classification_state_temp""")

    yield req_class_state_file
Fixture to bootstrap DB with known classification_state data.
Fixture to bootstrap DB with known classification_state data.
Fixture to bootstrap DB with known classification_state data .
def classification_data(db_conn, request):
    req_class_state_file = request.param
    assert req_class_state_file is not None
    class_state_import_sql = """
        CREATE TEMP TABLE classification_state_temp (LIKE classification_state INCLUDING DEFAULTS);
        ALTER TABLE classification_state_temp ALTER COLUMN run_id SET DEFAULT 1;
        ALTER TABLE classification_state_temp ALTER COLUMN virt_imei_shard DROP NOT NULL;
        COPY classification_state_temp(
            imei_norm,
            cond_name,
            start_date,
            end_date,
            block_date
        )
        FROM
            STDIN WITH CSV HEADER DELIMITER AS ','
    """
    with db_conn, db_conn.cursor() as cur:
        fn = req_class_state_file
        with open(path.join(path.abspath(path.dirname(__file__) + '/unittest_data'), fn)) as f:
            cur.copy_expert(sql=class_state_import_sql, file=f)
            cur.execute("""INSERT INTO classification_state(run_id,
                                                            imei_norm,
                                                            cond_name,
                                                            start_date,
                                                            end_date,
                                                            block_date,
                                                            virt_imei_shard)
                                SELECT run_id,
                                       imei_norm,
                                       cond_name,
                                       start_date,
                                       end_date,
                                       block_date,
                                       calc_virt_imei_shard(imei_norm)
                                  FROM classification_state_temp""")
    yield req_class_state_file
def classification_data ( db_conn , request ) : req_class_state_file = request . param assert req_class_state_file is not None class_state_import_sql = """
        CREATE TEMP TABLE classification_state_temp (LIKE classification_state INCLUDING DEFAULTS);
        ALTER TABLE classification_state_temp ALTER COLUMN run_id SET DEFAULT 1;
        ALTER TABLE classification_state_temp ALTER COLUMN virt_imei_shard DROP NOT NULL;
        COPY classification_state_temp(
            imei_norm,
            cond_name,
            start_date,
            end_date,
            block_date
        )
        FROM
            STDIN WITH CSV HEADER DELIMITER AS ','
    """ with db_conn , db_conn . cursor ( ) as cur : fn = req_class_state_file with open ( path . join ( path . abspath ( path . dirname ( __file__ ) + '/unittest_data' ) , fn ) ) as f : cur . copy_expert ( sql = class_state_import_sql , file = f ) cur . execute ( """INSERT INTO classification_state(run_id,
                                                            imei_norm,
                                                            cond_name,
                                                            start_date,
                                                            end_date,
                                                            block_date,
                                                            virt_imei_shard)
                                SELECT run_id,
                                       imei_norm,
                                       cond_name,
                                       start_date,
                                       end_date,
                                       block_date,
                                       calc_virt_imei_shard(imei_norm)
                                  FROM classification_state_temp""" ) yield req_class_state_file
Fixture to bootstrap DB with known classification_state data.
Fixture to bootstrap DB with known classification_state data .
"""Fixture to bootstrap DB with known classification_state data."""
db_conn
request