system-tests/test_driver.py at master · alegrey91/system-tests · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
import os
import shutil
import sys
import time
import traceback

import junit_xml

from configurations.system import customers, tests, backends
from infrastructure.backend_api import ControlPanelAPI
from systest_utils import Logger, statics
from systest_utils import systests_utilities


class TestDriver(object):
    def __init__(self,
                 test_name: str,
                 backend: str,
                 customer: str,
                 temp_dir: str = "temp",
                 fresh: bool = True,
                 # duration: int = 3,
                 **kwargs):
        # set objects
        self.test_name = test_name
        self.credentials_obj: customers.Credentials = customers.CREDENTIALS
        self.backend_obj: backends.Backend = backends.BACKENDS[backend]

        # other test features
        self.agent_location = kwargs["agent"] if "agent" in kwargs else None
        self.temp_dir = os.path.abspath(temp_dir)
        self.refresh = fresh
        self.duration = systests_utilities.TestUtil.get_arg_from_dict(kwargs, "duration", 3) * 60

        self.kwargs = self.parse_kwargs(kwargs)

    def main(self):

        # ControlPanelAPI
        backend = ControlPanelAPI(user_name=self.credentials_obj.get_name(),
                                  password=self.credentials_obj.get_password(),
                                  customer=self.credentials_obj.get_customer(),
                                  client_id=self.credentials_obj.get_client_id(),
                                  secret_key=self.credentials_obj.get_secret_key(),
                                  url=self.backend_obj.get_dashboard_url(),
                                  auth_url=self.backend_obj.get_auth_url(),
                                  login_method=self.backend_obj.get_login_method())

        status = statics.FAILURE
        summary = ""
        err = ""

        try:
            systests_utilities.TestUtil.create_dir(self.temp_dir)
            systests_utilities.TestUtil.create_dir(statics.DEFAULT_XML_PATH, override=False)

            status, summary = self.run_test(backend=backend)
        except Exception as e:
            status = statics.FAILURE
            err = e
            Logger.logger.error(e)
            summary = e
        finally:
            self.clear()
            self.final_report(status=status, err=err, summary=summary)
        return not status

    def run_test(self, backend: ControlPanelAPI = None):
        test_obj = tests.get_test(self.test_name)
        test_class_obj = test_obj.test_obj(test_driver=self, backend=backend, test_obj=test_obj)
        start = time.time()
        try:
            status, summary = test_class_obj.start()
        except Exception as ex:
            status = statics.FAILURE
            _, _ = test_class_obj.cleanup()
            summary = ex
            Logger.logger.error("error: {}".format(traceback.print_exc()))
        finally:
            Logger.logger.info('time: {}'.format(systests_utilities.TestUtil.get_time(start, time.time())))
        return status, summary

    def clear(self):
        Logger.logger.info("test driver clearing")
        # remove temp directory
        try:
            shutil.rmtree(self.temp_dir)
        except Exception as e:
            Logger.logger.error(e)

    # ======================== report ==================================
    def final_report(self, status, err, summary):
        test_case = junit_xml.TestCase(name=self.test_name,
                                       classname=self.test_name,
                                       stdout=sys.stdout.output_buf,
                                       stderr=sys.stderr.output_buf)
        if status == statics.SUCCESS:
            Logger.logger.success("test {} status: SUCCESS".format(self.test_name))
        else:
            Logger.logger.error("test {} status: FAILURE".format(self.test_name))
            Logger.logger.error(summary)
            test_case.add_error_info(message=err, output=summary)

        test_suite = junit_xml.TestSuite(name="system-test", test_cases=[test_case])

        xml_file = os.path.join(statics.DEFAULT_XML_PATH, "{}.xml".format(self.test_name))
        with open(xml_file, 'w') as result:
            result.write(junit_xml.TestSuite.to_xml_string([test_suite]))
        Logger.logger.debug("xml file saved {}".format(xml_file))

    @staticmethod
    def parse_kwargs(kwargs: dict):
        if not kwargs["kwargs"]:
            return kwargs
        for i in kwargs["kwargs"]:
            arg = i.split('=')
            val = arg[1].split(",") if len(arg) > 1 else arg
            kwargs[arg[0]] = val if len(val) > 1 else val[0]
        return kwargs