pyr0ball
6039ab2464
- app/services/ticket_export.py: plugin-dispatch architecture; Notion
exporter (Notion API v1, blocks-based, 50 entry cap, 2000-char
truncation per block); Jira exporter (REST API v3, Basic Auth, ADF
description, configurable issue type defaulting to Bug)
- app/rest.py: POST /api/incidents/{id}/export endpoint; Notion/Jira
credential fields added to SettingsBody and PATCH /api/settings handler
- web/src/views/IncidentsView.vue: "Export ticket ▾" dropdown in
incident detail drawer — click-outside close, inline URL link on success
- web/src/views/SettingsView.vue: Ticket Trackers section with Notion
token + database ID, Jira URL/email/token/project/issue-type; show/hide
for secret fields
- tests/test_ticket_export.py: 17 tests covering dispatch, Notion
success/error/config/payload/truncation paths, Jira success/error/
auth/project/summary/default-issue-type
224 lines
10 KiB
Python
224 lines
10 KiB
Python
"""Tests for ticket_export service — Notion and Jira exporters."""
|
|
from __future__ import annotations
|
|
|
|
from unittest.mock import MagicMock, patch
|
|
import pytest
|
|
|
|
|
|
INCIDENT = {
|
|
"id": "inc-1",
|
|
"label": "Redis OOM — key eviction flood",
|
|
"issue_type": "memory",
|
|
"started_at": "2026-06-01T03:00:00Z",
|
|
"ended_at": "2026-06-01T03:45:00Z",
|
|
"notes": "Triggered by batch job at 03:00",
|
|
"severity": "high",
|
|
}
|
|
|
|
ENTRIES = [
|
|
{"entry_id": "e1", "source_id": "host:redis", "severity": "ERROR", "text": "maxmemory reached, evicting keys"},
|
|
{"entry_id": "e2", "source_id": "host:app", "severity": "WARN", "text": "Redis NOEVICTION response"},
|
|
]
|
|
|
|
|
|
def _mock_response(status_code: int, body: dict):
|
|
resp = MagicMock()
|
|
resp.is_success = (status_code < 400)
|
|
resp.status_code = status_code
|
|
resp.json.return_value = body
|
|
resp.text = str(body)
|
|
return resp
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# available_targets
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_available_targets_lists_known_integrations():
|
|
from app.services.ticket_export import available_targets
|
|
targets = available_targets()
|
|
assert "notion" in targets
|
|
assert "jira" in targets
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# export_incident dispatch
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_export_incident_raises_for_unknown_target():
|
|
from app.services.ticket_export import export_incident
|
|
with pytest.raises(ValueError, match="Unknown ticket target"):
|
|
export_incident("linear", INCIDENT, ENTRIES, {})
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Notion exporter
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestNotionExport:
|
|
def test_successful_export_returns_url_and_id(self):
|
|
from app.services.ticket_export import export_incident
|
|
page_id = "a1b2c3d4-e5f6-7890-abcd-ef1234567890"
|
|
mock_resp = _mock_response(200, {"id": page_id, "url": f"https://notion.so/{page_id}"})
|
|
with patch("app.services.ticket_export.httpx.post", return_value=mock_resp) as mock_post:
|
|
result = export_incident("notion", INCIDENT, ENTRIES, {
|
|
"notion_token": "secret_abc123",
|
|
"notion_database_id": "db-id-xyz",
|
|
})
|
|
assert result["ticket_id"] == page_id
|
|
assert "notion.so" in result["url"]
|
|
mock_post.assert_called_once()
|
|
|
|
def test_raises_value_error_when_not_configured(self):
|
|
from app.services.ticket_export import export_incident
|
|
with pytest.raises(ValueError, match="Notion not configured"):
|
|
export_incident("notion", INCIDENT, ENTRIES, {
|
|
"notion_token": "",
|
|
"notion_database_id": "db-id",
|
|
})
|
|
|
|
def test_raises_value_error_when_database_id_missing(self):
|
|
from app.services.ticket_export import export_incident
|
|
with pytest.raises(ValueError, match="Notion not configured"):
|
|
export_incident("notion", INCIDENT, ENTRIES, {
|
|
"notion_token": "secret_abc",
|
|
"notion_database_id": "",
|
|
})
|
|
|
|
def test_raises_runtime_error_on_api_error(self):
|
|
from app.services.ticket_export import export_incident
|
|
mock_resp = _mock_response(401, {"message": "Unauthorized"})
|
|
with patch("app.services.ticket_export.httpx.post", return_value=mock_resp):
|
|
with pytest.raises(RuntimeError, match="Notion API error 401"):
|
|
export_incident("notion", INCIDENT, ENTRIES, {
|
|
"notion_token": "bad-token",
|
|
"notion_database_id": "db-id",
|
|
})
|
|
|
|
def test_sends_correct_database_id(self):
|
|
from app.services.ticket_export import export_incident
|
|
db_id = "my-database-uuid"
|
|
mock_resp = _mock_response(200, {"id": "page-id", "url": "https://notion.so/page-id"})
|
|
with patch("app.services.ticket_export.httpx.post", return_value=mock_resp) as mock_post:
|
|
export_incident("notion", INCIDENT, ENTRIES, {
|
|
"notion_token": "secret_abc123",
|
|
"notion_database_id": db_id,
|
|
})
|
|
call_kwargs = mock_post.call_args
|
|
payload = call_kwargs.kwargs.get("json") or call_kwargs.args[1] if len(call_kwargs.args) > 1 else call_kwargs.kwargs["json"]
|
|
assert payload["parent"]["database_id"] == db_id
|
|
|
|
def test_incident_label_becomes_page_title(self):
|
|
from app.services.ticket_export import export_incident
|
|
mock_resp = _mock_response(200, {"id": "pid", "url": "https://notion.so/pid"})
|
|
with patch("app.services.ticket_export.httpx.post", return_value=mock_resp) as mock_post:
|
|
export_incident("notion", INCIDENT, ENTRIES, {
|
|
"notion_token": "tok",
|
|
"notion_database_id": "dbid",
|
|
})
|
|
payload = mock_post.call_args.kwargs["json"]
|
|
title_text = payload["properties"]["title"]["title"][0]["text"]["content"]
|
|
assert INCIDENT["label"] in title_text
|
|
|
|
def test_url_falls_back_to_constructed_url(self):
|
|
from app.services.ticket_export import export_incident
|
|
page_id = "abc123"
|
|
mock_resp = _mock_response(200, {"id": page_id}) # no 'url' in response
|
|
with patch("app.services.ticket_export.httpx.post", return_value=mock_resp):
|
|
result = export_incident("notion", INCIDENT, ENTRIES, {
|
|
"notion_token": "tok",
|
|
"notion_database_id": "dbid",
|
|
})
|
|
assert "notion.so" in result["url"] or page_id in result["url"]
|
|
|
|
def test_long_text_truncated_to_notion_limit(self):
|
|
from app.services.ticket_export import export_incident
|
|
mock_resp = _mock_response(200, {"id": "pid", "url": "https://notion.so/pid"})
|
|
long_entries = [{"entry_id": f"e{i}", "source_id": "host:svc", "severity": "ERROR",
|
|
"text": "x" * 300} for i in range(60)]
|
|
with patch("app.services.ticket_export.httpx.post", return_value=mock_resp) as mock_post:
|
|
export_incident("notion", INCIDENT, long_entries, {
|
|
"notion_token": "tok",
|
|
"notion_database_id": "dbid",
|
|
})
|
|
payload = mock_post.call_args.kwargs["json"]
|
|
for block in payload.get("children", []):
|
|
for rt in block.get("bulleted_list_item", {}).get("rich_text", []):
|
|
assert len(rt["text"]["content"]) <= 2000
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Jira exporter
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestJiraExport:
|
|
_config = {
|
|
"jira_url": "https://myorg.atlassian.net",
|
|
"jira_email": "ops@example.com",
|
|
"jira_api_token": "ATATT3xFfGF0abc123",
|
|
"jira_project_key": "OPS",
|
|
"jira_issue_type": "Bug",
|
|
}
|
|
|
|
def test_successful_export_returns_url_and_key(self):
|
|
from app.services.ticket_export import export_incident
|
|
mock_resp = _mock_response(201, {"id": "10042", "key": "OPS-42", "self": "https://myorg.atlassian.net/rest/api/3/issue/10042"})
|
|
with patch("app.services.ticket_export.httpx.post", return_value=mock_resp):
|
|
result = export_incident("jira", INCIDENT, ENTRIES, self._config)
|
|
assert result["ticket_id"] == "OPS-42"
|
|
assert "OPS-42" in result["url"]
|
|
assert "myorg.atlassian.net" in result["url"]
|
|
|
|
def test_raises_value_error_when_not_configured(self):
|
|
from app.services.ticket_export import export_incident
|
|
with pytest.raises(ValueError, match="Jira not configured"):
|
|
export_incident("jira", INCIDENT, ENTRIES, {
|
|
"jira_url": "",
|
|
"jira_email": "a@b.com",
|
|
"jira_api_token": "tok",
|
|
"jira_project_key": "OPS",
|
|
})
|
|
|
|
def test_raises_runtime_error_on_api_error(self):
|
|
from app.services.ticket_export import export_incident
|
|
mock_resp = _mock_response(403, {"errorMessages": ["Forbidden"]})
|
|
with patch("app.services.ticket_export.httpx.post", return_value=mock_resp):
|
|
with pytest.raises(RuntimeError, match="Jira API error 403"):
|
|
export_incident("jira", INCIDENT, ENTRIES, self._config)
|
|
|
|
def test_sends_basic_auth_header(self):
|
|
from app.services.ticket_export import export_incident
|
|
import base64
|
|
mock_resp = _mock_response(201, {"key": "OPS-1", "id": "1"})
|
|
with patch("app.services.ticket_export.httpx.post", return_value=mock_resp) as mock_post:
|
|
export_incident("jira", INCIDENT, ENTRIES, self._config)
|
|
call_kwargs = mock_post.call_args.kwargs
|
|
auth_header = call_kwargs["headers"]["Authorization"]
|
|
assert auth_header.startswith("Basic ")
|
|
decoded = base64.b64decode(auth_header[6:]).decode()
|
|
assert "ops@example.com" in decoded
|
|
|
|
def test_uses_correct_project_key(self):
|
|
from app.services.ticket_export import export_incident
|
|
mock_resp = _mock_response(201, {"key": "OPS-7", "id": "7"})
|
|
with patch("app.services.ticket_export.httpx.post", return_value=mock_resp) as mock_post:
|
|
export_incident("jira", INCIDENT, ENTRIES, self._config)
|
|
payload = mock_post.call_args.kwargs["json"]
|
|
assert payload["fields"]["project"]["key"] == "OPS"
|
|
|
|
def test_incident_label_becomes_summary(self):
|
|
from app.services.ticket_export import export_incident
|
|
mock_resp = _mock_response(201, {"key": "OPS-8", "id": "8"})
|
|
with patch("app.services.ticket_export.httpx.post", return_value=mock_resp) as mock_post:
|
|
export_incident("jira", INCIDENT, ENTRIES, self._config)
|
|
payload = mock_post.call_args.kwargs["json"]
|
|
assert payload["fields"]["summary"] == INCIDENT["label"]
|
|
|
|
def test_default_issue_type_is_bug(self):
|
|
from app.services.ticket_export import export_incident
|
|
config = {k: v for k, v in self._config.items() if k != "jira_issue_type"}
|
|
mock_resp = _mock_response(201, {"key": "OPS-9", "id": "9"})
|
|
with patch("app.services.ticket_export.httpx.post", return_value=mock_resp) as mock_post:
|
|
export_incident("jira", INCIDENT, ENTRIES, config)
|
|
payload = mock_post.call_args.kwargs["json"]
|
|
assert payload["fields"]["issuetype"]["name"] == "Bug"
|