How to use the perceval.backends.core.jenkins.Jenkins function in perceval

"""Test whether attributes are initializated"""

        jenkins = Jenkins(SERVER_URL, tag='test', sleep_time=60, detail_depth=2)
        self.assertEqual(jenkins.url, SERVER_URL)
        self.assertIsNone(jenkins.user)
        self.assertIsNone(jenkins.api_token)
        self.assertEqual(jenkins.origin, SERVER_URL)
        self.assertEqual(jenkins.sleep_time, 60)
        self.assertEqual(jenkins.detail_depth, 2)
        self.assertEqual(jenkins.tag, 'test')
        self.assertIsNone(jenkins.client)
        self.assertIsNone(jenkins.blacklist_ids)

        # When tag is empty or None it will be set to
        # the value in url
        jenkins = Jenkins(SERVER_URL)
        self.assertEqual(jenkins.url, SERVER_URL)
        self.assertEqual(jenkins.origin, SERVER_URL)
        self.assertEqual(jenkins.tag, SERVER_URL)
        self.assertEqual(jenkins.sleep_time, SLEEP_TIME)
        self.assertEqual(jenkins.detail_depth, DETAIL_DEPTH)
        self.assertIsNone(jenkins.blacklist_ids)

        jenkins = Jenkins(SERVER_URL, tag='')
        self.assertEqual(jenkins.url, SERVER_URL)
        self.assertEqual(jenkins.origin, SERVER_URL)
        self.assertEqual(jenkins.tag, SERVER_URL)
        self.assertIsNone(jenkins.blacklist_ids)

        jenkins = Jenkins(SERVER_URL, user=USER, api_token=TOKEN)
        self.assertEqual(jenkins.url, SERVER_URL)
        self.assertEqual(jenkins.origin, SERVER_URL)

def test_fetch_blacklist(self):
        """Test whether jobs in blacklist are not retrieved"""

        blacklist = [JOB_BUILDS_1]

        configure_http_server()

        jenkins = Jenkins(SERVER_URL, blacklist_ids=blacklist)
        nrequests = len(requests_http)

        with self.assertLogs(logger, level='WARNING') as cm:
            builds = [build for build in jenkins.fetch()]
            self.assertEqual(cm.output[0], 'WARNING:perceval.backends.core.jenkins:'
                                           'Not getting blacklisted job: apex-build-brahmaputra')

        # No HTTP calls at all must be done for JOB_BUILDS_1
        # Just the first call for all jobs and one for each job,
        # including those jobs that raise errors
        self.assertEqual(len(requests_http) - nrequests, 6)

        # Builds just from JOB_BUILDS_2
        self.assertEqual(len(builds), 37)

