PyLodge sample implementation

conftest.py

__author__ = 'ashwin'
import os
import re
import json
from selenium import webdriver
import pytest
import redis
import inspect

desired_caps = []
APP_URL = "Your Application URL"
redis_client = redis.StrictRedis(host=os.environ['REDIS_HOST_IP'], port=6379, db=0)

def set_redis(key, value):
    redis_client.set(key, value)


class ResourceHandler(object):

    def __init__(self, desired_cap):
		self.driver =webdriver.Firefox()
       self.driver.maximize_window()
        self.driver.get(APP_URL)


    def release(self):
        self.driver.quit()

def _get_resource_handler():
	resource_handler_obj = ResourceHandler()
	return resource_handler_obj

@pytest.fixture(scope="session")
def resource_handler(request):
    handler = _get_resource_handler()
   def _release_resource_handler():
        """teardown resource_handler"""
        handler.release()
	request.addfinalizer(_release_resource_handler)
    return handler
"""This part of code will get the name of the test and status at runtime"""
@pytest.mark.tryfirst
def pytest_runtest_makereport(item,__multicall__):
	"""Get the browser name, version and platform"""
	browser_with_platform =item.funcargs['resource_handler'].driver.desired_capabilities['platform']+\
		'-' + item.funcargs['resource_handler'].driver.desired_capabilities['browserName']+ \
		'-'+item.funcargs['resource_handler'].driver.desired_capabilities['version']
   	rep = __multicall__.execute()
    	current_test_name = ''
    	current_test_log =''
    	if rep.when == 'call':
    		"""Extract the test name from the stack """
        	nodeid = rep.nodeid
        	callerlist = nodeid.split("::")
        	test_name_full = callerlist[3]
        	substring = re.search('test_tc(\d+)_(.*)', test_name_full)

        	if substring:
            		test_name_full = substring.group(2)
            		test_name_list = test_name_full.split('[')
            		test_name = test_name_list[0]
            		current_test_name = "MYWEBAPP_" + test_name + "||" + browser_with_platform
            		current_test_log = 'LOG_' + current_test_name
            		current_session_id = 'Sauce_' + session_id

        	if rep.passed:
            		logdata = 'Test Case Passed'
            		set_redis(current_test_log,logdata)
            		set_redis(current_test_name, "Passed")

		elif rep.failed:
            		logdata = rep.longrepr
            		set_redis(current_test_log,str(logdata))
            		set_redis(current_test_name, "Failed")

        	else:
            		logdata = rep.longrepr
            		set_redis(current_test_log,str(logdata))
            		set_redis(current_test_name, "Skipped")
    	return rep

updatetestlodge.py

__author__ = 'ashwin'
from multiprocessing import Pool
import redis
import os
from pylodge import pylodge
import datetime

testlodge_username = os.environ['TESTLODGE_USERNAME']
testlodge_password = os.environ['TESTLODGE_PASSWORD']
testlodge_project = os.environ['TESTLODGE_PROJECT']
testlodge_api_url = os.environ['TESTLODGE_API_URL']
“””You can either save Test Lodge details in environment variables and get them or directly input them below”””
pylodge_obj = pylodge.PyLodge(os.environ['TESTLODGE_USERNAME'], os.environ['TESTLODGE_PASSWORD'],
							os.environ['TESTLODGE_PROJECT'], os.environ['TESTLODGE_API_URL'])
redis_client = redis.StrictRedis(host=os.environ['REDIS_HOST_IP'], port=6379, db=0)
all_redis_keys = redis_client.keys()
print "ALL REDIS KEYS: %s" % all_redis_keys
try:
	test_case_logs_list = redis_client.keys("LOG_ MYWEBAPP_*")
	test_case_list = redis_client.keys("MYWEBAPP_*")
	print "test case list: %s" % test_case_list
	result_dict_list = []
	log_dict_list = []

	browser_list = []
	for prefixed_test_case_and_browser in test_case_list:
		test_status = redis_client.get(prefixed_test_case_and_browser)
		prefixed_test_case, user_agent = prefixed_test_case_and_browser.split('||')
		test_case = prefixed_test_case.replace("MYWEBAPP_", "")
		result_dict = {'Test Case': test_case, 'User Agent': user_agent, 'Test Status': test_status}
		result_dict_list.append(result_dict)
		if user_agent not in browser_list:
			browser_list.append(user_agent)
	for prefixed_test_case_and_browser in test_case_logs_list:
		test_log = redis_client.get(prefixed_test_case_and_browser)
		prefixed_test_case, user_agent = prefixed_test_case_and_browser.split('||')
		test_case = prefixed_test_case.replace("LOG_ MYWEBAPP_", "")
		log_dict = {'Test Case': test_case, 'User Agent': user_agent, 'Test Log': test_log}
		log_dict_list.append(log_dict)
		if user_agent not in browser_list:
			browser_list.append(user_agent)
	def mark_test_status_of_test_case(test_case, test_status, pylodge_obj,test_log):
		pylodge_obj.mark_test_status(test_case, test_status,test_log=test_log)
	def mark_test_status_of_remaining_tests(test_run_name, pylodge_obj):
		test_case_list=pylodge_obj.fetch_and_save_not_run_test_case_ids(test_run_name)
		print 'Skipped test caselist is %s' %test_case_list
		pylodge_obj.mark_test_status_multiple(test_run_name=test_run_name,test_case_ids=test_case_list,status='skipped')
	pool = Pool(processes=6)
	for browser in browser_list:
		result_list = [filtered_result_dict for filtered_result_dict in result_dict_list if
					filtered_result_dict['User Agent'] == browser]
		log_list= [filtered_log_dict for filtered_log_dict in log_dict_list if
					filtered_log_dict['User Agent'] == browser]
		time_stamp = datetime.datetime.now().strftime('%Y_%m_%d_%H_%M_%S').format()
		run_name = 'Selenium-%s-%s' % (time_stamp, browser)
		run_id = pylodge_obj.create_test_run(run_name)
		print "Scheduling processes for %s test case results" % len(result_list)
		apply_results = []
		for result in result_list:
			for test_log in log_list:
				if result['Test Case']== test_log['Test Case']:
					test_log_str=test_log['Test Log']
					break
			print "Test Log is %s" %test_log_str
			apply_results.append(pool.apply_async(func=mark_test_status_of_test_case,
												args=(result['Test Case'], result['Test Status'], pylodge_obj,test_log_str)))
		pool.apply(func=mark_test_status_of_remaining_tests,args=(run_name,pylodge_obj))
		print "Waiting for results to come back"
		for apply_result in apply_results:
			apply_result.get(timeout=30)

		print "Finished marking tests for browser %s" % browser
		redis_client.flushall()
except Exception, e:
	print 'exception ###### %s'%e
	redis_client.flushall()