class SeleniumTestCase(BZTestCase):
    """
    :type obj: SubprocessedExecutor
    """
    def __init__(self, methodName='runTest'):
        super(SeleniumTestCase, self).__init__(methodName)
        self.obj = None

    def setUp(self):
        super(SeleniumTestCase, self).setUp()
        self.engine = EngineEmul()
        paths = [local_paths_config()]
        self.engine.configure(paths)  # FIXME: avoid using whole engine in particular module test!

        self.virtual_display = VirtualDisplay()
        self.virtual_display.engine = self.engine
        self.virtual_display.startup()

        self.obj = SeleniumExecutor()
        self.obj.engine = self.engine
        self.obj.settings = self.engine.config.get("modules").get("selenium")
        self.obj.env = self.obj.engine.env

    def tearDown(self):
        self.virtual_display.shutdown()
        super(SeleniumTestCase, self).tearDown()

    def configure(self, config):
        self.obj.engine.config.merge(config)
        self.obj.execution = self.obj.engine.config.get('execution')
        if isinstance(self.obj.execution, list):
            self.obj.execution = self.obj.execution[0]

class SeleniumTestCase(BZTestCase):
    def setUp(self):
        super(SeleniumTestCase, self).setUp()
        self.engine_obj = EngineEmul()
        self.paths = [ABS_PATH("/../../bzt/10-base.json"), local_paths_config()]
        self.engine_obj.configure(self.paths)
        self.selenium_config = self.engine_obj.config["modules"]["selenium"]

class SeleniumTestCase(BZTestCase):
    def setUp(self):
        super(SeleniumTestCase, self).setUp()
        self.engine_obj = EngineEmul()
        self.paths = [__dir__() + "/../../bzt/10-base.json", local_paths_config()]
        self.engine_obj.configure(self.paths)  # FIXME: avoid using whole engine in particular module test!
        self.engine_obj.config.get("modules").get("selenium").merge({"virtual-display": {"width": 1024, "height": 768}})
        self.selenium_config = self.engine_obj.config["modules"]["selenium"]

 def setUp(self):
     super(TestTestNGTester, self).setUp()
     engine_obj = EngineEmul()
     paths = [local_paths_config()]
     engine_obj.configure(paths)
     self.obj = TestNGTester()
     self.obj.settings = engine_obj.config.get("modules").get("testng")
     self.obj.engine = engine_obj
     self.obj.env = self.obj.engine.env

 def test_encode_decode_infinities(self):
     engine = EngineEmul()
     obj = Configuration()
     obj.merge({
         "foo": float("inf"),
     })
     cfg = engine.create_artifact("config", ".json")
     obj.dump(cfg, Configuration.JSON)
     with open(cfg) as fds:
         dump = json.loads(fds.read())
     self.assertEqual(dump["foo"], "inf")
     self.assertEqual(dehumanize_time(dump["foo"]), float("inf"))

 def setUp(self):
     super(SeleniumTestCase, self).setUp()
     self.engine_obj = EngineEmul()
     self.engine_obj.artifacts_base_dir = ABS_PATH("/../../build/test")
     self.paths = [ABS_PATH("/../../bzt/10-base.json"), local_paths_config()]
     self.engine_obj.configure(self.paths)
     self.selenium_config = self.engine_obj.config["modules"]["selenium"]

    def setUp(self):
        super(TestJUnitTester, self).setUp()
        engine_obj = EngineEmul()
        paths = [local_paths_config()]
        engine_obj.configure(paths)

        # just download geckodriver & chromedriver with selenium
        selenium = SeleniumExecutor()
        selenium.engine = engine_obj
        selenium.install_required_tools()
        for driver in selenium.webdrivers:
            selenium.env.add_path({"PATH": driver.get_driver_dir()})

        self.obj = JUnitTester()
        self.obj.env = selenium.env
        self.obj.settings = engine_obj.config.get("modules").get("junit")
        self.obj.engine = engine_obj

    def test_functional_report(self):
        engine = EngineEmul()
        aggregator = FunctionalAggregator()
        aggregator.engine = engine
        engine.aggregator = aggregator

        obj = JUnitXMLReporter()
        obj.engine = engine
        obj.parameters = BetterDict()

        reader = FuncSamplesReader(RESOURCES_DIR + "functional/nose.ldjson", engine, ROOT_LOGGER)
        aggregator.add_underling(reader)

        aggregator.prepare()
        obj.prepare()
        aggregator.post_process()
        obj.post_process()

        self.assertFilesEqual(obj.report_file_path, RESOURCES_DIR + "functional/xunit-report.xml")

    def setUp(self):
        super(SeleniumTestCase, self).setUp()
        self.engine = EngineEmul()
        paths = [local_paths_config()]
        self.engine.configure(paths)  # FIXME: avoid using whole engine in particular module test!

        self.virtual_display = VirtualDisplay()
        self.virtual_display.engine = self.engine
        self.virtual_display.startup()

        self.obj = SeleniumExecutor()
        self.obj.engine = self.engine
        self.obj.settings = self.engine.config.get("modules").get("selenium")
        self.obj.env = self.obj.engine.env