def test_fetch_blacklist_from_archive(self):
        """Test whether jobs in balcklist are not retrieved from archive"""

        blacklist = [JOB_BUILDS_1]

        configure_http_server()
        self.backend_write_archive = Jenkins(SERVER_URL, blacklist_ids=blacklist, archive=self.archive)
        self.backend_read_archive = Jenkins(SERVER_URL, blacklist_ids=blacklist, archive=self.archive)

        with self.assertLogs(logger, level='WARNING') as cm:
            self._test_fetch_from_archive()
            self.assertEqual(cm.output[0], 'WARNING:perceval.backends.core.jenkins:'
                                           'Not getting blacklisted job: apex-build-brahmaputra')
            self.assertEqual(cm.output[1], 'WARNING:perceval.backends.core.jenkins:500 Server Error: '
                                           'Internal Server Error for url: '
                                           'http://example.com/ci/job/500-error-job/api/json?depth=1')
            self.assertEqual(cm.output[2], 'WARNING:perceval.backends.core.jenkins:Unable to fetch builds from job '
                                           'http://example.com/ci/job/500-error-job/; skipping')
            self.assertEqual(cm.output[3], 'WARNING:perceval.backends.core.jenkins:Unable to parse builds from job '
                                           'http://example.com/ci/job/invalid-json-job/; skipping')
            self.assertEqual(cm.output[4], 'WARNING:perceval.backends.core.jenkins:'
                                           'Not getting blacklisted job: apex-build-brahmaputra')
            self.assertEqual(cm.output[5], 'WARNING:perceval.backends.core.jenkins:500 Server Error: '
                                           'Internal Server Error for url: '

def setUp(self):
        super().setUp()
        self.backend_write_archive = Jenkins(SERVER_URL, archive=self.archive)
        self.backend_read_archive = Jenkins(SERVER_URL, archive=self.archive)

def test_fetch_auth_api_token(self):
        """Test whether a list of builds is returned using username and API token"""

        configure_http_server()

        # Test fetch builds from jobs list
        jenkins = Jenkins(SERVER_URL, user=USER, api_token=TOKEN)

        with self.assertLogs(logger, level='WARNING') as cm:
            builds = [build for build in jenkins.fetch()]
            self.assertEqual(cm.output[0], 'WARNING:perceval.backends.core.jenkins:500 Server Error: '
                                           'Internal Server Error for url: '
                                           'http://example.com/ci/job/500-error-job/api/json?depth=1')
            self.assertEqual(cm.output[1], 'WARNING:perceval.backends.core.jenkins:Unable to fetch builds from job '
                                           'http://example.com/ci/job/500-error-job/; skipping')
            self.assertEqual(cm.output[2], 'WARNING:perceval.backends.core.jenkins:Unable to parse builds from job '
                                           'http://example.com/ci/job/invalid-json-job/; skipping')

        self.assertEqual(len(builds), 69)

        with open(os.path.join(os.path.dirname(os.path.abspath(__file__)), "data/jenkins/jenkins_build.json")) \
                as build_json:
            first_build = json.load(build_json)

self.assertEqual(jenkins.detail_depth, 2)
        self.assertEqual(jenkins.tag, 'test')
        self.assertIsNone(jenkins.client)
        self.assertIsNone(jenkins.blacklist_ids)

        # When tag is empty or None it will be set to
        # the value in url
        jenkins = Jenkins(SERVER_URL)
        self.assertEqual(jenkins.url, SERVER_URL)
        self.assertEqual(jenkins.origin, SERVER_URL)
        self.assertEqual(jenkins.tag, SERVER_URL)
        self.assertEqual(jenkins.sleep_time, SLEEP_TIME)
        self.assertEqual(jenkins.detail_depth, DETAIL_DEPTH)
        self.assertIsNone(jenkins.blacklist_ids)

        jenkins = Jenkins(SERVER_URL, tag='')
        self.assertEqual(jenkins.url, SERVER_URL)
        self.assertEqual(jenkins.origin, SERVER_URL)
        self.assertEqual(jenkins.tag, SERVER_URL)
        self.assertIsNone(jenkins.blacklist_ids)

        jenkins = Jenkins(SERVER_URL, user=USER, api_token=TOKEN)
        self.assertEqual(jenkins.url, SERVER_URL)
        self.assertEqual(jenkins.origin, SERVER_URL)
        self.assertEqual(jenkins.tag, SERVER_URL)
        self.assertEqual(jenkins.user, USER)
        self.assertEqual(jenkins.api_token, TOKEN)
        self.assertIsNone(jenkins.blacklist_ids)

        jenkins = Jenkins(SERVER_URL, blacklist_ids=[JOB_BUILDS_1])
        self.assertEqual(jenkins.url, SERVER_URL)
        self.assertEqual(jenkins.origin, SERVER_URL)

def test_fetch_no_job_builds(self):
        """Test whether a warning message is logged when no job builds exist"""

        configure_http_server(builds_job=False)

        # Test fetch builds from jobs list
        jenkins = Jenkins(SERVER_URL)

        with self.assertLogs(logger, level='DEBUG') as cm:
            builds = [build for build in jenkins.fetch()]
            self.assertRegex(cm.output[1], 'DEBUG:perceval.backends.core.jenkins:No builds for job.*')
            self.assertRegex(cm.output[2], 'DEBUG:perceval.backends.core.jenkins:No builds for job.*')

        self.assertEqual(builds, [])

        # Check request params
        expected = {
            'depth': ['1']
        }

        req = httpretty.last_request()

        self.assertEqual(req.method, 'GET')

def test_backend_class(self):
        """Test if the backend class is Jenkins"""

        self.assertIs(JenkinsCommand.BACKEND, Jenkins)

def test_initialization_error(self):
        """Test whether an exeception is thrown when the user and api_token are not initialized together"""

        with self.assertLogs(logger) as cm:
            with self.assertRaises(BackendError):
                _ = Jenkins(SERVER_URL, user=USER)

            self.assertEqual(cm.output[0], 'ERROR:perceval.backends.core.jenkins:'
                                           'Authentication method requires user and api_token')

        with self.assertLogs(logger) as cm:
            with self.assertRaises(BackendError):
                _ = Jenkins(SERVER_URL, api_token=TOKEN)

            self.assertEqual(cm.output[0], 'ERROR:perceval.backends.core.jenkins:'
                                           'Authentication method requires user and api_token')

def test_initialization(self):
        """Test whether attributes are initializated"""

        jenkins = Jenkins(SERVER_URL, tag='test', sleep_time=60, detail_depth=2)
        self.assertEqual(jenkins.url, SERVER_URL)
        self.assertIsNone(jenkins.user)
        self.assertIsNone(jenkins.api_token)
        self.assertEqual(jenkins.origin, SERVER_URL)
        self.assertEqual(jenkins.sleep_time, 60)
        self.assertEqual(jenkins.detail_depth, 2)
        self.assertEqual(jenkins.tag, 'test')
        self.assertIsNone(jenkins.client)
        self.assertIsNone(jenkins.blacklist_ids)

        # When tag is empty or None it will be set to
        # the value in url
        jenkins = Jenkins(SERVER_URL)
        self.assertEqual(jenkins.url, SERVER_URL)
        self.assertEqual(jenkins.origin, SERVER_URL)
        self.assertEqual(jenkins.tag, SERVER_URL)