Notes

"""
This is knooble filewatcher windows task service, gets executed on windows startup. It monitors given directory
for file changes(cretion/deletion/modifications) and update statistics to mongodb database
"""
import os
import logging
import shutil
import subprocess
import time
import glob
import re

import hashlib
import zipfile

import gridfs

from collections import OrderedDict
from datetime import datetime

from RQ4_Integrity.InegrityHandler import IntegrityHandler
from databaseinterface import DatabaseManagement
from html_parser import HtmlParser
from rsp_report_parser import RSPHtmlParser
from sql_dbManagement import SQLDBManagement
import configparser

from svn_management import SVNManager

NETWORK_LOCATIONS = {
"TBM_CHERRY": [
r"C:KBDataTrailer"],
# "RSP_ENDURANCE_RUN": [r"C:KBAppsKnorr-BremseTestFlowWip"],
}

RSP_REPORT_PATH = r"C:KBDataJenkinsSandboxVT_SimulatorMeasurement_DataPDF_Reportsreport"
BATCH_DB_PATH = r'C:KBDataJenkinsSandboxVT_SimulatorMeasurement_DataPDF_Reports'

SUPPORTED_FILE_FORMATS = [".html", ".htm"]
BACKUP_FILE_FORMATS = [".pdf"]
COMMON_PATH = r'Automation\Trailer_endurance@adpx\reports'
BUD_SERVER_PATH = r'\file.corp.knorr-bremse.comknorrProjectTEBS_G3SoftwareCherryDaily_Build_RSP_Test_Report'

logger = logging.getLogger(__name__)


def setup_logger(is_debugging_needed):
if not os.path.exists(r"logs"):
os.makedirs(r"logs")
# setup file logger
log_filename = r".logswatcher" + time.strftime("%Y%m%d_%H%M%S") + ".log"
debugging_option = logging.INFO

if is_debugging_needed is True:
debugging_option = logging.DEBUG
logging.basicConfig(
filename=log_filename,
format="%(asctime)s, " "[%(levelname)s], " "%(name)s, " "%(message)s",
level=logging.DEBUG,
)

# setup console logger
console = logging.StreamHandler()
console.setLevel(debugging_option)
formatter = logging.Formatter(
"%(asctime)s, " "[%(levelname)s], " "%(name)s:, %(message)s"
)
console.setFormatter(formatter)

logging.getLogger("").addHandler(console)


class FileWalker:
"""
Walks through entire directory and cross check it with database records
"""

def __init__(self, logger=None):
super().__init__()
setup_logger(True)
self.svn = SVNManager()
self.databaseManagement = DatabaseManagement()
self.sqlDBManagement = SQLDBManagement()
self.svn_repository = "https://BU2S0069.corp.knorr-bremse.com/svn/Trailerdashboard/"
self.logger = logger or logging.root
self.mainproduct = None
self.software_version = None
self.SubProduct = None

def updateStatistics(self, hashcode):
result = self.databaseManagement.RSPEnduranceRunResults(hashcode)
totalnumberTC = 0
totalpassedTC = 0
totalfailedTC = 0
totalNotExecutedTC = 0
totalnumberTC = len(result["testCaseResults"])
for res in result["testCaseResults"]:
if res["verdict"] == "Passed":
totalpassedTC = totalpassedTC + 1
elif res["verdict"] in ["Fail", "Failed"]:
totalfailedTC = totalfailedTC + 1
elif res["verdict"] == "Skipped":
totalNotExecutedTC = totalNotExecutedTC + 1

statistic = []
tempList = []
tempList.append("Overall test cases")
tempList.append(totalnumberTC)
tempList.append("None")
statistic.append(tempList)

excecuted_TC_count = totalnumberTC - totalNotExecutedTC
executed_percentage = round(
(excecuted_TC_count / totalnumberTC) * 100) if totalnumberTC != 0 else 0
tempList = []
tempList.append("Executed test cases")
tempList.append(excecuted_TC_count)
tempList.append(f"{executed_percentage}%")
statistic.append(tempList)

tempList = []
passed_TC_percentage = round(
(totalpassedTC / totalnumberTC) * 100) if totalnumberTC != 0 else 0

tempList.append("Test cases Passed")
tempList.append(totalpassedTC)
tempList.append(f"{passed_TC_percentage}%")
statistic.append(tempList)

tempList = []
failed_percentage = round(
(totalfailedTC / totalnumberTC) * 100) if totalnumberTC != 0 else 0

