228 lines
8.8 KiB
Python
228 lines
8.8 KiB
Python
from django.test import TestCase
|
|
from django.urls import reverse
|
|
from unittest.mock import patch
|
|
|
|
from core.models import (
|
|
AIRequest,
|
|
ChatSession,
|
|
ConversationEvent,
|
|
MemoryItem,
|
|
Person,
|
|
PersonIdentifier,
|
|
User,
|
|
WorkspaceConversation,
|
|
)
|
|
|
|
|
|
class SystemDiagnosticsAPITests(TestCase):
|
|
def setUp(self):
|
|
self.user = User.objects.create_superuser(
|
|
username="sys-diag-admin",
|
|
email="sys-diag@example.com",
|
|
password="pw",
|
|
)
|
|
person = Person.objects.create(user=self.user, name="System Diagnostics Person")
|
|
identifier = PersonIdentifier.objects.create(
|
|
user=self.user,
|
|
person=person,
|
|
service="signal",
|
|
identifier="+15554443333",
|
|
)
|
|
self.session = ChatSession.objects.create(user=self.user, identifier=identifier)
|
|
self.workspace_conversation = WorkspaceConversation.objects.create(
|
|
user=self.user,
|
|
platform_type="signal",
|
|
title="Diag Memory Scope",
|
|
platform_thread_id=str(self.session.id),
|
|
)
|
|
self.client.force_login(self.user)
|
|
|
|
def test_event_ledger_smoke_api_returns_counts_and_missing_required(self):
|
|
ConversationEvent.objects.create(
|
|
user=self.user,
|
|
session=self.session,
|
|
ts=1700000000000,
|
|
event_type="message_created",
|
|
direction="in",
|
|
origin_transport="signal",
|
|
payload={"message_id": "m1"},
|
|
raw_payload={},
|
|
)
|
|
response = self.client.get(
|
|
reverse("system_event_ledger_smoke"),
|
|
{
|
|
"minutes": "999999",
|
|
"service": "signal",
|
|
"require_types": "message_created,reaction_added",
|
|
},
|
|
)
|
|
self.assertEqual(200, response.status_code)
|
|
payload = response.json()
|
|
self.assertTrue(payload.get("ok"))
|
|
self.assertEqual("signal", payload.get("service"))
|
|
self.assertIn("event_type_counts", payload)
|
|
self.assertIn("missing_required_types", payload)
|
|
self.assertIn("reaction_added", payload.get("missing_required_types") or [])
|
|
|
|
@patch("core.views.system.get_recent_event_rows")
|
|
def test_event_ledger_smoke_api_can_use_manticore_source(self, mocked_rows):
|
|
mocked_rows.return_value = [
|
|
{
|
|
"id": "",
|
|
"user_id": int(self.user.id),
|
|
"session_id": str(self.session.id),
|
|
"ts": 1700000000000,
|
|
"event_type": "message_created",
|
|
"kind": "message_sent",
|
|
"direction": "in",
|
|
"origin_transport": "signal",
|
|
"trace_id": "",
|
|
}
|
|
]
|
|
response = self.client.get(
|
|
reverse("system_event_ledger_smoke"),
|
|
{"minutes": "999999", "service": "signal"},
|
|
)
|
|
self.assertEqual(200, response.status_code)
|
|
payload = response.json()
|
|
self.assertTrue(payload.get("ok"))
|
|
self.assertEqual("manticore", payload.get("data_source"))
|
|
self.assertEqual(1, int(payload.get("count") or 0))
|
|
|
|
def test_trace_diagnostics_includes_projection_shadow_links(self):
|
|
trace_id = "trace-system-diag-1"
|
|
event = ConversationEvent.objects.create(
|
|
user=self.user,
|
|
session=self.session,
|
|
ts=1700000001000,
|
|
event_type="message_created",
|
|
direction="in",
|
|
origin_transport="signal",
|
|
trace_id=trace_id,
|
|
payload={"message_id": "m2"},
|
|
raw_payload={},
|
|
)
|
|
response = self.client.get(
|
|
reverse("system_trace_diagnostics"),
|
|
{"trace_id": trace_id},
|
|
)
|
|
self.assertEqual(200, response.status_code)
|
|
payload = response.json()
|
|
self.assertTrue(payload.get("ok"))
|
|
self.assertEqual(1, payload.get("count"))
|
|
self.assertIn(str(self.session.id), payload.get("related_session_ids") or [])
|
|
urls = payload.get("projection_shadow_urls") or []
|
|
self.assertTrue(urls)
|
|
self.assertIn(str(self.session.id), str(urls[0]))
|
|
events = payload.get("events") or []
|
|
self.assertEqual(str(event.id), str(events[0].get("id")))
|
|
self.assertIn(
|
|
str(self.session.id),
|
|
str(events[0].get("projection_shadow_url") or ""),
|
|
)
|
|
|
|
@patch("core.views.system.get_trace_event_rows")
|
|
def test_trace_diagnostics_can_use_manticore_source(self, mocked_rows):
|
|
mocked_rows.return_value = [
|
|
{
|
|
"id": "",
|
|
"ts": 1700000001000,
|
|
"event_type": "message_created",
|
|
"direction": "in",
|
|
"session_id": str(self.session.id),
|
|
"origin_transport": "signal",
|
|
"origin_message_id": "m2",
|
|
"payload": {"trace_id": "trace-system-diag-m"},
|
|
}
|
|
]
|
|
response = self.client.get(
|
|
reverse("system_trace_diagnostics"),
|
|
{"trace_id": "trace-system-diag-m"},
|
|
)
|
|
self.assertEqual(200, response.status_code)
|
|
payload = response.json()
|
|
self.assertTrue(payload.get("ok"))
|
|
self.assertEqual("manticore", payload.get("data_source"))
|
|
self.assertEqual(1, int(payload.get("count") or 0))
|
|
self.assertIn(str(self.session.id), payload.get("related_session_ids") or [])
|
|
|
|
@patch("core.views.system.get_trace_ids")
|
|
@patch("core.views.system.count_behavioral_events")
|
|
def test_system_settings_page_includes_manticore_trace_ids(
|
|
self, mocked_behavioral_count, mocked_trace_ids
|
|
):
|
|
mocked_behavioral_count.return_value = 7
|
|
mocked_trace_ids.return_value = ["trace-from-manticore"]
|
|
response = self.client.get(reverse("system_settings"))
|
|
self.assertEqual(200, response.status_code)
|
|
content = response.content.decode("utf-8")
|
|
self.assertIn("trace-from-manticore", content)
|
|
self.assertIn("Behavioral Events: 7", content)
|
|
|
|
def test_memory_search_status_and_query_api(self):
|
|
request = AIRequest.objects.create(
|
|
user=self.user,
|
|
conversation=self.workspace_conversation,
|
|
window_spec={},
|
|
operation="memory_propose",
|
|
)
|
|
memory = MemoryItem.objects.create(
|
|
user=self.user,
|
|
conversation=self.workspace_conversation,
|
|
memory_kind="fact",
|
|
status="active",
|
|
content={"text": "User prefers concise status updates on WhatsApp."},
|
|
source_request=request,
|
|
)
|
|
status_response = self.client.get(reverse("system_memory_search_status"))
|
|
self.assertEqual(200, status_response.status_code)
|
|
status_payload = status_response.json()
|
|
self.assertTrue(status_payload.get("ok"))
|
|
self.assertIn("status", status_payload)
|
|
|
|
query_response = self.client.get(
|
|
reverse("system_memory_search_query"),
|
|
{"q": "concise status updates"},
|
|
)
|
|
self.assertEqual(200, query_response.status_code)
|
|
query_payload = query_response.json()
|
|
self.assertTrue(query_payload.get("ok"))
|
|
self.assertGreaterEqual(int(query_payload.get("count") or 0), 1)
|
|
first_hit = (query_payload.get("hits") or [{}])[0]
|
|
self.assertEqual(str(memory.id), str(first_hit.get("memory_id") or ""))
|
|
|
|
@patch("core.views.system.get_recent_event_rows")
|
|
@patch("core.views.system.count_behavioral_events")
|
|
def test_system_settings_page_renders_searchable_datalists(
|
|
self, mocked_behavioral_count, mocked_recent_rows
|
|
):
|
|
mocked_behavioral_count.return_value = 3
|
|
mocked_recent_rows.return_value = [
|
|
{
|
|
"event_type": "presence_available",
|
|
"origin_transport": "whatsapp",
|
|
}
|
|
]
|
|
ConversationEvent.objects.create(
|
|
user=self.user,
|
|
session=self.session,
|
|
ts=1700000002000,
|
|
event_type="reaction_added",
|
|
direction="system",
|
|
origin_transport="signal",
|
|
trace_id="trace-system-diag-2",
|
|
payload={"message_id": "m3"},
|
|
raw_payload={},
|
|
)
|
|
response = self.client.get(reverse("system_settings"))
|
|
self.assertEqual(200, response.status_code)
|
|
content = response.content.decode("utf-8")
|
|
self.assertIn('datalist id="diagnostics-session-options"', content)
|
|
self.assertIn('datalist id="diagnostics-trace-options"', content)
|
|
self.assertIn('datalist id="diagnostics-service-options"', content)
|
|
self.assertIn('datalist id="diagnostics-event-type-options"', content)
|
|
self.assertIn(str(self.session.id), content)
|
|
self.assertIn("trace-system-diag-2", content)
|
|
self.assertIn("whatsapp", content)
|
|
self.assertIn("presence_available", content)
|