"""Every input a session will read waits in its inbox, where people see,
reorder and drop it before the session takes it."""

import base64
import json
import threading
import time
import unittest
import urllib.error

from harness import Albedo, Provider, operation_id, python, text, wait_until
from image_limits_test import png


class QueueTests(unittest.TestCase):
    def setUp(self):
        self.gate = threading.Event()
        self.started = threading.Event()
        self.first = True

        def reply(_):
            if self.first:
                self.first = False
                self.started.set()
                self.gate.wait(40)
            return text("done")

        self.provider = Provider(reply)
        self.addCleanup(self.provider.close)
        self.addCleanup(self.gate.set)
        self.app = Albedo(self.provider, protocol="responses")
        self.app.__enter__()
        self.addCleanup(self.app.__exit__, None, None, None)

    def put_input(self, session, intent):
        input_id = operation_id()
        with self.app.api(
            f"/sessions/{session}/inputs/{input_id}", intent, method="PUT"
        ) as response:
            self.assertEqual(json.load(response)["delivery"], "pending")
        return input_id

    def queue(self, session, path="", body=None, method="GET"):
        with self.app.api(
            f"/extensions/mail/sessions/{session}/queue{path}", body, method=method
        ) as response:
            return json.load(response)

    def input_action(self, session, input_id, action, body=None, method="POST"):
        with self.app.api(
            f"/sessions/{session}/inputs/{input_id}/{action}", body, method=method
        ) as response:
            raw = response.read()
            return json.loads(raw) if raw else None

    def phase(self, session):
        with self.app.api(f"/sessions/{session}?tail=0") as response:
            return json.load(response)["status"]["phase"]

    def said(self):
        return [
            json.dumps(record["request"]["input"]) for record in self.provider.requests
        ]

    def test_waiting_inputs_are_listed_reordered_and_dropped(self):
        session = self.app.session()
        self.app.prompt(session, "hold the turn").close()
        self.assertTrue(self.started.wait(10))
        first = self.put_input(session, {"kind": "message", "text": "first waiting"})
        image = self.put_input(
            session,
            {
                "kind": "message",
                "text": "with an image",
                "images": [
                    {
                        "mime_type": "image/png",
                        "data": base64.b64encode(png(2, 2)).decode(),
                    }
                ],
            },
        )
        last = self.put_input(session, {"kind": "message", "text": "last waiting"})

        queue = self.queue(session)
        self.assertEqual([item["id"] for item in queue["items"]], [first, image, last])
        self.assertEqual(queue["total"], 3)
        self.assertEqual(
            {(item["kind"], item["from"], item["delivery"]) for item in queue["items"]},
            {("chat", "you", "steer")},
        )
        self.assertEqual([len(item["images"]) for item in queue["items"]], [0, 1, 0])
        self.assertEqual(
            [item["editable"] for item in queue["items"]], [True, False, True]
        )
        self.assertEqual(
            [row["id"] for row in queue["page"]["rows"]], [first, image, last]
        )
        self.assertEqual(
            [action["id"] for action in queue["page"]["actions"]],
            ["drop", "first", "clear"],
        )

        moved = self.queue(session, f"/{last}/first", {}, method="POST")
        self.assertEqual([item["id"] for item in moved["items"]], [last, first, image])
        dropped = self.queue(session, f"/{image}", method="DELETE")
        self.assertEqual([item["id"] for item in dropped["items"]], [last, first])
        with self.app.api(f"/sessions/{session}/inputs/{image}") as response:
            self.assertEqual(json.load(response)["delivery"], "cancelled")
        with self.assertRaises(urllib.error.HTTPError) as missing:
            self.queue(session, f"/{image}", method="DELETE")
        self.assertEqual(missing.exception.code, 404)

        self.gate.set()
        self.app.idle(session)
        self.assertEqual(self.queue(session)["items"], [])
        said = [
            json.dumps(record["request"]["input"]) for record in self.provider.requests
        ]
        later = next(request for request in said if "first waiting" in request)
        self.assertLess(later.index("last waiting"), later.index("first waiting"))
        self.assertFalse(any("with an image" in request for request in said))

    def test_clearing_the_queue_cancels_what_waits(self):
        session = self.app.session()
        self.app.prompt(session, "hold the turn").close()
        self.assertTrue(self.started.wait(10))
        waiting = [
            self.put_input(session, {"kind": "message", "text": f"waiting {index}"})
            for index in range(3)
        ]
        cleared = self.queue(session, method="DELETE")
        self.assertEqual(cleared["items"], [])
        for input_id in waiting:
            with self.app.api(f"/sessions/{session}/inputs/{input_id}") as response:
                self.assertEqual(json.load(response)["delivery"], "cancelled")
        self.gate.set()
        self.app.idle(session)
        said = json.dumps([record["request"] for record in self.provider.requests])
        self.assertFalse(any(f"waiting {index}" in said for index in range(3)))

    def test_a_long_queue_reaches_the_model_in_bounded_batches_in_order(self):
        session = self.app.session()
        self.app.prompt(session, "hold the turn").close()
        self.assertTrue(self.started.wait(10))
        for index in range(70):
            self.put_input(session, {"kind": "message", "text": f"queued {index:02} ."})
        self.assertEqual(self.queue(session)["total"], 70)
        self.gate.set()
        self.app.idle(session)
        self.assertEqual(self.queue(session)["items"], [])
        seen = []
        for record in self.provider.requests:
            said = json.dumps(record["request"]["input"])
            present = [index for index in range(70) if f"queued {index:02} ." in said]
            new = [index for index in present if index not in seen]
            # No step takes more than one batch from the inbox.
            self.assertLessEqual(len(new), 64)
            seen.extend(new)
            # What a request repeats from history keeps its order.
            self.assertEqual(present, sorted(present))
        self.assertEqual(seen, list(range(70)))

    def test_a_held_message_waits_for_its_edit_and_keeps_its_place(self):
        def script(request):
            if len(self.provider.requests) == 1:
                self.started.set()
                self.gate.wait(40)
                return python("print('one more step')")
            return text("done")

        self.provider.script = script
        session = self.app.session()
        self.app.prompt(session, "hold the turn").close()
        self.assertTrue(self.started.wait(10))
        typo = self.put_input(
            session, {"kind": "message", "text": "use the wrnog branch"}
        )
        behind = self.put_input(session, {"kind": "message", "text": "then push it"})
        held = self.input_action(session, typo, "hold", {})
        self.assertGreater(held["lease_ms"], 0)

        # The next step and the end of the turn go without the held message,
        # and without what waits behind it.
        self.gate.set()
        wait_until(
            lambda: len(self.provider.requests) >= 2 and self.phase(session) == "idle",
            20,
            "the turn did not end",
        )
        time.sleep(0.5)
        self.assertEqual(len(self.provider.requests), 2)
        self.assertFalse(
            any("wrnog" in said or "then push it" in said for said in self.said())
        )
        self.assertEqual(
            [item["id"] for item in self.queue(session)["items"]], [typo, behind]
        )

        edited = self.input_action(
            session, typo, "edit", {"kind": "message", "text": "use the right branch"}
        )
        self.assertEqual(edited["input"]["id"], typo)
        self.app.idle(session)
        said = self.said()
        self.assertFalse(any("wrnog" in request for request in said))
        later = next(request for request in said if "use the right branch" in request)
        self.assertLess(
            later.index("use the right branch"), later.index("then push it")
        )
        with self.app.api(f"/sessions/{session}/inputs/{typo}") as response:
            self.assertEqual(json.load(response)["delivery"], "committed")

        with self.assertRaises(urllib.error.HTTPError) as read:
            self.input_action(
                session, typo, "edit", {"kind": "message", "text": "too late"}
            )
        self.assertEqual(read.exception.code, 409)
        self.assertEqual(json.load(read.exception)["code"], "input_not_waiting")

    def test_a_released_hold_reads_the_message_as_it_was(self):
        session = self.app.session()
        self.app.prompt(session, "hold the turn").close()
        self.assertTrue(self.started.wait(10))
        waiting = self.put_input(session, {"kind": "message", "text": "as it was"})
        self.input_action(session, waiting, "hold", {})
        self.gate.set()
        wait_until(lambda: self.phase(session) == "idle", 20, "the turn did not end")
        time.sleep(0.5)
        self.assertFalse(any("as it was" in said for said in self.said()))

        self.assertIsNone(self.input_action(session, waiting, "hold", method="DELETE"))
        self.app.idle(session)
        self.assertTrue(any("as it was" in said for said in self.said()))


if __name__ == "__main__":
    unittest.main()