from __future__ import annotations """AGY headful 브라우저 수집 RED. 사용자 요구: AGY가 headful Chrome을 사람처럼 조작해 nedrug 엑셀을 내려받는 경로가 있어야 한다. 요약용 AI 옵션(agy.enabled)과 수집용 headful AGY는 별개다. """ import json from datetime import datetime from pathlib import Path from typing import Any import pytest from dmf_crawler.config import AgyCfg, Config, SourceCfg from dmf_crawler.models import ( AgyEnvelope, AgyStatus, COL_APPLICANT, COL_INGREDIENT, COL_PERMIT_DATE, COL_PERMIT_NO, TriggerKind, ) HEADERS = ["등록번호", "성분명", "업체명", "제조소명", "제조소소재지", "제조국가명", "발급일자"] ROW = ["20260101-1-A-1-1", "테스트성분", "테스트제약", "테스트제조소", "서울", "대한민국", "2026-01-01"] CCBAC03_HEADERS = [ "대상의약품", "등록번호", "성분명", "신청인", "제조소명", "제조소소재지", "제조국가", "최초등록일자", "최종변경일자", "최종연차보고년도", "취소/취하구분", "취소/취하일자", "문서번호", "연계심사문서번호", ] CCBAC03_ROW = [ "신물질", "20260101-1-A-1-1", "테스트성분", "테스트신청인", "테스트제조소", "서울", "대한민국", "2026-01-01", "2026-02-03", "2026", "", "", "DOC-1", "LINK-1", ] def _write_xlsx(path: Path, headers: list[str], row: list[str]) -> None: import xlsxwriter wb = xlsxwriter.Workbook(str(path), {"strings_to_urls": False}) ws = wb.add_worksheet("전체") for col, value in enumerate(headers): ws.write(0, col, value) for col, value in enumerate(row): ws.write(1, col, value) wb.close() def _write_minimal_dmf_xlsx(path: Path) -> None: _write_xlsx(path, HEADERS, ROW) def _write_ccbac03_dmf_xlsx(path: Path) -> None: _write_xlsx(path, CCBAC03_HEADERS, CCBAC03_ROW) def test_headful_browser_prompt_contract_forbids_headless_http_and_secrets(tmp_path: Path) -> None: from dmf_crawler.agy import browser_collect prompt = browser_collect.build_headful_prompt(download_dir=tmp_path, run_id="run_headful_contract") assert "headful" in prompt.lower() or "visible Chrome" in prompt assert "Chrome" in prompt assert "chrome-devtools" in prompt assert "curl" in prompt and "금지" in prompt assert "headless" in prompt and "금지" in prompt assert "serviceKey" not in prompt assert "비밀번호" in prompt and "프롬프트" in prompt and "넣지" in prompt assert "crdownload" in prompt assert "xlsx" in prompt.lower() assert "JSON" in prompt def test_headful_browser_prompt_blocks_wrong_review_result_export(tmp_path: Path) -> None: from dmf_crawler.agy import browser_collect prompt = browser_collect.build_headful_prompt(download_dir=tmp_path, run_id="run_wrong_export_guard") assert "CCBAC03" in prompt assert "/pbp/CCBAC03/getExcel" in prompt assert "원료의약품 등록(DMF) 현황" in prompt assert "의약품등심사결과공개" in prompt and "금지" in prompt for header in HEADERS: assert header in prompt assert "헤더" in prompt and "확인" in prompt assert "삭제" in prompt and "다시" in prompt def test_headful_browser_collect_validates_download_and_returns_raw_records( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: from dmf_crawler.agy import browser_collect download_dir = tmp_path / "downloads" download_dir.mkdir() xlsx_path = download_dir / "dmf_download.xlsx" _write_minimal_dmf_xlsx(xlsx_path) captured: dict[str, Any] = {} def fake_run_agy(cfg: Any, prompt: str, run_id: str, log_dir: Path) -> AgyEnvelope: captured["cfg"] = cfg captured["prompt"] = prompt captured["run_id"] = run_id captured["log_dir"] = log_dir return AgyEnvelope( status=AgyStatus.SUCCESS, response=json.dumps({"status": "ok", "downloaded_xlsx": str(xlsx_path)}, ensure_ascii=False), exit_code=0, ) monkeypatch.setattr(browser_collect.client, "run_agy", fake_run_agy) cfg = Config().with_overrides( source=SourceCfg(mode="browser"), # AI 요약은 꺼져 있어도 수집용 AGY는 source.mode=browser 때문에 동작해야 한다. agy=AgyCfg(enabled=False, binary_path=str(tmp_path / "agy.exe")), ) result = browser_collect.fetch_all(cfg, run_id="run_headful_ok", log_dir=tmp_path / "logs", download_dir=download_dir) assert result.records_received == 1 assert result.total_count_reported == 1 assert result.body_signature_ok is True assert result.archive_dir == download_dir assert result.endpoint == "agy:headful-browser" assert result.records[0].get(COL_PERMIT_NO) == ROW[0] assert result.records[0].get(COL_INGREDIENT) == ROW[1] assert "headless" in captured["prompt"].lower() assert "금지" in captured["prompt"] def test_headful_browser_accepts_public_ccbac03_xlsx_headers(tmp_path: Path) -> None: from dmf_crawler.agy import browser_collect from dmf_crawler.importers.prototype_xlsx import read_rows download_dir = tmp_path / "downloads" download_dir.mkdir() xlsx_path = download_dir / "CCBAC03_DMF.xlsx" _write_ccbac03_dmf_xlsx(xlsx_path) _target, row_count, _sha = browser_collect.validate_downloaded_xlsx(xlsx_path, download_dir=download_dir) _run_date, records = read_rows(xlsx_path) assert row_count == 1 assert records[0].get(COL_PERMIT_NO) == CCBAC03_ROW[1] assert records[0].get(COL_INGREDIENT) == CCBAC03_ROW[2] assert records[0].get(COL_APPLICANT) == CCBAC03_ROW[3] assert records[0].get(COL_PERMIT_DATE) == CCBAC03_ROW[8] def test_headful_browser_collect_rejects_incomplete_downloads(tmp_path: Path) -> None: from dmf_crawler.agy import browser_collect download_dir = tmp_path / "downloads" download_dir.mkdir() (download_dir / "dmf.xlsx.crdownload").write_bytes(b"partial") cfg = Config().with_overrides(source=SourceCfg(mode="browser")) with pytest.raises(browser_collect.HeadfulDownloadError, match="다운로드가 아직 끝나지 않았습니다"): browser_collect.validate_downloaded_xlsx(download_dir / "dmf.xlsx", download_dir=download_dir) def test_headful_browser_collect_reports_execute_url_permission_denial_from_stderr( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: from dmf_crawler.agy import browser_collect stderr = tmp_path / "agy.stderr.log" stderr.write_text( 'jetski: no output produced — a tool required the "execute_url" permission that headless mode cannot prompt for, so it was auto-denied.', encoding="utf-8", ) def fake_run_agy(cfg: Any, prompt: str, run_id: str, log_dir: Path) -> AgyEnvelope: return AgyEnvelope(status=AgyStatus.SUCCESS, response="", exit_code=0, stderr_path=stderr) monkeypatch.setattr(browser_collect.client, "run_agy", fake_run_agy) cfg = Config().with_overrides(source=SourceCfg(mode="browser")) with pytest.raises(browser_collect.HeadfulBrowserError, match="nedrug URL 실행 권한"): browser_collect.fetch_all(cfg, run_id="run_execute_denied", log_dir=tmp_path / "logs", download_dir=tmp_path / "downloads") def test_headful_browser_collect_reports_mcp_permission_denial_from_stderr( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: from dmf_crawler.agy import browser_collect stderr = tmp_path / "agy.stderr.log" stderr.write_text( 'jetski: no output produced — a tool required the "mcp" permission that headless mode cannot prompt for, so it was auto-denied.', encoding="utf-8", ) def fake_run_agy(cfg: Any, prompt: str, run_id: str, log_dir: Path) -> AgyEnvelope: return AgyEnvelope(status=AgyStatus.SUCCESS, response="", exit_code=0, stderr_path=stderr) monkeypatch.setattr(browser_collect.client, "run_agy", fake_run_agy) cfg = Config().with_overrides(source=SourceCfg(mode="browser")) with pytest.raises(browser_collect.HeadfulBrowserError, match="chrome-devtools MCP 권한"): browser_collect.fetch_all(cfg, run_id="run_mcp_denied", log_dir=tmp_path / "logs", download_dir=tmp_path / "downloads") def test_headful_mcp_allow_rule_is_scoped_and_does_not_use_dangerous_skip(tmp_path: Path) -> None: from dmf_crawler.agy import browser_collect config_path = tmp_path / "config.json" config_path.write_text( json.dumps( { "permissions": {"allow": ["command(start)"]}, "userSettings": {"globalPermissionGrants": {"allow": ["command(npm run test)"]}}, }, ensure_ascii=False, ), encoding="utf-8", ) changed = browser_collect.ensure_headful_mcp_permission(config_path) changed_again = browser_collect.ensure_headful_mcp_permission(config_path) parsed = json.loads(config_path.read_text(encoding="utf-8")) top_allow = parsed["permissions"]["allow"] global_allow = parsed["userSettings"]["globalPermissionGrants"]["allow"] assert changed is True assert changed_again is False assert "mcp(chrome-devtools/*)" in top_allow assert "mcp(chrome-devtools/*)" in global_allow assert "execute_url(nedrug.mfds.go.kr)" in top_allow assert "execute_url(nedrug.mfds.go.kr)" in global_allow assert "mcp(*)" not in top_allow assert "mcp(*)" not in global_allow assert "execute_url(*)" not in top_allow assert "execute_url(*)" not in global_allow assert not any("dangerously-skip-permissions" in item for item in top_allow + global_allow) def test_source_auto_without_api_key_routes_to_headful_browser(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: from dmf_crawler import pipeline, secrets_dpapi cfg = Config().with_overrides(source=SourceCfg(mode="auto"), agy=AgyCfg(enabled=False)) calls: list[str] = [] monkeypatch.setattr(secrets_dpapi, "load_service_key", lambda: None) monkeypatch.setattr(pipeline, "_fetch_with_headful_browser", lambda ctx: calls.append(ctx.run_id)) ctx = pipeline.RunContext( run_id="run_auto_headful", run_date="2026-09-03", started_at=datetime.now().astimezone(), trigger=TriggerKind.MANUAL, cfg=cfg, log_dir=tmp_path / "logs", app_version="test", ) pipeline.stage_fetch(ctx) assert calls == ["run_auto_headful"] def test_source_browser_mode_routes_to_headful_even_when_api_key_exists(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: from dmf_crawler import pipeline, secrets_dpapi cfg = Config().with_overrides(source=SourceCfg(mode="browser"), agy=AgyCfg(enabled=False)) calls: list[str] = [] monkeypatch.setattr(secrets_dpapi, "load_service_key", lambda: "dummy-key") monkeypatch.setattr(pipeline, "_fetch_with_headful_browser", lambda ctx: calls.append(ctx.run_id)) ctx = pipeline.RunContext( run_id="run_browser_forced", run_date="2026-09-03", started_at=datetime.now().astimezone(), trigger=TriggerKind.MANUAL, cfg=cfg, log_dir=tmp_path / "logs", app_version="test", ) pipeline.stage_fetch(ctx) assert calls == ["run_browser_forced"]