Coverage for jaypore_ci/reporters/text.py: 92%

1import pendulum 

2from jaypore_ci.interfaces import Reporter, Status 

3 

4 

5def __get_time_format__(job): 

6 time = " --:--" 

7 if job.run_state is not None: 

8 if ( 

9 job.run_state.finished_at is not None 

10 and job.run_state.started_at is not None 

11 ): 

12 s = job.run_state.finished_at - job.run_state.started_at 

13 elif job.run_state.started_at is not None: 13 ↛ 16line 13 didn't jump to line 16, because the condition on line 13 was never false

14 s = pendulum.now() - job.run_state.started_at 

15 else: 

16 s = None 

17 s = s.in_seconds() if s is not None else 0 

18 m = s // 60 

19 time = f"{m:>3}:{s % 60:>2}" 

20 return time 

21 

22 

23def get_job_report(jobname): 

24 with open(f"/jaypore_ci/run/{jobname}.txt", "r", encoding="utf-8") as fl: 24 ↛ 25line 24 didn't jump to line 25

25 return fl.read() 

26 

27 

28__ST_MAP__ = { 

29 Status.RUNNING: "🔵", 

30 Status.FAILED: "🔴", 

31 Status.PASSED: "🟢", 

32} 

33 

34 

35class Text(Reporter): 

36 def render(self, pipeline): 

37 """ 

38 Returns a human readable report for a given pipeline. 

39 """ 

40 max_name = max(len(job.name) for job in pipeline.jobs.values()) 

41 max_name = max(max_name, len("jayporeci")) 

42 max_report = 10 

43 name = ("JayporeCI" + " " * max_name)[:max_name] 

44 graph = [ 

45 "", 

46 "```jayporeci", 

47 f"╔ {pipeline.get_status_dot()} : {name} [sha {pipeline.remote.sha[:10]}]", 

48 ] 

49 closer = "┗" + ("━" * (len(" O : ") + max_name + 1 + 1 + 8 + 1)) + "┛" 

50 for stage in pipeline.stages: 

51 nodes, edges = set(), set() 

52 for job in pipeline.jobs.values(): 

53 if job.stage != stage: 

54 continue 

55 nodes.add(job.name) 

56 edges |= {(p, job.name) for p in job.parents} 

57 if not nodes: 

58 continue 

59 graph += [f"┏━ {stage}", "┃"] 

60 for n in sorted( 

61 nodes, key=lambda x: (len(pipeline.jobs[x].parents), x) 

62 ): # Fewer parents first 

63 n = pipeline.jobs[n] 

64 name = (n.name + " " * max_name)[:max_name] 

65 status = __ST_MAP__.get(n.status, "🟡") 

66 run_id = f"{n.run_id}"[:8] if n.run_id is not None else "" 

67 graph += [f"┃ {status} : {name} [{run_id:<8}] {__get_time_format__(n)}"] 

68 try: 

69 report = get_job_report(n.name) 

70 report = " ".join(report.strip().split()) 

71 report = (report + " " * max_report)[:max_report] 

72 except FileNotFoundError: 

73 report = " " * max_report 

74 graph[-1] += f" {report}" 

75 if n.parents: 

76 graph[-1] += f" ❮-- {n.parents}" 

77 graph += [closer] 

78 graph += ["```"] 

79 graph = "\n".join(graph) 

80 return f"\n{graph}"