110 lines
3.9 KiB
Python
110 lines
3.9 KiB
Python
import asyncio
|
|
import json
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from config import Settings
|
|
from n8n_client import N8NClient, WebhookError
|
|
|
|
|
|
CHAT_URL = "https://n8n.example.test/webhook/126bdb74-0630-415d-a8f7-6f699eac1a2c"
|
|
UPLOAD_URL = "https://n8n.example.test/webhook/993d3a50-19a7-4c64-9d36-67c0ced130dd"
|
|
|
|
|
|
def settings() -> Settings:
|
|
return Settings(chat_webhook_url=CHAT_URL, upload_webhook_url=UPLOAD_URL)
|
|
|
|
|
|
def test_chat_sends_exact_contract_and_reads_plain_text() -> None:
|
|
received: list[httpx.Request] = []
|
|
|
|
def respond(request: httpx.Request) -> httpx.Response:
|
|
received.append(request)
|
|
return httpx.Response(200, text="Ответ специалиста или агента")
|
|
|
|
client = N8NClient(settings(), httpx.MockTransport(respond))
|
|
first = asyncio.run(client.ask(" Вопрос 1 ", "my-store", "session-1"))
|
|
second = asyncio.run(client.ask("Вопрос 2", "my-store", "session-1"))
|
|
|
|
assert first == second == "Ответ специалиста или агента"
|
|
assert [request.method for request in received] == ["POST", "POST"]
|
|
assert all(str(request.url) == CHAT_URL for request in received)
|
|
assert [json.loads(request.content) for request in received] == [
|
|
{"message": "Вопрос 1", "collection_name": "my-store", "session_id": "session-1"},
|
|
{"message": "Вопрос 2", "collection_name": "my-store", "session_id": "session-1"},
|
|
]
|
|
|
|
|
|
def test_upload_uses_one_multipart_file_and_collection() -> None:
|
|
received: list[httpx.Request] = []
|
|
|
|
def respond(request: httpx.Request) -> httpx.Response:
|
|
received.append(request)
|
|
return httpx.Response(200, json={"unverified": "response shape"})
|
|
|
|
client = N8NClient(settings(), httpx.MockTransport(respond))
|
|
assert asyncio.run(client.upload("notes.md", b"# Notes\n", "my-store")) is None
|
|
|
|
assert len(received) == 1
|
|
request = received[0]
|
|
assert request.method == "POST"
|
|
assert str(request.url) == UPLOAD_URL
|
|
assert request.headers["content-type"].startswith("multipart/form-data; boundary=")
|
|
assert b'name="collection_name"' in request.content
|
|
assert b"my-store" in request.content
|
|
assert b'name="file"; filename="notes.md"' in request.content
|
|
assert b"Content-Type: text/markdown" in request.content
|
|
assert b"# Notes\n" in request.content
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"filename,contents,collection",
|
|
[
|
|
("notes.txt", b"text", "my-store"),
|
|
("notes.md", b"", "my-store"),
|
|
("notes.md", b"text", " "),
|
|
("notes.md", b"12345", "my-store"),
|
|
],
|
|
)
|
|
def test_invalid_upload_never_calls_webhook(
|
|
filename: str, contents: bytes, collection: str
|
|
) -> None:
|
|
called = False
|
|
|
|
def respond(request: httpx.Request) -> httpx.Response:
|
|
nonlocal called
|
|
called = True
|
|
return httpx.Response(200)
|
|
|
|
options = Settings(
|
|
chat_webhook_url=CHAT_URL,
|
|
upload_webhook_url=UPLOAD_URL,
|
|
max_file_bytes=4,
|
|
)
|
|
client = N8NClient(options, httpx.MockTransport(respond))
|
|
with pytest.raises(ValueError):
|
|
asyncio.run(client.upload(filename, contents, collection))
|
|
assert not called
|
|
|
|
|
|
def test_chat_http_error_and_empty_reply_are_not_answers() -> None:
|
|
for response in (httpx.Response(502, text="failure"), httpx.Response(200, text=" ")):
|
|
client = N8NClient(settings(), httpx.MockTransport(lambda _: response))
|
|
with pytest.raises(WebhookError):
|
|
asyncio.run(client.ask("Вопрос", "my-store", "session-1"))
|
|
|
|
|
|
def test_timeout_does_not_retry_upload() -> None:
|
|
attempts = 0
|
|
|
|
def respond(request: httpx.Request) -> httpx.Response:
|
|
nonlocal attempts
|
|
attempts += 1
|
|
raise httpx.ReadTimeout("timeout", request=request)
|
|
|
|
client = N8NClient(settings(), httpx.MockTransport(respond))
|
|
with pytest.raises(WebhookError, match="ожидания загрузки"):
|
|
asyncio.run(client.upload("notes.md", b"# Notes", "my-store"))
|
|
assert attempts == 1
|