# -*- encoding: latin-1 -*-

#============================================================================ 

#Name        : test_ats4.py 

#Part of     : Helium 

#Copyright (c) 2009 Nokia Corporation and/or its subsidiary(-ies).

#All rights reserved.

#This component and the accompanying materials are made available

#under the terms of the License "Eclipse Public License v1.0"

#which accompanies this distribution, and is available

#at the URL "http://www.eclipse.org/legal/epl-v10.html".

#

#Initial Contributors:

#Nokia Corporation - initial contribution.

#

#Contributors:

#

#Description:

#===============================================================================

""" Testing ats4 framework. """

#E1101 => Mocker shows mockery

#C0302 => too many lines

#W0142 => used * or ** magic 

#W0603 => used global

#R* during refactoring

from cStringIO import StringIO

from xml.etree.ElementTree import fromstring

from xml.etree import ElementTree as et

import difflib

import logging

logging.getLogger().setLevel(logging.ERROR)

import tempfile

import zipfile

import os

import re

import subprocess

import fileutils

import amara

import ntpath

import ats3

import ats3.testconfigurator as atc

import ats3.dropgenerator as adg

import ats3.parsers as parser

TEST_PATH = None

TEST_FILES = {}

TSRC = None

OUTPUT = None

# Shortcuts

E = et.Element

SE = et.SubElement

class Bunch(object):

    """ Configuration object. Argument from constructor are converted into class attributes. """

    def __init__(self, **kwargs):

        self.__dict__.update(kwargs)

def equal_xml(xml1, xml2):

    """Check the equality of the given XML snippets.

    Tag name equality:

    >>> equal_xml('<a/>', '<a/>')

    True

    >>> equal_xml('<a/>', '<b/>')

    False

    Attribute equality:

    >>> equal_xml('<a k="v"/>', '<a k="v"/>')

    True

    >>> equal_xml('<a k="v"/>', '<a k="w"/>')

    False

    Text content equality:

    >>> equal_xml('<a>v</a>', '<a>v</a>')

    True

    >>> equal_xml('<a>v</a>', '<a>w</a>')

    False

    >>> equal_xml('<a>v</a>', '<a></a>')

    False

    Text content equality when whitespace differs:

    >>> equal_xml('<a>v</a>', '<a>v </a>')

    True

    Equality of child elements:

    >>> equal_xml('<a><b><c k="v"/></b></a>', '<a><b><c k="v"/></b></a>')

    True

    >>> equal_xml('<a><b><c k="v"/></b></a>', '<a><b><c k="w"/></b></a>')

    False

    >>> equal_xml('<a><b><c k="v"/>v</b></a>', '<a><b><c k="v"/>w</b></a>')

    False

    >>> equal_xml('<a><b><c k="v"/>v</b></a>', '<a><b><c k="v"/>v </b></a>')

    True

    """

    def __init__():

        check_instance(xml1, xml2)

    def check_instance(xml1, xml2):

        """if xml1 and xml2 are instances, converts to strings"""

        if isinstance(xml1, basestring):

            xml1 = fromstring(xml1)

        if isinstance(xml2, basestring):

            xml2 = fromstring(xml2)

        check_tags(xml1, xml2)

    def check_tags(xml1, xml2):

        """check xml tags and text equality"""

        if xml1.tag != xml2.tag:

            return False

        if xml1.attrib != xml2.attrib:

            return False

        if xml1.text:

            if not xml2.text:

                return False

        if xml2.text:

            if not xml1.text:

                return False

        produce_xml_children(xml1, xml2)

    def produce_xml_children(xml1, xml2):

        """checks if xml children are of same length and are equal?"""

        if xml1.text and xml2.text and xml1.text.strip() != xml2.text.strip():

            return False

        if xml1.tail is not None and xml2.tail is not None:

            if xml1.tail.strip() != xml2.tail.strip():

                return False

        elif xml1.tail != xml2.tail:

            return False

        children1 = list(xml1.getchildren())

        children2 = list(xml2.getchildren())

        if len(children1) != len(children2):

            return False

        for child1, child2 in zip(children1, children2):

            return equal_xml(child1, child2)

    return True        

