temp

Author: riturajFi

tests/e2e/__init__.py

@@ -0,0 +1,6 @@
+"""End-to-end tests runnable as standalone scripts.
+
+Run from repo root, e.g.:
+  python tests/e2e/test_middle_out_basic.py
+"""
+

tests/e2e/_client.py

@@ -0,0 +1,35 @@
+import os
+import sys
+from pathlib import Path
+
+
+def _import_client():
+    try:
+        from acontext import AcontextClient  # type: ignore
+
+        return AcontextClient
+    except ModuleNotFoundError:
+        repo_root = Path(__file__).resolve().parents[2]
+        sdk_src = repo_root / "src" / "client" / "acontext-py" / "src"
+        sys.path.insert(0, str(sdk_src))
+        from acontext import AcontextClient  # type: ignore
+
+        return AcontextClient
+
+
+AcontextClient = _import_client()
+
+BASE_URL = os.environ.get("BASE_URL", "http://localhost:8080")
+
+if "API_KEY" not in os.environ:
+    raise RuntimeError(
+        "Missing API_KEY. Run with e.g. `API_KEY=... python tests/e2e/test_middle_out_basic.py`"
+    )
+
+API_KEY = os.environ["API_KEY"]
+
+client = AcontextClient(
+    api_key=API_KEY,
+    base_url=BASE_URL,
+)
+

tests/e2e/_helpers.py

@@ -0,0 +1,57 @@
+from __future__ import annotations
+
+from typing import Any, List
+
+from tests.e2e._client import client
+
+
+def new_session() -> str:
+    session = client.sessions.create()
+    return session.id
+
+
+def store_text(session_id: str, text: str) -> None:
+    client.sessions.store_message(
+        session_id,
+        format="acontext",
+        blob={
+            "role": "user",
+            "parts": [{"type": "text", "text": text}],
+        },
+    )
+
+
+def _extract_texts_from_acontext_messages(items: list[Any]) -> list[str]:
+    texts: list[str] = []
+    for message in items:
+        for part in getattr(message, "parts", []) or []:
+            if getattr(part, "type", None) == "text" and getattr(part, "text", None):
+                texts.append(part.text)
+    return texts
+
+
+def _extract_texts_from_openai_messages(items: list[Any]) -> list[str]:
+    texts: list[str] = []
+    for message in items:
+        if isinstance(message, dict):
+            content = message.get("content")
+            if isinstance(content, str) and content:
+                texts.append(content)
+    return texts
+
+
+def get_texts(session_id: str, **kwargs: Any) -> List[str]:
+    resp = client.sessions.get_messages(session_id, **kwargs)
+    format_ = kwargs.get("format", "openai")
+    if format_ == "acontext":
+        return _extract_texts_from_acontext_messages(resp.items)
+    if format_ == "openai":
+        return _extract_texts_from_openai_messages(resp.items)
+    return []
+
+
+def banner(title: str) -> None:
+    print("\n" + "=" * 80)
+    print(title)
+    print("=" * 80)
+

tests/e2e/test_middle_out_basic.py

@@ -0,0 +1,34 @@
+from tests.e2e._helpers import banner, get_texts, new_session, store_text
+
+
+def main() -> None:
+    banner("middle_out preserves head and tail")
+
+    sid = new_session()
+
+    for i in range(30):
+        store_text(sid, f"msg-{i} " + "x" * 200)
+
+    texts = get_texts(
+        sid,
+        format="acontext",
+        edit_strategies=[
+            {
+                "type": "middle_out",
+                "params": {"token_reduce_to": 500},
+            }
+        ],
+    )
+
+    for t in texts:
+        print(t[:20])
+
+    print("\nEXPECT:")
+    print("- msg-0, msg-1 present")
+    print("- msg-28, msg-29 present")
+    print("- gaps in the middle")
+
+
+if __name__ == "__main__":
+    main()
+

tests/e2e/test_middle_out_even_determinism.py

@@ -0,0 +1,30 @@
+from tests.e2e._helpers import banner, get_texts, new_session, store_text
+
+
+def main() -> None:
+    banner("even-count determinism (right-middle removed)")
+
+    sid = new_session()
+
+    for m in ["m0", "m1", "m2", "m3"]:
+        store_text(sid, m)
+
+    texts = get_texts(
+        sid,
+        format="acontext",
+        edit_strategies=[
+            {
+                "type": "middle_out",
+                "params": {"token_reduce_to": 10},
+            }
+        ],
+    )
+
+    print("Returned:", texts)
+    print("\nEXPECT:")
+    print("- m2 is missing")
+
+
+if __name__ == "__main__":
+    main()
+

tests/e2e/test_middle_out_keep_tail.py

@@ -0,0 +1,30 @@
+from tests.e2e._helpers import banner, get_texts, new_session, store_text
+
+
+def main() -> None:
+    banner("keep-tail fallback")
+
+    sid = new_session()
+
+    store_text(sid, "old " + "x" * 200)
+    store_text(sid, "new " + "x" * 200)
+
+    texts = get_texts(
+        sid,
+        format="acontext",
+        edit_strategies=[
+            {
+                "type": "middle_out",
+                "params": {"token_reduce_to": 10},
+            }
+        ],
+    )
+
+    print("Returned:", texts)
+    print("\nEXPECT:")
+    print("- only 'new' remains")
+
+
+if __name__ == "__main__":
+    main()
+

tests/e2e/test_middle_out_tool_pairing.py

@@ -0,0 +1,58 @@
+import json
+
+from tests.e2e._client import client
+from tests.e2e._helpers import banner, new_session, store_text
+
+
+def main() -> None:
+    banner("tool-call / tool-result pairing")
+
+    sid = new_session()
+
+    client.sessions.store_message(
+        sid,
+        format="openai",
+        blob={
+            "role": "assistant",
+            "tool_calls": [
+                {
+                    "id": "call_1",
+                    "type": "function",
+                    "function": {"name": "f", "arguments": "{}"},
+                }
+            ],
+        },
+    )
+
+    client.sessions.store_message(
+        sid,
+        format="openai",
+        blob={
+            "role": "tool",
+            "tool_call_id": "call_1",
+            "content": "ok",
+        },
+    )
+
+    store_text(sid, "noise")
+
+    resp = client.sessions.get_messages(
+        sid,
+        format="openai",
+        edit_strategies=[
+            {
+                "type": "middle_out",
+                "params": {"token_reduce_to": 50},
+            }
+        ],
+    )
+
+    print(json.dumps(resp.model_dump(), indent=2))
+
+    print("\nEXPECT:")
+    print("- tool-call and tool-result appear together or not at all")
+
+
+if __name__ == "__main__":
+    main()
+

tests/e2e/test_middle_out_validation_errors.py

@@ -0,0 +1,31 @@
+from tests.e2e._client import client
+from tests.e2e._helpers import banner, new_session
+
+
+def main() -> None:
+    banner("validation errors")
+
+    sid = new_session()
+
+    cases = [
+        {"type": "middle_out", "params": {}},
+        {"type": "middle_out", "params": {"token_reduce_to": "x"}},
+        {"type": "middle_out", "params": {"token_reduce_to": 0}},
+    ]
+
+    for c in cases:
+        print("\nCase:", c)
+        try:
+            client.sessions.get_messages(
+                sid,
+                format="acontext",
+                edit_strategies=[c],  # type: ignore[arg-type]
+            )
+            print("❌ Unexpected success")
+        except Exception as e:
+            print("✅ Error:", e)
+
+
+if __name__ == "__main__":
+    main()
+