[WIP] chore/offline tests

.coveragerc

@@ -0,0 +1,7 @@
+[run]
+source = regzbot
+omit = regzbot/testing*.py
+
+[report]
+fail_under = 60
+show_missing = true

.gitignore

@@ -3,3 +3,4 @@ __pycache__
 *.pyc
 *.pyo
 *.swp
+.coverage

regzbot/export_mail.py

@@ -6,6 +6,7 @@
 
 from collections import Counter
 import datetime
+from email import generator as email_generator
 from email.message import EmailMessage
 import email.utils
 import tempfile
@@ -17,6 +18,17 @@
 logger = regzbot.logger
 
 
+def _mail_now():
+    mail_now = regzbot._TESTING.get("mail_now")
+    if mail_now is not None:
+        return mail_now
+    return regzbot.timendate_now()
+
+
+def _mail_days_delta(past):
+    return (_mail_now() - datetime.datetime.fromtimestamp(past, datetime.timezone.utc)).days
+
+
 class RegLinkMailReport(regzbot.RegLink):
     def __init__(self, *args):
         super().__init__(*args)
@@ -31,7 +43,7 @@ def mailreport(self):
             ):
                 monitored = "; thread monitored."
             authored = "\n  %s days ago, by %s%s" % (
-                regzbot.days_delta(self.gmtime),
+                _mail_days_delta(self.gmtime),
                 self.author,
                 monitored,
             )
@@ -95,17 +107,17 @@ def compile(self, lastreport_gmtime):
                 statusline.append(", ")
 
         statusline.append("; ")
-        statusline.append(str(regzbot.days_delta(self.gmtime)))
+        statusline.append(str(_mail_days_delta(self.gmtime)))
         statusline.append(" days ago; ")
         statusline.append(str(len(self._actievents)))
         statusline.append(" activities")
         if len(self._actievents) > 0:
             statusline.append(", latest ")
-            statusline.append(str(regzbot.days_delta(self._actievents[-1].gmtime)))
+            statusline.append(str(_mail_days_delta(self._actievents[-1].gmtime)))
             statusline.append(" days ago")
 
         if self.poked:
-            statusline.append("; poked %s days ago" % regzbot.days_delta(self.poked.gmtime))
+            statusline.append("; poked %s days ago" % _mail_days_delta(self.poked.gmtime))
         statusline.append(".")
         report.append("".join(statusline))
 
@@ -166,7 +178,7 @@ def add_latestpatch(self, report):
             report.append("* %s" % actievent.subject)
             report.append("  %s" % actievent.url())
             report.append(
-                "  %s days ago, by %s" % (regzbot.days_delta(actievent.gmtime), actievent.author)
+                "  %s days ago, by %s" % (_mail_days_delta(actievent.gmtime), actievent.author)
             )
 
             break
