mirror of
https://github.com/wahyd4/links.git
synced 2026-08-08 21:04:53 +10:00
203 lines
8.7 KiB
Python
203 lines
8.7 KiB
Python
"""
|
|
Tests for Action links (link_type=ACTION, e.g. send-to-Discord).
|
|
|
|
Covers: model persistence, form validation/assembly, and the
|
|
redirect_to_original execution path (with mocked webhook).
|
|
"""
|
|
import pytest
|
|
from django.test import Client
|
|
from unittest.mock import patch
|
|
|
|
from links.models import Link
|
|
from links.forms import LinkForm
|
|
from links.actions import render_template, execute_action
|
|
from links import views
|
|
|
|
|
|
@pytest.fixture
|
|
def action_link(db):
|
|
return Link.objects.create(
|
|
alias='testaction',
|
|
link_type=Link.LinkType.ACTION,
|
|
action_config={
|
|
'action_type': 'discord_send',
|
|
'message_template': '@小黑 {query}',
|
|
},
|
|
)
|
|
|
|
|
|
# ── render_template ──────────────────────────────────────────────────────────
|
|
|
|
class TestRenderTemplate:
|
|
def test_replaces_query_placeholder(self):
|
|
assert render_template('@小黑 {query}', '今天天气') == '@小黑 今天天气'
|
|
|
|
def test_empty_template_returns_query(self):
|
|
assert render_template('', 'hello') == 'hello'
|
|
|
|
def test_template_without_placeholder_keeps_query_appended(self):
|
|
# 模板里没有 {query} 时，至少保留 query 内容（用默认拼接）
|
|
assert render_template('@小黑', 'hi') == '@小黑'
|
|
|
|
|
|
# ── execute_action ────────────────────────────────────────────────────────────
|
|
|
|
class TestExecuteAction:
|
|
@patch('links.actions.send_discord_message')
|
|
def test_discord_send_success(self, mock_send):
|
|
result = execute_action(
|
|
{'action_type': 'discord_send', 'message_template': '@小黑 {query}'},
|
|
'帮我查天气',
|
|
)
|
|
mock_send.assert_called_once_with('@小黑 帮我查天气')
|
|
assert result['success'] is True
|
|
assert 'discord_deeplink' in result
|
|
|
|
@patch('links.actions.send_discord_message')
|
|
def test_unknown_action_type(self, mock_send):
|
|
result = execute_action({'action_type': 'teleport'}, 'x')
|
|
mock_send.assert_not_called()
|
|
assert result['success'] is False
|
|
|
|
@patch('links.actions.send_discord_message')
|
|
def test_empty_query_does_not_send(self, mock_send):
|
|
result = execute_action(
|
|
{'action_type': 'discord_send', 'message_template': '@小黑 {query}'},
|
|
' ',
|
|
)
|
|
mock_send.assert_not_called()
|
|
assert result['success'] is False
|
|
assert '缺少参数' in result['message']
|
|
|
|
@patch('links.actions.send_discord_message', side_effect=RuntimeError('boom'))
|
|
def test_webhook_failure_propagates(self, mock_send):
|
|
with pytest.raises(RuntimeError):
|
|
execute_action({'action_type': 'discord_send', 'message_template': '{query}'}, 'x')
|
|
|
|
|
|
# ── LinkForm validation ───────────────────────────────────────────────────────
|
|
|
|
class TestLinkFormActionValidation:
|
|
def _form(self, **overrides):
|
|
data = {
|
|
'alias': 'formaction',
|
|
'link_type': Link.LinkType.ACTION,
|
|
'action_type': 'discord_send',
|
|
'action_message_template': '@小黑 {query}',
|
|
'description': '',
|
|
}
|
|
data.update(overrides)
|
|
return LinkForm(data)
|
|
|
|
def test_valid_action_form_assembles_config(self, db):
|
|
form = self._form()
|
|
assert form.is_valid(), form.errors
|
|
assert form.cleaned_data['action_config'] == {
|
|
'action_type': 'discord_send',
|
|
'message_template': '@小黑 {query}',
|
|
}
|
|
|
|
def test_action_requires_type(self, db):
|
|
form = self._form(action_type='')
|
|
assert not form.is_valid()
|
|
assert 'Please choose an action' in str(form.errors)
|
|
|
|
def test_action_requires_template(self, db):
|
|
form = self._form(action_message_template='')
|
|
assert not form.is_valid()
|
|
assert 'Message template is required' in str(form.errors)
|
|
|
|
def test_link_type_clears_action_config(self, db):
|
|
form = self._form(link_type=Link.LinkType.LINK, original_url='https://example.com')
|
|
assert form.is_valid(), form.errors
|
|
assert form.cleaned_data['action_config'] == {}
|
|
|
|
def test_link_type_requires_url(self, db):
|
|
form = self._form(link_type=Link.LinkType.LINK, original_url='')
|
|
assert not form.is_valid()
|
|
assert 'Original URL is required' in str(form.errors)
|
|
|
|
|
|
# ── redirect_to_original execution ────────────────────────────────────────────
|
|
|
|
class TestRedirectAction:
|
|
def test_action_link_executes_and_redirects(self, action_link):
|
|
"""成功：发送消息并 302 跳转 Discord，不渲染页面。"""
|
|
client = Client()
|
|
with patch('links.actions.send_discord_message') as mock_send:
|
|
resp = client.get('/testaction/今天有什么新闻/')
|
|
assert resp.status_code == 302
|
|
assert resp['Location'].startswith('https://discord.com/channels/')
|
|
mock_send.assert_called_once_with('@小黑 今天有什么新闻')
|
|
|
|
def test_action_link_with_query_param(self, action_link):
|
|
client = Client()
|
|
with patch('links.actions.send_discord_message') as mock_send:
|
|
resp = client.get('/testaction/', {'q': '用查询参数提问'})
|
|
assert resp.status_code == 302
|
|
mock_send.assert_called_once_with('@小黑 用查询参数提问')
|
|
|
|
def test_action_link_dedup_same_query_in_window(self, action_link, monkeypatch):
|
|
"""10 分钟内同一 query 只发一次，刷新不重复发送。"""
|
|
monkeypatch.setenv('DISCORD_WEBHOOK_URL', 'https://example.com/hook')
|
|
client = Client()
|
|
with patch('links.actions.requests.post') as mock_post:
|
|
resp1 = client.get('/testaction/相同问题/')
|
|
resp2 = client.get('/testaction/相同问题/') # 刷新/重复访问
|
|
assert resp1.status_code == 302
|
|
assert resp2.status_code == 302
|
|
assert mock_post.call_count == 1, '同一问题 10 分钟内只应发送一次'
|
|
|
|
def test_action_link_different_query_not_deduped(self, action_link, monkeypatch):
|
|
monkeypatch.setenv('DISCORD_WEBHOOK_URL', 'https://example.com/hook')
|
|
client = Client()
|
|
with patch('links.actions.requests.post') as mock_post:
|
|
client.get('/testaction/问题甲/')
|
|
client.get('/testaction/问题乙/')
|
|
assert mock_post.call_count == 2, '不同问题应各自发送'
|
|
|
|
def test_action_link_webhook_failure_shows_error(self, action_link, monkeypatch):
|
|
import requests as req_lib
|
|
monkeypatch.setenv('DISCORD_WEBHOOK_URL', 'https://example.com/hook')
|
|
client = Client()
|
|
with patch('links.actions.requests.post', side_effect=req_lib.exceptions.RequestException('connection refused')):
|
|
resp = client.get('/testaction/测试/')
|
|
assert resp.status_code == 200
|
|
assert '发送到 Discord 失败' in resp.content.decode()
|
|
assert '打开 Discord' not in resp.content.decode()
|
|
|
|
def test_normal_link_unaffected(self, db):
|
|
Link.objects.create(alias='plainlink', link_type=Link.LinkType.LINK, original_url='https://example.com/path')
|
|
client = Client()
|
|
resp = client.get('/plainlink/')
|
|
assert resp.status_code in (301, 302)
|
|
assert resp['Location'] == 'https://example.com/path'
|
|
|
|
|
|
# ── /ask/ 路由（带问题 = 发送+跳转；不带 = 输入页）─────────────────────────
|
|
|
|
class TestAskRoute:
|
|
def test_ask_with_question_redirects(self, monkeypatch):
|
|
monkeypatch.setenv('DISCORD_WEBHOOK_URL', 'https://example.com/hook')
|
|
client = Client()
|
|
with patch('links.actions.requests.post') as mock_post:
|
|
resp = client.get('/ask/今天有什么新闻/')
|
|
assert resp.status_code == 302
|
|
assert resp['Location'].startswith('https://discord.com/channels/')
|
|
body = mock_post.call_args.kwargs['json']['content']
|
|
assert '今天有什么新闻' in body
|
|
|
|
def test_ask_with_question_dedup(self, monkeypatch):
|
|
monkeypatch.setenv('DISCORD_WEBHOOK_URL', 'https://example.com/hook')
|
|
client = Client()
|
|
with patch('links.actions.requests.post') as mock_post:
|
|
client.get('/ask/相同问题/')
|
|
client.get('/ask/相同问题/')
|
|
assert mock_post.call_count == 1
|
|
|
|
def test_ask_empty_renders_input_page(self):
|
|
client = Client()
|
|
resp = client.get('/ask/')
|
|
assert resp.status_code == 200
|
|
assert '问小黑' in resp.content.decode()
|