scripts/check_minimal_install.py
Rama main · réplica de lectura del repositorio original.
#!/usr/bin/env python3
"""Exercise core chat, SQLite persistence and replay with a local model fixture.
Run outside the checkout with an interpreter containing only the core wheel.
--source-core additionally blocks optional source imports for fast development.
No provider credentials or external network calls are used.
"""
import argparse
import asyncio
import importlib.abc
import importlib.machinery
import json
import os
from pathlib import Path
import sys
import tempfile
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--source-core", action="store_true")
parser.add_argument("--capability", default="chat")
parser.add_argument("--home", type=Path)
parser.add_argument("--full", action="store_true", help="Allow the full application's capabilities")
parser.add_argument("--save-replay", type=Path, help="Save a durable turn to verify after upgrade")
parser.add_argument("--verify-replay", type=Path, help="Replay a turn saved before the upgrade")
args = parser.parse_args()
runtime_home = tempfile.TemporaryDirectory(prefix="deeptutor-core-smoke-")
os.environ["DEEPTUTOR_HOME"] = str(args.home) if args.home else runtime_home.name
if args.source_core:
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
if args.source_core:
from deeptutor.modules import catalog
catalog.installed = lambda name: name == "core"
class BlockOptional(importlib.abc.MetaPathFinder):
def find_spec(self, fullname, path=None, target=None):
if fullname.startswith(("deeptutor.", "deeptutor_cli.")):
relative = fullname.replace(".", "/")
owner = catalog.module_for_path(relative + ".py")
if not (Path(__file__).resolve().parents[1] / (relative + ".py")).is_file():
owner = catalog.module_for_path(relative + "/__init__.py")
if owner != "core":
# Wheels retain namespace parents for core-owned shared metadata,
# without shipping the optional application's __init__.py.
if any(
item.startswith(relative + "/")
for item in catalog.manifests()["core"]["paths"]
):
spec = importlib.machinery.ModuleSpec(
fullname, loader=None, is_package=True
)
spec.submodule_search_locations = [
str(Path(__file__).resolve().parents[1] / relative)
]
return spec
raise ModuleNotFoundError(f"Absent module {owner}: {fullname}", name=fullname)
sys.meta_path.insert(0, BlockOptional())
from deeptutor.app import DeepTutorApp
async def main():
app = DeepTutorApp()
try:
if args.verify_replay:
saved = json.loads(args.verify_replay.read_text())
assert await app.get_session(saved["session_id"]) is not None
replay = [item async for item in app.stream_turn(saved["turn_id"])]
assert [
[e["seq"], e["type"], e["content"]]
for e in replay
if e["seq"] <= saved["terminal_seq"]
] == saved["events"]
print("Pre-upgrade session and durable turn replay: OK")
contracts = {row["name"] for row in app.get_capability_contracts()}
assert {"chat", "ask_questions", args.capability} <= contracts
if args.capability == "chat" and not args.full:
assert contracts == {"chat", "ask_questions"}, contracts
session, turn = await app.start_turn(
{"content": "Hello", "capability": args.capability, "tools": []}
)
events = [item async for item in app.stream_turn(turn["id"])]
assert not [event for event in events if event["type"] == "error"], events
assert any(
event["type"] == "content" and "minimal core" in event["content"] for event in events
), events
terminal = next(event for event in events if event["type"] == "done")
assert terminal["metadata"]["status"] == "completed", events
assert await app.get_session(session["id"]) is not None
await app.container.close()
from deeptutor.app.container import set_application_container
set_application_container(None)
app = DeepTutorApp()
replay = [item async for item in app.stream_turn(turn["id"])]
assert [
(e["seq"], e["type"], e["content"]) for e in replay if e["seq"] <= terminal["seq"]
] == [(e["seq"], e["type"], e["content"]) for e in events if e["seq"] <= terminal["seq"]]
if args.save_replay:
args.save_replay.write_text(
json.dumps(
{
"session_id": session["id"],
"turn_id": turn["id"],
"terminal_seq": terminal["seq"],
"events": [
[e["seq"], e["type"], e["content"]]
for e in events
if e["seq"] <= terminal["seq"]
],
}
)
)
print("Core chat, durable session and replay: OK")
finally:
await app.container.close()
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
import threading
class Handler(BaseHTTPRequestHandler):
def log_message(self, *args):
pass
def do_POST(self):
body = json.loads(self.rfile.read(int(self.headers["Content-Length"])))
self.send_response(200)
self.send_header(
"Content-Type", "text/event-stream" if body.get("stream") else "application/json"
)
self.end_headers()
if body.get("stream"):
for delta, finish in [
({"content": "Hello from the minimal core."}, None),
({}, "stop"),
]:
value = {
"id": "smoke",
"object": "chat.completion.chunk",
"created": 0,
"model": "smoke",
"choices": [{"index": 0, "delta": delta, "finish_reason": finish}],
}
self.wfile.write(("data: " + json.dumps(value) + "\n\n").encode())
self.wfile.write(b"data: [DONE]\n\n")
else:
self.wfile.write(
json.dumps(
{
"id": "smoke",
"object": "chat.completion",
"created": 0,
"model": "smoke",
"choices": [
{
"index": 0,
"message": {
"role": "assistant",
"content": "Hello from the minimal core.",
},
"finish_reason": "stop",
}
],
"usage": {"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2},
}
).encode()
)
server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
threading.Thread(target=server.serve_forever, daemon=True).start()
import deeptutor.services.llm.config as config
config._LLM_CONFIG_CACHE = config.LLMConfig(
model="smoke",
api_key="local-test",
base_url=f"http://127.0.0.1:{server.server_port}/v1",
wire_api="chat_completions",
)
try:
asyncio.run(asyncio.wait_for(main(), timeout=30))
finally:
server.shutdown()
runtime_home.cleanup()