def setup_module():

    """ Setup test environment. """

    global TEST_PATH, TSRC, OUTPUT

    TEST_PATH = path(tempfile.mkdtemp())

    OUTPUT = TEST_PATH.joinpath(r"output")

    component = TEST_PATH

    component.joinpath("group").makedirs()

    for path_parts in (("tsrc", "group", "bld.inf"),

                       ("tsrc", "group", "test.pkg"),

                       ("tsrc", "testmodules", "file1.dll"),

                       ("tsrc", "testmodules", "file2.dll"),

                       ("tsrc", "data", "file1"),

                       ("tsrc", "data", "file2"),

                       ("tsrc", "data", "file3"),

                       ("tsrc", "data", "mmc", "file4"),

                       ("tsrc", "data", "c", "file5"),

                       ("tsrc", "conf", "file1.cfg"),

                       ("tsrc", "conf", "file2.cfg"),

                       ("tsrc", "init", "TestFramework.ini"),

                       ("tsrc", "custom", "prepostaction.xml"),

                       ("tsrc", "custom", "postpostaction.xml"),

                       # These do not have to be under 'tsrc':

                       ("tsrc", "output", "images", "file1.fpsx"),

                       ("tsrc", "output", "images", "file2.fpsx"),

                       ("tsrc", "sis", "file1.sisx"),

                       ("tsrc", "sis", "file2.sisx"),

                       ("tsrc", "sis", "file3.sisx"),

                       ("tsrc", "trace_init", "trace_activation_1.xml")):

        filepath = component.joinpath(*path_parts)

        if not filepath.parent.exists():

            filepath.parent.makedirs()

        filepath.touch()

        TEST_FILES.setdefault(path_parts[1], []).append(filepath)

    TSRC = component.joinpath("tsrc")

    filepath = OUTPUT.joinpath("pmd", "pmd_file.pmd")

    if not filepath.parent.exists():

        filepath.parent.makedirs()

    filepath.touch()

    TEST_FILES.setdefault("pmd_file", []).append(filepath)

    tracing = component.joinpath("tsrc", "trace_init")

    root = E('ConfigurationFile')

    confs = E("Configurations")

    trace_act = SE(confs, "TraceActivation")

    conf = SE(trace_act, "Configuration")

    conf.set('Name', 'MCU')

    mcu = SE(conf, "MCU")

    sett = SE(mcu, "settings")

    SE(sett, "timestamp")

    root.append(confs)

    ettree = et.ElementTree(root)

    doc = amara.parse(et.tostring(ettree.getroot()))

    handle = open(tracing.joinpath("trace_activation_1.xml"), "w")

    handle.write(doc.xml(indent="yes"))

    handle.close()

#    tracing.writestr("trace_activation_1.xml", doc.xml(indent=u"yes"))

    group = component.joinpath("tsrc", "group")

    group.joinpath("bld.inf").write_text(

        r"""

        PRJ_TESTMMPFILES

        stif.mmp /* xyz.mmp */ abcd.mmp

        /*xyz.mmp*/

        eunit.mmp /* xyz.mmp */

        both.mmp

        ..\sub-component\group\sub-component.mmp

        """.replace('\\', os.sep))

    group.joinpath("test.pkg").write_text(

        r"""

        ;Language - standard language definitions

        &EN

        ; standard SIS file header

        #{"BTEngTestApp"},(0x04DA27D5),1,0,0

        ;Supports Series 60 v 3.0

        (0x101F7961), 0, 0, 0, {"Series60ProductID"}

        ;Localized Vendor Name

        %{"BTEngTestApp"}

        ;Unique Vendor name

        :"Nokia"

        ; Files to copy

        "\tsrc\testmodules\file1.dll"-"c:\sys\bin\file1.dll"

        "\tsrc\testmodules\file2.dll"-"c:\sys\bin\file2.dll"

        "\tsrc\data\file1"-"e:\sys\bin\file1"

        "\tsrc\data\file2"-"e:\sys\bin\file2"

        "\tsrc\data\file3"-"e:\sys\bin\file3"

        "\tsrc\conf\file1.cfg"-"!:\sys\bin\file1.cfg"

        "\tsrc\conf\file2.cfg"-"!:\sys\bin\file2.cfg"

        "\tsrc\init\TestFramework.ini"-"!:\sys\bin\TestFramework.ini"

        "\tsrc\sis\file1.sisx"-"!:\sys\bin\file1.sisx"

        "\tsrc\sis\file2.sisx"-"!:\sys\bin\file2.sisx"

        """.replace('\\', os.sep))

    group.joinpath("stif.mmp").write_text("LIBRARY stiftestinterface.lib")

    group.joinpath("eunit.mmp").write_text("LIBRARY eunit.lib")

    group.joinpath("both.mmp").write_text("LIBRARY stiftestinterface.lib eunit.lib")

    init = component.joinpath("tsrc", "init")

    init.joinpath("TestFramework.ini").write_text(

        r"""

#     - Sets a device reset module's dll name(Reboot).

#        + If Nokia specific reset module is not available or it is not correct one

#          StifHWResetStub module may use as a template for user specific reset

#          module. 

[Engine_Defaults]

TestReportMode= FullReport        # Possible values are: 'Empty', 'Summary', 'Environment',

                                                               'TestCases' or 'FullReport'

CreateTestReport= YES            # Possible values: YES or NO

TestReportFilePath= C:\LOGS\TestFramework\

TestReportFileName= TestReport

TestReportFormat= TXT            # Possible values: TXT or HTML

        """)

