cudrop_test.TestDataCleanupScripts Class Reference

Inheritance diagram for cudrop_test.TestDataCleanupScripts:

Public Member Functions
def	setUp (self)
def	test_misc_counts (self)
def	test_drop_cu_hist (self)
def	test_trans_hdr_dtl (self)
def	test_drop_cu_misc (self)
def	test_cuauditadmin (self)
def	test_cuuser (self)
def	test_audituser (self)
def	tearDown (self)

Detailed Description

Test suite for database cleanup scripts

Definition at line 207 of file cudrop_test.py.

Member Function Documentation

setUp()

def cudrop_test.TestDataCleanupScripts.setUp

(

self

)

Load fixture

Definition at line 210 of file cudrop_test.py.

    def setUp(self):
        """Load fixture"""
        fixture_cleanup()
        fixture_load()

tearDown()

def cudrop_test.TestDataCleanupScripts.tearDown

(

self

)

Cleanup fixture

Definition at line 790 of file cudrop_test.py.

    def tearDown(self):
        """Cleanup fixture"""
        fixture_cleanup()

test_audituser()

def cudrop_test.TestDataCleanupScripts.test_audituser

(

self

)

Test <cu>audituser cleanup script

Cleanup Script:
    - cudropcuaudituser.pl

Definition at line 718 of file cudrop_test.py.

    def test_audituser(self):
        """Test <cu>audituser cleanup script

        Cleanup Script:
            - cudropcuaudituser.pl
        """
        run_cmd_subprocess("{} {} {} clean SCRUBCU memdata {} {}".format(
            "/usr/bin/python3",
            "/opt/odyssey/tools/bin/ody_migr_executor.py",
            TESTING_SERVER,
            TEST_USERNAME,
            TEST_PASSWORD
        ))

        run_cmd_subprocess("{} {} {} migrate SCRUBCU memdata {} {}".format(
            "/usr/bin/python3",
            "/opt/odyssey/tools/bin/ody_migr_executor.py",
            TESTING_SERVER,
            TEST_USERNAME,
            TEST_PASSWORD
        ))

        time.sleep(FIXTURE_READY_TIME)

        with pg_handler.PGSession() as conn:
            with conn.cursor(cursor_factory=DictCursor) as dict_cur:
                cuauditadmin_transaction = PgTransaction(conn, dict_cur)

                dict_cur.execute("SELECT * from scrubcuaudituser "
                                 "limit {}".format(LIMIT_SELECT))

                initial_records = dict_cur.fetchall()
                initial_records = [dict(rec)
                                   for rec in initial_records]

                cuauditadmin_transaction(
                    DELETE,
                    "scrubcuaudituser"
                )
                cuauditadmin_transaction.commit()

                # insert fixture data (to be deleted + recent data)
                recent_records = []
                for i, rec in enumerate(initial_records):
                    rec_recent = rec.copy()
                    rec_recent["auditdate"] = TEST_DATE_NOW
                    rec_recent["accountnumber"] += chr(i * 2 + 97)
                    recent_records.append(rec_recent)

                    rec_recent_del = rec.copy()
                    rec_recent_del["auditdate"] = '20140801'
                    rec_recent_del["accountnumber"] += chr(i * 2 + 1 + 97)
                    recent_records.append(rec_recent_del)

                cuauditadmin_transaction(
                    INSERT,
                    "scrubcuaudituser",
                    collection=recent_records
                )
                cuauditadmin_transaction.commit()

                run_cmd_subprocess("{}/{} SCRUBCU".format(
                    PERL_SCRIPTS_ROOT,
                    table_to_perlscript_map["cuaudituser"]))

                after_cleanup_count = cuauditadmin_transaction(
                    SELECT_RETURN,
                    "scrubcuaudituser",
                    return_only_count=True
                )
                self.assertEqual(after_cleanup_count, len(recent_records) / 2)

test_cuauditadmin()

def cudrop_test.TestDataCleanupScripts.test_cuauditadmin

(

self

)

Test cuauditadmin cleanup script

Cleanup Script:
    - cudropcuauditadmin.pl

