#!/usr/bin/python3
# coding: utf-8

#################################################################
#
# Copyright (C) ИП Пуляев Григорий Васильевич, 2024
# email: rodegast@xmail.ru
# telegram: t.me/kybyc_meta_test
#
# Расширение добавляет поддержку Allure Framework
# в платформу автоматизированного тестирования Meta Test.
#
# Этот файл НЕ является частью платформы Meta Test.
# Лицензия расширения - Apache License 2.0
# https://www.apache.org/licenses/LICENSE-2.0
#
#################################################################

import os
import json
import time

from allure_commons import plugin_manager
from allure_commons.logger import AllureFileLogger
from allure_commons.reporter import AllureReporter
from allure_commons.utils import md5, uuid4, represent
from allure_commons.types import LabelType, AttachmentType, LinkType
from allure_commons.model2 import Status, Parameter, StatusDetails, TestResult, TestStepResult, Label, Link

from report import BaseReport
from utilites import rMkDir, rmPath
from core_dsl import ItemTestConfig
from parameter import CONFIG, TEST_STATUS, COMMAND_STATUS, ItemConfig

__all__ = (
	"Allure",
)

def get_history_id(report):
	return md5(
		report.test_name, report.browser_name, report.file_line
		, report.browser_version, repr(report.test_parameters)
	)

def get_status(status):
	if status in (TEST_STATUS.UNKNOWN, TEST_STATUS.STARTED):
		return Status.UNKNOWN
	elif status in (TEST_STATUS.FAILED, TEST_STATUS.FATAL, TEST_STATUS.XFAILED, TEST_STATUS.TIMEOUT):
		return Status.FAILED
	elif status == TEST_STATUS.ERROR:
		return Status.BROKEN
	elif status in (TEST_STATUS.SKIPPED, TEST_STATUS.STOPPED):
		return Status.SKIPPED
	return Status.PASSED

def get_time(report=None):
	return int(round(1000 * report.timestamp if report else time.time()))

