Vito Sansevero
e7fb2d8e76
Rule R (#329): a file is a rotated copy only when it is a direct entry of the log folder named prompt_manager.log.<n> (n >= 1, no leading zero, compared after normcase) and a regular file that is not a symlink or a junction. Everything else is never deleted. - Clear (POST /logs/truncate) empties the active log and deletes only rotated copies; a .bak, a .1.gz, any other file and every link are kept. It runs off the event loop under one files lock and the file handler's lock, so no rollover renames a file mid-way, and logs only afterwards. - delete_rotated(expected) deletes exactly the reviewed copies, or nothing when one of them changed (size, mtime_ns, inode, missing, now a link). - The download serves only a listed regular, non-link file, opened without following links and read off the loop; a link or folder is a 404. - /logs/files rows gain kind; files and stats read links with lstat and stats no longer count a link's target. - The config update runs off the loop and swaps handlers under the files lock.
582 lines
21 KiB
Python
582 lines
21 KiB
Python
"""Route tests for the logging API (py/api/logging_routes.py)."""
|
|
|
|
import json
|
|
import os
|
|
import shutil
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
_mock_server = MagicMock()
|
|
_mock_server.PromptServer.instance.routes = MagicMock()
|
|
sys.modules.setdefault("server", _mock_server)
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
from aiohttp import web # noqa: E402
|
|
from aiohttp.test_utils import AioHTTPTestCase # noqa: E402
|
|
|
|
from database.operations import PromptDatabase # noqa: E402
|
|
from py.api import PromptManagerAPI # noqa: E402
|
|
from py.api import logging_routes # noqa: E402
|
|
from py.api.logging_routes import LOG_CONFIG_KEYS # noqa: E402
|
|
from utils import logging_config # noqa: E402
|
|
|
|
|
|
class FakeLoggerManager:
|
|
"""Stand-in for PromptManagerLogger with a scratch log directory."""
|
|
|
|
def __init__(self, log_dir):
|
|
self.log_dir = Path(log_dir)
|
|
self.config = {
|
|
"level": "INFO",
|
|
"max_file_size": 10 * 1024 * 1024,
|
|
"backup_count": 5,
|
|
"console_logging": True,
|
|
"file_logging": True,
|
|
"buffer_size": 1000,
|
|
}
|
|
self.update_calls = []
|
|
self.logs = []
|
|
self.threads = {}
|
|
|
|
def update_config(self, new_config):
|
|
self.threads["update_config"] = threading.current_thread()
|
|
self.update_calls.append(dict(new_config))
|
|
self.config.update(new_config)
|
|
|
|
def log_file_entry(self, name):
|
|
return logging_config.log_file_entry(self.log_dir, name)
|
|
|
|
def get_config(self):
|
|
return dict(self.config)
|
|
|
|
def get_recent_logs(self, limit=100, level=None):
|
|
logs = self.logs
|
|
if level:
|
|
logs = [entry for entry in logs if entry["level"] == level.upper()]
|
|
return list(reversed(logs[-limit:]))
|
|
|
|
def get_log_files(self):
|
|
files = []
|
|
for row in sorted(
|
|
logging_config.scan_log_folder(self.log_dir), key=lambda r: r["name"]
|
|
):
|
|
if row["_dir"]:
|
|
continue
|
|
files.append(
|
|
{
|
|
"filename": row["name"],
|
|
"path": str(self.log_dir / row["name"]),
|
|
"size": row["size"],
|
|
"modified": "2026-10-07T00:00:00",
|
|
"is_main": row["kind"] == "active",
|
|
"kind": row["kind"],
|
|
}
|
|
)
|
|
return files
|
|
|
|
def truncate_logs(self):
|
|
"""Rule R: empty the active log, delete rotated copies, keep the rest."""
|
|
self.threads["truncate_logs"] = threading.current_thread()
|
|
truncated = []
|
|
for row in logging_config.scan_log_folder(self.log_dir):
|
|
path = self.log_dir / row["name"]
|
|
if row["kind"] == "active":
|
|
path.write_text("")
|
|
elif row["kind"] == "rotated":
|
|
path.unlink()
|
|
else:
|
|
continue
|
|
truncated.append(row["name"])
|
|
return {"truncated": truncated, "errors": []}
|
|
|
|
def get_log_stats(self):
|
|
return {
|
|
"buffer_count": len(self.logs),
|
|
"level_counts": {},
|
|
"log_files_count": len(self.get_log_files()),
|
|
"total_log_size": 0,
|
|
"log_directory": str(self.log_dir),
|
|
"current_level": self.config["level"],
|
|
}
|
|
|
|
|
|
class LoggingAPITestCase(AioHTTPTestCase):
|
|
|
|
async def get_application(self):
|
|
self.tmpdir = tempfile.mkdtemp()
|
|
self.log_dir = Path(self.tmpdir) / "logs"
|
|
self.log_dir.mkdir()
|
|
self.manager = FakeLoggerManager(self.log_dir)
|
|
# Public paths are rendered relative to ComfyUI's base directory. Other
|
|
# test modules leave their own ``folder_paths`` stub behind, so declare
|
|
# ours explicitly instead of depending on import order.
|
|
self._saved_folder_paths = sys.modules.get("folder_paths")
|
|
fake_folder_paths = MagicMock()
|
|
fake_folder_paths.base_path = self.tmpdir
|
|
fake_folder_paths.get_output_directory.return_value = os.path.join(
|
|
self.tmpdir, "output"
|
|
)
|
|
sys.modules["folder_paths"] = fake_folder_paths
|
|
|
|
app = web.Application()
|
|
routes = web.RouteTableDef()
|
|
self.api = PromptManagerAPI(
|
|
db=PromptDatabase(os.path.join(self.tmpdir, "prompts.db"))
|
|
)
|
|
self.api._get_logger_manager = lambda: self.manager
|
|
self.api.add_routes(routes)
|
|
app.router.add_routes(routes)
|
|
return app
|
|
|
|
async def tearDownAsync(self):
|
|
await super().tearDownAsync() # closes the aiohttp test client
|
|
if self._saved_folder_paths is None:
|
|
sys.modules.pop("folder_paths", None)
|
|
else:
|
|
sys.modules["folder_paths"] = self._saved_folder_paths
|
|
self.api.db.close_all()
|
|
shutil.rmtree(self.tmpdir, ignore_errors=True)
|
|
|
|
async def _post_config(self, payload):
|
|
return await self.client.request(
|
|
"POST", "/prompt_manager/logs/config", json=payload
|
|
)
|
|
|
|
|
|
class TestUpdateLogConfigWhitelist(LoggingAPITestCase):
|
|
|
|
def test_whitelist_matches_logger_manager_keys(self):
|
|
self.assertEqual(LOG_CONFIG_KEYS, frozenset(self.manager.config.keys()))
|
|
|
|
async def test_unknown_keys_rejected_and_listed(self):
|
|
resp = await self._post_config({"level": "INFO", "evil": 1, "log_dir": "/x"})
|
|
|
|
self.assertEqual(resp.status, 400)
|
|
data = await resp.json()
|
|
self.assertFalse(data["success"])
|
|
self.assertIn("evil", data["error"])
|
|
self.assertIn("log_dir", data["error"])
|
|
self.assertEqual(self.manager.update_calls, [])
|
|
|
|
async def test_invalid_level_rejected(self):
|
|
resp = await self._post_config({"level": "VERBOSE"})
|
|
self.assertEqual(resp.status, 400)
|
|
self.assertEqual(self.manager.update_calls, [])
|
|
|
|
async def test_level_must_be_a_string(self):
|
|
resp = await self._post_config({"level": 10})
|
|
self.assertEqual(resp.status, 400)
|
|
self.assertEqual(self.manager.update_calls, [])
|
|
|
|
async def test_level_is_normalised_to_upper_case(self):
|
|
resp = await self._post_config({"level": "debug"})
|
|
self.assertEqual(resp.status, 200)
|
|
self.assertEqual(self.manager.update_calls, [{"level": "DEBUG"}])
|
|
|
|
async def test_negative_int_rejected(self):
|
|
resp = await self._post_config({"backup_count": -1})
|
|
self.assertEqual(resp.status, 400)
|
|
self.assertEqual(self.manager.update_calls, [])
|
|
|
|
async def test_numeric_string_rejected(self):
|
|
resp = await self._post_config({"max_file_size": "1024"})
|
|
self.assertEqual(resp.status, 400)
|
|
self.assertEqual(self.manager.update_calls, [])
|
|
|
|
async def test_bool_is_not_an_int(self):
|
|
resp = await self._post_config({"buffer_size": True})
|
|
self.assertEqual(resp.status, 400)
|
|
self.assertEqual(self.manager.update_calls, [])
|
|
|
|
async def test_flags_must_be_bool(self):
|
|
resp = await self._post_config({"console_logging": "yes"})
|
|
self.assertEqual(resp.status, 400)
|
|
self.assertEqual(self.manager.update_calls, [])
|
|
|
|
async def test_valid_payload_is_applied(self):
|
|
payload = {
|
|
"level": "WARNING",
|
|
"max_file_size": 2 * 1024 * 1024,
|
|
"backup_count": 0,
|
|
"console_logging": False,
|
|
"file_logging": True,
|
|
"buffer_size": 50,
|
|
}
|
|
resp = await self._post_config(payload)
|
|
|
|
self.assertEqual(resp.status, 200)
|
|
data = await resp.json()
|
|
self.assertTrue(data["success"])
|
|
self.assertEqual(self.manager.update_calls, [payload])
|
|
self.assertEqual(data["config"]["level"], "WARNING")
|
|
|
|
async def test_max_file_size_is_bounded_to_1_to_1024_mb(self):
|
|
mb = 1024 * 1024
|
|
for bad in (0, mb - 1, 1024 * mb + 1, 10**12):
|
|
resp = await self._post_config({"max_file_size": bad})
|
|
self.assertEqual(resp.status, 400, bad)
|
|
self.assertIn("max_file_size", (await resp.json())["error"])
|
|
self.assertEqual(self.manager.update_calls, [])
|
|
for ok in (mb, 1024 * mb):
|
|
resp = await self._post_config({"max_file_size": ok})
|
|
self.assertEqual(resp.status, 200, ok)
|
|
self.assertEqual(
|
|
self.manager.update_calls,
|
|
[{"max_file_size": mb}, {"max_file_size": 1024 * mb}],
|
|
)
|
|
|
|
async def test_backup_count_is_bounded_to_0_to_50(self):
|
|
for bad in (-1, 51, 10**6):
|
|
resp = await self._post_config({"backup_count": bad})
|
|
self.assertEqual(resp.status, 400, bad)
|
|
self.assertIn("backup_count", (await resp.json())["error"])
|
|
self.assertEqual(self.manager.update_calls, [])
|
|
for ok in (0, 50):
|
|
resp = await self._post_config({"backup_count": ok})
|
|
self.assertEqual(resp.status, 200, ok)
|
|
|
|
async def test_body_must_be_an_object(self):
|
|
resp = await self._post_config(["level", "DEBUG"])
|
|
self.assertEqual(resp.status, 400)
|
|
self.assertEqual(self.manager.update_calls, [])
|
|
|
|
async def test_malformed_json_is_400(self):
|
|
resp = await self.client.request(
|
|
"POST",
|
|
"/prompt_manager/logs/config",
|
|
data=b"{not json",
|
|
headers={"Content-Type": "application/json"},
|
|
)
|
|
self.assertEqual(resp.status, 400)
|
|
self.assertEqual(self.manager.update_calls, [])
|
|
|
|
async def test_oversized_body_is_413(self):
|
|
body = json.dumps({"level": "x" * 1_000_100}).encode()
|
|
resp = await self.client.request(
|
|
"POST",
|
|
"/prompt_manager/logs/config",
|
|
data=body,
|
|
headers={"Content-Type": "application/json"},
|
|
)
|
|
self.assertEqual(resp.status, 413)
|
|
self.assertFalse((await resp.json())["success"])
|
|
self.assertEqual(self.manager.update_calls, [])
|
|
|
|
|
|
class TestNoAbsolutePathsInLogResponses(LoggingAPITestCase):
|
|
|
|
async def test_log_files_have_no_absolute_path(self):
|
|
(self.log_dir / "prompt_manager.log").write_text("hello")
|
|
|
|
resp = await self.client.request("GET", "/prompt_manager/logs/files")
|
|
|
|
body = await resp.text()
|
|
self.assertEqual(resp.status, 200)
|
|
self.assertNotIn(self.tmpdir, body)
|
|
data = json.loads(body)
|
|
self.assertEqual(data["files"][0]["filename"], "prompt_manager.log")
|
|
|
|
async def test_log_stats_directory_is_not_absolute(self):
|
|
resp = await self.client.request("GET", "/prompt_manager/logs/stats")
|
|
|
|
body = await resp.text()
|
|
self.assertEqual(resp.status, 200)
|
|
self.assertNotIn(self.tmpdir, body)
|
|
self.assertEqual(json.loads(body)["stats"]["log_directory"], "logs")
|
|
|
|
async def test_error_from_manager_does_not_leak_path(self):
|
|
def boom():
|
|
raise FileNotFoundError(
|
|
2, "No such file", os.path.join(self.tmpdir, "x.log")
|
|
)
|
|
|
|
self.manager.get_log_stats = boom
|
|
|
|
resp = await self.client.request("GET", "/prompt_manager/logs/stats")
|
|
|
|
body = await resp.text()
|
|
self.assertEqual(resp.status, 500)
|
|
self.assertNotIn(self.tmpdir, body)
|
|
self.assertIn("x.log", json.loads(body)["error"])
|
|
|
|
|
|
class TestGetLogs(LoggingAPITestCase):
|
|
|
|
def _fill(self, n):
|
|
for i in range(n):
|
|
self.manager.logs.append(
|
|
{"level": "INFO" if i % 2 else "ERROR", "message": f"m{i}"}
|
|
)
|
|
|
|
async def test_default_limit_and_count(self):
|
|
self._fill(3)
|
|
resp = await self.client.request("GET", "/prompt_manager/logs")
|
|
data = await resp.json()
|
|
self.assertEqual(resp.status, 200)
|
|
self.assertTrue(data["success"])
|
|
self.assertEqual(data["count"], 3)
|
|
self.assertEqual(data["limit"], 100)
|
|
self.assertIsNone(data["level_filter"])
|
|
|
|
async def test_level_filter_passed_through(self):
|
|
self._fill(4)
|
|
resp = await self.client.request("GET", "/prompt_manager/logs?level=error")
|
|
data = await resp.json()
|
|
self.assertEqual(data["level_filter"], "error")
|
|
self.assertEqual(data["count"], 2)
|
|
|
|
async def test_limit_is_clamped_to_500(self):
|
|
resp = await self.client.request("GET", "/prompt_manager/logs?limit=99999")
|
|
self.assertEqual((await resp.json())["limit"], 500)
|
|
|
|
async def test_limit_below_one_becomes_one(self):
|
|
self._fill(3)
|
|
resp = await self.client.request("GET", "/prompt_manager/logs?limit=0")
|
|
data = await resp.json()
|
|
self.assertEqual(data["limit"], 1)
|
|
self.assertEqual(data["count"], 1)
|
|
|
|
async def test_non_integer_limit_is_400(self):
|
|
resp = await self.client.request("GET", "/prompt_manager/logs?limit=ten")
|
|
self.assertEqual(resp.status, 400)
|
|
self.assertFalse((await resp.json())["success"])
|
|
|
|
async def test_manager_failure_is_500_with_empty_logs(self):
|
|
def boom(**kwargs):
|
|
raise RuntimeError("buffer gone")
|
|
|
|
self.manager.get_recent_logs = boom
|
|
resp = await self.client.request("GET", "/prompt_manager/logs")
|
|
data = await resp.json()
|
|
self.assertEqual(resp.status, 500)
|
|
self.assertEqual(data["logs"], [])
|
|
|
|
|
|
class TestLogFilesAndDownload(LoggingAPITestCase):
|
|
|
|
async def test_files_failure_is_500(self):
|
|
def boom():
|
|
raise RuntimeError("no dir")
|
|
|
|
self.manager.get_log_files = boom
|
|
resp = await self.client.request("GET", "/prompt_manager/logs/files")
|
|
data = await resp.json()
|
|
self.assertEqual(resp.status, 500)
|
|
self.assertEqual(data["files"], [])
|
|
|
|
async def test_download_existing_file(self):
|
|
(self.log_dir / "prompt_manager.log").write_bytes(b"line1\nline2\n")
|
|
|
|
resp = await self.client.request(
|
|
"GET", "/prompt_manager/logs/download/prompt_manager.log"
|
|
)
|
|
|
|
self.assertEqual(resp.status, 200)
|
|
self.assertIn("attachment", resp.headers["Content-Disposition"])
|
|
self.assertEqual(await resp.text(), "line1\nline2\n")
|
|
|
|
async def test_download_missing_file_is_404(self):
|
|
resp = await self.client.request(
|
|
"GET", "/prompt_manager/logs/download/nope.log"
|
|
)
|
|
self.assertEqual(resp.status, 404)
|
|
|
|
async def test_download_rejects_traversal(self):
|
|
resp = await self.client.request(
|
|
"GET", "/prompt_manager/logs/download/..%2F..%2Fetc%2Fpasswd"
|
|
)
|
|
self.assertEqual(resp.status, 400)
|
|
|
|
async def test_download_rejects_backslash(self):
|
|
resp = await self.client.request(
|
|
"GET", "/prompt_manager/logs/download/..%5Cx.log"
|
|
)
|
|
self.assertEqual(resp.status, 400)
|
|
|
|
async def test_download_read_failure_is_500(self):
|
|
# Rewritten (#329): a directory is now a 404; inject a read error.
|
|
(self.log_dir / "prompt_manager.log").write_bytes(b"x")
|
|
secret = os.path.join(self.tmpdir, "logs", "prompt_manager.log")
|
|
|
|
def fail(path):
|
|
raise PermissionError(13, "Permission denied", secret)
|
|
|
|
with patch.object(logging_routes, "_read_log_bytes", fail):
|
|
resp = await self.client.request(
|
|
"GET", "/prompt_manager/logs/download/prompt_manager.log"
|
|
)
|
|
body = await resp.text()
|
|
self.assertEqual(resp.status, 500)
|
|
self.assertNotIn(self.tmpdir, body)
|
|
|
|
async def test_directory_download_is_404(self):
|
|
(self.log_dir / "dir.log").mkdir()
|
|
resp = await self.client.request("GET", "/prompt_manager/logs/download/dir.log")
|
|
self.assertEqual(resp.status, 404)
|
|
|
|
async def test_download_refuses_a_symlink(self):
|
|
secret = Path(self.tmpdir) / "secret.txt"
|
|
secret.write_text("the key")
|
|
try:
|
|
os.symlink(secret, self.log_dir / "prompt_manager.log.3")
|
|
except (OSError, NotImplementedError): # pragma: no cover
|
|
self.skipTest("this account cannot create symlinks")
|
|
resp = await self.client.request(
|
|
"GET", "/prompt_manager/logs/download/prompt_manager.log.3"
|
|
)
|
|
body = await resp.text()
|
|
self.assertEqual(resp.status, 404)
|
|
self.assertNotIn("the key", body)
|
|
|
|
async def test_download_reads_off_the_loop(self):
|
|
(self.log_dir / "prompt_manager.log").write_bytes(b"line\n")
|
|
seen = []
|
|
real = logging_routes._read_log_bytes
|
|
|
|
def read(path):
|
|
seen.append(threading.current_thread())
|
|
return real(path)
|
|
|
|
with patch.object(logging_routes, "_read_log_bytes", read):
|
|
resp = await self.client.request(
|
|
"GET", "/prompt_manager/logs/download/prompt_manager.log"
|
|
)
|
|
self.assertEqual(resp.status, 200)
|
|
self.assertEqual(len(seen), 1)
|
|
self.assertIsNot(seen[0], threading.current_thread())
|
|
|
|
async def test_download_keeps_its_response_shape(self):
|
|
(self.log_dir / "prompt_manager.log.2").write_bytes(b"a\nb\n")
|
|
resp = await self.client.request(
|
|
"GET", "/prompt_manager/logs/download/prompt_manager.log.2"
|
|
)
|
|
self.assertEqual(resp.status, 200)
|
|
self.assertEqual(resp.content_type, "text/plain")
|
|
self.assertEqual(
|
|
resp.headers["Content-Disposition"],
|
|
'attachment; filename="prompt_manager.log.2"',
|
|
)
|
|
self.assertEqual(resp.headers["Content-Length"], "4")
|
|
self.assertEqual(await resp.read(), b"a\nb\n")
|
|
|
|
async def test_log_files_rows_carry_their_kind(self):
|
|
for name in (
|
|
"prompt_manager.log",
|
|
"prompt_manager.log.1",
|
|
"prompt_manager.log.bak",
|
|
):
|
|
(self.log_dir / name).write_text("x")
|
|
resp = await self.client.request("GET", "/prompt_manager/logs/files")
|
|
data = await resp.json()
|
|
kinds = {row["filename"]: row["kind"] for row in data["files"]}
|
|
self.assertEqual(
|
|
kinds,
|
|
{
|
|
"prompt_manager.log": "active",
|
|
"prompt_manager.log.1": "rotated",
|
|
"prompt_manager.log.bak": "other",
|
|
},
|
|
)
|
|
|
|
|
|
class TestTruncateConfigAndStats(LoggingAPITestCase):
|
|
|
|
async def test_truncate_success(self):
|
|
(self.log_dir / "prompt_manager.log").write_text("data")
|
|
resp = await self.client.request("POST", "/prompt_manager/logs/truncate")
|
|
data = await resp.json()
|
|
self.assertEqual(resp.status, 200)
|
|
self.assertEqual(data["results"]["truncated"], ["prompt_manager.log"])
|
|
self.assertEqual((self.log_dir / "prompt_manager.log").read_text(), "")
|
|
|
|
async def test_truncate_keeps_other_files(self):
|
|
for name in (
|
|
"prompt_manager.log",
|
|
"prompt_manager.log.1",
|
|
"prompt_manager.log.bak",
|
|
):
|
|
(self.log_dir / name).write_text("data")
|
|
resp = await self.client.request("POST", "/prompt_manager/logs/truncate")
|
|
data = await resp.json()
|
|
self.assertEqual(resp.status, 200)
|
|
self.assertEqual(set(data), {"success", "message", "results"})
|
|
self.assertEqual(
|
|
sorted(data["results"]["truncated"]),
|
|
["prompt_manager.log", "prompt_manager.log.1"],
|
|
)
|
|
self.assertEqual(
|
|
sorted(p.name for p in self.log_dir.iterdir()),
|
|
["prompt_manager.log", "prompt_manager.log.bak"],
|
|
)
|
|
|
|
async def test_truncate_runs_off_the_loop(self):
|
|
resp = await self.client.request("POST", "/prompt_manager/logs/truncate")
|
|
self.assertEqual(resp.status, 200)
|
|
self.assertIsNot(
|
|
self.manager.threads["truncate_logs"], threading.current_thread()
|
|
)
|
|
|
|
async def test_update_config_runs_off_the_loop(self):
|
|
resp = await self._post_config({"level": "DEBUG"})
|
|
self.assertEqual(resp.status, 200)
|
|
self.assertIsNot(
|
|
self.manager.threads["update_config"], threading.current_thread()
|
|
)
|
|
|
|
async def test_truncate_failure_is_500(self):
|
|
def boom():
|
|
raise RuntimeError("locked")
|
|
|
|
self.manager.truncate_logs = boom
|
|
resp = await self.client.request("POST", "/prompt_manager/logs/truncate")
|
|
self.assertEqual(resp.status, 500)
|
|
|
|
async def test_get_config(self):
|
|
resp = await self.client.request("GET", "/prompt_manager/logs/config")
|
|
data = await resp.json()
|
|
self.assertEqual(resp.status, 200)
|
|
self.assertEqual(data["config"]["level"], "INFO")
|
|
|
|
async def test_get_config_failure_is_500(self):
|
|
def boom():
|
|
raise RuntimeError("gone")
|
|
|
|
self.manager.get_config = boom
|
|
resp = await self.client.request("GET", "/prompt_manager/logs/config")
|
|
self.assertEqual(resp.status, 500)
|
|
|
|
async def test_update_config_manager_failure_is_500(self):
|
|
def boom(config):
|
|
raise RuntimeError("handler error")
|
|
|
|
self.manager.update_config = boom
|
|
resp = await self.client.request(
|
|
"POST", "/prompt_manager/logs/config", json={"level": "INFO"}
|
|
)
|
|
self.assertEqual(resp.status, 500)
|
|
|
|
async def test_stats_success(self):
|
|
resp = await self.client.request("GET", "/prompt_manager/logs/stats")
|
|
data = await resp.json()
|
|
self.assertEqual(resp.status, 200)
|
|
self.assertEqual(data["stats"]["current_level"], "INFO")
|
|
|
|
|
|
class TestLoggerManagerResolution(LoggingAPITestCase):
|
|
|
|
def test_default_manager_is_the_singleton(self):
|
|
from py.api.logging_routes import LoggingRoutesMixin
|
|
from utils.logging_config import get_logger_manager
|
|
|
|
manager = LoggingRoutesMixin._get_logger_manager(self.api)
|
|
self.assertIs(manager, get_logger_manager())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|