Definition at line 565 of file cudrop_test.py.

    def test_cuauditadmin(self):
        """Test cuauditadmin cleanup script

        Cleanup Script:
            - cudropcuauditadmin.pl
        """
        run_cmd_subprocess("{} {} {} clean SCRUBCU admin {} {}".format(
            "/usr/bin/python3",
            "/opt/odyssey/tools/bin/ody_migr_executor.py",
            TESTING_SERVER,
            TEST_USERNAME,
            TEST_PASSWORD
        ))

        run_cmd_subprocess("{} {} {} migrate SCRUBCU admin {} {}".format(
            "/usr/bin/python3",
            "/opt/odyssey/tools/bin/ody_migr_executor.py",
            TESTING_SERVER,
            TEST_USERNAME,
            TEST_PASSWORD
        ))

        time.sleep(FIXTURE_READY_TIME)

        with pg_handler.PGSession() as conn:
            with conn.cursor(cursor_factory=DictCursor) as dict_cur:
                cuauditadmin_transaction = PgTransaction(conn, dict_cur)

                dict_cur.execute("SELECT * from cuauditadmin where "
                                 "auditdate<'20170801' limit {}".format(
                                     LIMIT_SELECT))

                to_be_deleted_records = dict_cur.fetchall()
                to_be_deleted_records = [dict(rec)
                                         for rec in to_be_deleted_records]
                # initial_to_be_deleted_count = min(
                #     len(to_be_deleted_records), LIMIT_SELECT)

                dict_cur.execute("delete from cuauditadmin")
                cuauditadmin_transaction.commit()

                # insert fixture data (to be deleted + recent data)
                recent_records = []
                for i, rec in enumerate(to_be_deleted_records):
                    rec_recent = rec.copy()
                    rec_recent["auditdate"] = TEST_DATE_NOW
                    rec_recent["user_name"] = rec_recent["user_name"] + str(i)
                    recent_records.append(rec_recent)

                cuauditadmin_transaction(
                    INSERT,
                    "cuauditadmin",
                    collection=to_be_deleted_records + recent_records
                )
                cuauditadmin_transaction.commit()

                run_cmd_subprocess("{}/{}".format(
                    PERL_SCRIPTS_ROOT,
                    table_to_perlscript_map["cuauditadmin"]))

                after_cleanup_count = cuauditadmin_transaction(
                    SELECT_RETURN,
                    "cuauditadmin",
                    return_only_count=True
                )

                self.assertEqual(after_cleanup_count, len(recent_records))

test_cuuser()

def cudrop_test.TestDataCleanupScripts.test_cuuser

(

self

)

Verify deletion of <cu>user (only) table

Cleanup Scripts:
    - cudropcuusers.pl

Definition at line 633 of file cudrop_test.py.

    def test_cuuser(self):
        """Verify deletion of <cu>user (only) table

        Cleanup Scripts:
            - cudropcuusers.pl
        """
        run_cmd_subprocess("{} {} {} clean SCRUBCU memdata {} {}".format(
            "/usr/bin/python3",
            "/opt/odyssey/tools/bin/ody_migr_executor.py",
            TESTING_SERVER,
            TEST_USERNAME,
            TEST_PASSWORD
        ))
        run_cmd_subprocess("{} {} {} migrate SCRUBCU memdata {} {}".format(
            "/usr/bin/python3",
            "/opt/odyssey/tools/bin/ody_migr_executor.py",
            TESTING_SERVER,
            TEST_USERNAME,
            TEST_PASSWORD
        ))

        time.sleep(FIXTURE_READY_TIME)

        with pg_handler.PGSession() as conn:
            with conn.cursor(cursor_factory=DictCursor) as dict_cur:
                cuuser_transaction = PgTransaction(conn, dict_cur)

                # 1. prepare fixture to delete
                dict_cur.execute("SELECT * from scrubcuuser "
                                 "limit {}".format(LIMIT_SELECT))

                initial_records = dict_cur.fetchall()
                initial_records = [dict(rec)
                                   for rec in initial_records]

                # make sure we have a few records in scrubcuuser already
                if len(initial_records) != 0:

                    # empty the table
                    cuuser_transaction(
                        DELETE,
                        "scrubcuuser"
                    )
                    cuuser_transaction.commit()

                    new_records = []
                    # add one record that should not be deleted
                    record_not_to_delete = initial_records[0].copy()
                    record_not_to_delete["pwchange"] = TEST_DATE_NOW
                    record_not_to_delete["passwd"] = 's0m3^@l!dPassW0rD'
                    if "user_id" in record_not_to_delete:
                        del record_not_to_delete["user_id"]
                    record_not_to_delete["user_name"] += str(SERIAL_OFFSET)
                    new_records.append(record_not_to_delete)

                    # prepare records eligible for deletion
                    for i, rec in enumerate(initial_records):
                        updated_rec = rec.copy()
                        updated_rec["pwchange"] = '20160801'
                        updated_rec["passwd"] = 'NULL PASSWORD'
                        # handled by auto-increment
                        if "user_id" in updated_rec:
                            del updated_rec["user_id"]
                        updated_rec["user_name"] += str(
                            SERIAL_OFFSET + i * 5 + 1)
                        new_records.append(updated_rec)

                    cuuser_transaction(
                        INSERT,
                        "scrubcuuser",
                        collection=new_records
                    )
                    cuuser_transaction.commit()

                    run_cmd_subprocess("{}/{} SCRUBCU".format(
                        PERL_SCRIPTS_ROOT,
                        table_to_perlscript_map["cuuser"]))

                    count_after_cleanup = cuuser_transaction(
                        SELECT_RETURN,
                        "scrubcuuser",
                        return_only_count=True
                    )
                    self.assertEqual(count_after_cleanup, 1)

