Labyricorn/ThinkStorm
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
ThinkStorm/tests/test_signal_gateway.py
T
xAdmin 41e08611c9 Implement optional single-image intake, Signal ingestion, and multimodal vision analysis 
- Add image intake service with format validation (JPEG, PNG, WebP) and EXIF/GPS stripping
- Enforce strict single-image rule across Web and Signal attachment channels
- Implement token-optimized vision downscaling and JPEG compression
- Add IMAGE_CONTEXT pipeline stage with OmniRoute vision routing and resilient failover
- Seed and manage versioned idea-image-interpreter prompt in catalog
- Update Web UI with responsive image picker, preview chip, and Visual Context tab
- Add comprehensive automated test suite in test_image_intake.py
- Update README and Labyricorn devlog
2026-08-23 01:40:22 -07:00

451 lines
19 KiB
Python
Raw Permalink Blame History

"""
Comprehensive Tests for ThinkStorm Signal Gateway Integration
Covers:
1. Valid callback-test request returns 2xx and causes no message-processing side effects.
2. Valid Signal event with correct bearer secret is durably accepted.
3. Missing bearer secret returns 401.
4. Wrong bearer secret returns 401.
5. Nullable sender_number, sender_name, group_id, and reply_to are accepted.
6. Unsupported schema_version is rejected with 400.
7. Missing or mismatched Idempotency-Key is rejected for actual events.
8. Delivering the same event_id twice returns success but produces exactly one downstream action.
9. Direct message maps to expected ThinkStorm conversation/metadata.
10. Group message maps to expected ThinkStorm conversation/metadata.
11. Callback-test payload is not parsed as a real Signal message.
12. Outbound client sends correct URL, authorization header, and JSON payload.
13. Outbound 202, 400, 401, 413, 503, and timeout behaviors are handled intentionally.
14. Secrets are absent from error messages and captured logs.
15. Service health checks.
"""
import pytest
import json
import uuid
import urllib.request
import urllib.error
from unittest.mock import patch, MagicMock
from io import BytesIO
from starlette.testclient import TestClient
from thinkstorm.main import app
from thinkstorm.database import init_db, get_db
from thinkstorm.config import config
from thinkstorm.services.signal_gateway import (
SignalGatewayAdapter,
SignalGatewayAuthError,
SignalGatewayClientError,
SignalGatewayPayloadTooLargeError,
SignalGatewayUnavailableError,
SignalGatewayTimeoutError,
mask_secret
)
TEST_SECRET = "sgw_callback_secret_test_2026_unit_testing"
TEST_API_KEY = "sgw_apikey_test_2026_unit_testing"
@pytest.fixture(autouse=True)
def setup_test_environment(monkeypatch):
init_db()
monkeypatch.setattr(config.services, "signal_gateway_callback_secret", TEST_SECRET)
monkeypatch.setattr(config.services, "signal_gateway_api_key", TEST_API_KEY)
monkeypatch.setattr(config.services, "signal_gateway_base_url", "http://10.138.4.46:8000")
@pytest.fixture
def client():
return TestClient(app)
# -----------------------------------------------------------------------------
# 1. Callback Test Request Tests
# -----------------------------------------------------------------------------
def test_callback_test_request_success(client):
"""Test callback probe returns 2xx and creates no ideas or queue jobs."""
headers = {
"Authorization": f"Bearer {TEST_SECRET}",
"Content-Type": "application/json"
}
payload = {
"schema_version": 1,
"event_type": "signal_gateway.callback_test"
}
with get_db() as conn:
ideas_before = conn.execute("SELECT COUNT(*) FROM ideas").fetchone()[0]
events_before = conn.execute("SELECT COUNT(*) FROM signal_inbound_events").fetchone()[0]
resp = client.post("/api/integrations/signal/events", headers=headers, json=payload)
assert resp.status_code == 200
data = resp.json()
assert data["accepted"] is True
assert data["event_type"] == "signal_gateway.callback_test"
with get_db() as conn:
ideas_after = conn.execute("SELECT COUNT(*) FROM ideas").fetchone()[0]
events_after = conn.execute("SELECT COUNT(*) FROM signal_inbound_events").fetchone()[0]
assert ideas_after == ideas_before
assert events_after == events_before
# -----------------------------------------------------------------------------
# 2. Callback Authentication Tests
# -----------------------------------------------------------------------------
def test_callback_missing_bearer_secret(client):
"""Missing bearer secret returns 401."""
payload = {
"schema_version": 1,
"event_type": "signal_gateway.callback_test"
}
resp = client.post("/api/integrations/signal/events", json=payload)
assert resp.status_code == 401
assert "detail" in resp.json()
def test_callback_wrong_bearer_secret(client):
"""Wrong bearer secret returns 401."""
headers = {"Authorization": "Bearer wrong_secret_token_value"}
payload = {
"schema_version": 1,
"event_type": "signal_gateway.callback_test"
}
resp = client.post("/api/integrations/signal/events", headers=headers, json=payload)
assert resp.status_code == 401
def test_callback_malformed_auth_header(client):
"""Malformed auth header (e.g. Basic or token only) returns 401."""
headers = {"Authorization": f"Basic {TEST_SECRET}"}
payload = {"schema_version": 1, "event_type": "signal_gateway.callback_test"}
resp = client.post("/api/integrations/signal/events", headers=headers, json=payload)
assert resp.status_code == 401
# -----------------------------------------------------------------------------
# 3. Schema & Idempotency Key Validation
# -----------------------------------------------------------------------------
def test_callback_unsupported_schema_version(client):
"""Unsupported schema_version is rejected with 400."""
headers = {
"Authorization": f"Bearer {TEST_SECRET}",
"Idempotency-Key": "sig_test_ver"
}
payload = {
"schema_version": 2,
"event_id": "sig_test_ver",
"channel": "signal",
"sender_uuid": "12345678-1234-1234-1234-123456789abc",
"text": "Hello"
}
resp = client.post("/api/integrations/signal/events", headers=headers, json=payload)
assert resp.status_code == 400
assert "schema_version" in resp.json()["detail"]
def test_callback_missing_or_mismatched_idempotency_key(client):
"""Missing or mismatched Idempotency-Key header is rejected for actual events."""
event_id = "sig_idemp_001"
payload = {
"schema_version": 1,
"event_id": event_id,
"channel": "signal",
"sender_uuid": "12345678-1234-1234-1234-123456789abc",
"text": "An idea about autonomous systems."
}
# Missing header
resp_missing = client.post(
"/api/integrations/signal/events",
headers={"Authorization": f"Bearer {TEST_SECRET}"},
json=payload
)
assert resp_missing.status_code == 400
assert "Idempotency-Key" in resp_missing.json()["detail"]
# Mismatched header
resp_mismatch = client.post(
"/api/integrations/signal/events",
headers={
"Authorization": f"Bearer {TEST_SECRET}",
"Idempotency-Key": "different_id_123"
},
json=payload
)
assert resp_mismatch.status_code == 400
assert "Idempotency-Key" in resp_mismatch.json()["detail"]
# -----------------------------------------------------------------------------
# 4. Actual Signal Event Ingestion & Durable Idempotency
# -----------------------------------------------------------------------------
def test_valid_signal_event_ingestion_and_idempotency(client):
"""Delivering a valid Signal event creates idea and inbound record; duplicate is harmless."""
event_id = f"sig_event_valid_{uuid.uuid4().hex[:12]}"
headers = {
"Authorization": f"Bearer {TEST_SECRET}",
"Idempotency-Key": event_id
}
payload = {
"schema_version": 1,
"event_id": event_id,
"channel": "signal",
"sender_uuid": "a1b2c3d4-e5f6-7a8b-9c0d-1e2f3a4b5c6d",
"sender_number": "+15551234567",
"sender_name": "Alice Developer",
"group_id": None,
"message_id": "1787300000001",
"reply_to": None,
"text": "Automated pipeline for tracking real-time API contract drifts.",
"received_at": "2026-08-21T20:15:00+00:00"
}
# 1. First delivery -> durably accepted and enqueued
resp1 = client.post("/api/integrations/signal/events", headers=headers, json=payload)
assert resp1.status_code == 200
data1 = resp1.json()
assert data1["accepted"] is True
assert data1["event_id"] == event_id
idea_id = data1["idea_id"]
assert idea_id.startswith("TS-")
assert data1.get("duplicate") is not True
# Verify DB state
with get_db() as conn:
ev_row = conn.execute("SELECT * FROM signal_inbound_events WHERE event_id = ?", (event_id,)).fetchone()
assert ev_row is not None
assert ev_row["sender_uuid"] == "a1b2c3d4-e5f6-7a8b-9c0d-1e2f3a4b5c6d"
assert ev_row["sender_name"] == "Alice Developer"
assert ev_row["sender_number"] == "+15551234567"
assert ev_row["group_id"] is None
assert ev_row["idea_id"] == idea_id
idea_row = conn.execute("SELECT * FROM ideas WHERE id = ?", (idea_id,)).fetchone()
assert idea_row is not None
assert idea_row["original_text"] == "Automated pipeline for tracking real-time API contract drifts."
assert idea_row["source_channel"] == "signal"
assert idea_row["source_sender_uuid"] == "a1b2c3d4-e5f6-7a8b-9c0d-1e2f3a4b5c6d"
assert idea_row["source_event_id"] == event_id
assert idea_row["lifecycle_state"] == "SUBMITTED"
assert idea_row["processing_state"] == "QUEUED"
# 2. Duplicate delivery -> returns 200 success without creating duplicate idea
with get_db() as conn:
ideas_before_dup = conn.execute("SELECT COUNT(*) FROM ideas").fetchone()[0]
resp2 = client.post("/api/integrations/signal/events", headers=headers, json=payload)
assert resp2.status_code == 200
data2 = resp2.json()
assert data2["accepted"] is True
assert data2["event_id"] == event_id
assert data2.get("duplicate") is True
assert data2["idea_id"] == idea_id
with get_db() as conn:
ideas_after_dup = conn.execute("SELECT COUNT(*) FROM ideas").fetchone()[0]
assert ideas_after_dup == ideas_before_dup
# -----------------------------------------------------------------------------
# 5. Direct vs Group Message Mapping & Nullable Fields
# -----------------------------------------------------------------------------
def test_direct_message_mapping_with_null_fields(client):
"""Direct message with null sender_number, sender_name, group_id, reply_to is accepted."""
event_id = f"sig_event_direct_nulls_{uuid.uuid4().hex[:12]}"
headers = {
"Authorization": f"Bearer {TEST_SECRET}",
"Idempotency-Key": event_id
}
payload = {
"schema_version": 1,
"event_id": event_id,
"channel": "signal",
"sender_uuid": "bbbbbbbb-1111-2222-3333-444444444444",
"sender_number": None,
"sender_name": None,
"group_id": None,
"message_id": "1787300000002",
"reply_to": None,
"text": "Direct message with completely null optional metadata.",
"received_at": "2026-08-21T20:20:00+00:00"
}
resp = client.post("/api/integrations/signal/events", headers=headers, json=payload)
assert resp.status_code == 200
idea_id = resp.json()["idea_id"]
with get_db() as conn:
idea = conn.execute("SELECT * FROM ideas WHERE id = ?", (idea_id,)).fetchone()
assert idea["source_channel"] == "signal"
assert idea["source_sender_uuid"] == "bbbbbbbb-1111-2222-3333-444444444444"
assert idea["source_group_id"] is None
def test_group_message_mapping(client):
"""Group message sets source_group_id and source_sender_uuid appropriately."""
event_id = f"sig_event_group_{uuid.uuid4().hex[:12]}"
group_id = "EdG5w+Xq1eAbcDeFgHiJkLmNoPqRsTuVwXyZ1234567="
headers = {
"Authorization": f"Bearer {TEST_SECRET}",
"Idempotency-Key": event_id
}
payload = {
"schema_version": 1,
"event_id": event_id,
"channel": "signal",
"sender_uuid": "cccccccc-2222-3333-4444-555555555555",
"sender_number": "+15559876543",
"sender_name": "Bob TeamLead",
"group_id": group_id,
"message_id": "1787300000003",
"reply_to": None,
"text": "Team brainstorming: Distributed event sourcing with sqlite-vss.",
"received_at": "2026-08-21T20:25:00+00:00"
}
resp = client.post("/api/integrations/signal/events", headers=headers, json=payload)
assert resp.status_code == 200
idea_id = resp.json()["idea_id"]
with get_db() as conn:
idea = conn.execute("SELECT * FROM ideas WHERE id = ?", (idea_id,)).fetchone()
assert idea["source_channel"] == "signal"
assert idea["source_sender_uuid"] == "cccccccc-2222-3333-4444-555555555555"
assert idea["source_group_id"] == group_id
ev = conn.execute("SELECT * FROM signal_inbound_events WHERE event_id = ?", (event_id,)).fetchone()
assert ev["group_id"] == group_id
# -----------------------------------------------------------------------------
# 6. Outbound Signal Gateway Client Tests
# -----------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_outbound_client_success_202():
"""Outbound client sends valid payload and handles 202 Accepted."""
adapter = SignalGatewayAdapter(endpoint="http://10.138.4.46:8000", api_key=TEST_API_KEY)
mock_resp = MagicMock()
mock_resp.status = 202
mock_resp.__enter__.return_value = mock_resp
mock_resp.read.return_value = json.dumps({
"accepted": True,
"application_id": "thinkstorm",
"message_id": "sgw_1787300000000_0",
"status": "queued"
}).encode("utf-8")
with patch("urllib.request.urlopen", return_value=mock_resp) as mock_urlopen:
res = await adapter.send_message(
recipient="12345678-1234-1234-1234-123456789abc",
text="ThinkStorm response idea processed"
)
assert res["accepted"] is True
assert res["status"] == "queued"
assert res["message_id"] == "sgw_1787300000000_0"
# Verify request details
req = mock_urlopen.call_args[0][0]
assert req.full_url == "http://10.138.4.46:8000/api/v1/messages"
assert req.headers["Authorization"] == f"Bearer {TEST_API_KEY}"
assert req.headers["Content-type"] == "application/json"
body = json.loads(req.data.decode("utf-8"))
assert body["recipient"] == "12345678-1234-1234-1234-123456789abc"
assert body["text"] == "ThinkStorm response idea processed"
@pytest.mark.asyncio
async def test_outbound_client_invalid_recipient_or_text():
"""Outbound client validates recipient and text length constraints before sending."""
adapter = SignalGatewayAdapter(endpoint="http://10.138.4.46:8000", api_key=TEST_API_KEY)
# Empty recipient
with pytest.raises(SignalGatewayClientError):
await adapter.send_message(recipient="", text="Hello")
# Recipient > 256 bytes
with pytest.raises(SignalGatewayClientError):
await adapter.send_message(recipient="a" * 257, text="Hello")
# Empty text
with pytest.raises(SignalGatewayClientError):
await adapter.send_message(recipient="uuid-123", text="")
# Text > 16000 bytes
with pytest.raises(SignalGatewayPayloadTooLargeError):
await adapter.send_message(recipient="uuid-123", text="x" * 16001)
@pytest.mark.asyncio
async def test_outbound_client_error_handling():
"""Outbound client maps HTTP status codes 400, 401, 413, 503 and timeouts."""
adapter = SignalGatewayAdapter(endpoint="http://10.138.4.46:8000", api_key=TEST_API_KEY)
# 400 Bad Request
err_400 = urllib.error.HTTPError(
url="http://10.138.4.46:8000/api/v1/messages",
code=400,
msg="Bad Request",
hdrs={},
fp=BytesIO(json.dumps({"accepted": False, "error": {"code": "bad_request", "message": "Invalid recipient format"}}).encode("utf-8"))
)
with patch("urllib.request.urlopen", side_effect=err_400):
with pytest.raises(SignalGatewayClientError) as exc:
await adapter.send_message(recipient="bad-recipient", text="Test")
assert "400" in str(exc.value)
# 401 Unauthorized
err_401 = urllib.error.HTTPError(
url="http://10.138.4.46:8000/api/v1/messages",
code=401,
msg="Unauthorized",
hdrs={},
fp=BytesIO(json.dumps({"accepted": False, "error": {"code": "unauthorized", "message": "Invalid key"}}).encode("utf-8"))
)
with patch("urllib.request.urlopen", side_effect=err_401):
with pytest.raises(SignalGatewayAuthError):
await adapter.send_message(recipient="uuid-123", text="Test")
# 413 Payload Too Large
err_413 = urllib.error.HTTPError(
url="http://10.138.4.46:8000/api/v1/messages",
code=413,
msg="Payload Too Large",
hdrs={},
fp=BytesIO(json.dumps({"accepted": False, "error": {"code": "payload_too_large", "message": "Too large"}}).encode("utf-8"))
)
with patch("urllib.request.urlopen", side_effect=err_413):
with pytest.raises(SignalGatewayPayloadTooLargeError):
await adapter.send_message(recipient="uuid-123", text="Test")
# 503 Service Unavailable (retries and fails)
err_503 = urllib.error.HTTPError(
url="http://10.138.4.46:8000/api/v1/messages",
code=503,
msg="Service Unavailable",
hdrs={},
fp=BytesIO(json.dumps({"accepted": False, "error": {"code": "unavailable", "message": "Queue full"}}).encode("utf-8"))
)
with patch("urllib.request.urlopen", side_effect=err_503):
with pytest.raises(SignalGatewayUnavailableError):
await adapter.send_message(recipient="uuid-123", text="Test", max_retries=1)
# Timeout Error
timeout_err = urllib.error.URLError(reason="Connection timed out")
with patch("urllib.request.urlopen", side_effect=timeout_err):
with pytest.raises(SignalGatewayTimeoutError):
await adapter.send_message(recipient="uuid-123", text="Test")
# -----------------------------------------------------------------------------
# 7. Secret Redaction / Safety Tests
# -----------------------------------------------------------------------------
def test_secrets_redacted_in_errors_and_masking():
"""Secrets are properly masked and not leaked."""
masked = mask_secret("sgw_callback_secret_123456789")
assert "sgw_" in masked
assert "6789" in masked
assert "callback_secret" not in masked
assert mask_secret("") == ""
assert mask_secret("short") == "***"
# -----------------------------------------------------------------------------
# 8. Admin Service Health Test
# -----------------------------------------------------------------------------
def test_admin_signal_gateway_health_test(client):
"""Admin can trigger health probe for signal_gateway."""
# Login as admin
login_resp = client.post("/api/auth/login", json={"username": "admin", "password": config.admin_bootstrap_key})
token = login_resp.json()["token"]
resp = client.post("/api/admin/services/signal_gateway/test", headers={"Authorization": f"Bearer {token}"})
assert resp.status_code == 200
data = resp.json()
assert data["service_id"] == "signal_gateway"
assert "healthy" in data
Reference in New Issue View Git Blame Copy Permalink