aboutsummaryrefslogtreecommitdiff
path: root/app/tests/test_backends_servers.py
diff options
context:
space:
mode:
Diffstat (limited to 'app/tests/test_backends_servers.py')
-rw-r--r--app/tests/test_backends_servers.py52
1 files changed, 52 insertions, 0 deletions
diff --git a/app/tests/test_backends_servers.py b/app/tests/test_backends_servers.py
index 61897d4..4065a34 100644
--- a/app/tests/test_backends_servers.py
+++ b/app/tests/test_backends_servers.py
@@ -1,9 +1,11 @@
"""Tests for the server lifecycle module (backends/servers.py)."""
+import io
import os
import signal
import tempfile
import unittest
+from contextlib import redirect_stdout
from pathlib import Path
from unittest.mock import MagicMock, patch
@@ -118,6 +120,56 @@ class StartTests(unittest.TestCase):
# Pid file cleaned up after early exit.
self.assertFalse((self.dir / "test-server.pid").exists())
+ def test_exited_event_carries_a_known_crash_hint(self):
+ """The exited event's log tail is scanned for known signatures."""
+ (self.dir / "test-server.log").write_text(
+ "triton.compiler.errors.CompilationError:\n"
+ 'ValueError("type fp8e4nv not supported in this architecture. '
+ 'The supported fp8 dtypes are")\n', encoding="utf-8")
+ proc = MagicMock()
+ proc.pid = 99
+ proc.poll.return_value = 1
+ events = []
+ with patch.object(servers, "LOG_DIR", self.dir), \
+ patch("subprocess.Popen", return_value=proc), \
+ patch("backends.common.server_running", return_value=False), \
+ patch("time.sleep"):
+ ok = servers.start(self.spec, progress=events.append)
+ self.assertFalse(ok)
+ exited = next(e for e in events if e.get("kind") == "exited")
+ self.assertIn("FP8", exited["hint"])
+ self.assertIn("8.9", exited["hint"])
+
+ def test_exited_event_has_no_hint_for_unknown_crashes(self):
+ proc = MagicMock()
+ proc.pid = 99
+ proc.poll.return_value = 1
+ events = []
+ with patch.object(servers, "LOG_DIR", self.dir), \
+ patch("subprocess.Popen", return_value=proc), \
+ patch("backends.common.server_running", return_value=False), \
+ patch("time.sleep"):
+ servers.start(self.spec, progress=events.append)
+ exited = next(e for e in events if e.get("kind") == "exited")
+ self.assertIsNone(exited["hint"])
+
+ def test_boot_hint_reads_the_log_tail(self):
+ self.assertIsNone(servers._boot_hint(["everything fine"]))
+ self.assertIn("FP8", servers._boot_hint(
+ ["x", 'ValueError("type fp8e4nv not supported in this '
+ 'architecture")', "y"]))
+ self.assertIsNone(servers._boot_hint([]))
+
+ def test_console_progress_prints_the_hint(self):
+ out = io.StringIO()
+ with redirect_stdout(out):
+ servers._console_progress({
+ "kind": "exited", "name": "test", "returncode": 1,
+ "log_tail": ["boom"], "hint": "FP8 needs compute "
+ "capability 8.9+"})
+ self.assertIn("hint: FP8 needs compute capability 8.9+",
+ out.getvalue())
+
def test_returns_false_on_timeout(self):
proc = MagicMock()
proc.pid = 7