class TestSwagger2YAML(BZTestCase):
    def setUp(self):
        super(TestSwagger2YAML, self).setUp()
        self.engine = EngineEmul()

    def _get_swagger2yaml(self, path, file_name=None):
        return Swagger2YAML(FakeOptions(file_name=file_name), RESOURCES_DIR + path)

    def _get_tmp(self, prefix='test', suffix='.yml'):
        return self.engine.create_artifact(prefix, suffix)

    def test_convert(self):
        self.maxDiff = None
        source = RESOURCES_DIR + "/swagger/petstore.json"
        result = self._get_tmp()
        options = FakeOptions(file_name=result)
        process(options, [source])
        actual = yaml.load(open(result).read())
        expected = yaml.load(open(RESOURCES_DIR + "/swagger/petstore-converted.yaml").read())
        self.assertEqual(actual, expected)

class TestConverter(BZTestCase):
    def setUp(self):
        super(TestConverter, self).setUp()
        self.engine = EngineEmul()

    def _get_soapui2yaml(self, path, file_name=None, test_case=None):
        return SoapUI2YAML(FakeOptions(file_name=file_name, test_case=test_case), __dir__() + path)

    def _get_tmp(self, prefix='test', suffix='.yml'):
        return self.engine.create_artifact(prefix, suffix)

    def test_convert(self):
        source = RESOURCES_DIR + "soapui/project.xml"
        result = self._get_tmp()
        options = FakeOptions(file_name=result, test_case="index")
        process(options, [source])
        actual = yaml.load(open(result).read())
        expected = yaml.load(open(RESOURCES_DIR + "soapui/project.xml.yml").read())
        self.assertEqual(actual, expected)

    def test_flickr(self):
        source = RESOURCES_DIR + "soapui/flickr-sample.xml"
        result = self._get_tmp()
        options = FakeOptions(file_name=result)
        process(options, [source])
        actual = yaml.load(open(result).read())
        expected = yaml.load(open(RESOURCES_DIR + "soapui/flickr-sample.xml.yml").read())
        self.assertEqual(actual, expected)

    def test_egalaxy(self):
        source = RESOURCES_DIR + "soapui/egalaxy.xml"
        result = self._get_tmp()
        options = FakeOptions(file_name=result)
        process(options, [source])
        actual = yaml.load(open(result).read())
        expected = yaml.load(open(RESOURCES_DIR + "soapui/egalaxy.xml.yml").read())
        self.assertEqual(actual, expected)

 def setUp(self):
     super(TestEngine, self).setUp()
     self.obj = EngineEmul()
     self.paths = local_paths_config()

class TestEngine(BZTestCase):
    def setUp(self):
        super(TestEngine, self).setUp()
        self.obj = EngineEmul()
        self.paths = local_paths_config()

    def test_requests(self):
        configs = [
            __dir__() + "/../bzt/10-base.json",
            __dir__() + "/json/get-post.json",
            __dir__() + "/json/reporting.json",
            self.paths
        ]
        self.obj.configure(configs)
        self.obj.prepare()
        self.obj.prepare()
        self.obj.run()
        self.obj.post_process()

    def test_double_exec(self):
        configs = [
            __dir__() + "/../bzt/10-base.json",
            __dir__() + "/yaml/triple.yml",
            __dir__() + "/json/reporting.json",
            self.paths
        ]
        self.obj.configure(configs)
        self.obj.prepare()
        self.obj.run()
        self.obj.post_process()

    def test_unknown_module(self):
        configs = [
            __dir__() + "/../bzt/10-base.json",
            __dir__() + "/json/gatling.json",
            self.paths
        ]
        self.obj.configure(configs)
        self.obj.config["provisioning"] = "unknown"
        self.obj.config["modules"]["unknown"] = BetterDict()

        try:
            self.obj.prepare()
            self.fail()
        except ValueError:
            pass

 def setUp(self):
     super(TestConverter, self).setUp()
     self.engine = EngineEmul()

 def setUp(self):
     self.engine = EngineEmul()

 def temp_yaml(self):
     emul = EngineEmul()
     return emul.create_artifact("test", ".yml")

 def setUp(self):
     super(SeleniumTestCase, self).setUp()
     self.engine_obj = EngineEmul()
     self.paths = [__dir__() + "/../../bzt/10-base.json", local_paths_config()]
     self.engine_obj.configure(self.paths)
     self.selenium_config = self.engine_obj.config["modules"]["selenium"]

