eezd
f67a8ca402
Replace cookie authentication with revocable Bearer sessions, optional one-use PNG challenges, bounded login limits, and short-lived market WebSocket tickets. Keep captcha disabled by default. Standardize ordinary JSON responses while preserving HTTP statuses, raw OpenAPI, redirects, SSE, and WebSocket semantics. Revoke long connections with their owning session. Expose official calendar data and collector status to the frontend and research tools, support bilingual news topics, and keep summaries and forward-only labels explicit. Verify the staged snapshot with all 305 backend tests.
287 lines
13 KiB
Python
287 lines
13 KiB
Python
import asyncio
|
|
import json
|
|
import uuid
|
|
from datetime import UTC, datetime, timedelta
|
|
from types import SimpleNamespace
|
|
|
|
import httpx
|
|
import httpx2
|
|
import pytest
|
|
|
|
|
|
def install_model_transport(llm, monkeypatch, protocol, handler):
|
|
monkeypatch.setattr(llm, "decrypt", lambda value: "test-only")
|
|
module = llm.openai if protocol == "openai" else llm.anthropic
|
|
name = "AsyncOpenAI" if protocol == "openai" else "AsyncAnthropic"
|
|
constructor = getattr(module, name)
|
|
http = httpx if protocol == "openai" else httpx2
|
|
monkeypatch.setattr(module, name, lambda **kwargs: constructor(**kwargs, http_client=http.AsyncClient(transport=http.MockTransport(handler))))
|
|
return SimpleNamespace(protocol=protocol, base_url="https://model.invalid", api_key_encrypted=b"test", model="test", max_tokens=100)
|
|
|
|
|
|
@pytest.mark.parametrize("protocol", ["openai", "anthropic"])
|
|
@pytest.mark.parametrize("complete", [False, True])
|
|
def test_model_requires_protocol_end_marker(trading, monkeypatch, protocol, complete):
|
|
from alphapulse.ai import llm
|
|
if protocol == "openai":
|
|
chunks = [{"id": "1", "object": "chat.completion.chunk", "created": 0, "model": "test", "choices": [{"index": 0, "delta": {"content": "部分回答"}, "finish_reason": None}]}]
|
|
if complete:
|
|
chunks.append(chunks[0] | {"choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}]})
|
|
body = "".join(f"data: {json.dumps(c)}\n\n" for c in chunks) + "data: [DONE]\n\n"
|
|
else:
|
|
events = [
|
|
{"type": "message_start", "message": {"id": "1", "type": "message", "role": "assistant", "model": "test", "content": [], "stop_reason": None, "stop_sequence": None, "usage": {"input_tokens": 1, "output_tokens": 0}}},
|
|
{"type": "content_block_start", "index": 0, "content_block": {"type": "text", "text": ""}},
|
|
{"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": "部分回答"}},
|
|
{"type": "message_delta", "delta": {"stop_reason": "end_turn", "stop_sequence": None}, "usage": {"output_tokens": 1}},
|
|
]
|
|
if complete:
|
|
events.append({"type": "message_stop"})
|
|
body = "".join(f"event: {e['type']}\ndata: {json.dumps(e)}\n\n" for e in events)
|
|
|
|
def handle(request):
|
|
http = httpx if protocol == "openai" else httpx2
|
|
return http.Response(200, headers={"content-type": "text/event-stream"}, content=body.encode())
|
|
|
|
provider = install_model_transport(llm, monkeypatch, protocol, handle)
|
|
async def check():
|
|
output = []
|
|
try:
|
|
async for item in llm.stream(provider, "system", [], []):
|
|
output.append(item)
|
|
except llm.LlmError:
|
|
assert not complete
|
|
else:
|
|
assert complete
|
|
assert [x.text for x in output if isinstance(x, llm.TextDelta)] == ["部分回答"]
|
|
assert any(isinstance(x, llm.Completion) for x in output) == complete
|
|
asyncio.run(check())
|
|
|
|
|
|
def test_connection_failure_uses_only_sdk_retry_budget(trading, monkeypatch):
|
|
from alphapulse.ai import llm
|
|
attempts = 0
|
|
def handle(request):
|
|
nonlocal attempts
|
|
attempts += 1
|
|
raise httpx.ConnectError("测试连接失败", request=request)
|
|
provider = install_model_transport(llm, monkeypatch, "openai", handle)
|
|
async def check():
|
|
with pytest.raises(llm.LlmError, match="无法连接"):
|
|
async for _ in llm.stream(provider, "system", [], []):
|
|
pass
|
|
asyncio.run(check())
|
|
assert attempts == llm.MAX_RETRIES + 1
|
|
|
|
|
|
def test_pipeline_error_keeps_history_checkpoint_and_marks_run_failed(trading, monkeypatch):
|
|
from scrapy import signals
|
|
from scrapy.crawler import Crawler
|
|
from twisted.python.failure import Failure
|
|
from alphapulse.crawler.spiders import base
|
|
|
|
class Spider(base.NewsSpider):
|
|
name = "test"
|
|
|
|
now = datetime.now(UTC)
|
|
target, checkpoint = now - timedelta(days=10), now - timedelta(days=1)
|
|
source = SimpleNamespace(history_target=target, history_position=checkpoint, consecutive_failures=0)
|
|
writes = []
|
|
class Session:
|
|
async def __aenter__(self):
|
|
return self
|
|
async def __aexit__(self, *args):
|
|
pass
|
|
async def execute(self, statement):
|
|
writes.append(statement.compile().params)
|
|
async def get(self, *args, **kwargs):
|
|
return source
|
|
async def commit(self):
|
|
pass
|
|
monkeypatch.setattr(base, "SessionLocal", Session)
|
|
async def check():
|
|
crawler = Crawler(Spider)
|
|
spider = Spider.from_crawler(crawler)
|
|
run = base.RunState(uuid.uuid4(), "history", now, target, history_target=target, history_position=target)
|
|
spider.run = run
|
|
crawler.signals.send_catch_log(signals.item_error, item={}, response=None, spider=spider, failure=Failure(RuntimeError("写库失败")))
|
|
spider._close_run()
|
|
await asyncio.gather(*spider._tasks)
|
|
assert spider.history_checkpoint is None and source.history_position == checkpoint
|
|
assert source.consecutive_failures == 1
|
|
assert writes[0]["status"] == "failed" and "写库失败" in writes[0]["error"]
|
|
asyncio.run(check())
|
|
|
|
|
|
@pytest.mark.parametrize("failure_at", ["usage", "second_tool"])
|
|
def test_chat_completes_tool_results_on_disconnect_or_exception(trading, monkeypatch, failure_at):
|
|
from alphapulse.api import chat
|
|
provider = SimpleNamespace(id=1, context_window=10000, max_tokens=100, name="test", reasoning_effort=None)
|
|
class Session:
|
|
async def __aenter__(self):
|
|
return self
|
|
async def __aexit__(self, *args):
|
|
pass
|
|
async def scalar(self, *args):
|
|
return provider
|
|
def expunge(self, *args):
|
|
pass
|
|
async def current(*args):
|
|
return SimpleNamespace(id=1)
|
|
async def system(*args):
|
|
return "system"
|
|
async def history(*args):
|
|
return []
|
|
saved = []
|
|
async def save(*messages):
|
|
saved.extend(messages)
|
|
calls = [{"id": "a", "name": "first", "arguments": {}}, {"id": "b", "name": "second", "arguments": {}}]
|
|
async def stream(*args):
|
|
yield chat.llm.Completion(tool_calls=calls)
|
|
async def tool(name, args):
|
|
if name == "second":
|
|
raise RuntimeError("工具失败")
|
|
return chat.ToolOutcome("实际结果")
|
|
monkeypatch.setattr(chat, "SessionLocal", Session)
|
|
monkeypatch.setattr(chat, "active_session", current)
|
|
monkeypatch.setattr(chat, "build_system_prompt", system)
|
|
monkeypatch.setattr(chat, "session_messages", history)
|
|
monkeypatch.setattr(chat, "save_messages", save)
|
|
monkeypatch.setattr(chat.llm, "stream", stream)
|
|
monkeypatch.setattr(chat, "run_tool", tool)
|
|
async def check():
|
|
generator = chat._message_events(chat.SendIn(text="test"))
|
|
events = []
|
|
async for event in generator:
|
|
events.append(event.event)
|
|
if failure_at == "usage" and event.event == "usage":
|
|
await generator.aclose()
|
|
break
|
|
results = [m.content for m in saved if m.role == "tool"]
|
|
assert [r["tool_call_id"] for r in results] == ["a", "b"]
|
|
if failure_at == "second_tool":
|
|
assert results[0]["result"] == "实际结果"
|
|
assert "error" in json.loads(results[1]["result"])
|
|
assert events[-1] == "error"
|
|
asyncio.run(check())
|
|
|
|
|
|
def test_skill_rename_rejects_whitespace_without_writing(trading):
|
|
from fastapi import HTTPException
|
|
from alphapulse.api import skills
|
|
node = SimpleNamespace(name="原名称")
|
|
class Session:
|
|
async def get(self, *args):
|
|
return node
|
|
async def check():
|
|
with pytest.raises(HTTPException) as error:
|
|
await skills.patch(1, skills.NodePatch(name=" \t "), Session())
|
|
assert error.value.status_code == 400
|
|
assert node.name == "原名称"
|
|
asyncio.run(check())
|
|
|
|
|
|
def test_chat_reset_blocks_new_generation_until_commit(trading, monkeypatch):
|
|
from alphapulse.api import chat
|
|
|
|
async def check():
|
|
monkeypatch.setattr(chat, "generating", asyncio.Lock())
|
|
entered, finish = asyncio.Event(), asyncio.Event()
|
|
monkeypatch.setattr(chat, "SessionLocal", lambda: pytest.fail("重置期间不能进入生成的数据库路径"))
|
|
saved = []
|
|
class Session:
|
|
async def execute(self, statement):
|
|
entered.set()
|
|
await finish.wait()
|
|
def add(self, row):
|
|
saved.append(row)
|
|
async def commit(self):
|
|
assert chat.generating.locked()
|
|
async def reset_and_send():
|
|
task = asyncio.create_task(chat.reset(chat.ResetIn(clear_context=True), Session()))
|
|
try:
|
|
await entered.wait()
|
|
events = [event async for event in chat._message_events(chat.SendIn(text="重置期间的新消息"))]
|
|
assert len(events) == 1 and events[0].event == "error"
|
|
assert not task.done()
|
|
finally:
|
|
finish.set()
|
|
await task
|
|
await asyncio.wait_for(reset_and_send(), timeout=2)
|
|
assert len(saved) == 1 and saved[0].content == "" and saved[0].source == "reset"
|
|
assert not chat.generating.locked()
|
|
asyncio.run(check())
|
|
|
|
|
|
def test_chat_reset_rejects_busy_generation_and_releases_lock_on_failure(trading, monkeypatch):
|
|
from fastapi import HTTPException
|
|
from alphapulse.api import chat
|
|
|
|
async def check():
|
|
monkeypatch.setattr(chat, "generating", asyncio.Lock())
|
|
class Session:
|
|
async def execute(self, statement):
|
|
raise RuntimeError("重置写库失败")
|
|
async with chat.generating:
|
|
with pytest.raises(HTTPException) as error:
|
|
await chat.reset(chat.ResetIn(), Session())
|
|
assert error.value.status_code == 409
|
|
with pytest.raises(RuntimeError, match="重置写库失败"):
|
|
await chat.reset(chat.ResetIn(), Session())
|
|
assert not chat.generating.locked()
|
|
asyncio.run(check())
|
|
|
|
|
|
@pytest.mark.parametrize("change", ["metadata", "content", "unchanged"])
|
|
def test_news_updates_metadata_without_creating_content_revision(trading, monkeypatch, change):
|
|
from alphapulse.crawler import pipelines
|
|
from alphapulse.models import NewsItem
|
|
|
|
original = {
|
|
"url": "https://example.invalid/1", "title": "原标题", "summary": "摘要", "content_html": "<p>正文</p>",
|
|
"content_text": "正文", "author": "原作者", "tags": ["旧标签"], "is_paywalled": False,
|
|
"published_at": datetime(2026, 10, 20, tzinfo=UTC), "source_updated_at": None, "extra": {},
|
|
}
|
|
news = NewsItem(id=1, source_code="cls_telegraph", external_id="1", revision_no=1,
|
|
content_hash=pipelines.content_hash(original), **original)
|
|
incoming = dict(original)
|
|
if change != "unchanged":
|
|
incoming.update(author="更正作者", tags=["新标签"], is_paywalled=True, extra={"level": 1},
|
|
published_at=original["published_at"] + timedelta(minutes=1),
|
|
source_updated_at=original["published_at"] + timedelta(minutes=2))
|
|
if change == "content":
|
|
incoming.update(title="修订标题", content_html="<p>修订正文</p>", content_text="修订正文")
|
|
revisions, events = [], []
|
|
class Session:
|
|
async def __aenter__(self):
|
|
return self
|
|
async def __aexit__(self, *args):
|
|
pass
|
|
async def scalar(self, statement):
|
|
return news
|
|
def add(self, revision):
|
|
revisions.append(revision)
|
|
async def commit(self):
|
|
pass
|
|
async def publish(channel, data):
|
|
events.append(json.loads(data))
|
|
monkeypatch.setattr(pipelines, "SessionLocal", Session)
|
|
monkeypatch.setattr(pipelines.redis, "publish", publish)
|
|
run = SimpleNamespace(items_seen=0, items_new=0, items_revised=0)
|
|
pipeline = pipelines.NewsPipeline(SimpleNamespace(spider=SimpleNamespace(run=run)))
|
|
|
|
async def check():
|
|
for _ in range(2):
|
|
await pipeline.process_item({**incoming, "source_code": "cls_telegraph", "external_id": "1", "raw_response_id": 10})
|
|
assert {key: getattr(news, key) for key in pipelines.UPDATABLE_FIELDS} == incoming
|
|
assert news.content_hash == pipelines.content_hash(incoming)
|
|
assert news.revision_no == (2 if change == "content" else 1)
|
|
assert run.items_seen == 2 and run.items_new == 0
|
|
if change == "content":
|
|
assert len(revisions) == run.items_revised == 1
|
|
assert revisions[0].revision_no == 2 and revisions[0].content_html == incoming["content_html"]
|
|
assert events == [{"id": 1, "source_code": "cls_telegraph", "kind": "revised"}]
|
|
else:
|
|
assert not revisions and not events and run.items_revised == 0
|
|
asyncio.run(check())
|