from __future__ import annotations import ast import json from pathlib import Path import tempfile 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-01", "FED-02", "FED-03"} & set(by_id), ) self.assertTrue( all("선사" in by_id[scenario_id].question for scenario_id in ( "FED-01", "FED-02", "FED-03" )) ) 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_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.assertTrue(all(item["mcp_token_env"] == "HMM_MCP_BEARER_TOKEN" 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()