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()