tempList.append("Test cases Failed")
tempList.append(totalfailedTC)
tempList.append(f"{failed_percentage}%")
statistic.append(tempList)

Statistics = []
for statistic in statistic:
statisticDict = OrderedDict()
statisticDict["statName"] = statistic[0]
statisticDict["stat"] = statistic[1]
statisticDict["statStatus"] = statistic[2]
Statistics.append(statisticDict)
return Statistics

def uploadPDFdocuments(self, rsp_report_folder_path):
if os.path.exists(rsp_report_folder_path):
rsp_report_path = [os.path.join(rsp_report_folder_path, filename) for filename in
os.listdir(rsp_report_folder_path)]
index_to_update_record = 0
for file_name in os.listdir(rsp_report_path[0]):
file_path = os.path.join(rsp_report_path[0], file_name)
sourceFileName, sourceFileType = os.path.splitext(os.path.basename(file_path))
if (
sourceFileType.lower() in BACKUP_FILE_FORMATS
and sourceFileName.startswith("RSP_Test")
):
zipFileName, _ = os.path.splitext(os.path.basename(file_path))
sourceFilePath = os.path.dirname(file_path)
zipFileNameWithExtension = zipFileName + ".zip"

destinationFileName = os.path.join(sourceFilePath, zipFileName)
try:
shutil.make_archive(
destinationFileName,
"zip",
sourceFilePath,
os.path.basename(file_path)
)
gridConnDB = self.databaseManagement.mongoClient.grid_file

fs = gridfs.GridFS(gridConnDB)

ti_m = os.path.getmtime(file_path)
m_ti = time.ctime(ti_m)
t_obj = time.strptime(m_ti)
T_stamp = time.strftime("%Y-%m-%d %H:%M:%S", t_obj)
testcaseuniqueid = hashlib.sha224(
time.strftime("%Y-%m-%d", t_obj).encode("utf-8")
).hexdigest()

existing_file = self.databaseManagement.get_existing_report_filename(testcaseuniqueid,
index_to_update_record)
if existing_file:
fs.delete(existing_file._id)

fileLocation = os.path.join(sourceFilePath, zipFileNameWithExtension)
with open(fileLocation, "rb") as fileData:
data = fileData.read()

fs.put(data, filename=zipFileNameWithExtension)

self.databaseManagement.updatefilenameinDB(zipFileNameWithExtension, testcaseuniqueid,
index_to_update_record)
index_to_update_record = index_to_update_record + 1

destination_path = os.path.join(BUD_SERVER_PATH, time.strftime("%Y-%m-%d", t_obj))
if os.path.exists(destination_path):
if (datetime.now() - datetime.fromtimestamp(
os.path.getctime(destination_path))).total_seconds() / 60 > 30:
if len(os.listdir(destination_path)) == 0:
os.rmdir(destination_path)
else:
for file in os.listdir(destination_path):
os.remove(os.path.join(destination_path, file))
os.rmdir(destination_path)
os.makedirs(destination_path, exist_ok=True)
else:
os.makedirs(destination_path, exist_ok=True)

shutil.move(os.path.join(sourceFilePath, zipFileNameWithExtension),
os.path.join(destination_path, zipFileNameWithExtension))

except:
logger.error("Error while creating archive for RSP report")

logger.info("RSP report archive transfer successfully")
else:
logger.info("RSP report folder not available")

def uploadAllinonePDFdocuments(self, rsp_report_folder_path):

if os.path.exists(rsp_report_folder_path):
rsp_report_path = [os.path.join(rsp_report_folder_path, filename) for filename in
os.listdir(rsp_report_folder_path)]
all_reports_zip_path = os.path.join(rsp_report_path[0], f"{self.fileName}.zip")

with zipfile.ZipFile(all_reports_zip_path, "w") as all_reports_zip:
for file_name in os.listdir(rsp_report_path[0]):
file_path = os.path.join(rsp_report_path[0], file_name)
sourceFileName, sourceFileType = os.path.splitext(os.path.basename(file_path))
if (
sourceFileType.lower() in BACKUP_FILE_FORMATS
and sourceFileName.startswith("RSP_Test")
):
try:
all_reports_zip.write(file_path, os.path.basename(file_path))

except Exception as e:
logger.error(f"Error while adding {file_name} to the zip file: {e}")

# Upload file to SVN
self.svn.svn_file_upload(r"" + self.svn_repository + "RSP/RSP", file_path, file_name)

os.remove(all_reports_zip_path)
logger.info(f"uploaded RSP report with name {self.fileName}")
shutil.rmtree(BATCH_DB_PATH)

