Coverage for jaypore_ci/remotes/email.py: 97%

1""" 

2An email remote. 

3 

4This is used to report pipeline status via email. 

5Multiple updates appear as a single thread. 

6""" 

7import os 

8import time 

9import smtplib 

10from html import escape as html_escape 

11 

12from email.headerregistry import Address 

13from email.message import EmailMessage 

14from pathlib import Path 

15from urllib.parse import urlparse 

16 

17 

18from jaypore_ci.interfaces import Remote, Repo 

19from jaypore_ci.logging import logger 

20 

21 

22class Email(Remote): # pylint: disable=too-many-instance-attributes 

23 """ 

24 You can send pipeline status via email using this remote. In order to use it you 

25 can specify the following environment variables in your secrets: 

26 

27 .. code-block:: console 

28 

29 JAYPORE_EMAIL_ADDR=email-account@gmail.com 

30 JAYPORE_EMAIL_PASSWORD=some-app-password 

31 JAYPORE_EMAIL_TO=myself@gmail.com,mailing-list@gmail.com 

32 JAYPORE_EMAIL_FROM=noreply@gmail.com 

33 

34 If you're using something other than gmail, you can specify 

35 `JAYPORE_EMAIL_HOST` and `JAYPORE_EMAIL_PORT` as well. 

36 

37 Once that is done you can supply this remote to your pipeline instead of 

38 the usual gitea one. 

39 

40 .. code-block:: python 

41 

42 from jaypore_ci import jci, remotes, repos 

43 

44 git = repos.Git.from_env() 

45 email = remotes.Email.from_env(repo=git) 

46 with jci.Pipeline(repo=git, remote=email) as p: 

47 pass 

48 # Do something 

49 

50 :param host: What smtp host to use. 

51 :param port: Smtp port to use. 

52 :param addr: Smtp address to use for login. 

53 :param password: Smtp password to use for login. 

54 :param email_to: Which address the email should go to. 

55 :param email_from: Which address should be the sender of this email. 

56 :param subject: The subject line of the email. 

57 :param only_on_failure: If set to True, a single email will be sent when 

58 the pipeline fails. In all other cases no email is 

59 sent. 

60 :param publish_interval: Determines the delay in sending another email when 

61 we are sending multiple email updates in a single 

62 email thread. If `only_on_failure` is set, this 

63 option is ignored. 

64 """ 

65 

66 @classmethod 

67 def from_env(cls, *, repo: Repo) -> "Email": 

68 """ 

69 Creates a remote instance from the environment. 

70 """ 

71 remote = urlparse(repo.remote) 

72 owner = Path(remote.path).parts[1] 

73 name = Path(remote.path).parts[2].replace(".git", "") 

74 return cls( 

75 host=os.environ.get("JAYPORE_EMAIL_HOST", "smtp.gmail.com"), 

76 port=int(os.environ.get("JAYPORE_EMAIL_PORT", 465)), 

77 addr=os.environ["JAYPORE_EMAIL_ADDR"], 

78 password=os.environ["JAYPORE_EMAIL_PASSWORD"], 

79 email_to=os.environ["JAYPORE_EMAIL_TO"], 

80 email_from=os.environ.get( 

81 "JAYPORE_EMAIL_FROM", os.environ["JAYPORE_EMAIL_ADDR"] 

82 ), 

83 subject=f"JCI [{owner}/{name}] [{repo.branch} {repo.sha[:8]}]", 

84 branch=repo.branch, 

85 sha=repo.sha, 

86 ) 

87 

88 def __init__( 

89 self, 

90 *, 

91 host: str, 

92 port: int, 

93 addr: str, 

94 password: str, 

95 email_to: str, 

96 email_from: str, 

97 subject: str, 

98 only_on_failure: bool = False, 

99 publish_interval: int = 30, 

100 **kwargs, 

101 ): # pylint: disable=too-many-arguments 

102 super().__init__(**kwargs) 

103 # --- customer 

104 self.host = host 

105 self.port = port 

106 self.addr = addr 

107 self.password = password 

108 self.email_to = email_to 

109 self.email_from = email_from 

110 self.subject = subject 

111 self.timeout = 10 

112 self.publish_interval = publish_interval 

113 self.only_on_failure = only_on_failure 

114 # --- 

115 self.__smtp__ = None 

116 self.__last_published_at__ = None 

117 self.__last_report__ = None 

118 

119 @property 

120 def smtp(self): 

121 if self.__smtp__ is None: 

122 smtp = smtplib.SMTP_SSL(self.host, self.port) 

123 smtp.ehlo() 

124 smtp.login(self.addr, self.password) 

125 self.__smtp__ = smtp 

126 return self.__smtp__ 

127 

128 def logging(self): 

129 """ 

130 Return's a logging instance with information about gitea bound to it. 

131 """ 

132 return logger.bind(addr=self.addr, host=self.host, port=self.port) 

133 

134 def publish(self, report: str, status: str) -> None: 

135 """ 

136 Will publish the report via email. 

137 

138 :param report: Report to write to remote. 

139 :param status: One of ["pending", "success", "error", "failure", 

140 "warning"] This is the dot next to each commit in gitea. 

141 """ 

142 assert status in ("pending", "success", "error", "failure", "warning") 

143 if ( 

144 self.__last_published_at__ is not None 

145 and (time.time() - self.__last_published_at__) < self.publish_interval 

146 and status not in ("success", "failure") 

147 ) or (self.only_on_failure and status != "failure"): 

148 return 

149 if self.__last_report__ == report: 

150 return 

151 self.__last_report__ = report 

152 self.__last_published_at__ = time.time() 

153 # Let's send the email 

154 msg = EmailMessage() 

155 msg["Subject"] = self.subject 

156 msg["From"] = Address("JayporeCI", "JayporeCI", self.email_from) 

157 msg["To"] = self.email_to 

158 msg.set_content(report) 

159 msg.add_alternative( 

160 f"<html><body><pre>{html_escape(report)}</pre></body></html>", 

161 subtype="html", 

162 ) 

163 try: 

164 self.smtp.send_message(msg) 

165 except Exception as e: # pylint: disable=broad-except 

166 self.logging().exception(e) 

167 self.__last_published_at__ = time.time() 

168 self.logging().info( 

169 "Report published", 

170 subject=self.subject, 

171 email_from=self.email_from, 

172 email_to=self.email_to, 

173 )