def teardown_module():

    """ Cleanup environment after testing. """    

    def __init__():

        path(TEST_PATH).rmtree()

# CTC related functions    

def check_ctc_write(steps):

    """Checks if CTC data is written on the CTC log"""

    step = steps.next()

    assert step.findtext("./type") == "NonTestExecuteTask"

    params = step.findall("./parameters/parameter")

    assert params[0].get("value") == "writelocal"

    assert params[1].get("value") == path(r"z:\sys\bin\ctcman.exe")

    step = steps.next()

    assert step.findtext("./type") == "NonTestExecuteTask"

    params = step.findall("./parameters/parameter")

    assert params[0].get("value") == "writefile"

    assert params[1].get("value") == path(r"z:\sys\bin\ctcman.exe")

def check_ctc_log(steps, testtype=""):

    """Fetches CTC Log"""

    #For the ctcdata.txt to be published on the ATS network drive

    step = steps.next()

    assert step.findtext("./type") == "FileDownloadTask"

    params = step.findall("./parameters/parameter")

    #assert params[0].get("value") == "false"

    assert params[0].get("value") == path(r"c:\data\ctc\ctcdata.txt")

def check_ctc_start(steps):

    """Checks if CTC starts in ATS"""

    step = steps.next()

    assert step.findtext("./type") == "CreateDirTask"

    params = step.findall("./parameters/parameter")

    assert params[0].get("value") == path(r"c:\data\ctc")

    step = steps.next()

    assert step.findtext("./type") == "NonTestExecuteTask"

    params = step.findall("./parameters/parameter")

    assert params[0].get("value") == path(r"z:\sys\bin\ctcman.exe")   

def check_fetch_logs(steps, harness="STIF"):

    """Checks fetching log directory is created"""

    step = steps.next()

    assert step.findtext("./type") == "FileDownloadTask"

    params = step.findall("./parameters/parameter")

    if harness == "STIF":

        assert params[0].get("value") == path(r"c:\logs\testframework\*")  

    else:

        assert params[0].get("value") == path(r"c:\Shared\EUnit\logs\*")

def check_diamonds_action(action):

    """ Testing Diamonds action. """

    assert action.findtext("./type") == "DiamondsAction"

def check_send_email_action(action, reportemail):

    """verifies if sening email option works"""

    assert action.findtext("./type") == "EmailAction"

    params = action.findall("./parameters/parameter")

    assert params[0].get("name") == "subject"

    #assert params[0].get("value") == "email subject"

    assert params[1].get("name") == "to"

    assert params[1].get("value") == reportemail

def check_ctc_run_process_action(action):

    """verifies if CTC run process action works"""

    #The parameters for this test are intended to execute on a windows machine

    assert action.findtext("./type") == "RunProcessAction"

    params = action.findall("./parameters/parameter")

    assert params[0].get("name") == "file"

    assert params[0].get("value") == "catsctc2html/catsctc2html.exe"

    assert params[1].get("name") == "parameters"

    assert params[1].get("value") == r"--ctcdata_files=\\10.0.0.1\ctc_helium\1234\drop0\set0\ctcdata --monsym_files=\\10.0.0.1\ctc_helium\1234\mon_syms\1\MON.sym --diamonds_build_id=1234 --drop_id=0 --total_amount_of_drops=1"

def check_ats_report_action(action, filestore):

    """verifies if sening ATS report option works"""

    assert action.findtext("./type") == "FileStoreAction"

    params = action.findall("./parameters/parameter")

    assert params[0].get("name") == "to-folder"

    assert params[0].get("value") == filestore

    assert params[1].get("name") == "report-type"

    assert params[1].get("value") == "ATS_REPORT"

    assert params[2].get("name") == "date-format"

    assert params[2].get("value") == "yyyyMMdd"

    assert params[3].get("name") == "time-format"

    assert params[3].get("value") == "HHmmss"

