222 lines
8.3 KiB
Python
222 lines
8.3 KiB
Python
import asyncio
|
|
|
|
import httpx
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
|
|
from B01_Dashboard import B01_Dashboard_Knowledge_Chat as chat
|
|
|
|
|
|
def test_greeting_answers_without_search_or_api_key(monkeypatch):
|
|
monkeypatch.delenv("NVIDIA_API_KEY", raising=False)
|
|
monkeypatch.setattr(
|
|
chat, "search", lambda _question: (_ for _ in ()).throw(AssertionError("searched"))
|
|
)
|
|
result = asyncio.run(chat.ask_knowledge_chat(chat.ChatRequest(question="안녕!"), {}))
|
|
assert result == {
|
|
"answer": "안녕하세요! 법령이나 지침에 대해 궁금한 점을 물어보세요.",
|
|
"sources": [],
|
|
}
|
|
|
|
|
|
def test_natural_question_finds_term_without_particle(monkeypatch):
|
|
monkeypatch.setattr(
|
|
chat,
|
|
"_sections",
|
|
lambda: [("original/원가계산/품셈.md", "# 일위대가\n공사 한 단위에 필요한 비용을 계산")],
|
|
)
|
|
assert chat.search("일위대가가 뭐야")[0]["path"] == "original/원가계산/품셈.md"
|
|
|
|
|
|
def test_source_endpoint_only_reads_allowed_original_markdown(monkeypatch, tmp_path):
|
|
original = tmp_path / "original"
|
|
allowed = original / "법령" / "현행.md"
|
|
excluded = original / "실무문서" / "사례.md"
|
|
outside = tmp_path / "technical_info" / "가이드.md"
|
|
allowed.parent.mkdir(parents=True)
|
|
excluded.parent.mkdir()
|
|
outside.parent.mkdir()
|
|
allowed.write_text("# 허용 원문", encoding="utf-8")
|
|
excluded.write_text("비공개", encoding="utf-8")
|
|
outside.write_text("중복 자료", encoding="utf-8")
|
|
monkeypatch.setattr(chat, "KNOWLEDGE", tmp_path)
|
|
|
|
result = asyncio.run(chat.read_source(chat.SourceRequest(path="original/법령/현행.md"), {}))
|
|
assert result["content"] == "# 허용 원문"
|
|
for path in (
|
|
"original/실무문서/사례.md",
|
|
"technical_info/가이드.md",
|
|
"../technical_info/가이드.md",
|
|
):
|
|
with pytest.raises(HTTPException) as error:
|
|
asyncio.run(chat.read_source(chat.SourceRequest(path=path), {}))
|
|
assert error.value.status_code == 404
|
|
|
|
|
|
def test_source_request_words_do_not_displace_matching_rule(monkeypatch):
|
|
monkeypatch.setattr(
|
|
chat,
|
|
"_sections",
|
|
lambda: [
|
|
("original/법률/일반기준/현행_2026.md", "임도 문서 근거로 " * 10),
|
|
("original/행정규칙/임도시설기준/별표7.md", "임도 종단기울기 기준 7%"),
|
|
],
|
|
)
|
|
assert chat.search("임도 종단기울기 기준을 문서 근거로 설명해줘")[0]["path"].endswith(
|
|
"별표7.md"
|
|
)
|
|
|
|
|
|
def test_missing_api_key_is_reported_before_search(monkeypatch):
|
|
monkeypatch.delenv("NVIDIA_API_KEY", raising=False)
|
|
monkeypatch.setattr(chat, "search", lambda _question: [])
|
|
with pytest.raises(HTTPException) as error:
|
|
asyncio.run(chat.ask_knowledge_chat(chat.ChatRequest(question="일위대가가 뭐야"), {}))
|
|
assert error.value.status_code == 503
|
|
|
|
|
|
def test_busy_nvidia_is_retried_and_reported(monkeypatch, caplog):
|
|
monkeypatch.setenv("NVIDIA_API_KEY", "test-key")
|
|
monkeypatch.setattr(chat, "search", lambda _question: [])
|
|
calls = []
|
|
|
|
class Client:
|
|
def __init__(self, timeout):
|
|
pass
|
|
|
|
async def __aenter__(self):
|
|
return self
|
|
|
|
async def __aexit__(self, *_args):
|
|
pass
|
|
|
|
async def post(self, *_args, **_kwargs):
|
|
calls.append(1)
|
|
return type("Response", (), {"status_code": 503})()
|
|
|
|
async def no_wait(_seconds):
|
|
pass
|
|
|
|
monkeypatch.setattr(chat.httpx, "AsyncClient", Client)
|
|
monkeypatch.setattr(chat.asyncio, "sleep", no_wait)
|
|
with pytest.raises(HTTPException) as error:
|
|
asyncio.run(chat.ask_knowledge_chat(chat.ChatRequest(question="일위대가가 뭐야"), {}))
|
|
assert len(calls) == 3
|
|
assert "요청이 몰려" in error.value.detail
|
|
assert [f"attempt={n} status=503" in caplog.text for n in (1, 2, 3)] == [True] * 3
|
|
assert "test-key" not in caplog.text
|
|
assert "일위대가가 뭐야" not in caplog.text
|
|
|
|
|
|
def test_nvidia_timeout_is_reported(monkeypatch):
|
|
monkeypatch.setenv("NVIDIA_API_KEY", "test-key")
|
|
monkeypatch.setattr(chat, "search", lambda _question: [])
|
|
|
|
class Client:
|
|
def __init__(self, timeout):
|
|
pass
|
|
|
|
async def __aenter__(self):
|
|
return self
|
|
|
|
async def __aexit__(self, *_args):
|
|
pass
|
|
|
|
async def post(self, *_args, **_kwargs):
|
|
raise httpx.ReadTimeout("timed out")
|
|
|
|
monkeypatch.setattr(chat.httpx, "AsyncClient", Client)
|
|
with pytest.raises(HTTPException) as error:
|
|
asyncio.run(chat.ask_knowledge_chat(chat.ChatRequest(question="일위대가가 뭐야"), {}))
|
|
assert error.value.status_code == 504
|
|
assert "시간 초과" in error.value.detail
|
|
|
|
|
|
def test_chat_reads_only_original_except_excluded_folders(monkeypatch, tmp_path):
|
|
public = tmp_path / "original" / "법률"
|
|
guide = tmp_path / "technical_info"
|
|
outside = tmp_path / "별도자료"
|
|
private = tmp_path / "original" / "실무문서"
|
|
stmate = tmp_path / "original" / "원가계산" / "STmate"
|
|
stmate_analysis = tmp_path / "original" / "경쟁사 프로그램" / "STmate 분석"
|
|
dasan = tmp_path / "original" / "원가계산" / "다산소프트"
|
|
public.mkdir(parents=True)
|
|
guide.mkdir()
|
|
outside.mkdir()
|
|
private.mkdir()
|
|
stmate.mkdir(parents=True)
|
|
stmate_analysis.mkdir(parents=True)
|
|
dasan.mkdir()
|
|
(public / "현행.md").write_text("# 임도 기준\n" + "임도 설치 기준 내용 " * 10, encoding="utf-8")
|
|
(guide / "_검토.md").write_text("# 임도 기준\n기술정보 안내 " * 10, encoding="utf-8")
|
|
(outside / "자료.md").write_text("# 임도 기준\n별도 자료 " * 10, encoding="utf-8")
|
|
(private / "사례.md").write_text("# 임도 기준\n비공개 사례 " * 10, encoding="utf-8")
|
|
(stmate / "자료.md").write_text("# 임도 기준\nSTmate 자료 " * 10, encoding="utf-8")
|
|
(stmate_analysis / "자료.md").write_text(
|
|
"# 임도 기준\nSTmate 분석 자료 " * 10, encoding="utf-8"
|
|
)
|
|
(dasan / "자료.md").write_text("# 임도 기준\n다산소프트 자료 " * 10, encoding="utf-8")
|
|
monkeypatch.setattr(chat, "KNOWLEDGE", tmp_path)
|
|
monkeypatch.setenv("NVIDIA_API_KEY", "test-key")
|
|
chat._sections.cache_clear()
|
|
sent = {}
|
|
|
|
class Response:
|
|
status_code = 200
|
|
|
|
def raise_for_status(self):
|
|
pass
|
|
|
|
def json(self):
|
|
return {"choices": [{"message": {"content": "설명 [1]"}}]}
|
|
|
|
class Client:
|
|
def __init__(self, timeout):
|
|
pass
|
|
|
|
async def __aenter__(self):
|
|
return self
|
|
|
|
async def __aexit__(self, *_args):
|
|
pass
|
|
|
|
async def post(self, url, *, headers, json):
|
|
sent["url"] = url
|
|
sent["authorization"] = headers["Authorization"]
|
|
sent["model"] = json["model"]
|
|
sent["prompt"] = json["messages"][0]["content"]
|
|
sent["reasoning_effort"] = json["reasoning_effort"]
|
|
sent["stream"] = json["stream"]
|
|
return Response()
|
|
|
|
monkeypatch.setattr(chat.httpx, "AsyncClient", Client)
|
|
result = asyncio.run(
|
|
chat.ask_knowledge_chat(
|
|
chat.ChatRequest(
|
|
question="그중 몇 퍼센트야?",
|
|
history=[
|
|
chat.ChatTurn(role="user", content="임도 기준 알려줘"),
|
|
chat.ChatTurn(role="assistant", content="임도 기준입니다."),
|
|
],
|
|
),
|
|
{},
|
|
)
|
|
)
|
|
assert result["answer"] == "설명 [1]"
|
|
assert sent["url"] == "https://integrate.api.nvidia.com/v1/chat/completions"
|
|
assert sent["authorization"] == "Bearer test-key"
|
|
assert sent["model"] == "nvidia/nemotron-3-super-120b-a12b"
|
|
assert sent["reasoning_effort"] == "none"
|
|
assert sent["stream"] is False
|
|
assert "이전 대화" in sent["prompt"]
|
|
assert "사용자: 임도 기준 알려줘" in sent["prompt"]
|
|
paths = {source["path"] for source in result["sources"]}
|
|
assert paths == {"original/법률/현행.md"}
|
|
assert "실무문서" not in sent["prompt"]
|
|
assert "비공개 사례" not in sent["prompt"]
|
|
assert "STmate" not in sent["prompt"]
|
|
assert "다산소프트" not in sent["prompt"]
|
|
assert "기술정보 안내" not in sent["prompt"]
|
|
assert "별도 자료" not in sent["prompt"]
|
|
chat._sections.cache_clear()
|