Notes

import hashlib
import logging
import math
import os

import time
from collections import OrderedDict
from datetime import datetime
import datetime
import re
from lxml import etree

from databaseinterface import DatabaseManagement
from sql_dbManagement import SQLDBManagement

logger = logging.getLogger(__name__)


class RSPHtmlParser:
lastUpdated = None
lastParserdHTML = None

def __init__(self, filePath, dbObj):
# Try to read the file until it completes copying

# if os.path.basename(filePath).startswith("Regression_"):
self.file_name = None
self.Url = None
self.subproduct = None
self.mainproduct = None
self.software_version = "Not_Available"
self.filePath = filePath
self.dbObj = dbObj
self.sqlDBManagement = SQLDBManagement()
self.Start_DateTime = None
while True:
try:

time.sleep(3)
self.data = open(filePath, "r", encoding="utf-16").read()
break
except:
pass
self.doc = etree.HTML(self.data)
self.originalFileName = os.path.basename(filePath)

def getTestCaseDate(self):
test_case_datetime = []
try:
test_case_datetime = self.doc.xpath(
'//table[@class="execution-information"]//th[text()="Start time:"]/following-sibling::td/text()')[0]
return test_case_datetime
except:
return test_case_datetime

def getTestCaseVerdict(self):
testcase_verdict = []
testcase_verdict = self.doc.xpath('//table[@class="sequence-statistic"]//td/@class')[0]
return testcase_verdict

def getTestSessionId(self):
test_session_id = []
try:
test_session_id =
self.doc.xpath('//h3[text()="Test Session Informations"]/following-sibling::table//tr/td/p/text()')[1]
return test_session_id
except:
return False

def getTestCaseResult(self):
test_case_result = []
try:
test_caseid =
self.doc.xpath('//h3[text()="Test Case Informations"]/following-sibling::table//tr/td/p/text()')[
1]
test_casename =
self.doc.xpath('//h3[text()="Test Case Informations"]/following-sibling::table//tr/td/p/text()')[5]

testcasedict = OrderedDict()
testcasedict["testsessionid"] = self.getTestSessionId()
testcasedict["testCaseID"] = test_caseid
testcasedict["testName"] = test_casename
testcasedict["verdict"] = self.getTestCaseVerdict()
testcasedict["testPTCparameter"] = self.get_ptc_parameter()

test_case_result.append(testcasedict)
return test_case_result
except:
logger.info("session not available")

def getTestCaseResultsql(self):
test_case_result = []
try:
try:
test_caseid =
self.doc.xpath('//h3[text()="Test Case Informations"]/following-sibling::table//tr/td/p/text()')[
1]
except:
test_caseid = "Not_Available"
try:
test_casename =
self.doc.xpath('//h3[text()="Test Case Informations"]/following-sibling::table//tr/td/p/text()')[5]
except:
test_casename = "Not_Available"

if (test_caseid != "Not_Available" or test_casename != "Not_Available"):
Heading_Name = "null"
Category = "null"
ExecutionStatus = "null"
Heading_Name = test_casename
Testcases = test_casename
Verdict = self.getTestCaseVerdict()
Category = Category
if Verdict in ["Passed", "Fail", "Failed"]:
if Verdict == "Passed":
Verdict = "pass"
elif Verdict in ["Fail", "Failed"]:
Verdict = "fail"
ExecutionStatus = "Executed"
elif Verdict in ["not executed", "Skipped", "Error"]:
ExecutionStatus = "NotExecuted"
Verdict = "error"

self.mainproduct = "RSP"
self.subproduct = "RSP"
data = {
"Testcase_Id": test_caseid,
"StartDateTimeStamp": self.Start_DateTime,
"Heading_Name": str("TC_" + test_caseid + "_" + Heading_Name),
"Testcases": str("TC_" + test_caseid + "_" + Testcases),
"Verdict": Verdict,
"Category": Category,
"ExecutionStatus": ExecutionStatus,
"KeyID": hashlib.sha224(
str(self.Start_DateTime.strftime(
"%Y-%m-%d %H:%M:%S") + self.mainproduct + "_" + self.subproduct).encode(
"utf-8")).hexdigest(),
"PTCKeyID": hashlib.sha224(
str(self.Start_DateTime.strftime("%Y-%m-%d %H:%M:%S") + test_caseid).encode(
"utf-8")).hexdigest()
}
result = self.sqlDBManagement.insert_data("Testcases", data)
if result != False:
logger.info("Inserted data into testcase table.")
self.get_ptc_parametersql(test_caseid)
return test_case_result
except Exception as e:
logger.info(print(str(e)))
logger.info("session not available")

def get_ptc_parameter(self):
ptc_parameter = []
ptc_parameter = self.doc.xpath('//h3[text()="PTC Parameters"]/following-sibling::table//tr/td/p/text()')
ptc_parameter_dict = {}

for i in range(0, len(ptc_parameter), 2):
key = ptc_parameter[i]
value = ptc_parameter[i + 1]
ptc_parameter_dict[key] = str(value)

return ptc_parameter_dict

