Coverage for /home/runner/work/coxbuild/coxbuild/src/coxbuild/runtime.py: 51%

1from typing import TYPE_CHECKING 

2 

3from coxbuild.configurations import Configuration, ConfigurationAccessor 

4from coxbuild.tasks import TaskContext 

5 

6if TYPE_CHECKING: 

7 from coxbuild.managers import Manager 

8 from coxbuild.pipelines import Pipeline 

9 from coxbuild.services import EventContext, EventHandler, Service 

10 from coxbuild.tasks import Task 

11 

12 

13class ExecutionState(ConfigurationAccessor): 

14 __configname__ = "execution" 

15 

16 @property 

17 def matchedTasks(self) -> "list[Task]": 

18 """Matched task names.""" 

19 return self.config.get("matchedTasks") or [] 

20 

21 @matchedTasks.setter 

22 def matchedTasks(self, value: "list[Task]") -> None: 

23 self.config["matchedTasks"] = value 

24 

25 @property 

26 def unmatchedTasks(self) -> list[str]: 

27 """Unmatched task names.""" 

28 return self.config.get("unmatchedTasks") or [] 

29 

30 @unmatchedTasks.setter 

31 def unmatchedTasks(self, value: list[str]) -> None: 

32 self.config["unmatchedTasks"] = value 

33 

34 @property 

35 def manager(self) -> "Manager | None": 

36 """Get manager.""" 

37 return self.config.get("manager") or None 

38 

39 @manager.setter 

40 def manager(self, value: "Manager | None") -> None: 

41 self.config["manager"] = value 

42 

43 @property 

44 def pipeline(self) -> "Pipeline | None": 

45 """Get pipeline.""" 

46 return self.config.get("pipeline") or None 

47 

48 @pipeline.setter 

49 def pipeline(self, value: "Pipeline | None") -> None: 

50 self.config["pipeline"] = value 

51 

52 @property 

53 def task(self) -> "Task | None": 

54 """Get task.""" 

55 return self.config.get("task") or None 

56 

57 @task.setter 

58 def task(self, value: "Task | None") -> None: 

59 self.config["task"] = value 

60 

61 @property 

62 def service(self) -> "Service | None": 

63 """Get service.""" 

64 return self.config.get("service") or None 

65 

66 @service.setter 

67 def service(self, value: "Service | None") -> None: 

68 self.config["service"] = value 

69 

70 @property 

71 def handler(self) -> "EventHandler | None": 

72 """Get event handler.""" 

73 return self.config.get("handler") or None 

74 

75 @handler.setter 

76 def handler(self, value: "EventHandler | None") -> None: 

77 self.config["handler"] = value 

78 

79 @property 

80 def event(self) -> "EventContext | None": 

81 """Get event context.""" 

82 return self.config.get("event") or None 

83 

84 @event.setter 

85 def event(self, value: "EventContext | None") -> None: 

86 self.config["event"] = value 

87 

88 @property 

89 def configuration(self) -> Configuration | None: 

90 """Get configuration.""" 

91 return self.config.get("config") or None 

92 

93 @configuration.setter 

94 def configuration(self, value: Configuration | None) -> None: 

95 self.config["config"] = value 

96 

97 

98def withExecutionState(task: "Task") -> "Task": 

99 """Decorator to add executionState argument to task context.""" 

100 def hook(context: TaskContext): 

101 if context.config: 

102 context.kwds.update(executionState=ExecutionState(context.config)) 

103 

104 task.before(hook) 

105 return task 

106 

107 

108def withManager(task: "Task") -> "Task": 

109 """Decorator to add manager argument to task context.""" 

110 def hook(context: TaskContext): 

111 if context.config: 

112 context.kwds.update(manager=ExecutionState(context.config).manager) 

113 

114 task.before(hook) 

115 return task 

116 

117 

118def withPipeline(task: "Task") -> "Task": 

119 """Decorator to add pipeline argument to task context.""" 

120 def hook(context: TaskContext): 

121 if context.config: 

122 context.kwds.update( 

123 pipeline=ExecutionState(context.config).pipeline) 

124 

125 task.before(hook) 

126 return task 

127 

128 

129def withService(task: "Task") -> "Task": 

130 """Decorator to add service argument to task context.""" 

131 def hook(context: TaskContext): 

132 if context.config: 

133 context.kwds.update(service=ExecutionState(context.config).service) 

134 

135 task.before(hook) 

136 return task 

137 

138 

139def withTask(task: "Task") -> "Task": 

140 """Decorator to add task argument to task context.""" 

141 def hook(context: TaskContext): 

142 if context.config: 

143 context.kwds.update(task=ExecutionState(context.config).task) 

144 

145 task.before(hook) 

146 return task 

147 

148 

149def withHandler(task: "Task") -> "Task": 

150 """Decorator to add handler argument to task context.""" 

151 def hook(context: TaskContext): 

152 if context.config: 

153 context.kwds.update(handler=ExecutionState(context.config).handler) 

154 

155 task.before(hook) 

156 return task 

157 

158 

159def withEvent(task: "Task") -> "Task": 

160 """Decorator to add event argument to task context.""" 

161 def hook(context: TaskContext): 

162 if context.config: 

163 context.kwds.update(event=ExecutionState(context.config).event) 

164 

165 task.before(hook) 

166 return task 

167 

168 

169def withConfig(task: "Task") -> "Task": 

170 """Decorator to add config argument to task context.""" 

171 def hook(context: TaskContext): 

172 if context.config: 

173 context.kwds.update(config=ExecutionState( 

174 context.config).configuration) 

175 

176 task.before(hook) 

177 return task