def check_stif_report_action(action, filestore):

    """verifies if sening STIF report option works"""

    assert action.findtext("./type") == "FileStoreAction"

    params = action.findall("./parameters/parameter")

    assert params[0].get("name") == "to-folder"

    assert params[0].get("value") == filestore

    assert params[1].get("name") == "report-type"

    assert params[1].get("value") == "STIF_COMPONENT_REPORT_ALL_CASES"

    assert params[2].get("name") == "run-log"

    assert params[2].get("value") == "true"

    assert params[3].get("name") == "date-format"

    assert params[3].get("value") == "yyyyMMdd"

    assert params[4].get("name") == "time-format"

    assert params[4].get("value") == "HHmmss"

def check_log_dir(steps):

    """ Test the log dir. """

    step = steps.next()

    assert step.findtext("./type") == "CreateDirTask"

def check_trace_start(steps, filestore):

    """Checks if tracing is started on the ATS"""

    step = steps.next()

    assert step.findtext("./type") == "trace-start"

    params = step.findall("./parameters/parameter")

    assert params[0].get("value") == path(r"ATS3Drop\set0\trace_activation\trace_activation_1.xml")

    assert params[1].get("value") == "MCU"

    assert params[2].get("value") == path(r"ATS3Drop\pmds\pmd_file.pmd")

    assert params[3].get("value") == filestore.joinpath("traces", "set0", "tracelog.blx")

    assert params[4].get("value") == "60"

    assert params[5].get("value") == "yyyyMMdd"

    assert params[6].get("value") == "HHmmss"

def check_trace_end_steps(steps, filestore):

    """ Test trace end step. """

    step = steps.next()

    assert step.findtext("./type") == "trace-stop"

    params = step.findall("./parameters/parameter")

    assert params[0].get("value") == "60"

    step = steps.next()

    assert step.findtext("./type") == "trace-convert"

    params = step.findall("./parameters/parameter")

    assert params[0].get("value") == path(r"ATS3Drop\pmds\pmd_file.pmd")

    assert params[1].get("value") == filestore.joinpath("traces", "set0", "tracelog.blx")

    assert params[2].get("value") == "60"

    assert params[3].get("value") == "yyyyMMdd"

    assert params[4].get("value") == "HHmmss"

