# encoding: utf-8
import asyncio
import re
import time
from .artifact import Artifact, async_save_response_to, save_response_to
from .input import PendingInputAction
from .item import AsyncItem, Item
from .mix import (ActionsMixIn, AsyncActionsMixIn, AsyncDeletionMixIn,
AsyncDescriptionMixIn, DeletionMixIn, DescriptionMixIn)
from .report import (AsyncCoverageReport, AsyncCoverageResult,
AsyncCoverageTrends, AsyncTestReport, CoverageReport,
CoverageResult, CoverageTrends, TestReport)
[docs]
class Build(Item, DescriptionMixIn, DeletionMixIn, ActionsMixIn):
[docs]
def console_text(self):
with self.handle_stream('GET', 'consoleText') as resp:
yield from resp.iter_lines()
[docs]
def progressive_output(self, html=False):
url = 'logText/progressiveHtml' if html else 'logText/progressiveText'
start = 0
while True:
resp = self.handle_req('GET', url, params={'start': start})
time.sleep(1)
if start == resp.headers.get('X-Text-Size'):
continue
yield from resp.iter_lines()
if not resp.headers.get('X-More-Data'):
break
start = resp.headers['X-Text-Size']
[docs]
def stop(self):
return self.handle_req('POST', 'stop')
[docs]
def term(self):
return self.handle_req('POST', 'term')
[docs]
def kill(self):
return self.handle_req('POST', 'kill')
[docs]
def get_next_build(self):
if item := self.api_json(tree='nextBuild[url]')['nextBuild']:
return self.__class__(self.jenkins, item['url'])
return None
[docs]
def get_previous_build(self):
if item := self.api_json(tree='previousBuild[url]')['previousBuild']:
return self.__class__(self.jenkins, item['url'])
return None
@property
def project(self):
job_name = self.jenkins._url2name(re.sub(r'\w+[/]?$', '', self.url))
return self.jenkins.get_job(job_name)
[docs]
def get_test_report(self):
tr = TestReport(self.jenkins, f'{self.url}testReport/')
return tr if tr.exists() else None
[docs]
def get_coverage_report(self):
'''Access coverage report generated by `JaCoCo <https://plugins.jenkins.io/jacoco/>`_'''
cr = CoverageReport(self.jenkins, f'{self.url}jacoco/')
return cr if cr.exists() else None
[docs]
def get_coverage_result(self):
'''Access coverage result generated by `Code Coverage API <https://plugins.jenkins.io/code-coverage-api/>`_'''
cr = CoverageResult(self.jenkins, f'{self.url}coverage/result/')
return cr if cr.exists() else None
[docs]
def get_coverage_trends(self):
ct = CoverageTrends(self.jenkins, f'{self.url}coverage/trend/')
return ct if ct.exists() else None
[docs]
class WorkflowRun(Build):
@property
def artifacts(self):
artifacts = self.handle_req('GET', 'wfapi/artifacts').json()
return [Artifact(self.jenkins, art) for art in artifacts]
[docs]
def save_artifacts(self, filename='archive.zip'):
with self.handle_stream('GET', 'artifact/*zip*/archive.zip') as resp:
save_response_to(resp, filename)
[docs]
class FreeStyleBuild(Build):
pass
[docs]
class MatrixBuild(Build):
pass
# async class
[docs]
class AsyncBuild(AsyncItem, AsyncDescriptionMixIn, AsyncDeletionMixIn, AsyncActionsMixIn):
[docs]
async def console_text(self):
async with self.handle_stream('GET', 'consoleText') as resp:
async for line in resp.aiter_lines():
yield line
[docs]
async def progressive_output(self, html=False):
url = 'logText/progressiveHtml' if html else 'logText/progressiveText'
start = 0
while True:
resp = await self.handle_req('GET', url, params={'start': start})
await asyncio.sleep(1)
if start == resp.headers.get('X-Text-Size'):
continue
async for line in resp.aiter_lines():
yield line
if not resp.headers.get('X-More-Data'):
break
start = resp.headers['X-Text-Size']
[docs]
async def stop(self):
return await self.handle_req('POST', 'stop')
[docs]
async def term(self):
return await self.handle_req('POST', 'term')
[docs]
async def kill(self):
return await self.handle_req('POST', 'kill')
[docs]
async def get_next_build(self):
data = await self.api_json(tree='nextBuild[url]')
return self.__class__(self.jenkins, data['nextBuild']['url']) if data['nextBuild'] else None
[docs]
async def get_previous_build(self):
data = await self.api_json(tree='previousBuild[url]')
return self.__class__(self.jenkins, data['previousBuild']['url']) if data['previousBuild'] else None
@property
async def project(self):
job_name = self.jenkins._url2name(re.sub(r'\w+[/]?$', '', self.url))
return await self.jenkins.get_job(job_name)
[docs]
async def get_test_report(self):
tr = AsyncTestReport(self.jenkins, f'{self.url}testReport/')
return tr if await tr.exists() else None
[docs]
async def get_coverage_report(self):
'''Access coverage report generated by `JaCoCo <https://plugins.jenkins.io/jacoco/>`_'''
cr = AsyncCoverageReport(self.jenkins, f'{self.url}jacoco/')
return cr if await cr.exists() else None
[docs]
async def get_coverage_result(self):
'''Access coverage result generated by `Code Coverage API <https://plugins.jenkins.io/code-coverage-api/>`_'''
cr = AsyncCoverageResult(self.jenkins, f'{self.url}coverage/result/')
return cr if await cr.exists() else None
[docs]
async def get_coverage_trends(self):
ct = AsyncCoverageTrends(self.jenkins, f'{self.url}coverage/trend/')
return ct if await ct.exists() else None
[docs]
class AsyncWorkflowRun(AsyncBuild):
@property
async def artifacts(self):
artifacts = (await self.handle_req('GET', 'wfapi/artifacts')).json()
return [Artifact(self.jenkins, art) for art in artifacts]
[docs]
async def save_artifacts(self, filename='archive.zip'):
async with self.handle_stream('GET', 'artifact/*zip*/archive.zip') as resp:
await async_save_response_to(resp, filename)
[docs]
class AsyncFreeStyleBuild(AsyncBuild):
pass
[docs]
class AsyncMatrixBuild(AsyncBuild):
pass