test_drop_cu_hist()

def cudrop_test.TestDataCleanupScripts.test_drop_cu_hist

(

self

)

Test cleanupscripts for history tables:

    - scrubculoanhistory
    - scrubcuaccounthistory

Cleanup Scripts:
    - cudrophist.pl

Definition at line 264 of file cudrop_test.py.

    def test_drop_cu_hist(self):
        """Test cleanupscripts for history tables:

            - scrubculoanhistory
            - scrubcuaccounthistory

        Cleanup Scripts:
            - cudrophist.pl
        """
        with pg_handler.PGSession() as conn:
            with conn.cursor(cursor_factory=DictCursor) as dict_cur:
                hist_transaction = PgTransaction(conn, dict_cur)
                # pre-cleanup
                for tbl_account, tbl_loan in [("scrubcuaccounthistory",
                                               "scrubculoanhistory")]:

                    # 1. get the initial records from setUp
                    initial_records_loan = hist_transaction(
                        SELECT_RETURN,
                        tbl_loan
                    )

                    initial_records_account = hist_transaction(
                        SELECT_RETURN,
                        tbl_account
                    )

                    assert len(initial_records_account) >= NUM_OF_NEW_RECORDS
                    assert len(initial_records_loan) >= NUM_OF_NEW_RECORDS

                    if VERBOSE:
                        print("\ninitial records [{}]\n".format("history"))
                        for rec in initial_records_loan +\
                                initial_records_account:
                            print(rec["date"])

                    # 2. add NUM_OF_NEW_RECORDS with today's updatedate (these
                    # NUM_OF_NEW_RECORDS records must not be scraped by the db
                    # cleanup cronjob cudropcurdcstatus.pl)

                    latest_records_account = []
                    latest_records_loan = []

                    for i, record in enumerate(initial_records_account[
                                               :NUM_OF_NEW_RECORDS]):
                        latest_rec = record.copy()
                        latest_rec["date"] = TEST_DATE_NOW

                        latest_rec["accountnumber"] = int(
                            latest_rec["accountnumber"]) + SERIAL_OFFSET + i

                        latest_rec["tracenumber"] = int(
                            latest_rec["tracenumber"]) + SERIAL_OFFSET + i

                        latest_rec["certnumber"] += SERIAL_OFFSET + i

                        latest_records_account.append(latest_rec)

                    for record in initial_records_loan[:NUM_OF_NEW_RECORDS]:
                        latest_rec = record.copy()

                        latest_rec["date"] = TEST_DATE_NOW

                        latest_rec["accountnumber"] = int(
                            latest_rec["accountnumber"]) + SERIAL_OFFSET + i

                        latest_rec["tracenumber"] = int(
                            latest_rec["tracenumber"]) + SERIAL_OFFSET + i

                        latest_records_loan.append(latest_rec)

                    if VERBOSE:
                        print(("\nnewly added records with "
                               "today's date [{}]\n").format("history"))
                        for rec in latest_records_loan +\
                                latest_records_account:
                            print(rec["date"])

                    hist_transaction(
                        INSERT,
                        tbl_account,
                        collection=latest_records_account
                    )

                    hist_transaction(
                        INSERT,
                        tbl_loan,
                        collection=latest_records_loan
                    )

                    loan_records_after_insertion = hist_transaction(
                        SELECT_RETURN,
                        tbl_loan
                    )

                    account_records_after_insertion = hist_transaction(
                        SELECT_RETURN,
                        tbl_account
                    )

                    assert len(loan_records_after_insertion)\
                        == len(initial_records_loan) + NUM_OF_NEW_RECORDS

                    assert len(account_records_after_insertion)\
                        == len(initial_records_account) + NUM_OF_NEW_RECORDS

                # 3. execute cleanup script (outside the for loop, we only
                # run this once)
                run_cmd_subprocess("{}/{} SCRUBCU".format(
                    PERL_SCRIPTS_ROOT,
                    table_to_perlscript_map["history"]))

                # 4. post cleanup
                for tbl_account, tbl_loan in [("scrubcuaccounthistory",
                                               "scrubculoanhistory")]:
                    loan_records_after_cleanup = hist_transaction(
                        SELECT_RETURN,
                        tbl_loan
                    )

                    account_records_after_cleanup = hist_transaction(
                        SELECT_RETURN,
                        tbl_account
                    )

                    assert len(
                        loan_records_after_cleanup) == NUM_OF_NEW_RECORDS
                    assert len(
                        account_records_after_cleanup) == NUM_OF_NEW_RECORDS

                    if VERBOSE:
                        print(
                            "\nrecords after cleanup [{}]\n".format("history"))
                        for rec in loan_records_after_cleanup +\
                                account_records_after_cleanup:
                            print(rec["date"])