else:
logger.warning("RSP report folder not available")

def collect_measurements(self):
for networkLocation, netWorkDrivePaths in NETWORK_LOCATIONS.items():
for networkPath in netWorkDrivePaths:
if networkLocation == 'TBM_CHERRY':
if os.path.exists(networkPath):
for (root, dirs, files) in os.walk(networkPath, topdown=False, followlinks=True):
for idx, fileName in enumerate(files):
sourceFileName, sourceFileType = os.path.splitext(fileName)
if (
sourceFileType.lower() in SUPPORTED_FILE_FORMATS
and (fileName.startswith("Regression_") or fileName.__contains__(
"Regression_Test_Report")) and (
fileName.endswith(".html") or fileName.endswith(".htm"))
):
filePath = os.path.join(root, fileName)
parser = HtmlParser(filePath, self.sqlDBManagement)

# Parse report
parser.parse_statistics_data()
parser.parse_test_case_result()

self.software_version = parser.software_version
self.mainproduct = parser.mainproduct
self.SubProduct = parser.subproduct

# update test session in integrity
integrity = IntegrityHandler(parser)
res = integrity.aa_connect()
if res == 0:
integrity.update_integrity()
integrity.aa_disconnect()

# It archives all files from sourceFilePath directory
source_file_path = root
destination_file_path = root

destination_filename = os.path.join(destination_file_path, parser.filename[:-4])

shutil.make_archive(
destination_filename,
"zip",
source_file_path,
fileName,
)

# Upload report to SVN server

res = self.svn.svn_file_upload(parser.repository_url,
f"{destination_filename}.zip", parser.filename)

if res:
# Once file is uploaded to SVN , update database
try:

result = parser.update_database()

if not result:
continue

self.sqlDBManagement.refinetables(self.mainproduct, self.software_version,
self.SubProduct)
os.remove(f"{destination_filename}.zip")

except Exception as e:
logger.warning(e.__str__())
else:
logger.warning("SQL DB not updated on Server : ".format(parser.filename))

else:
logger.warning("Path Not Available: {} ".format(networkPath))

if networkLocation == 'RSP_ENDURANCE_RUN':
statistics_info_flag = False
end_time_stamp = False
try:
if os.path.exists(networkPath):
test_flow_dir_path = networkPath + '\' + max(os.listdir(networkPath))
report_creation_date = datetime.fromtimestamp(
os.path.getctime(test_flow_dir_path)).isoformat()
date_time = re.findall('d{4}-d{2}-d{2}', report_creation_date)[0] + ' ' + '00:00:00'
for test_session_id in os.listdir(test_flow_dir_path):
report_path = test_flow_dir_path + '\' + test_session_id
# report_path= ""
for test_cases in os.listdir(report_path):
if os.path.exists(os.path.join(report_path, test_cases, COMMON_PATH)):
report_data = os.listdir(os.path.join(report_path, test_cases, COMMON_PATH))

for id, path in enumerate(report_data):
file_path = report_path + '\' + test_cases + '\' + COMMON_PATH + '\' + path
file_data = glob.glob(file_path + r'data.html')
for reports in file_data:
parser = RSPHtmlParser(reports, self.databaseManagement)
# parser = RSPHtmlParser(reports, self.sqlDBManagement)

reportdate = parser.getTestCaseDate()
if not bool(reportdate):
reportdate = date_time
testdate = datetime.strptime(reportdate, "%Y-%m-%d %H:%M:%S")
uniqueID = hashlib.sha224(
testdate.strftime("%Y-%m-%d").encode("utf-8")
).hexdigest()
if len(report_data) == id + 1:
end_time_stamp = True
# Insert RSP data to SQL DB
result = parser.getTestCaseMetaDataSql(test_flow_dir_path,
statistics_info_flag,
end_time_stamp)
if result:
statistics_info_flag = True
self.fileName = parser.file_name
self.mainproduct = parser.mainproduct
self.SubProduct = parser.subproduct
self.software_version = parser.software_version
try:
self.uploadAllinonePDFdocuments(RSP_REPORT_PATH)
# if duplicate date data is present
self.sqlDBManagement.refinetables(self.mainproduct, self.software_version,
self.SubProduct)
except Exception as e:
logger.error(str(e))

else:
logger.warning("Path Not Available: {} ".format(networkPath))
except Exception as e:
logger.error(str(e))


if __name__ == "__main__":
obj = FileWalker()
obj.collect_measurements()
obj.sqlDBManagement.close_connection()