@@ -235,33 +247,63 @@ def __init__(
         self.reporttext = reporttext
 
     @classmethod
-    def __create_mail(cls, content, treename):
+    def listed(cls, lastreport_gmtime):
+        regressionslist = list()
+        for regression in RegressionMailReport.get_all(only_unsolved=True):
+            # ignore some
+            if regression._actievents:
+                last_activity = regression._actievents[-1].gmtime
+            else:
+                last_activity = regression._histevents[-1].gmtime
+            regressionslist.append(
+                cls(
+                    regression._actim_report.entry,
+                    regression.gmtime,
+                    regression.gmtime_filed,
+                    last_activity,
+                    regression.treename,
+                    regression.versionline,
+                    regression.backburner,
+                    regression.identified,
+                    regression.mailreport(lastreport_gmtime),
+                )
+            )
+
+        regressionslist.sort(key=lambda x: x.gmtime_activity, reverse=True)
+        return regressionslist
+
+    @classmethod
+    def __create_mail(cls, content, treename, *, fixed_message_id=False):
         msg = EmailMessage()
         msg["To"] = (
             "LKML <linux-kernel@vger.kernel.org>, Linus Torvalds <torvalds@linux-foundation.org>, Linux regressions mailing list <regressions@lists.linux.dev>"
         )
         msg["Subject"] = "%s for %s [%s]" % (
             regzbot.REPORT_SUBJECT_PREFIX,
             treename,
-            datetime.date.today(),
+            _mail_now().date(),
         )
-        msg["Date"] = email.utils.localtime()
-        msg["Message-ID"] = email.utils.make_msgid(domain="leemhuis.info")
+        msg["Date"] = email.utils.formatdate(timeval=_mail_now().timestamp(), localtime=True)
+        if fixed_message_id:
+            msg["Message-ID"] = "<regzbot-testing-mailreport@example.com>"
+        else:
+            msg["Message-ID"] = email.utils.make_msgid(domain="leemhuis.info")
         msg.set_content(content, cte="quoted-printable")
         return msg
 
     @classmethod
-    def pagecreate(cls, categories, treename, lastreport_msgid):
+    def pagecreate(cls, categories, treename, lastreport_msgid, *, interactive=True):
         def repintro(report, number_issues, treename):
             intro = list()
 
-            print("Enter/Paste your intro for %s and hit Ctrl-D to save it." % treename)
-            while True:
-                try:
-                    line = input()
-                except EOFError:
-                    break
-                intro.append(line)
+            if interactive:
+                print("Enter/Paste your intro for %s and hit Ctrl-D to save it." % treename)
+                while True:
+                    try:
+                        line = input()
+                    except EOFError:
+                        break
+                    intro.append(line)
             if report:
                 intro.append("\n---\n")
 
@@ -450,11 +492,7 @@ def categorize(cls, regressionlist, lastreport_gmtime):
         }
 
         for regression in regressionlist:
-            filed_days = (
-                datetime.datetime.now(datetime.timezone.utc)
-                - datetime.datetime.fromtimestamp(regression.gmtime_filed, datetime.timezone.utc)
-            ).days
-            last_activity_days = regzbot.days_delta(regression.gmtime_activity)
+            last_activity_days = _mail_days_delta(regression.gmtime_activity)
 
             if regression.backburner:
                 if lastreport_gmtime > regression.gmtime_activity:
@@ -512,80 +550,97 @@ def categorize(cls, regressionlist, lastreport_gmtime):
         return categories
 
     @classmethod
-    def compile(cls):
-        logger.debug("[reportmail] generating")
+    def prepare_reports(cls, categories, lastreport_msgid, *, interactive=True):
+        reports = dict()
+        for treename in categories.keys():
+            reports[treename] = cls.pagecreate(
+                categories[treename], treename, lastreport_msgid, interactive=interactive
+            )
+        return reports
+
+    @classmethod
+    def lastreport(cls):
+        if regzbot.is_running_citesting("offline"):
+            lastreport_gmtime = regzbot._TESTING["mail_lastreport_gmtime"]
+            lastreport_msgid = regzbot._TESTING.get("mail_lastreport_msgid")
+        else:
+            lastreport_msgid = regzbot.RegzbotState.get("lastreport_mainline_msgid")
+            lastreport_gmtime = regzbot.RegzbotState.get("lastreport_mainline_gmtime")
 
-        lastreport_msgid = regzbot.RegzbotState.get("lastreport_mainline_msgid")
-        lastreport_gmtime = regzbot.RegzbotState.get("lastreport_mainline_gmtime")
         if lastreport_gmtime:
             lastreport_gmtime = int(lastreport_gmtime)
         else:
-            lastreport_gmtime = int(datetime.datetime.now(datetime.timezone.utc).timestamp())
+            lastreport_gmtime = int(_mail_now().timestamp())
 
+        return lastreport_gmtime, lastreport_msgid
+
+    @classmethod
+    def prepare(cls, *, interactive=True):
+        lastreport_gmtime, lastreport_msgid = cls.lastreport()
         logger.debug("[reportmail] lastreport was %s" % lastreport_gmtime)
 
         # gather everything we need
