539 lines
20 KiB
Python
539 lines
20 KiB
Python
from __future__ import annotations
|
|
|
|
import ast
|
|
import html
|
|
import json
|
|
from pathlib import Path
|
|
import re
|
|
import tempfile
|
|
from types import SimpleNamespace
|
|
from typing import Any, Mapping
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from ai_web_agent_console.scenarios import ScenarioConfigError, load_demo_scenarios
|
|
from ai_web_agent_console.profile import load_app_profile
|
|
from ai_web_agent_console.mcp_tool_router import McpTool, build_mcp_tool_arguments
|
|
from ai_web_agent_console.mcp_result import (
|
|
has_actionable_text_result,
|
|
status_result_evidence,
|
|
status_result_summary,
|
|
)
|
|
from ai_web_agent_console.model_registry import load_model_registry
|
|
|
|
|
|
def _load_console_query_helpers():
|
|
"""Load the Streamlit entrypoint only when its optional runtime is installed."""
|
|
|
|
try:
|
|
from app import _prepare_hmm_hr_tool_query
|
|
except ModuleNotFoundError:
|
|
return None
|
|
return _prepare_hmm_hr_tool_query
|
|
|
|
|
|
class DemoScenarioConfigTest(unittest.TestCase):
|
|
def test_model_registry_uses_console_names(self) -> None:
|
|
registry = load_model_registry()
|
|
|
|
self.assertEqual(
|
|
registry.registry_name,
|
|
"AI_WEB_AGENT_CONSOLE_MODEL_PROFILES",
|
|
)
|
|
self.assertTrue(registry.default_profile.default_for_console)
|
|
|
|
def test_profile_environment_overrides_json_defaults(self) -> None:
|
|
path = Path(__file__).parents[1] / "config" / "app_profile.json"
|
|
with patch.dict(
|
|
"os.environ",
|
|
{
|
|
"AGENT_CONSOLE_SHORT_NAME": "HMM",
|
|
"AGENT_CONSOLE_PAGE_TITLE": "HMM AI 업무 에이전트",
|
|
"AGENT_CONSOLE_PRIMARY_COLOR": "#003b70",
|
|
},
|
|
clear=False,
|
|
):
|
|
profile = load_app_profile(path)
|
|
|
|
self.assertEqual(profile.short_name, "HMM")
|
|
self.assertEqual(profile.page_title, "HMM AI 업무 에이전트")
|
|
self.assertEqual(profile.primary_color, "#003b70")
|
|
|
|
def test_profile_reads_dotenv_values(self) -> None:
|
|
path = Path(__file__).parents[1] / "config" / "app_profile.json"
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
env_file = Path(temp_dir) / ".env"
|
|
env_file.write_text("AGENT_CONSOLE_SHORT_NAME=HMM\n", encoding="utf-8")
|
|
profile = load_app_profile(path, env_file)
|
|
|
|
self.assertEqual(profile.short_name, "HMM")
|
|
|
|
def test_common_theme_covers_lists_expanders_and_secondary_buttons(self) -> None:
|
|
path = Path(__file__).parents[1] / "ai_web_agent_console" / "presentation.py"
|
|
source = path.read_text(encoding="utf-8")
|
|
|
|
self.assertIn('[data-testid="stAppViewContainer"] li', source)
|
|
self.assertIn('[data-testid="stExpander"] summary', source)
|
|
self.assertIn('div[data-testid="stButton"] > button', source)
|
|
self.assertIn('[data-baseweb="tab-list"] [role="tab"]', source)
|
|
self.assertIn('[data-testid="stTab"]', source)
|
|
self.assertIn('[role="tab"][aria-selected="true"]', source)
|
|
|
|
def test_audit_tab_uses_hmm_access_audit_loaders(self) -> None:
|
|
root = Path(__file__).parents[1]
|
|
entrypoint = (root / "app.py").read_text(
|
|
encoding="utf-8"
|
|
)
|
|
renderer = (root / "ai_web_agent_console" / "audit.py").read_text(
|
|
encoding="utf-8"
|
|
)
|
|
|
|
self.assertIn("FROM ADMIN.HMM_ACCESS_AUDIT", entrypoint)
|
|
self.assertIn("_load_hmm_audit_inventory", entrypoint)
|
|
self.assertIn("(protocol=tcps)(port=1521)", entrypoint)
|
|
self.assertIn(
|
|
"AI_WEB_AGENT_CONSOLE_AUDIT_WALLET_PASSWORD",
|
|
entrypoint,
|
|
)
|
|
self.assertIn("HMM 접근 관리", renderer)
|
|
self.assertNotIn('AUDIT_SCHEMA = "POC_2"', entrypoint)
|
|
|
|
def test_hmm_scenarios_are_enabled_and_unique(self) -> None:
|
|
path = Path(__file__).parents[1] / "config" / "hmm_demo_scenarios.json"
|
|
scenarios = load_demo_scenarios(path)
|
|
|
|
self.assertGreaterEqual(len(scenarios), 3)
|
|
self.assertEqual(len(scenarios), len({item.scenario_id for item in scenarios}))
|
|
self.assertTrue(all(item.question.strip() for item in scenarios))
|
|
by_id = {item.scenario_id: item for item in scenarios}
|
|
self.assertEqual(
|
|
{"FED-01", "FED-02", "FED-03", "FED-04", "FED-05"},
|
|
{"FED-01", "FED-02", "FED-03", "FED-04", "FED-05"} & set(by_id),
|
|
)
|
|
self.assertTrue(
|
|
all("선사" in by_id[scenario_id].question for scenario_id in (
|
|
"FED-01", "FED-02", "FED-03", "FED-04", "FED-05"
|
|
))
|
|
)
|
|
carrier_scenarios = [
|
|
item for item in scenarios if item.scenario_id.startswith("FED-")
|
|
]
|
|
self.assertTrue(
|
|
all(
|
|
re.search(r"\bE\d{4,}\b", item.question, flags=re.IGNORECASE) is None
|
|
for item in carrier_scenarios
|
|
)
|
|
)
|
|
report_scenarios = [
|
|
item for item in carrier_scenarios if "리포트로 보여줘" in item.question
|
|
]
|
|
self.assertGreaterEqual(len(report_scenarios), 2)
|
|
self.assertTrue(
|
|
all(item.category == "선사 실적 리포트" for item in report_scenarios)
|
|
)
|
|
|
|
def test_hmm_mcp_allows_carrier_federation_tool(self) -> None:
|
|
path = Path(__file__).parents[1] / "config" / "mcp_servers.json"
|
|
payload = json.loads(path.read_text(encoding="utf-8"))
|
|
server = next(
|
|
item for item in payload["servers"] if item["id"] == "hmm_hr_mcp"
|
|
)
|
|
|
|
self.assertIn(
|
|
"search_carrier_performance",
|
|
server["tool_allowlist"],
|
|
)
|
|
|
|
def test_hmm_mcp_allows_dynamic_html_renderer(self) -> None:
|
|
root = Path(__file__).parents[1]
|
|
payload = json.loads(
|
|
(root / "config" / "mcp_servers.json").read_text(encoding="utf-8")
|
|
)
|
|
server = next(
|
|
item for item in payload["servers"] if item["id"] == "hmm_hr_mcp"
|
|
)
|
|
source = (root / "app.py").read_text(encoding="utf-8")
|
|
|
|
self.assertEqual(
|
|
server["endpoint_url"],
|
|
"https://hmm-backoffice.cloud-handson.com/mcp",
|
|
)
|
|
self.assertIn("render_hmm_carrier_report", server["tool_allowlist"])
|
|
self.assertNotIn('tool.name == "render_hmm_carrier_report"', source)
|
|
|
|
def test_hmm_report_rows_accept_nested_select_ai_json_array(self) -> None:
|
|
source = (Path(__file__).parents[1] / "app.py").read_text(encoding="utf-8")
|
|
tree = ast.parse(source)
|
|
helper = next(
|
|
node
|
|
for node in tree.body
|
|
if isinstance(node, ast.FunctionDef)
|
|
and node.name == "_mcp_structured_rows"
|
|
)
|
|
namespace: dict[str, Any] = {
|
|
"Any": Any,
|
|
"Mapping": Mapping,
|
|
"json": json,
|
|
}
|
|
exec(compile(ast.Module(body=[helper], type_ignores=[]), "app.py", "exec"), namespace)
|
|
|
|
rows = namespace["_mcp_structured_rows"](
|
|
{
|
|
"response": {
|
|
"result": json.dumps(
|
|
[
|
|
{"EMPLOYEE_CODE": "E9001", "CARRIER_CODE": "C901"},
|
|
{"EMPLOYEE_CODE": "E9002", "CARRIER_CODE": "C902"},
|
|
]
|
|
)
|
|
}
|
|
}
|
|
)
|
|
|
|
self.assertEqual(len(rows), 2)
|
|
self.assertEqual(rows[0]["CARRIER_CODE"], "C901")
|
|
|
|
def test_hmm_report_normalizes_repeated_select_ai_column_labels(self) -> None:
|
|
source = (Path(__file__).parents[1] / "app.py").read_text(encoding="utf-8")
|
|
tree = ast.parse(source)
|
|
helpers = [
|
|
node
|
|
for node in tree.body
|
|
if isinstance(node, ast.FunctionDef)
|
|
and node.name in {"_camel_case_key", "_normalize_presentation_value"}
|
|
]
|
|
namespace: dict[str, Any] = {
|
|
"Any": Any,
|
|
"Mapping": Mapping,
|
|
"re": re,
|
|
}
|
|
exec(compile(ast.Module(body=helpers, type_ignores=[]), "app.py", "exec"), namespace)
|
|
|
|
first = namespace["_normalize_presentation_value"](
|
|
{"CARRIER_CODE": "C001", "LATEST_REVENUE_USD": 100}
|
|
)
|
|
repeated = namespace["_normalize_presentation_value"](
|
|
{"carrier Code": "C002", "LATEST REVENUE USD": 200}
|
|
)
|
|
|
|
self.assertEqual(first, {"carrierCode": "C001", "latestRevenueUsd": 100})
|
|
self.assertEqual(repeated, {"carrierCode": "C002", "latestRevenueUsd": 200})
|
|
|
|
def test_hmm_report_title_and_answer_follow_presentation_contract(self) -> None:
|
|
source = (Path(__file__).parents[1] / "app.py").read_text(encoding="utf-8")
|
|
tree = ast.parse(source)
|
|
helper = next(
|
|
node
|
|
for node in tree.body
|
|
if isinstance(node, ast.FunctionDef)
|
|
and node.name == "_clean_presentation_title"
|
|
)
|
|
namespace: dict[str, Any] = {
|
|
"Any": Any,
|
|
"html": html,
|
|
"re": re,
|
|
}
|
|
exec(compile(ast.Module(body=[helper], type_ignores=[]), "app.py", "exec"), namespace)
|
|
|
|
title = namespace["_clean_presentation_title"](
|
|
"<b>E1001 팀 포트폴리오</b>"
|
|
)
|
|
|
|
self.assertEqual(title, "E1001 팀 포트폴리오")
|
|
self.assertIn('"title": _clean_presentation_title(title)', source)
|
|
self.assertIn(
|
|
'assistant_message["content"] = presentation_answer',
|
|
source,
|
|
)
|
|
self.assertIn(
|
|
"presentation_answer = _presentation_completion_answer(",
|
|
source,
|
|
)
|
|
self.assertIn(
|
|
"do not reproduce HTML tags, Markdown tables",
|
|
source,
|
|
)
|
|
|
|
def test_chat_context_is_scoped_to_current_selected_user(self) -> None:
|
|
source = (Path(__file__).parents[1] / "app.py").read_text(encoding="utf-8")
|
|
tree = ast.parse(source)
|
|
helper = next(
|
|
node
|
|
for node in tree.body
|
|
if isinstance(node, ast.FunctionDef)
|
|
and node.name == "load_chat_context"
|
|
)
|
|
captured: dict[str, Any] = {}
|
|
|
|
class FakeConnection:
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *_args):
|
|
return False
|
|
|
|
def execute(self, sql, params):
|
|
captured["sql"] = sql
|
|
captured["params"] = params
|
|
return self
|
|
|
|
def fetchall(self):
|
|
return [{"question": "내 담당 선사", "answer": "2건"}]
|
|
|
|
namespace: dict[str, Any] = {
|
|
"CHAT_CONTEXT_TURNS": 8,
|
|
"MAX_CONVERSATION_MESSAGES": 16,
|
|
"_chat_db_connect": FakeConnection,
|
|
"_is_failed_synthesis_answer": lambda _value: False,
|
|
}
|
|
exec(compile(ast.Module(body=[helper], type_ignores=[]), "app.py", "exec"), namespace)
|
|
|
|
messages = namespace["load_chat_context"](
|
|
"conversation-1",
|
|
selected_user_id="E1002",
|
|
)
|
|
|
|
self.assertIn("selected_user_id = ?", captured["sql"])
|
|
self.assertEqual(
|
|
captured["params"],
|
|
("conversation-1", "E1002", "E1002", 8),
|
|
)
|
|
self.assertEqual(messages[0]["content"], "내 담당 선사")
|
|
|
|
def test_standalone_question_uses_current_user_without_prior_context(self) -> None:
|
|
source = (Path(__file__).parents[1] / "app.py").read_text(encoding="utf-8")
|
|
tree = ast.parse(source)
|
|
helpers = [
|
|
node
|
|
for node in tree.body
|
|
if isinstance(node, ast.FunctionDef)
|
|
and node.name in {"_conversation_context", "resolve_standalone_question"}
|
|
]
|
|
captured: dict[str, Any] = {}
|
|
|
|
class FakeClient:
|
|
def complete(self, **kwargs):
|
|
captured.update(kwargs)
|
|
return json.dumps(
|
|
{
|
|
"standalone_question": (
|
|
"E1002 사용자의 담당 선사 최신 실적을 조회해줘"
|
|
)
|
|
},
|
|
ensure_ascii=False,
|
|
)
|
|
|
|
namespace: dict[str, Any] = {
|
|
"Any": Any,
|
|
"Mapping": Mapping,
|
|
"MAX_CONVERSATION_MESSAGES": 16,
|
|
"json": json,
|
|
"resolve_model_profile": lambda _key: SimpleNamespace(
|
|
model_id="model",
|
|
answer_model_region="region",
|
|
answer_model_endpoint="endpoint",
|
|
),
|
|
"build_oci_genai_completion_client": lambda *_args: FakeClient(),
|
|
"temperature_for_model_profile": lambda _profile: 0.0,
|
|
}
|
|
exec(compile(ast.Module(body=helpers, type_ignores=[]), "app.py", "exec"), namespace)
|
|
|
|
rewritten = namespace["resolve_standalone_question"](
|
|
question="내 담당 선사 최신 실적을 리포트로 보여줘",
|
|
messages=[],
|
|
model_profile_key="test",
|
|
selected_user_id="E1002",
|
|
)
|
|
prompt_payload = json.loads(captured["user_prompt"])
|
|
|
|
self.assertTrue(rewritten.startswith("E1002"))
|
|
self.assertEqual(prompt_payload["current_selected_user_id"], "E1002")
|
|
self.assertIn("authoritative", captured["system_prompt"])
|
|
|
|
def test_report_payload_requester_prefers_current_selected_user(self) -> None:
|
|
source = (Path(__file__).parents[1] / "app.py").read_text(encoding="utf-8")
|
|
tree = ast.parse(source)
|
|
helper = next(
|
|
node
|
|
for node in tree.body
|
|
if isinstance(node, ast.FunctionDef)
|
|
and node.name == "_presentation_payload"
|
|
)
|
|
namespace: dict[str, Any] = {
|
|
"Any": Any,
|
|
"Mapping": Mapping,
|
|
"datetime": __import__("datetime").datetime,
|
|
"timezone": __import__("datetime").timezone,
|
|
"re": re,
|
|
"_mcp_structured_rows": lambda _value: [],
|
|
"_normalize_presentation_value": lambda value: value,
|
|
"_clean_presentation_title": lambda value: str(value),
|
|
}
|
|
exec(compile(ast.Module(body=[helper], type_ignores=[]), "app.py", "exec"), namespace)
|
|
|
|
payload = namespace["_presentation_payload"](
|
|
"E1001 팀장 문맥이 남은 질문",
|
|
[],
|
|
title="담당 선사 실적",
|
|
selected_user_id="E1002",
|
|
)
|
|
|
|
self.assertEqual(payload["report"]["requestedBy"], "E1002")
|
|
|
|
def test_hmm_report_data_query_drops_html_format_request(self) -> None:
|
|
source = (Path(__file__).parents[1] / "app.py").read_text(encoding="utf-8")
|
|
tree = ast.parse(source)
|
|
helper = next(
|
|
node
|
|
for node in tree.body
|
|
if isinstance(node, ast.FunctionDef)
|
|
and node.name == "_fallback_presentation_data_query"
|
|
)
|
|
namespace: dict[str, Any] = {"re": re}
|
|
exec(compile(ast.Module(body=[helper], type_ignores=[]), "app.py", "exec"), namespace)
|
|
|
|
query = namespace["_fallback_presentation_data_query"](
|
|
"E1001 팀장의 담당 선사 최신 매출, 매출총이익, 정시 운항률, "
|
|
"위험 등급을 HTML로 보여줘"
|
|
)
|
|
|
|
self.assertNotIn("HTML", query.upper())
|
|
self.assertIn("E1001", query)
|
|
self.assertIn("매출총이익", query)
|
|
self.assertIn("정시 운항률", query)
|
|
self.assertIn("위험 등급", query)
|
|
self.assertIn("_mcp_rows_contain_presentation_markup(mcp_result)", source)
|
|
|
|
personal_query = namespace["_fallback_presentation_data_query"](
|
|
"내 담당 선사와 최신 매출을 리포트로 보여줘"
|
|
)
|
|
self.assertEqual(personal_query, "내 담당 선사와 최신 매출을 보여줘")
|
|
|
|
def test_hmm_report_template_contains_only_dynamic_payload_slot(self) -> None:
|
|
template = (
|
|
Path(__file__).parents[1]
|
|
/ "assets"
|
|
/ "hmm-carrier-performance-report.html"
|
|
).read_text(encoding="utf-8")
|
|
|
|
self.assertEqual(template.count("__REPORT_DATA__"), 1)
|
|
self.assertNotIn("Bluewave Maritime", template)
|
|
self.assertNotIn("Southern Cross Marine", template)
|
|
|
|
def test_hmm_demo_user_presets_reference_runtime_token_only(self) -> None:
|
|
path = Path(__file__).parents[1] / "config" / "vpd_token_presets.json"
|
|
payload = json.loads(path.read_text(encoding="utf-8"))
|
|
presets = payload["presets"]
|
|
|
|
self.assertEqual(payload["version"], 2)
|
|
self.assertEqual({item["user_id"] for item in presets}, {
|
|
"E1001", "E1002", "E1003", "E1005", "E1007"
|
|
})
|
|
self.assertEqual(
|
|
{item["mcp_token_env"] for item in presets},
|
|
{f"HMM_MCP_BEARER_TOKEN_{item['user_id']}" for item in presets},
|
|
)
|
|
self.assertTrue(all("token" not in item for item in presets))
|
|
|
|
def test_duplicate_id_is_rejected(self) -> None:
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
path = Path(temp_dir) / "scenarios.json"
|
|
path.write_text(
|
|
json.dumps(
|
|
{
|
|
"scenarios": [
|
|
{"id": "HR-01", "title": "one", "question": "q1"},
|
|
{"id": "HR-01", "title": "two", "question": "q2"},
|
|
]
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
with self.assertRaises(ScenarioConfigError):
|
|
load_demo_scenarios(path)
|
|
|
|
def test_default_mcp_tool_arguments_follow_discovered_query_schema(self) -> None:
|
|
tool = McpTool(
|
|
name="search_hr_data",
|
|
description="",
|
|
schema={
|
|
"type": "object",
|
|
"properties": {"query": {"type": "string"}},
|
|
"required": ["query"],
|
|
},
|
|
read_only=True,
|
|
)
|
|
|
|
arguments = build_mcp_tool_arguments(
|
|
tool, "직원 E1005의 휴가 신청 내역", 50, preferred_tool="search_hr_data"
|
|
)
|
|
|
|
self.assertEqual(arguments, {"query": "직원 E1005의 휴가 신청 내역"})
|
|
|
|
def test_term_tool_arguments_follow_discovered_term_schema(self) -> None:
|
|
tool = McpTool(
|
|
name="resolve_hr_term",
|
|
description="",
|
|
schema={
|
|
"type": "object",
|
|
"properties": {"term": {"type": "string"}},
|
|
"required": ["term"],
|
|
},
|
|
read_only=True,
|
|
)
|
|
|
|
arguments = build_mcp_tool_arguments(
|
|
tool, "반차", 50, preferred_tool="search_hr_data"
|
|
)
|
|
|
|
self.assertEqual(arguments, {"term": "반차"})
|
|
|
|
def test_status_result_policy_text_is_preserved_as_answer_evidence(self) -> None:
|
|
result = {
|
|
"status": "success",
|
|
"result": (
|
|
"HR_POLICY_SEARCH_RESULT\n"
|
|
"EVIDENCE|file=KR_Leave_Policy.pdf|chunk=13|text=이월 기준"
|
|
),
|
|
}
|
|
|
|
summary = status_result_summary(result, excerpt_chars=40)
|
|
evidence = status_result_evidence(result)
|
|
|
|
self.assertEqual(summary["status"], "success")
|
|
self.assertGreater(summary["result_chars"], 40)
|
|
self.assertIn("KR_Leave_Policy.pdf", evidence["result"])
|
|
self.assertTrue(has_actionable_text_result(result))
|
|
|
|
def test_no_data_text_is_not_actionable(self) -> None:
|
|
self.assertFalse(
|
|
has_actionable_text_result({"status": "success", "result": "No data found"})
|
|
)
|
|
|
|
@unittest.skipIf(_load_console_query_helpers() is None, "Streamlit runtime is optional")
|
|
def test_policy_query_does_not_include_demo_user_context(self) -> None:
|
|
prepare = _load_console_query_helpers()
|
|
assert prepare is not None
|
|
tool = McpTool(
|
|
name="search_hr_policy",
|
|
description="Search policy documents",
|
|
schema={"properties": {"query": {"type": "string"}}},
|
|
read_only=True,
|
|
)
|
|
|
|
query = prepare(
|
|
question="연차 휴가 이월 기준과 제한을 알려줘",
|
|
tool=tool,
|
|
model_profile_key="gpt54_mini_oci",
|
|
selected_user_id="E1001",
|
|
selected_user_role="HR Team Manager",
|
|
selected_user_team="HMM HR Demo Team",
|
|
selected_user_scope="팀원 6명 관리",
|
|
)
|
|
|
|
self.assertEqual(query, "연차 휴가 이월 기준과 제한을 알려줘")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|