def get_ptc_parametersql(self, test_caseid):
PTCKeyID = hashlib.sha224(str(self.Start_DateTime.strftime("%Y-%m-%d %H:%M:%S") + test_caseid).encode(
"utf-8")).hexdigest()
ptc_parameter = self.doc.xpath('//h3[text()="PTC Parameters"]/following-sibling::table//tr/td/p/text()')
for i in range(0, len(ptc_parameter), 2):
data = {}
attribute = ptc_parameter[i]
value = ptc_parameter[i + 1]
data['PTCKeyID'] = PTCKeyID
data['Properties'] = attribute
data['Prop_Value'] = value
data['StartDateTimeStamp'] = self.Start_DateTime

if data:
self.sqlDBManagement.insert_data("PTCparameter", data)
logger.info("Inserted data into PTCparameter table")

def getTestCaseMetaData(self, test_flow_dir_path):
testresults = {}
testDateTime = self.getTestCaseDate()
# get report creation time
report_creation_date = datetime.datetime.fromtimestamp(os.path.getctime(test_flow_dir_path)).isoformat()
if not bool(testDateTime):
testDateTime = re.findall('d{4}-d{2}-d{2}', report_creation_date)[0] + ' ' + '00:00:00'
datetime_format = datetime.datetime.fromisoformat(report_creation_date)
testresults["testTimeStart"] = datetime_format
self.fileName = (
testresults["testTimeStart"]
)
testresults["fileName"] = self.fileName
testresults["testCaseResults"] = self.getTestCaseResult()
testresults["testCaseUniqueID"] = hashlib.sha224(
datetime.datetime.strptime(testDateTime, "%Y-%m-%d %H:%M:%S").strftime("%Y-%m-%d").encode("utf-8")
).hexdigest()
return testresults

def getTestCaseMetaDataSql(self, test_flow_dir_path, data_in_statistics_info_flag, end_time_stamp):
testDateTime = self.getTestCaseDate()
report_creation_date = datetime.datetime.fromtimestamp(os.path.getctime(test_flow_dir_path)).isoformat()
if not bool(testDateTime):
testDateTime = re.findall('d{4}-d{2}-d{2}', report_creation_date)[0] + ' ' + '00:00:00'
datetime_format = datetime.datetime.fromisoformat(report_creation_date)
self.Start_DateTime = datetime_format
self.session_id = self.getTestSessionId()
# Update enddateTimestamp of last executed report in SQLdb
if end_time_stamp:
self.mainproduct = "RSP"
self.subproduct = "RSP"
testDateTime = self.getTestCaseDate()
execution_end_date_time = datetime.datetime.strptime(
testDateTime, "%Y-%m-%d %H:%M:%S"
)
KeyID = hashlib.sha224(
str(self.Start_DateTime.strftime("%Y-%m-%d %H:%M:%S") + self.mainproduct + "_" + self.subproduct).encode(
"utf-8")).hexdigest()

result = self.sqlDBManagement.updateTableField("StatisticsInfo", "KeyID", KeyID, "EndDateTimeStamp",
execution_end_date_time)
if result:
logger.info("Updated end time stamp into statisticsinfo")
# ----

if self.session_id != False:
result = True
if data_in_statistics_info_flag == False:
self.SoftwareVersion = "Not_Available"
self.Start_DateTime = datetime_format
monthName = self.Start_DateTime.strftime("%B")
year = self.Start_DateTime.strftime("%Y")
mon = int(self.Start_DateTime.strftime("%m"))
self.mainproduct = "RSP"
subproduct = "RSP"
self.file_name = (self.Start_DateTime.strftime('%Y-%m-%d_%H-%M-%S') + "_RSP_Test.zip")
self.Url = f"http://pu2w2197:5014/download_svn_report?product={self.mainproduct}&subProduct={self.subproduct}&fileName={self.file_name}"
comments = "null"
data = {
"SoftwareVersion": self.SoftwareVersion,
"SessionID": self.session_id,
"EndDateTimeStamp": self.Start_DateTime,
"StartDateTimeStamp": self.Start_DateTime,
"Dates": str(self.Start_DateTime.strftime("%Y-%m-%d")),
"Qtr": math.ceil(mon / 3),
"Mon": mon,
"MonthName": monthName,
"FullYear": year,
"ReportFilename": self.file_name,
"Url": self.Url,
"MainProduct": self.mainproduct,
"SubProduct": self.subproduct,
"Comments": comments,
"KeyID": hashlib.sha224(
str(self.Start_DateTime.strftime(
"%Y-%m-%d %H:%M:%S") + self.mainproduct + "_" + subproduct).encode(
"utf-8")).hexdigest()
}
result = self.sqlDBManagement.insert_data("StatisticsInfo", data)
if result != False:
logger.info("Inserted data into statisticsinfo")
self.getTestCaseResultsql()
return result
else:
self.getTestCaseResultsql()

def uploadHTMLReport(self):
pass

def updateDatabase(self, test_flow_dir_path, statistics_info_flag, end_time_stamp):
updatedstatus = self.getTestCaseMetaDataSql(test_flow_dir_path, statistics_info_flag, end_time_stamp)
result = self.getTestCaseMetaData(test_flow_dir_path)
if result["testCaseResults"]:
self.dbObj.insertRSPTestResults(result)
logger.info("RSP_Results Database has been updated successfully.")
RSPHtmlParser.lastUpdated = datetime.datetime.now()
RSPHtmlParser.lastParserdHTML = self.originalFileName
return updatedstatus


if __name__ == "__main__":
dbObj = DatabaseManagement()
parser = RSPHtmlParser(r"C:KBAppspythonProject1data.html", dbObj)
parser.updateDatabase()