class TestTestPlan(mocker.MockerTestCase):

    """Creates TestPlan mocker setup"""

    def __init__(self, methodName="runTest"):

        mocker.MockerTestCase.__init__(self, methodName)  

    def setUp(self):

        """ Setup TestTestPlan testsuite. """

        opts = Bunch(testrun_name="testrun", harness="STIF", 

                     device_type="product", plan_name="ats3_test_plan",

                     diamonds_build_url="", report_email="", file_store=path(), test_timeout="60",

                     device_hwid="5425", trace_enabled="True", ctc_enabled="True", eunitexerunner_flags="/E S60AppEnv /R Off", 

                     multiset_enabled=False, ctc_run_process_params=r"10.0.0.1#x:\ats\drop0.zip#1", monsym_files="", hti="True")

        self.atp = ats3.Ats3TestPlan(opts)

        self.config_files = TEST_FILES["conf"]

        self.data_files = TEST_FILES["data"]

        self.engine_ini_file = TEST_FILES["init"][0]

        self.image_files = TEST_FILES["output"]

        self.sis_files = TEST_FILES["sis"]

        self.testmodule_files = TEST_FILES["testmodules"]

        self.ctc_enabled = self.atp["ctc_enabled"]

        self.custom_dir = "CustomD"

        self.eunitexerunner_flags = self.atp["eunitexerunner_flags"]

        if self.atp["trace_enabled"].lower() == "true":

            self.trace_activation_files = TEST_FILES["trace_init"]    

            self.pmd_files = TEST_FILES["pmd_file"]

        else:

            self.trace_activation_files = []    

            self.pmd_files = []

        self.test_timeout = self.atp["test_timeout"]

        self.eunitexerunner_flags = self.atp["eunitexerunner_flags"]

        self.device_hwid = self.atp["device_hwid"]

        self.test_harness = self.atp["harness"]

        self.src_dst = [("" + os.sep + "tsrc" + os.sep + "testmodules" + os.sep + "file1.dll", "c:\sys\bin\file1.dll", "testmodule"),

            ("" + os.sep + "tsrc" + os.sep + "testmodules" + os.sep + "file2.dll", "c:\sys\bin\file2.dll", "testmodule"),

            ("" + os.sep + "tsrc" + os.sep + "data" + os.sep + "file1", "e:\sys\bin\file1", "data"),

            ("" + os.sep + "tsrc" + os.sep + "data" + os.sep + "file2", "e:\sys\bin\file2", "data"),

            ("" + os.sep + "tsrc" + os.sep + "data" + os.sep + "file3", "e:\sys\bin\file3", "data"),

            ("" + os.sep + "tsrc" + os.sep + "conf" + os.sep + "file1.cfg", "c:\sys\bin\file1.cfg", "conf"),

            ("" + os.sep + "tsrc" + os.sep + "conf" + os.sep + "file2.cfg", "c:\sys\bin\file2.cfg", "conf"),

            ("" + os.sep + "tsrc" + os.sep + "init" + os.sep + "TestFramework.ini", "c:\sys\bin\TestFramework.ini", "engine_ini"),

            ("" + os.sep + "tsrc" + os.sep + "sis" + os.sep + "file1.sisx", "c:\sys\bin\file1.sisx", ""),

            ("" + os.sep + "tsrc" + os.sep + "sis" + os.sep + "file2.sisx", "c:\sys\bin\file2.sisx", ""),]

        self.component_path = str(TEST_PATH.joinpath("tsrc", "group"))

    def test_creation(self):

        """ Testing creation. """        

        assert self.atp["testrun_name"] == "testrun"

        assert self.atp["harness"] == "STIF"

        assert self.atp["device_type"] == "product"

    def test_insert_set(self):

        """ Inserting a set of file. """        

        self.atp.insert_set(data_files=self.data_files,

                           config_files=self.config_files,

                           engine_ini_file=self.engine_ini_file,

                           image_files=self.image_files,

                           testmodule_files=self.testmodule_files,

                           test_timeout=self.test_timeout,

                           eunitexerunner_flags=self.eunitexerunner_flags,

                           pmd_files=self.pmd_files,                            

                           trace_activation_files=self.trace_activation_files,

                           component_path=self.component_path)

        self.atp.insert_set(image_files=self.image_files,                           

                           engine_ini_file=self.engine_ini_file,

                           sis_files=self.sis_files,

                           test_timeout=self.test_timeout,

                           eunitexerunner_flags=self.eunitexerunner_flags,

                           pmd_files=self.pmd_files,                            

                           trace_activation_files=self.trace_activation_files,

                           component_path=self.component_path)

        self.atp.insert_set(data_files=self.data_files,

                           config_files=self.config_files,

                           engine_ini_file=self.engine_ini_file,

                           image_files=self.image_files,

                           testmodule_files=self.testmodule_files,

                           test_timeout=self.test_timeout,

                           eunitexerunner_flags=self.eunitexerunner_flags,

                           src_dst=self.src_dst,

                           pmd_files=self.pmd_files,                            

                           trace_activation_files=self.trace_activation_files,

                           component_path=self.component_path)

        self.atp.insert_set(engine_ini_file=self.engine_ini_file,

                           test_timeout=self.test_timeout,

                           eunitexerunner_flags=self.eunitexerunner_flags,

                           image_files=self.image_files,

                           test_harness=self.test_harness,

                           src_dst=self.src_dst,

                           pmd_files=self.pmd_files,                            

                           trace_activation_files=self.trace_activation_files,

                           component_path=self.component_path)

        self.atp.insert_set(test_timeout=self.test_timeout,      

                           eunitexerunner_flags=self.eunitexerunner_flags,               

                           image_files=self.image_files,                        

                           test_harness=self.test_harness,                      

                           src_dst=self.src_dst,                                

                           pmd_files=self.pmd_files,                            

                           trace_activation_files=self.trace_activation_files,

                           component_path=self.component_path)  

        assert self.atp.sets[0] == dict(name="set0",

                                       data_files=self.data_files,

                                       config_files=self.config_files,

                                       engine_ini_file=self.engine_ini_file,

                                       image_files=self.image_files,

                                       testmodule_files=self.testmodule_files,

                                       test_timeout=self.test_timeout,

                                       eunitexerunner_flags=self.eunitexerunner_flags,

                                       test_harness=self.test_harness,

                                       src_dst=[],

                                       pmd_files=self.pmd_files,

                                       trace_path=self.atp.file_store.joinpath(u"§RUN_NAME§" + os.sep + u"§RUN_START_DATE§_§RUN_START_TIME§", "traces", "set0", "tracelog.blx"),

                                       trace_activation_files=self.trace_activation_files,

                                       ctc_enabled=self.atp["ctc_enabled"],

                                       custom_dir=None,

                                       component_path=self.component_path)

        assert self.atp.sets[1] == dict(name="set1",