-        regressionslist = list()
+        categories = cls.categorize(cls.listed(lastreport_gmtime), lastreport_gmtime)
+        reports = cls.prepare_reports(categories, lastreport_msgid, interactive=interactive)
+        return reports, categories
 
-        for regression in RegressionMailReport.get_all(only_unsolved=True):
-            # ignore some
-            if regression._actievents:
-                last_activity = regression._actievents[-1].gmtime
-            else:
-                last_activity = regression._histevents[-1].gmtime
-            last_activity_days = regzbot.days_delta(last_activity)
-            if regression._actievents:
-                last_activity = regression._actievents[-1].gmtime
-            else:
-                last_activity = regression._histevents[-1].gmtime
-            regressionslist.append(
-                cls(
-                    regression._actim_report.entry,
-                    regression.gmtime,
-                    regression.gmtime_filed,
-                    last_activity,
-                    regression.treename,
-                    regression.versionline,
-                    regression.backburner,
-                    regression.identified,
-                    regression.mailreport(lastreport_gmtime),
-                )
-            )
-
-        regressionslist.sort(key=lambda x: x.gmtime_activity, reverse=True)
-        categories = cls.categorize(regressionslist, lastreport_gmtime)
-
-        report_gmtime = int(datetime.datetime.now(datetime.timezone.utc).timestamp())
+    @classmethod
+    def publish(cls, reports, *, interactive=True):
+        report_gmtime = int(_mail_now().timestamp())
+        lastreport_msgid = None
         with tempfile.TemporaryDirectory() as tmpdirname:
-            for counter, treename in enumerate(categories.keys()):
-                report = cls.pagecreate(categories[treename], treename, lastreport_msgid)
-
+            counter = 0
+            for treename, report in reports.items():
                 if not report:
                     logger.info("Nothing to report for %s" % treename)
                     continue
 
                 filename = os.path.join(tmpdirname, "%s-regzbotreport-%s" % (counter, treename))
-                msg = cls.__create_mail(report, treename)
+                msg = cls.__create_mail(report, treename, fixed_message_id=not interactive)
                 lastreport_msgid = msg["Message-ID"].strip("<>")
-                print("#" * 120)
-                print("\n%s\n" % filename)
-                print("#" * 120)
-                print(report)
+                if interactive:
+                    print("#" * 120)
+                    print("\n%s\n" % filename)
+                    print("#" * 120)
+                    print(report)
                 with open(filename, "w") as out:
-                    gen = email.generator.Generator(out)
+                    gen = email_generator.Generator(out)
                     gen.flatten(msg)
+                counter += 1
 
-            print("#" * 120)
+            if counter == 0:
+                return
+
+            if not interactive:
+                return
 
+            print("#" * 120)
             print(
                 "Review the reports in %s and sent them using \"git send-email --from='Regzbot (on behalf of Thorsten Leemhuis) <regressions@leemhuis.info>' --suppress-cc=self --to '' %s/*\""
                 % (tmpdirname, tmpdirname)
             )
             answer = input("Enter c to confirm you sent the report, anything else to abort: ")
             if answer.lower() != "c":
                 return
+
             regzbot.RegzbotState.set("lastreport_mainline_gmtime", report_gmtime)
             regzbot.RegzbotState.set("lastreport_mainline_msgid", lastreport_msgid)
-            lastreport_msgid = regzbot.RegzbotState.get("lastreport_mainline_msgid")
 
         logger.debug("[report] generated")
+
+    @classmethod
+    def compile(cls, *, interactive=True):
+        logger.debug("[reportmail] generating")
+        reports, _categories = cls.prepare(interactive=interactive)
+        cls.publish(reports, interactive=interactive)
+
+
+def dumpall_mail():
+    reports, categories = RegExportMailReport.prepare(interactive=False)
+    if not any(category["entries"] for category in categories.get("mainline", {}).values()):
+        yield "MAIL_MAINLINE: (empty)\n\n"
+        return
+
+    yield "MAIL_MAINLINE:\n"
+    yield reports.get("mainline", "")
+    yield "\n"