class TestConverter(BZTestCase):
    def setUp(self):
        self.engine = EngineEmul()

    def _get_jmx2yaml(self, path, file_name=None, dump_jmx=False):
        return JMX2YAML(FakeOptions(file_name=file_name, dump_jmx=dump_jmx), __dir__() + path)

    def _get_tmp(self, prefix='test', suffix='.yml'):
        return self.engine.create_artifact(prefix, suffix)

    def test_loadjmx1(self):
        log_recorder = RecordingHandler()
        obj = self._get_jmx2yaml("/jmeter/jmx/http.jmx", self._get_tmp())
        obj.log.addHandler(log_recorder)
        obj.process()
        self.assertIn("Loading jmx file", log_recorder.info_buff.getvalue())
        self.assertNotEqual("", log_recorder.debug_buff.getvalue())
        self.assertEqual("", log_recorder.err_buff.getvalue())
        obj.log.removeHandler(log_recorder)

    def test_loadjmx2(self):
        log_recorder = RecordingHandler()
        obj = self._get_jmx2yaml("/jmeter/jmx/notfound.jmx")
        obj.log.addHandler(log_recorder)
        try:
            obj.process()
            self.fail()
        except BaseException as exc:
            self.assertIn("File does not exist", exc.args[0])
        self.assertIn("Loading jmx file", log_recorder.info_buff.getvalue())
        self.assertIn("does not exist", log_recorder.err_buff.getvalue())
        self.assertEqual("", log_recorder.debug_buff.getvalue())
        obj.log.removeHandler(log_recorder)

    def test_loadjmx3(self):
        log_recorder = RecordingHandler()
        obj = self._get_jmx2yaml("/jmeter/jmx/broken.jmx")
        obj.log.addHandler(log_recorder)
        try:
            obj.process()
            self.fail()
        except BaseException as exc:
            self.assertIn("XML parsing failed", exc.args[0])
        self.assertIn("Loading jmx file", log_recorder.info_buff.getvalue())
        self.assertIn("Error while processing jmx file", log_recorder.err_buff.getvalue())
        self.assertIn("XML parsing error", log_recorder.debug_buff.getvalue())
        obj.log.removeHandler(log_recorder)

    def test_loadjmx4(self):
        log_recorder = RecordingHandler()
        obj = self._get_jmx2yaml("/jmeter/jmx/http.jmx", self._get_tmp('tmp', 'file'))
        obj.log.addHandler(log_recorder)
        obj.process()
        self.assertIn("Loading jmx file", log_recorder.info_buff.getvalue())
        self.assertIn("Done processing, result saved in", log_recorder.info_buff.getvalue())
        self.assertIn("Removing unknown element", log_recorder.warn_buff.getvalue())
        obj.log.removeHandler(log_recorder)

    def test_export_clean_jmx(self):
        tmp_jmx_name = self._get_tmp('tmp', '.jmx')
        open(tmp_jmx_name, 'w+').close()  # touch file

        yml = self._get_tmp()
        obj = self._get_jmx2yaml("/yaml/converter/disabled.jmx", yml, dump_jmx=tmp_jmx_name)
        log_recorder = RecordingHandler()
        obj.log.addHandler(log_recorder)
        obj.process()

        self.assertIn("Loading jmx file", log_recorder.info_buff.getvalue())
        self.assertIn("already exists and will be overwritten", log_recorder.warn_buff.getvalue())
        obj.log.removeHandler(log_recorder)

    def test_not_jmx(self):
        obj = self._get_jmx2yaml("/jmeter/jmx/not-jmx.xml")
        try:
            obj.process()
            self.fail()
        except BaseException as exc:
            self.assertIn("Bad jmx format", exc.args[0])

    def test_clean_disabled_jmx(self):
        yml = self._get_tmp()
        obj = self._get_jmx2yaml("/yaml/converter/disabled.jmx", yml)
        obj.process()
        disabled_elements = [element for element in obj.converter.dialect.tree.iter() if
                             element.get("enabled") == "false"]
        self.assertEquals(0, len(disabled_elements))

    def test_copy_global_csv_dataset(self):
        yml = self._get_tmp()
        obj = self._get_jmx2yaml("/yaml/converter/global_copy.jmx", yml)
        obj.process()
        yml = yaml.load(open(yml).read())
        datasets_first_tg = yml.get("scenarios").get("Thread Group one").get("data-sources")
        datasets_second_tg = yml.get("scenarios").get("Thread Group two").get("data-sources")

        global_csv_tg_one = [dataset for dataset in datasets_first_tg if dataset.get('path') == 'global.csv']
        global_csv_tg_two = [dataset for dataset in datasets_second_tg if dataset.get('path') == 'global.csv']

        local_csv_tg_one = [dataset for dataset in datasets_first_tg if dataset.get('path') == 'local.csv']