class Allure(BaseReport):
	def __init__(self):
		if CONFIG.ALLURE_CLEAR:
			rmPath(CONFIG.ALLURE_LOG_DIR)
		plugin_manager.register(AllureFileLogger(rMkDir(CONFIG.ALLURE_LOG_DIR)))
		
		self._tests     = {}
		self._test_data = {}
	
	def start(self, report):
		obj = TestResult(name=report.test_title, uuid=report.test_id, start=get_time(report))
		if report.is_parent:
			obj.testCaseId = report.parent_uuid
		else:
			obj.testCaseId = report.test_id
		obj.fullName    = report.test_title
		obj.historyId   = get_history_id(report)
		obj.description = report.description
		obj.parameters.extend([
			Parameter(name=k, value=represent(v)) for k, v in report.test_parameters.items()
		])
		obj.labels.append(Label(name=LabelType.FRAMEWORK, value="Meta Test"))
		obj.labels.append(Label(name=LabelType.THREAD, value=str(report.thread_id)))
		obj.labels.append(Label(name=LabelType.SEVERITY, value=report.test_args["severity"]))
		obj.labels.extend([ Label(name=LabelType.TAG, value=x) for x in report.test_args["tags"] ])
		
		if report.test_args["epic"]:
			if isinstance(report.test_args["epic"], str):
				obj.labels.append(Label(name=LabelType.EPIC, value=report.test_args["epic"]))
			else:
				obj.labels.extend([ Label(name=LabelType.EPIC, value=x) for x in report.test_args["epic"] ])
		if report.test_args["feature"]:
			if isinstance(report.test_args["feature"], str):
				obj.labels.append(Label(name=LabelType.FEATURE, value=report.test_args["feature"]))
			else:
				obj.labels.extend([ Label(name=LabelType.FEATURE, value=x) for x in report.test_args["feature"] ])
		if report.test_args["story"]:
			if isinstance(report.test_args["story"], str):
				obj.labels.append(Label(name=LabelType.STORY, value=report.test_args["story"]))
			else:
				obj.labels.extend([ Label(name=LabelType.STORY, value=x) for x in report.test_args["story"] ])
		if report.test_args["owner"]:
			obj.labels.append(Label(name="owner", value=report.test_args["owner"]))
		if report.test_args["issue"]:
			obj.links.append(Link(LinkType.LINK, report.test_args["issue"], "issue"))
		if report.test_args["tms_link"]:
			if isinstance(report.test_args["tms_link"], str):
				obj.links.append(Link(LinkType.TEST_CASE, report.test_args["tms_link"]))
			else:
				obj.links.extend([ Link(LinkType.TEST_CASE, x) for x in report.test_args["tms_link"] ])
		self._tests[report.test_id]     = obj
		self._test_data[report.test_id] = []
	
	def start_step(self, report, step_uuid=""):
		if report.test_id in self._test_data and not report.step_status == COMMAND_STATUS.SKIPPED:
			obj = TestStepResult(start=get_time(report))
			obj.name   = report.action_name
			obj.status = get_status(report.status)
			obj.description = report.description
			step_uuid  = step_uuid or report.step_uuid
			self._test_data[report.test_id].append({
				"method": "start_step", "parent_uuid": report.step_parent_uuid
				, "step_uuid": step_uuid, "step": obj
			})
			for x, y in report.screenshot:
				self._test_data[report.test_id].append({
					"method": "attach_data", "uuid": uuid4(), "name": x
					, "message": y, "attachment_type": AttachmentType.PNG
			})
			if report.page_source:
				self._test_data[report.test_id].append({
					"method": "attach_data", "uuid": uuid4(), "name": "page_source"
					, "message": report.page_source, "attachment_type": AttachmentType.HTML
				})
			if report.cookies:
				self._test_data[report.test_id].append({
					"method": "attach_data", "uuid": uuid4(), "name": "cookies"
					, "message": json.dumps(report.cookies), "attachment_type": AttachmentType.JSON
				})
		return step_uuid
	
	def stop_step(self, report, step_uuid=""):
		step_uuid = step_uuid or report.step_uuid
		if report.test_id in self._test_data and \
		   step_uuid in { x["step_uuid"] for x in self._test_data[report.test_id] if x["method"] == "start_step" }:
			self._test_data[report.test_id].append({
				"method": "stop_step", "time": get_time(report)+int(1000 * report.time)
				, "step_uuid": step_uuid, "status": get_status(report.step_status)
				, "detail": StatusDetails(message=report.message) if report.message else None
			})
	
	def action(self, report):
		if not report.status in (TEST_STATUS.FAILED, TEST_STATUS.FATAL, TEST_STATUS.XFAILED, TEST_STATUS.TIMEOUT, TEST_STATUS.ERROR):
			self.stop_step(report, self.start_step(report, uuid4()))
	
	def warning(self, report):
		if report.test_id in self._test_data:
			step_uuid = self.start_step(report, uuid4())
			self._test_data[report.test_id].append({
				"method": "attach_data", "uuid": uuid4(), "message": report.message
				, "name": "warning", "attachment_type": AttachmentType.TEXT
			})
			self.stop_step(report, step_uuid)
	
	def error(self, report):
		if report.test_id in self._test_data and report.test_id in self._tests:
			step_uuid = self.start_step(report, uuid4())
			
			debug = dict(report.debug)
			debug["Тайм-аут"] = report.timeout
			if report.test_scope:
				debug["scope"] = "\n".join( "%r : %r" % (x, y) for x, y in report.test_scope.items() )
			
			test = self._tests[report.test_id]
			traceback = debug.pop("traceback", "")
			test.statusDetails = StatusDetails(
				message=report.action_name+"\n"+report.message
				, trace=(report.file_line or traceback)
				, flaky=report.test_args["flaky"]
			)
			self._test_data[report.test_id].append({
				"method": "parameters", "uuid": step_uuid, "parameters": debug
			})
			if report.url and not report.url in [ x.url for x in test.links ]:
				test.links.append(Link(LinkType.LINK, report.url, report.page_name))
			test.status = get_status(report.status)
			self.stop_step(report, step_uuid)
	timeout_error = fatal_error = test_error = error
	
	def _stop_test(self, test_id, report=None):
		if test_id in self._tests and test_id in self._test_data:
			test     = self._tests[test_id]
			reporter = AllureReporter()
			reporter.schedule_test(test_id, test)
			for x in self._test_data.get(test_id, []):
				if x["method"]   == "start_step":
					reporter.start_step(x["parent_uuid"], x["step_uuid"], x["step"])
				elif x["method"] == "stop_step":
					reporter.stop_step(x["step_uuid"], stop=x["time"], status=x["status"], statusDetails=x["detail"])
				elif x["method"] == "attach_data" and x["message"]:
					reporter.attach_data(x["uuid"], x["message"], x["name"], attachment_type=x["attachment_type"])
				elif x["method"] == "parameters":
					reporter.get_item(x["uuid"]).parameters.extend([
						Parameter(name=k, value=represent(v)) for k, v in x["parameters"].items()
					])
			if report:
				for k, v in report.logs.items():
					reporter.attach_data(uuid4(), json.dumps(v), "log_"+k, attachment_type=AttachmentType.JSON)
			reporter.close_test(test_id)
			
			del self._test_data[test_id]
			del self._tests[test_id]
	
	def stop_test(self, report):
		if report.test_id in self._tests:
			test        = self._tests[report.test_id]
			test.status = get_status(report.test_status)
			test.stop   = get_time(report)
			for name, url in report.links.items():
				if url and not url in [ x.url for x in test.links ]:
					test.links.append(Link(LinkType.LINK, url, name))
			self._stop_test(report.test_id, report)
	skip_test = stop_test
	
	def close(self):
		for k, v in self._tests.copy().items():
			v.status = get_status(v.status or TEST_STATUS.STOPPED)
			v.stop   = get_time()
			self._stop_test(k)

ItemConfig.registrator(
	ItemConfig("ALLURE_LOG_DIR", str, os.path.join(CONFIG.REPORT_DIR, "allure_log"), "ALLURE: Путь к каталогу отчёта")
	, ItemConfig("ALLURE_CLEAR", bool, False, "ALLURE: Очищать каталог отчёта")
)

ItemTestConfig.registrator(
	ItemTestConfig("epic",       str, lambda test: test.filename[len(CONFIG.TEST_CONFIG_PATH):])
	, ItemTestConfig("feature",  str, "")
	, ItemTestConfig("story",    str, "")
	, ItemTestConfig("issue",    str, "")
	, ItemTestConfig("owner",    str, "")
	, ItemTestConfig("tms_link", str, "")
)
