i7lib/tests/test_sidecar_calls.py

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