DataOps-platform/scripts/verify_test_data.py

"""Verify test data script"""

import psycopg2

DB_CONFIG = {
    "host": "192.168.3.143",
    "port": 5432,
    "database": "dataops",
    "user": "postgres",
    "password": "dataOps",
}


def main():
    conn = psycopg2.connect(**DB_CONFIG)
    cur = conn.cursor()

    print("=" * 70)
    print("Database Test Data Verification")
    print("=" * 70)

    # data_products
    print("\n[data_products]")
    cur.execute(
        "SELECT id, product_name, target_table, record_count, column_count, status "
        "FROM public.data_products ORDER BY id"
    )
    for row in cur.fetchall():
        print(
            f"  ID={row[0]}, name={row[1]}, table={row[2]}, "
            f"records={row[3]}, columns={row[4]}, status={row[5]}"
        )

    # test_sales_data
    print("\n[test_sales_data - first 5 rows]")
    cur.execute(
        "SELECT order_id, order_date, customer_name, product_name, quantity, total_amount "
        "FROM public.test_sales_data LIMIT 5"
    )
    for row in cur.fetchall():
        print(
            f"  order={row[0]}, date={row[1]}, customer={row[2]}, "
            f"product={row[3]}, qty={row[4]}, amount={row[5]}"
        )

    # test_user_statistics
    print("\n[test_user_statistics - first 5 rows]")
    cur.execute(
        "SELECT user_id, username, login_count, total_orders, total_amount, user_level "
        "FROM public.test_user_statistics LIMIT 5"
    )
    for row in cur.fetchall():
        print(
            f"  user={row[0]}, name={row[1]}, logins={row[2]}, "
            f"orders={row[3]}, amount={row[4]}, level={row[5]}"
        )

    # test_product_inventory
    print("\n[test_product_inventory - first 5 rows]")
    cur.execute(
        "SELECT sku, product_name, brand, current_stock, stock_status, selling_price "
        "FROM public.test_product_inventory LIMIT 5"
    )
    for row in cur.fetchall():
        print(
            f"  sku={row[0]}, product={row[1]}, brand={row[2]}, "
            f"stock={row[3]}, status={row[4]}, price={row[5]}"
        )

    print("\n" + "=" * 70)
    print("Verification complete!")
    print("=" * 70)

    cur.close()
    conn.close()


if __name__ == "__main__":
    main()