| from __future__ import annotations |
|
|
| import json |
| import os |
| import tempfile |
| import unittest |
| from pathlib import Path |
| from unittest import mock |
|
|
| from fastapi import HTTPException |
|
|
| from services.log_service import LogService, LoggedCall |
|
|
|
|
| class LogServiceTests(unittest.TestCase): |
| def setUp(self): |
| self.tmp = tempfile.TemporaryDirectory() |
| self.addCleanup(self.tmp.cleanup) |
| self.path = Path(self.tmp.name) / "logs.jsonl" |
|
|
| def write_logs(self, count: int = 5) -> None: |
| rows = [] |
| for index in range(count): |
| rows.append(json.dumps({ |
| "id": f"log-{index}", |
| "time": f"2026-08-0{index + 1} 12:00:00", |
| "type": "call" if index % 2 == 0 else "account", |
| "summary": f"item {index}", |
| }, ensure_ascii=False)) |
| self.path.write_text("\n".join(rows) + "\n", encoding="utf-8") |
|
|
| def test_list_reads_newest_lines_without_reading_whole_file(self): |
| self.write_logs(5) |
| service = LogService(self.path) |
|
|
| with mock.patch.object(Path, "read_text", side_effect=AssertionError("read_text should not be used by list")): |
| items = service.list(limit=2) |
|
|
| self.assertEqual([item["id"] for item in items], ["log-4", "log-3"]) |
|
|
| def test_list_applies_filters_while_reading_backwards(self): |
| self.write_logs(5) |
| items = LogService(self.path).list(type="call", start_date="2026-08-03", limit=10) |
|
|
| self.assertEqual([item["id"] for item in items], ["log-4", "log-2"]) |
|
|
| def test_list_legacy_ids_can_be_deleted(self): |
| self.path.write_text( |
| "\n".join([ |
| json.dumps({"time": "2026-08-01 12:00:00", "type": "call", "summary": "old"}), |
| json.dumps({"time": "2026-08-02 12:00:00", "type": "call", "summary": "new"}), |
| ]) + "\n", |
| encoding="utf-8", |
| ) |
| service = LogService(self.path) |
|
|
| with mock.patch.object(Path, "read_text", side_effect=AssertionError("read_text should not be used by list/delete")): |
| newest = service.list(limit=1)[0] |
| result = service.delete([newest["id"]]) |
| remaining = service.list(limit=10) |
|
|
| self.assertEqual(result["removed"], 1) |
| self.assertEqual([item["summary"] for item in remaining], ["old"]) |
|
|
| def test_add_rotates_oversized_current_log(self): |
| service = LogService(self.path) |
|
|
| with mock.patch.dict(os.environ, {"CHATGPT2API_LOG_MAX_BYTES": "1"}): |
| service.add("call", "oversized", {"payload": "x"}) |
|
|
| archives = list(self.path.parent.glob("logs-*.jsonl")) |
| self.assertEqual(len(archives), 1) |
| self.assertEqual(self.path.read_text(encoding="utf-8"), "") |
| self.assertIn("oversized", archives[0].read_text(encoding="utf-8")) |
|
|
| def test_add_rotates_existing_oversized_log_before_writing_new_item(self): |
| self.path.write_text("x" * 600, encoding="utf-8") |
| service = LogService(self.path) |
|
|
| with mock.patch.dict(os.environ, {"CHATGPT2API_LOG_MAX_BYTES": "512"}): |
| service.add("call", "fresh", {"payload": "x"}) |
|
|
| archives = list(self.path.parent.glob("logs-*.jsonl")) |
| self.assertEqual(len(archives), 1) |
| self.assertEqual(archives[0].read_text(encoding="utf-8"), "x" * 600) |
| self.assertIn("fresh", self.path.read_text(encoding="utf-8")) |
|
|
| def test_delete_refuses_to_rewrite_oversized_log(self): |
| self.write_logs(5) |
| service = LogService(self.path) |
| newest = service.list(limit=1)[0] |
|
|
| with mock.patch.dict(os.environ, {"CHATGPT2API_LOG_DELETE_REWRITE_MAX_BYTES": "1"}): |
| with self.assertRaises(HTTPException) as raised: |
| service.delete([newest["id"]]) |
|
|
| self.assertEqual(raised.exception.status_code, 413) |
|
|
| def test_stream_hides_account_email_but_admin_log_keeps_it(self): |
| email = "stream-pool@example.com" |
| captured_logs: list[dict[str, object]] = [] |
|
|
| def capture_log(_type, _summary="", detail=None, **_data): |
| captured_logs.append(dict(detail or {})) |
|
|
| call = LoggedCall( |
| identity={"id": "normal-1", "name": "Normal", "role": "normal"}, |
| endpoint="/v1/chat/completions", |
| model="gpt-image-2", |
| summary="stream", |
| ) |
| chunks = [ |
| { |
| "delta": "done", |
| "account_email": email, |
| "_account_email": email, |
| "_conversation_id": "conversation-1", |
| } |
| ] |
|
|
| with mock.patch("services.log_service.log_service.add", capture_log): |
| public_chunks = list(call.stream(chunks)) |
|
|
| self.assertEqual(public_chunks, [{"delta": "done", "_conversation_id": "conversation-1"}]) |
| self.assertTrue( |
| any(item.get("account_email") == email for item in captured_logs), |
| captured_logs, |
| ) |
|
|
|
|
| if __name__ == "__main__": |
| unittest.main() |
|
|
