avocado-framework/avocado

import os
import tempfile
import unittest
from xml.dom import minidom

from avocado import Test
from avocado.core import job
from avocado.core.result import Result
from avocado.plugins import xunit
from selftests.utils import setup_avocado_loggers, temp_dir_prefix

try:
    import xmlschema

    SCHEMA_CAPABLE = True
except ImportError:
    SCHEMA_CAPABLE = False


setup_avocado_loggers()


UNIQUE_ID = "0000000000000000000000000000000000000000"
LOGFILE = None


class ParseXMLError(Exception):
    pass


class xUnitSucceedTest(unittest.TestCase):
    def setUp(self):
        class SimpleTest(Test):
            def test(self):
                pass

        self.tmpfile = tempfile.mkstemp()
        prefix = temp_dir_prefix(self)
        self.tmpdir = tempfile.TemporaryDirectory(prefix=prefix)
        config = {
            "job.run.result.xunit.output": self.tmpfile[1],
            "run.results_dir": self.tmpdir.name,
        }
        self.job = job.Job(config)
        self.job.setup()
        self.test_result = Result(UNIQUE_ID, LOGFILE)
        self.test_result.tests_total = 1
        self.test_result.logfile = (
            "/.../avocado/job-results/job-2018-11-28T16.27-8fef221/job.log"
        )
        self.test1 = SimpleTest(config=self.job.config, base_logdir=self.tmpdir.name)
        self.test1._Test__status = "PASS"
        self.test1._Test__logfile = ""
        self.test1.time_elapsed = 678.23689

    def tearDown(self):
        self.job.cleanup()
        errs = []
        cleanups = (
            lambda: os.close(self.tmpfile[0]),
            lambda: os.remove(self.tmpfile[1]),
            self.tmpdir.cleanup,
        )
        for cleanup in cleanups:
            try:
                cleanup()
            except Exception as exc:
                errs.append(str(exc))
        self.assertFalse(
            errs, "Failures occurred during cleanup:\n%s" % "\n".join(errs)
        )

    @unittest.skipUnless(
        SCHEMA_CAPABLE, "Unable to validate schema due to missing xmlschema library"
    )
    def test_add_success(self):
        self.test_result.start_test(self.test1)
        self.test_result.end_test(self.test1.get_state())
        self.test_result.end_tests()
        xunit_result = xunit.XUnitResult()
        xunit_result.render(self.test_result, self.job)
        xunit_output = self.job.config.get("job.run.result.xunit.output")
        with open(xunit_output, "rb") as fp:
            xml = fp.read()
        try:
            dom = minidom.parseString(xml)
        except Exception as details:
            raise ParseXMLError(
                f"Error parsing XML: '{details}'.\nXML Contents:\n{xml}"
            )
        self.assertTrue(dom)

        els = dom.getElementsByTagName("testsuite")
        self.assertEqual(len(els), 1)
        self.assertEqual(els[0].attributes["time"].value, "678.237")

        els = dom.getElementsByTagName("testcase")
        self.assertEqual(len(els), 1)
        self.assertEqual(els[0].attributes["time"].value, "678.237")

        junit_xsd = os.path.abspath(
            os.path.join(
                os.path.dirname(os.path.dirname(__file__)),
                os.path.pardir,
                ".data",
                "jenkins-junit.xsd",
            )
        )
        xml_schema = xmlschema.XMLSchema(junit_xsd)
        self.assertTrue(xml_schema.is_valid(xunit_output))

    def test_max_test_log_size(self):
        def get_system_out(out):
            return out[out.find(b"<system-out>") : out.find(b"<system-out/>")]

        log = tempfile.NamedTemporaryFile(dir=self.tmpdir.name, delete=False)
        log_content = b"1234567890" * 100
        log_content += b"this should not be present" + b"0987654321" * 100
        log.write(log_content)
        log_path = log.name
        log.close()
        self.test1._Test__status = "ERROR"
        self.test1._Test__logfile = log_path
        self.test_result.start_test(self.test1)
        self.test_result.end_test(self.test1.get_state())
        self.test_result.end_tests()
        xunit_result = xunit.XUnitResult()
        # setting the default value
        self.job.config["job.run.result.xunit.max_test_log_chars"] = 100000
        xunit_result.render(self.test_result, self.job)
        xunit_output = self.job.config.get("job.run.result.xunit.output")
        with open(xunit_output, "rb") as fp:
            unlimited = fp.read()
        # setting a small value
        self.job.config["job.run.result.xunit.max_test_log_chars"] = 10
        xunit_result.render(self.test_result, self.job)
        with open(xunit_output, "rb") as fp:
            limited = fp.read()
        # back to the default value
        self.job.config["job.run.result.xunit.max_test_log_chars"] = 100000
        xunit_result.render(self.test_result, self.job)
        with open(xunit_output, "rb") as fp:
            limited_but_fits = fp.read()
        self.assertLess(
            len(limited),
            len(unlimited) - 500,
            (
                f"Length of xunit limited to 10 chars was greater "
                f"than (unlimited - 500). "
                f"Unlimited output:\n{unlimited}\n\n"
                f"Limited output:\n{limited}"
            ),
        )
        unlimited_output = get_system_out(unlimited)
        self.assertIn(log_content, unlimited_output)
        self.assertEqual(unlimited_output, get_system_out(limited_but_fits))
        self.assertIn(b"this should not be present", unlimited)
        self.assertNotIn(b"this should not be present", limited)
        self.assertIn(b"1234567890", unlimited)
        self.assertNotIn(b"1234567890", limited)
        self.assertIn(b"12345", limited)
        self.assertIn(b"0987654321", unlimited)
        self.assertNotIn(b"0987654321", limited)
        self.assertIn(b"54321", limited)


if __name__ == "__main__":
    unittest.main()