test_drop_cu_misc()

def cudrop_test.TestDataCleanupScripts.test_drop_cu_misc

(

self

)

Validate expected counts for following tables:

    - curdcstatus
    - cu_vendorlog
    - cu_vendorcache
    - lnappuserresponse
    - culogtrack

Cleanup Scripts:
    - cudropcurdcstatus.pl
    - cudropcu_vendorlog.pl
    - cudropcu_vendorcache.pl
    - cudroplnapp.pl
    - cudropculogtrack.pl

Definition at line 450 of file cudrop_test.py.

    def test_drop_cu_misc(self):
        """Validate expected counts for following tables:

            - curdcstatus
            - cu_vendorlog
            - cu_vendorcache
            - lnappuserresponse
            - culogtrack

        Cleanup Scripts:
            - cudropcurdcstatus.pl
            - cudropcu_vendorlog.pl
            - cudropcu_vendorcache.pl
            - cudroplnapp.pl
            - cudropculogtrack.pl
        """
        with pg_handler.PGSession() as conn:
            with conn.cursor(cursor_factory=DictCursor) as dict_cur:
                curdcstatus_transaction = PgTransaction(conn, dict_cur)
                for table_name in MISC_TABLES:
                    # if table_name != "cu_vendorcache":
                    #     continue

                    # 1. get the initial records from setUp
                    initial_records = curdcstatus_transaction(
                        SELECT_RETURN,
                        table_name
                    )

                    if VERBOSE:
                        print("\ninitial records [{}]\n".format(table_name))
                        for rec in initial_records:
                            if table_name == "lnappuserresponse":
                                print(rec["respsubmiton"], rec["respstarton"])
                            else:
                                print(rec)

                    # 2. add NUM_OF_NEW_RECORDS with today's updatedate (these
                    # NUM_OF_NEW_RECORDS records must not be scraped by the db
                    # cleanup cronjob cudropcurdcstatus.pl)
                    latest_records = []
                    assert len(initial_records) >= NUM_OF_NEW_RECORDS

                    for i, record in enumerate(initial_records[
                                               :NUM_OF_NEW_RECORDS]):
                        latest_rec = record.copy()

                        if table_name == "curdcstatus":
                            latest_rec["lastupdate"] = TEST_DATE_NOW
                            latest_rec["depositid"] += SERIAL_OFFSET + i

                        elif table_name == "lnappuserresponse":
                            latest_rec["respid"] += SERIAL_OFFSET + i
                            latest_rec["respsubmiton"] = TEST_DATE_NOW
                            latest_rec["respstarton"] = TEST_DATE_NOW

                        elif table_name == "culogtrack":
                            latest_rec["accountnumber"] = int(
                                latest_rec["accountnumber"])\
                                + SERIAL_OFFSET\
                                + i
                            latest_rec["yearmo"] = TEST_DATE_NOW_YEARMO

                        elif table_name in ["cu_vendorlog", "cu_vendorcache"]:
                            latest_rec["entry"] = TEST_DATE_NOW

                        latest_records.append(latest_rec)

                    if VERBOSE:
                        print(("\nnewly added records with "
                               "today's date [{}]\n").format(table_name))
                        for rec in latest_records:
                            if table_name == "lnappuserresponse":
                                print(rec["respsubmiton"], rec["respstarton"])
                            else:
                                print(rec)

                    curdcstatus_transaction(
                        INSERT,
                        table_name,
                        collection=latest_records
                    )
                    curdcstatus_transaction.commit()

                    records_after_insertion = curdcstatus_transaction(
                        SELECT_RETURN,
                        table_name
                    )

                    assert len(records_after_insertion)\
                        == len(initial_records) + NUM_OF_NEW_RECORDS

                    # 3. execute cleanup script
                    run_cmd_subprocess("{}/{}".format(
                        PERL_SCRIPTS_ROOT,
                        table_to_perlscript_map[table_name]))

                    # 4. post cleanup
                    records_after_cleanup = curdcstatus_transaction(
                        SELECT_RETURN,
                        table_name
                    )

                    assert len(records_after_cleanup) == NUM_OF_NEW_RECORDS

                    if VERBOSE:
                        print(
                            "\nrecords after cleanup [{}]\n".format(
                                table_name))
                        for rec in records_after_cleanup:
                            if table_name == "lnappuserresponse":
                                print(rec["respsubmiton"], rec["respstarton"])
                            else:
                                print(rec)

