"""Synthetic approval exercise. No model, external CRM, credentials or network.
Tested with Python 3.13 and langgraph==1.2.19. Run this file to execute tests.
In-memory data and checkpoints deliberately do NOT survive a restart.
"""
from typing import TypedDict
from hashlib import sha256
import json
import unittest
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import interrupt, Command

class State(TypedDict, total=False):
    action: dict
    expected: str
    outcome: str

def fingerprint(action):
    return sha256(json.dumps(action, sort_keys=True).encode()).hexdigest()

class FakeCRM:
    def __init__(self):
        self.records = {"demo-001": {"stage": "new"}}
        self.applied = {}
        self.available = True

    def apply(self, action):
        if not self.available:
            raise ConnectionError("synthetic outage")
        digest = fingerprint(action)
        key = action["id"]
        if key in self.applied:
            if self.applied[key] != digest:
                return "id-conflict"
            return "already-applied"
        self.records[action["record"]]["stage"] = action["value"]
        self.applied[key] = digest
        return "applied"

def build(crm):
    def review(state):
        action = state["action"]
        # Whitelist a single synthetic operation and record; reject others.
        if (action.get("operation") != "set-stage"
            or action.get("record") not in crm.records
            or action.get("value") != "reviewed"
            or not isinstance(action.get("id"), str)):
            return {"outcome": "outside-scope"}
        digest = fingerprint(action)
        answer = interrupt({"action": action, "fingerprint": digest})
        # Real identity and authorization must be checked by the application.
        if not isinstance(answer, dict) or answer.get("approve") is not True:
            return {"outcome": "denied"}
        if answer.get("fingerprint") != digest:
            return {"outcome": "approval-mismatch"}
        return {"outcome": crm.apply(action)}

    builder = StateGraph(State)
    builder.add_node("review", review)
    builder.add_edge(START, "review")
    builder.add_edge("review", END)
    return builder.compile(checkpointer=InMemorySaver())

def action(**changes):
    return dict(id="change-001", operation="set-stage", record="demo-001",
                value="reviewed") | changes

class ApprovalTests(unittest.TestCase):
    def setUp(self):
        self.crm = FakeCRM()
        self.graph = build(self.crm)
        self.config = {"configurable": {"thread_id": "synthetic-thread"}}
        self.proposal = action()

    def pending(self):
        result = self.graph.invoke({"action": self.proposal}, self.config)
        self.assertIn("__interrupt__", result)
        self.assertEqual(self.crm.records["demo-001"]["stage"], "new")

    def resume(self, approve=True, digest=None):
        return self.graph.invoke(Command(resume={"approve": approve,
            "fingerprint": digest or fingerprint(self.proposal)}), self.config)

    def test_waits_before_write(self):
        self.pending()
        self.assertEqual(len(self.crm.applied), 0)

    def test_approved_exact_action(self):
        self.pending()
        self.assertEqual(self.resume()["outcome"], "applied")
        self.assertEqual(self.crm.records["demo-001"]["stage"], "reviewed")

    def test_denied_does_not_write(self):
        self.pending()
        self.assertEqual(self.resume(False)["outcome"], "denied")
        self.assertEqual(self.crm.records["demo-001"]["stage"], "new")

    def test_mismatched_approval(self):
        self.pending()
        self.assertEqual(self.resume(digest="different")["outcome"],
                         "approval-mismatch")
        self.assertEqual(len(self.crm.applied), 0)

    def test_duplicate_is_not_applied_twice(self):
        self.pending()
        self.resume()
        other = {"configurable": {"thread_id": "another-thread"}}
        self.graph.invoke({"action": self.proposal}, other)
        result = self.graph.invoke(Command(resume={"approve": True,
            "fingerprint": fingerprint(self.proposal)}), other)
        self.assertEqual(result["outcome"], "already-applied")
        self.assertEqual(len(self.crm.applied), 1)

    def test_outside_scope_never_asks_or_writes(self):
        bad = self.proposal | {"operation": "delete"}
        result = self.graph.invoke({"action": bad}, self.config)
        self.assertEqual(result["outcome"], "outside-scope")
        self.assertEqual(len(self.crm.applied), 0)

    def test_outage_then_retry(self):
        self.pending()
        self.crm.available = False
        with self.assertRaises(ConnectionError):
            self.resume()
        self.assertEqual(len(self.crm.applied), 0)
        self.crm.available = True
        result = self.graph.invoke(None, self.config)
        self.assertEqual(result["outcome"], "applied")
        self.assertEqual(len(self.crm.applied), 1)

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