nmpAgents/demo/tests/test_demo.py

46 lines
2.0 KiB
Python

"""Pruebas locales del cliente; no llaman a AWS."""
import importlib.util
import json
from pathlib import Path
import tempfile
import unittest
from unittest.mock import patch
import zipfile
spec = importlib.util.spec_from_file_location("demo_cli", Path(__file__).resolve().parents[1] / "scripts/demo.py")
demo = importlib.util.module_from_spec(spec)
spec.loader.exec_module(demo)
class DemoTests(unittest.TestCase):
def test_sse_joins_text_and_ignores_tool_events(self):
delta = lambda text: json.dumps({"event": {"contentBlockDelta": {"delta": {"text": text}}}})
raw = 'data: {"event":{"contentBlockStart":{"start":{"toolUse":{}}}}}\n'
raw += f'data: {delta("Hola ")}\n\ndata:{delta("NMP")}\ndata: [DONE]\n'
self.assertEqual(demo.extract_text(raw), "Hola NMP")
def test_stream_error_after_partial_text_is_not_success(self):
raw = 'data: {"event":{"contentBlockDelta":{"delta":{"text":"parcial"}}}}\n'
raw += 'data: {"event":{"modelStreamErrorException":{"message":"fallo"}}}\n'
with self.assertRaisesRegex(ValueError, "modelStreamErrorException"):
demo.extract_text(raw)
def test_package_contains_importable_module_at_zip_root(self):
with tempfile.TemporaryDirectory() as folder:
root = Path(folder)
source = root / "src/lambdas/example.py"
source.parent.mkdir(parents=True)
source.write_text("def lambda_handler(event, context): return event\n")
with patch.object(demo, "ROOT", root), patch.object(demo, "LAMBDAS", {"example": "src/lambdas/example.py"}), patch.object(demo, "validate"):
demo.package()
with zipfile.ZipFile(root / "artifacts/example.zip") as archive:
self.assertEqual(archive.namelist(), ["example.py"])
self.assertEqual(archive.read("example.py"), source.read_bytes())
def test_project_references(self):
demo.validate()
if __name__ == "__main__":
unittest.main()