test_misc_counts()

def cudrop_test.TestDataCleanupScripts.test_misc_counts

(

self

)

Validate expected counts for following tables:

    - curdcstatus
    - cu_vendorlog
    - cu_vendorcache
    - lnappuserresponse
    - culogtrack

Definition at line 215 of file cudrop_test.py.

    def test_misc_counts(self):
        """Validate expected counts for following tables:

            - curdcstatus
            - cu_vendorlog
            - cu_vendorcache
            - lnappuserresponse
            - culogtrack
        """
        with pg_handler.PGSession() as conn:
            with conn.cursor() as cur:
                select_transaction = PgTransaction(conn, cur)

                curdcstatus_count = select_transaction(
                    SELECT_RETURN,
                    "curdcstatus",
                    return_only_count=True
                )
                self.assertEqual(curdcstatus_count, LIMIT_SELECT)

                cuvendorlog_count = select_transaction(
                    SELECT_RETURN,
                    "cu_vendorlog",
                    return_only_count=True
                )
                self.assertEqual(cuvendorlog_count, LIMIT_SELECT)

                cuvendorcache_count = select_transaction(
                    SELECT_RETURN,
                    "cu_vendorcache",
                    return_only_count=True
                )
                self.assertEqual(cuvendorcache_count, min(LIMIT_SELECT, 23))

                lnappuserresponse_count = select_transaction(
                    SELECT_RETURN,
                    "lnappuserresponse",
                    return_only_count=True
                )
                self.assertEqual(lnappuserresponse_count,
                                 min(LIMIT_SELECT, 44))

                culogtrack_count = select_transaction(
                    SELECT_RETURN,
                    "culogtrack",
                    return_only_count=True
                )
                self.assertEqual(culogtrack_count, LIMIT_SELECT)

test_trans_hdr_dtl()

def cudrop_test.TestDataCleanupScripts.test_trans_hdr_dtl

(

self

)

Test cleanupscripts for trans{hdr,dtl} tables:

    - scrubcutranshdr
    - scrubcutransdtl

Cleanup Scripts:
    - cudropcutrans.pl

Definition at line 401 of file cudrop_test.py.

    def test_trans_hdr_dtl(self):
        """Test cleanupscripts for trans{hdr,dtl} tables:

            - scrubcutranshdr
            - scrubcutransdtl

        Cleanup Scripts:
            - cudropcutrans.pl
        """
        cu = "scrubcu"
        hdr_tablename = "{}transhdr".format(cu)
        dtl_tablename = "{}transdtl".format(cu)
        with pg_handler.PGSession() as conn:
            with conn.cursor(cursor_factory=DictCursor) as dict_cur:
                trans_transaction = PgTransaction(conn, dict_cur)

                initial_hdr_records_count = trans_transaction(
                    SELECT_RETURN,
                    hdr_tablename,
                    return_only_count=True
                )
                self.assertEqual(initial_hdr_records_count, 4)

                initial_dtl_records_count = trans_transaction(
                    SELECT_RETURN,
                    dtl_tablename,
                    return_only_count=True
                )
                self.assertEqual(initial_dtl_records_count, 5)

                # 3. execute cleanup script
                run_cmd_subprocess("{}/{} SCRUBCU".format(
                    PERL_SCRIPTS_ROOT,
                    table_to_perlscript_map["trans_hdr_dtl"]))

                hdr_records_count_after_cleanup = trans_transaction(
                    SELECT_RETURN,
                    hdr_tablename,
                    return_only_count=True
                )
                self.assertEqual(hdr_records_count_after_cleanup, 2)

                dtl_records_count_after_cleanup = trans_transaction(
                    SELECT_RETURN,
                    dtl_tablename,
                    return_only_count=True
                )
                self.assertEqual(dtl_records_count_after_cleanup, 2)

---

The documentation for this class was generated from the following file:

• app/tools/bin/cudrop_test.py