i7lib: send_email, email_status and read_history helpers over the sidecar (prompt 683, plan item 861)

New module sidecar_calls: retry on 503 and unreachable sidecar with a stable email id, SidecarError instead of silent failure, history series indexed by id and name with null gaps. README section and unittest tests with a stub sidecar (not run in the session: no action runs Python). Not tagged.
2026-10-28 11:11:52 +01:00
parent 52ee21aeff
commit 0d026e5d38
4 changed files with 332 additions and 2 deletions
@@ -0,0 +1,147 @@
+import base64
+import json
+import sys
+import threading
+import unittest
+from http.server import BaseHTTPRequestHandler, HTTPServer
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
+
+from i7lib import sidecar_calls
+from i7lib.sidecar_calls import SidecarError, email_status, read_history, send_email
+
+
+class Script:
+    def __init__(self, answers):
+        self.answers = list(answers)
+        self.seen = []
+
+
+def make_handler(script):
+    class Handler(BaseHTTPRequestHandler):
+        def _answer(self):
+            length = int(self.headers.get("Content-Length") or 0)
+            body = self.rfile.read(length) if length else b""
+            script.seen.append((self.command, self.path, json.loads(body) if body else None))
+            status, payload = script.answers.pop(0) if len(script.answers) > 1 else script.answers[0]
+            raw = json.dumps(payload).encode()
+            self.send_response(status)
+            self.send_header("Content-Type", "application/json")
+            self.send_header("Content-Length", str(len(raw)))
+            self.end_headers()
+            self.wfile.write(raw)
+
+        do_GET = _answer
+        do_POST = _answer
+
+        def log_message(self, *args):
+            pass
+
+    return Handler
+
+
+def ok(data):
+    return 200, {"data": data, "statusCode": 200, "errorCode": "", "errorMessage": ""}
+
+
+def failure(status, code, message="m"):
+    return status, {"data": None, "statusCode": status, "errorCode": code, "errorMessage": message}
+
+
+class SidecarCallsTest(unittest.TestCase):
+    def serve(self, answers):
+        script = Script(answers)
+        server = HTTPServer(("127.0.0.1", 0), make_handler(script))
+        thread = threading.Thread(target=server.serve_forever, daemon=True)
+        thread.start()
+        self.addCleanup(server.shutdown)
+        self.addCleanup(server.server_close)
+        sidecar_calls.SIDECAR_URL = f"http://127.0.0.1:{server.server_address[1]}"
+        return script
+
+    def setUp(self):
+        self.pauses = []
+        sidecar_calls._sleep = self.pauses.append
+        self.addCleanup(lambda: setattr(sidecar_calls, "_sleep", __import__("time").sleep))
+
+    def test_send_email_posts_the_encoded_request_and_returns_the_answer(self):
+        script = self.serve([ok({"emailId": "x", "state": "queued"})])
+        answer = send_email(
+            "a@example.com", "Report", body_text="hi",
+            attachments=[("r.csv", b"a,b", "text/csv")],
+            sender={"mode": "smtp", "host": "h", "port": 587, "passwordSealed": b"\x01\x02"},
+        )
+        self.assertEqual(answer["state"], "queued")
+        method, path, body = script.seen[0]
+        self.assertEqual((method, path), ("POST", "/api/email/send/"))
+        self.assertEqual(body["to"], ["a@example.com"])
+        self.assertEqual(body["attachments"][0]["dataBase64"], base64.b64encode(b"a,b").decode())
+        self.assertEqual(body["sender"]["passwordSealed"], base64.b64encode(b"\x01\x02").decode())
+        self.assertTrue(body["emailId"])
+
+    def test_send_email_retries_on_503_with_the_same_email_id(self):
+        script = self.serve([failure(503, "gateway_unreachable"), failure(503, "gateway_unreachable"), ok({"emailId": "x", "state": "queued"})])
+        answer = send_email(["a@example.com"], "Report", retry_seconds=60)
+        self.assertEqual(answer["state"], "queued")
+        ids = {body["emailId"] for _, _, body in script.seen}
+        self.assertEqual(len(script.seen), 3)
+        self.assertEqual(len(ids), 1)
+        self.assertEqual(self.pauses, [1.0, 2.0])
+
+    def test_send_email_gives_up_when_the_retry_time_is_used(self):
+        self.serve([failure(503, "gateway_unreachable")])
+        with self.assertRaises(SidecarError) as caught:
+            send_email(["a@example.com"], "Report", retry_seconds=0)
+        self.assertEqual(caught.exception.status, 503)
+        self.assertEqual(caught.exception.code, "gateway_unreachable")
+
+    def test_send_email_does_not_retry_refusals_that_will_not_change(self):
+        script = self.serve([failure(400, "invalid", "bad address")])
+        with self.assertRaises(SidecarError) as caught:
+            send_email(["bad"], "Report")
+        self.assertEqual(caught.exception.code, "invalid")
+        self.assertEqual(len(script.seen), 1)
+        self.assertEqual(self.pauses, [])
+
+    def test_a_full_outbox_and_an_oversized_email_raise_at_once(self):
+        for status, code in ((429, "queue_full"), (413, "too_large")):
+            script = self.serve([failure(status, code)])
+            with self.assertRaises(SidecarError) as caught:
+                send_email(["a@example.com"], "Report")
+            self.assertEqual((caught.exception.status, caught.exception.code), (status, code))
+            self.assertEqual(len(script.seen), 1)
+
+    def test_email_status_returns_the_state_and_raises_not_found(self):
+        self.serve([ok({"state": "sent", "detail": "", "attempts": 1, "updatedMs": 5})])
+        self.assertEqual(email_status("e1")["state"], "sent")
+        self.serve([failure(404, "not_found")])
+        with self.assertRaises(SidecarError) as caught:
+            email_status("nope")
+        self.assertEqual(caught.exception.status, 404)
+
+    def test_read_history_indexes_series_by_name_and_id_with_times(self):
+        series = {"signalId": 1, "name": "cpu", "startMs": 60000, "stepMs": 60000, "values": [1.5, None, 3]}
+        script = self.serve([ok({"series": [series]})])
+        result = read_history(["cpu", 2], 0, 240000, resolution="1min")
+        self.assertIs(result["cpu"], result[1])
+        self.assertEqual(result["cpu"]["values"], [1.5, None, 3])
+        self.assertEqual(result["cpu"]["times"], [60000, 120000, 180000])
+        _, path, _ = script.seen[0]
+        self.assertEqual(path, "/api/history/?signals=cpu,2&start=0&end=240000&resolution=1min")
+
+    def test_read_history_raises_403_for_a_signal_not_entitled(self):
+        self.serve([failure(403, "signal_not_entitled")])
+        with self.assertRaises(SidecarError) as caught:
+            read_history("param", 0, 60000)
+        self.assertEqual(caught.exception.status, 403)
+
+    def test_a_sidecar_that_is_down_is_retried_then_raised(self):
+        sidecar_calls.SIDECAR_URL = "http://127.0.0.1:1"
+        with self.assertRaises(SidecarError) as caught:
+            email_status("e1", retry_seconds=0)
+        self.assertEqual(caught.exception.code, "sidecar_unreachable")
+
+
+if __name__ == "__main__":
+    unittest.main()