#.........这里部分代码省略.........

 def setUp(self):
     super(TestEngine, self).setUp()
     self.obj = EngineEmul()
     self.obj.artifacts_base_dir = tempfile.gettempdir() + "/bzt"
     self.paths = local_paths_config()

class TestEngine(BZTestCase):
    def setUp(self):
        super(TestEngine, self).setUp()
        self.obj = EngineEmul()
        self.paths = local_paths_config()

    def test_find_file(self):
        self.sniff_log(self.obj.log)

        config = RESOURCES_DIR + "json/get-post.json"
        configs = [config, self.paths]
        self.obj.configure(configs)
        self.assertEqual(2, len(self.obj.file_search_paths))

        self.obj.find_file(config)
        self.assertEqual("", self.log_recorder.warn_buff.getvalue())

        self.obj.find_file("reporting.json")
        self.assertIn("Guessed location", self.log_recorder.warn_buff.getvalue())

        self.obj.find_file("definitely_missed.file")
        self.assertIn("Could not find", self.log_recorder.warn_buff.getvalue())

        self.obj.find_file("http://localhost:8000/BlazeDemo.html")
        self.assertIn("Downloading http://localhost:8000/BlazeDemo.html", self.log_recorder.info_buff.getvalue())

    def test_missed_config(self):
        configs = ['definitely_missed.file']
        try:
            self.obj.configure(configs)
            self.fail()
        except TaurusConfigError as exc:
            self.assertIn('reading config file', str(exc))

    def test_configuration_smoothness(self):
        def find_ad_dict_ed(*args):
            if isinstance(args[0], dict) and not isinstance(args[0], BetterDict):
                raise BaseException("dict found in Configuration")

        configs = [
            RESOURCES_DIR + "json/get-post.json",
            self.paths]
        self.obj.configure(configs)
        self.assertTrue(isinstance(self.obj.config, Configuration))
        BetterDict.traverse(self.obj.config, find_ad_dict_ed)

    def test_requests(self):
        configs = [
            RESOURCES_DIR + "json/get-post.json",
            RESOURCES_DIR + "json/reporting.json",
            self.paths]
        self.obj.configure(configs)
        self.obj.prepare()

        for executor in self.obj.provisioning.executors:
            executor.env.set({"TEST_MODE": "files"})

        self.obj.run()
        self.obj.post_process()

    def test_double_exec(self):
        configs = [
            RESOURCES_DIR + "yaml/triple.yml",
            RESOURCES_DIR + "json/reporting.json",
            self.paths
        ]
        self.obj.configure(configs)
        self.obj.prepare()

        self.assertEquals(1, len(self.obj.services))

        for executor in self.obj.provisioning.executors:
            executor.env.set({"TEST_MODE": "files"})

        self.obj.run()
        self.obj.post_process()

    def test_unknown_module(self):
        configs = [
            RESOURCES_DIR + "json/gatling.json",
            self.paths
        ]
        self.obj.configure(configs)
        self.obj.config["provisioning"] = "unknown"
        self.obj.config["modules"]["unknown"] = BetterDict()

        self.assertRaises(TaurusConfigError, self.obj.prepare)

    def test_null_aggregator(self):
        self.obj.config.merge({
            "execution": [{
                "scenario": {
                    "requests": [{"url": "http://example.com/"}],
                }}],
            "settings": {
                "aggregator": None,
                "default-executor": "jmeter",
            },
            "modules": {
                "local": "bzt.modules.provisioning.Local",
#.........这里部分代码省略.........