Data-driven migration testing with REST APIs and Python

sample_migration_test_suite.py

import csv

import lemoncheesecake.api as lcc
import requests
from lemoncheesecake.matching import check_that, equal_to

USERS_MIGRATION_CSV = "users.csv"
APP_BASE_URL = "http://my-account-management-system.com/"
USERS_ENDPOINT = "users/"
ACCOUNT_DETAILS_ENDPOINT = "accountdetails/"


def get_user_profile(base_url, phone_number):
    params = {"phoneNumber": phone_number}
    return requests.get(url=base_url + USERS_ENDPOINT, params=params)


def get_account_details(base_url, user_id):
    params = {"userId": user_id}
    return requests.get(url=base_url + ACCOUNT_DETAILS_ENDPOINT, params=params)


def verify_common_account_attributes(account_details, account_balance):
    """ Verify all the common credit account attributes """
    status = account_details['operationalStatus']
    check_that("operational status", status, equal_to("ACCOUNT_APPROVED"))
    activation_status = account_details['activationStatus']
    check_that("activation status", activation_status, equal_to("ACCOUNT_ACTIVATED"))
    account_balance_ = account_details['accountBalance']
    check_that("account balance amount", account_balance_, equal_to(float(account_balance)))


def perform_account_migration_tests(mobile_number, customer_id, user_id, account_balance, account_bonus):
    def func():
        lcc.log_info("mobile number: {} | account balance: {}".format(mobile_number, account_balance))
        """ Verify various attributes from the user's profile """
        response = get_user_profile(base_url=APP_BASE_URL, phone_number=mobile_number)
        check_that("value", response.status_code, equal_to(200))
        if response.status_code is 200:
            response_data = response.json()['responseData']
            user_id_ = response_data['userId']
            customer_id_ = response_data['customerId']
            used_balance = response_data['usedBalance']
            check_that("used balance", used_balance,
                       equal_to(account_balance - account_bonus))  # business logic applicable for all the users
            check_that("user id", user_id_, equal_to(user_id))
            check_that("customer id", customer_id_, equal_to(customer_id))
            """ Verify various attributes from the user's all loan requests """
            response = get_account_details(base_url=APP_BASE_URL, user_id=user_id_)
            check_that("value", response.status_code, equal_to(200))
            if response.status_code is 200:
                check_that("count", response.json()['responseData']['count'],
                           equal_to(1))  # business logic applicable for all the users
            else:
                lcc.log_error("Failed to fetch the account details for the user {}".format(user_id_))

        else:
            lcc.log_error("Failed to fetch the user profile for user {}.".format(user_id))

    return func


@lcc.suite("Simple migration test suite")
class sample_migration_test_suite:
    csv_data = list()

    def __init__(self):
        with open(USERS_MIGRATION_CSV) as csv_file:
            self.csv_data = csv.DictReader(csv_file)
            next(self.csv_data)  # skip the header
            for row in self.csv_data:
                mobile_number = row['phone_number']
                customer_id = int(row['customer_id'])
                user_id = int(row['user_id'])
                account_balance = float(row['account_balance'])  # assume this being in users.csv
                account_bonus = float(row['account_bonus'])  # assume this being in users.csv
                test = lcc.Test(name=mobile_number,
                                description="Bank account migration test for user {} with phone number".format(user_id,
                                                                                                               mobile_number),
                                callback=perform_account_migration_tests(mobile_number=mobile_number,
                                                                         customer_id=customer_id, user_id=user_id,
                                                                         account_balance=account_balance,
                                                                         account_bonus=account_bonus))
                lcc.add_test_in_suite(test=test, suite=self)