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()