# FILE: _St/WS8771/test_ws8771_module.py | ROLE: WS8771 정적·단위 회귀 테스트
from __future__ import annotations

import ast
import json
import os
import re
import shutil
from pathlib import Path

ROOT = Path(__file__).resolve().parent
SOURCE = (ROOT / 'ws8771_main.py').read_text(encoding='utf-8')
CONFIG = json.loads((ROOT / 'ws8771_config.json').read_text(encoding='utf-8'))


def test_ws_version_and_direct_finalize_handlers() -> None:
    assert 'WS_BUILD_VERSION = "2.3.8.324"' in SOURCE
    assert 'WS_BUILD_TAG = "v324"' in SOURCE
    assert 'WS_LABEL = "WS v324"' in SOURCE
    assert 'ws8771_intentional_stop.json' not in SOURCE
    assert 'write_ws_intentional_stop_marker' not in SOURCE
    assert 'def handle_ig_media_exists' in SOURCE
    assert 'def handle_ig_media_download_direct' in SOURCE
    assert 'def handle_ig_media_downloaded' in SOURCE
    assert 'def generate_account_html_report' in SOURCE
    assert 'def handle_ig_report_rebuild_all' in SOURCE
    assert 'if act == "ig_report_rebuild_all"' in SOURCE



def test_hourly_central_statistics_sync_handler() -> None:
    for token in [
        'def handle_ig_observation_sync_hour(self, req):',
        '"hour_bucket": hour_bucket',
        '"share_count": 0',
        'yellow_api_call(self.cfg, "media_stat_hour_upsert"',
        'yellow_api_call(self.cfg, "media_stat_hour_list"',
        '통계 1시간 연동 account=',
        'if act == "ig_observation_sync_hour"',
    ]:
        assert token in SOURCE, token
    actions = set((CONFIG.get('thread_queue_mode') or {}).get('actions') or [])
    assert 'ig_observation_sync_hour' in actions


def test_physical_mouse_modules_are_removed() -> None:
    for name in [
        'ws8771_ig_auto_click_download.py',
        'ws8771_ig_auto_download.py', 'ws8771_origin_crosscheck.py'
    ]:
        assert not (ROOT / name).exists(), name
    forbidden = [
        'MOUSEEVENTF_LEFTDOWN', 'mouse_left_click(', 'mouse_move_smooth(',
        'SetCursorPos', 'GetCursorPos', 'WindowFromPoint',
        'ig_rtc_auto_click_once', 'ig_rtc_mouse_sync', 'ig_rtc_hover_status',
        'hert8772position', 'ig_dom_calibration'
    ]
    for token in forbidden:
        assert token not in SOURCE, token


def test_config_has_no_retired_mouse_state_and_auto_enter_is_enabled() -> None:
    for key in ['user_mouse_idle_resume_sec', 'user_mouse_move_threshold_px', 'ig_dom_calibration', 'hert8772position', 'mouse_action_rest_sec']:
        assert key not in CONFIG
    actions = set((CONFIG.get('thread_queue_mode') or {}).get('actions') or [])
    assert 'ig_rtc_auto_click_once' not in actions
    assert {'ig_media_exists', 'ig_media_download_direct', 'ig_media_downloaded', 'ig_observation_sync_hour', 'ig_report_rebuild_all'}.issubset(actions)
    assert 'def handle_ig_rtc_enter_auto_arm' in SOURCE
    assert CONFIG['enabled'] is True
    assert CONFIG['enter_mode'] == 'auto'
    assert CONFIG['try_enter_after_idok'] is True
    assert CONFIG['match_contains'] is True


def test_instagram_save_dialog_enter_flow_is_wired() -> None:
    for token in [
        'def instagram_download_dialog_match',
        'self.auto_enter_expected = {',
        'self.arm_auto_enter(reason, data)',
        'find_save_dialog_filename_edit(hwnd, expected)',
        'ig_match, ig_match_reason = instagram_download_dialog_match(filename, expected)',
        'if auto_unarmed and not ig_match:',
        'auto_accept_save_dialog(hwnd, self.cfg, expected)',
        'press_enter_key()',
    ]:
        assert token in SOURCE, token
    # Instagram 파일 패턴만 자동 확정하고 다른 프로그램 Save As는 건드리지 않는다.
    for token in ['_No000_', '_No없음_', 'ig_filename_pattern', 'filename_pattern_mismatch']:
        assert token in SOURCE, token

    tree = ast.parse(SOURCE)
    node = next(n for n in tree.body if isinstance(n, ast.FunctionDef) and n.name == 'instagram_download_dialog_match')
    scope = {'os': os, 're': re}
    exec(compile(ast.Module(body=[node], type_ignores=[]), str(ROOT / 'ws8771_main.py'), 'exec'), scope)
    match = scope['instagram_download_dialog_match']
    assert match('owner_No000_Da0WTq-OosH.mp4', {}) == (True, 'ig_filename_pattern')
    assert match('owner_No003_Da0WTq-OosH.jpg', {'shortcode': 'Da0WTq-OosH'}) == (True, 'expected_shortcode')
    assert match('report.xlsx', {}) == (False, 'not_media_extension')
    assert match('vacation.jpg', {}) == (False, 'filename_pattern_mismatch')


def test_fixed_stdown_instagram_account_storage(tmp_path: Path) -> None:
    for token in [
        'derive_stdown_root(ST_ROOT) / "instagram"',
        'def relocate_download_to_expected_folder',
        'download_relocate_failed',
        'target_path_enter',
        'enter_relocate_fallback',
        'write_control_text(edit_hwnd, target_path)',
    ]:
        assert token in SOURCE, token

    tree = ast.parse(SOURCE)
    node = next(n for n in tree.body if isinstance(n, ast.FunctionDef) and n.name == 'expected_download_folder')
    scope = {
        'Path': Path,
        'ST_ROOT': Path('/tmp/bundle/_St'),
        'derive_stdown_root': lambda root: Path(root).parent / '_StDown',
        'safe_path_part': lambda value, fallback='unknown': str(value or fallback),
    }
    exec(compile(ast.Module(body=[node], type_ignores=[]), str(ROOT / 'ws8771_main.py'), 'exec'), scope)
    expected = scope['expected_download_folder']({'account': 'account_a'})
    assert expected.replace('\\', '/').endswith('/_StDown/instagram/account_a')
    merged = scope['expected_download_folder']({'account': 'account_a', 'account_folder_merge': True})
    assert merged.replace('\\', '/').endswith('/_StDown/instagram/account_a')

    relocate_node = next(n for n in tree.body if isinstance(n, ast.FunctionDef) and n.name == 'relocate_download_to_expected_folder')
    target_root = tmp_path / '_StDown' / 'instagram' / 'account_a'
    relocate_scope = {
        'Path': Path,
        'os': os,
        'shutil': shutil,
        'expected_download_folder': lambda data: str(target_root),
        'file_hashes': lambda path: ('md5', 'sha256', Path(path).stat().st_size),
    }
    exec(compile(ast.Module(body=[relocate_node], type_ignores=[]), str(ROOT / 'ws8771_main.py'), 'exec'), relocate_scope)
    source = tmp_path / 'Downloads' / 'owner_No001_CODE1.mp4'
    source.parent.mkdir(parents=True)
    source.write_bytes(b'media')
    moved_path, moved = relocate_scope['relocate_download_to_expected_folder'](source, {'account': 'account_a'})
    assert moved is True
    assert Path(moved_path) == target_root / source.name
    assert Path(moved_path).read_bytes() == b'media'
    assert not source.exists()


def test_ws_records_shared_media_saved_after_file_finalize() -> None:
    for token in [
        "from Shared.storage import canonical_instagram_post_url, derive_stdown_root, record_downloaded_file",
        "def record_shared_media_saved",
        "canonical_instagram_post_url(",
        "MEDIA_SAVED_OK",
        "storage_ok",
        "client_event_id",
    ]:
        assert token in SOURCE, token


def test_save_dialog_state_callback_exists_for_ui_diagnostics() -> None:
    assert 'def save_dialog_state(self, is_open, source, dialogs)' in SOURCE
    assert 'save_dialog_state' in SOURCE
    # v230 field omission caused repeated Tk callback exceptions; the callback must now be concrete.
    tree = ast.parse(SOURCE)
    floating = next(n for n in tree.body if isinstance(n, ast.ClassDef) and n.name == 'FloatingApp')
    assert any(isinstance(n, ast.FunctionDef) and n.name == 'save_dialog_state' for n in floating.body)


def test_ws_direct_download_streams_to_target(tmp_path: Path) -> None:
    assert 'def download_url_to_file' in SOURCE
    assert 'DIRECT_DOWNLOAD_BEGIN' in SOURCE
    assert 'DIRECT_DOWNLOAD_OK' in SOURCE

    tree = ast.parse(SOURCE)
    node = next(n for n in tree.body if isinstance(n, ast.FunctionDef) and n.name == 'download_url_to_file')
    import threading
    import urllib.request
    scope = {'Path': Path, 'os': os, 'threading': threading, 'urllib': __import__('urllib')}
    exec(compile(ast.Module(body=[node], type_ignores=[]), str(ROOT / 'ws8771_main.py'), 'exec'), scope)

    source = tmp_path / 'source.mp4'
    source.write_bytes(b'instagram-media-bytes')
    target = tmp_path / '_StDown' / 'instagram' / 'account_a' / 'source.mp4'
    result = scope['download_url_to_file'](source.as_uri(), target, headers={}, timeout=5)
    assert result['ok'] is True
    assert Path(result['file_path']) == target.resolve()
    assert target.read_bytes() == b'instagram-media-bytes'
    assert not list(target.parent.glob('*.part'))


def test_feature_check_log_is_korean_concise_and_deduplicated() -> None:
    for token in [
        'feature_check_mode = bool(data.get("feature_check_mode"))',
        'check_id = str(data.get("check_id") or "").strip()',
        'check_status = str(data.get("check_status") or "").strip()',
        'check_message_ko = str(data.get("check_message_ko") or "").strip()',
        'def project_progress_check(self, feature_name, check_id, status, message_ko, detail=None):',
        '_project_progress_seen',
        '_append_project_progress_line(judgement)',
        '검색어=*** 상세로그 ***',
        'ws_project_progress_log',
        'ws_failure_log',
        'elif (not feature_check_mode) and (stage or decision or action):',
    ]:
        assert token in SOURCE, token


def test_project_progress_and_failure_logs_are_linked(tmp_path: Path, capsys) -> None:
    tree = ast.parse(SOURCE)
    function_names = {
        'project_log_context_snapshot',
        '_append_project_progress_line',
        '_project_failure_keyword',
        '_write_project_failure_excerpt',
        '_write_run_detail_marker',
        '_prune_python_logs',
    }
    nodes = [n for n in tree.body if isinstance(n, ast.FunctionDef) and n.name in function_names]
    floating = next(n for n in tree.body if isinstance(n, ast.ClassDef) and n.name == 'FloatingApp')
    method = next(n for n in floating.body if isinstance(n, ast.FunctionDef) and n.name == 'project_progress_check')

    import threading
    from datetime import datetime
    from types import SimpleNamespace

    progress_path = tmp_path / 'ws_project_progress_log_test.log'
    run_path = tmp_path / 'ws_run_log_test.log'
    progress_file = progress_path.open('a', encoding='utf-8', buffering=1)
    scope = {
        'threading': threading,
        'datetime': datetime,
        'os': os,
        '_PROJECT_LOG_LOCK': threading.RLock(),
        '_PROJECT_LOG_CONTEXT': {
            'enabled': True,
            'run_id': 'TEST_RUN',
            'run_log': str(run_path),
            'error_log': str(tmp_path / 'ws_err_log_test.log'),
            'project_progress_log': str(progress_path),
            'failure_dir': str(tmp_path),
            'failure_prefix': 'ws_failure_log',
        },
        '_PROJECT_PROGRESS_FILE': progress_file,
        'now_text': lambda: '2026-07-22 12:00:00',
        'safe_path_part': lambda value, fallback='x': re.sub(r'[^0-9A-Za-z._-]+', '_', str(value or fallback)).strip('_') or fallback,
    }
    exec(compile(ast.Module(body=nodes + [method], type_ignores=[]), str(ROOT / 'ws8771_main.py'), 'exec'), scope)

    fake = SimpleNamespace()
    progress = scope['project_progress_check']
    progress(fake, '자동다운로드', 'AD-01', '[체크완료]', '자동 시작 활성화', {'cycle_id': 'run_1'})
    progress(fake, '자동다운로드', 'AD-01', '[체크완료]', '자동 시작 활성화', {'cycle_id': 'run_1'})
    progress(fake, '자동다운로드', 'AD-03', '[체크실패]', '대상 불일치', {'cycle_id': 'No001_CODE', 'error_code': 'target_mismatch'})
    progress(fake, '자동다운로드', 'AD-06', '[체크완료]', '파일 저장 완료', {'cycle_id': 'No001_CODE'})
    result = progress(fake, '서버등록', 'AD-06-S01', '[체크실패]', 'HTTP 500 / 다음점검=media_insert', {
        'cycle_id': 'No001_CODE',
        'shortcode': 'CODE',
        'asset_kind': 'video',
        'error_code': 'HTTP 500',
        'next_check': 'media_insert',
        'blocking': False,
    })
    progress_file.close()

    text = progress_path.read_text(encoding='utf-8')
    assert text.count('[자동다운로드][AD-01][체크완료]') == 1
    assert '[자동다운로드][AD-03][복구완료]' in text
    assert '[서버등록][AD-06-S01][체크실패]' in text
    assert '다음단계=로컬처리계속·실패항목별도추적' in text
    assert f'상세로그={run_path}' in text
    assert '검색어=*** 상세로그 *** PP-AD-06-S01-' in text
    failure_path = Path(result['failure_log'])
    assert failure_path.exists()
    failure_text = failure_path.read_text(encoding='utf-8')
    assert f'원본 상세로그: {run_path}' in failure_text
    assert '검색어: *** 상세로그 *** PP-AD-06-S01-' in failure_text
    captured = capsys.readouterr().out
    assert '*** 상세로그 *** PP-AD-06-S01-' in captured


def test_yellow_http_error_body_is_preserved(monkeypatch) -> None:
    import io
    import importlib.util
    import urllib.error

    spec = importlib.util.spec_from_file_location('ws8771_ig_sort_test', ROOT / 'ws8771_ig_sort.py')
    assert spec is not None and spec.loader is not None
    ws8771_ig_sort = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(ws8771_ig_sort)

    body = json.dumps({
        'ok': 0,
        'error': 'schema_mismatch',
        'error_stage': 'collect_event_schema',
        'table': 'ig_media_collect_event',
        'missing_columns': ['pass_id', 'rank_label'],
        'message': 'required columns are missing',
        'sql_error_code': 1054,
    }, ensure_ascii=False).encode('utf-8')

    def fail_urlopen(req, timeout=0):
        raise urllib.error.HTTPError(
            req.full_url,
            500,
            'Internal Server Error',
            hdrs=None,
            fp=io.BytesIO(body),
        )

    monkeypatch.setattr(ws8771_ig_sort.urllib.request, 'urlopen', fail_urlopen)
    logs = []
    result = ws8771_ig_sort.call_api(
        {'yellow_api': {'enabled': True, 'url': 'https://example.invalid/yellow.php', 'token': 'secret-token'}},
        'media_insert',
        {'shortcode': 'CODE1', 'asset_kind': 'video'},
        logger=logs.append,
    )

    assert result['ok'] is False
    assert result['http_status'] == 500
    assert result['error'] == 'schema_mismatch'
    assert result['error_stage'] == 'collect_event_schema'
    assert result['table'] == 'ig_media_collect_event'
    assert result['missing_columns'] == ['pass_id', 'rank_label']
    assert result['sql_error_code'] == 1054
    assert any('stage=collect_event_schema' in line for line in logs)
    assert all('secret-token' not in line for line in logs)


def test_server_failure_progress_contains_schema_diagnostics() -> None:
    for token in [
        '"http_status": detail_doc.get("http_status", "")',
        '"error_stage": detail_doc.get("error_stage", "")',
        '"server_message": detail_doc.get("server_message", "")',
        '"missing_columns": detail_doc.get("missing_columns", "")',
        '"http_status": yellow_res.get("http_status", "")',
        '"error_stage": str(yellow_res.get("error_stage") or "")',
    ]:
        assert token in SOURCE, token


def test_html_report_uses_published_time_fixed_thumbnail_and_full_view(tmp_path: Path) -> None:
    report_source = (ROOT / 'ws8771_html_report.py').read_text(encoding='utf-8')
    assert '<dt>게시 시각</dt>' in report_source
    assert '<dt>저장 시각</dt>' not in report_source
    assert 'item.get("upload_at") or item.get("published_at")' in report_source
    assert 'repeat(auto-fill,320px)' in report_source
    assert '.post-card{{width:320px' in report_source
    assert 'aspect-ratio:1/1' in report_source
    assert '>전체보기</a>' in report_source
    assert 'http://127.0.0.1:8780/' in report_source
    assert 'def _report_account_name' in report_source
    assert 'def rebuild_account_reports' in report_source
    assert 'REPORT_VERSION = 4' in report_source
    assert '"top_n": normalize_int(meta.get("top_n"), 0)' in SOURCE
    assert 'meta.get("top_n") or meta.get("rank_total")' not in SOURCE

    module_scope: dict[str, object] = {}
    exec(compile(report_source, str(ROOT / 'ws8771_html_report.py'), 'exec'), module_scope)
    display_time = module_scope['_display_time']
    assert display_time({'upload_at': '2026-07-20 12:34:56', 'downloaded_at': '2026-07-25 14:00:00'}) == '2026-07-20 12:34:56'
    assert display_time({'published_at': '2026-07-19 01:02:03', 'collected_at': '2026-07-25 14:00:00'}) == '2026-07-19 01:02:03'
    assert display_time({'downloaded_at': '2026-07-25 14:00:00'}) == ''


def test_report_rebuild_all_generates_account_specific_analyze_link(tmp_path: Path) -> None:
    import importlib.util

    module_path = ROOT / 'ws8771_html_report.py'
    spec = importlib.util.spec_from_file_location('ws8771_html_report_v249_test', module_path)
    assert spec and spec.loader
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)

    root = tmp_path / 'instagram'
    media = root / 'lilys_home'
    media.mkdir(parents=True)
    (media / 'sample.jpg').write_bytes(b'jpg')
    doc = {
        'folder_name': 'lilys_home',
        'items': {
            'CODE1|image|0': {
                'shortcode': 'CODE1',
                'asset_kind': 'image',
                'file_name': 'sample.jpg',
                'upload_at': '2026-07-20 12:34:56',
            }
        },
    }
    json_path = root / 'lilys_home.json'
    json_path.write_text(json.dumps(doc, ensure_ascii=False), encoding='utf-8')

    result = module.rebuild_account_reports(str(root))
    assert result['ok'] is True
    assert result['requested_count'] == 1
    assert result['success_count'] == 1
    html_text = (root / 'lilys_home.html').read_text(encoding='utf-8')
    legacy_text = (media / 'index.htm').read_text(encoding='utf-8')
    expected = 'http://127.0.0.1:8780/?account=lilys_home'
    assert expected in html_text
    assert expected in legacy_text


def test_observation_hour_sync_uses_atomic_hour_api() -> None:
    text = SOURCE
    for token in (
        'yellow_api_call(self.cfg, "media_stat_hour_upsert"',
        '"items_json": json.dumps(normalized',
        'yellow_api_call(self.cfg, "media_stat_hour_list"',
        'account_scope_mismatch',
        'saved_count != len(normalized)',
        'duplicate_shortcode',
        'missing_shortcodes = sorted(expected_shortcodes - received_shortcodes)',
    ):
        assert token in text, token


def test_injection_download_metadata_sort_fallback_and_merge_preservation() -> None:
    tree = ast.parse(SOURCE)
    wanted = {'download_meta_sort', 'download_meta_metric', 'merge_download_item_metadata'}
    nodes = [n for n in tree.body if isinstance(n, ast.FunctionDef) and n.name in wanted]
    scope = {'normalize_int': lambda value, default=0: int(value or default)}
    exec(compile(ast.Module(body=nodes, type_ignores=[]), str(ROOT / 'ws8771_main.py'), 'exec'), scope)
    sort_payload = scope['download_meta_sort']({'sort_snapshot': {'views': 321, 'likes': 12, 'comments': 3, 'reposts': 2}})
    assert sort_payload['views'] == 321
    current = {'shortcode': 'CODE1', 'views': 0, 'likes': 0, 'comments': 0, 'reposts': 0, 'thumbnail_url': '', 'sort': {}}
    previous = {'shortcode': 'CODE1', 'views': 111, 'likes': 9, 'comments': 2, 'reposts': 1, 'thumbnail_url': 'old.jpg', 'sort': {'views': 111}}
    missing_meta = {'shortcode': 'CODE1', 'metric_presence': {'views': False, 'likes': False, 'comments': False, 'reposts': False}}
    merged = scope['merge_download_item_metadata'](previous, current, missing_meta)
    assert merged['views'] == 111
    assert merged['likes'] == 9
    assert merged['thumbnail_url'] == 'old.jpg'
    explicit = {'shortcode': 'CODE1', 'views': 0, 'likes': 0, 'comments': 0, 'reposts': 0, 'metric_presence': {'views': True, 'likes': True, 'comments': True, 'reposts': True}, 'sort': {'views': 0, 'likes': 0, 'comments': 0, 'reposts': 0}}
    merged_zero = scope['merge_download_item_metadata'](previous, current, explicit)
    assert merged_zero['views'] == 0
    assert merged_zero['likes'] == 0


def test_duplicate_metadata_refresh_merges_current_shortcode_stats() -> None:
    for token in [
        'incoming = {',
        'merged_item = merge_download_item_metadata(it, incoming, meta)',
        'download_meta_metric(meta, sort, metric_name)',
        'sort = download_meta_sort(meta)',
        '"caption": str(meta.get("caption") or "")',
    ]:
        assert token in SOURCE, token


def test_v310_metadata_collect_action_and_json_record(tmp_path: Path) -> None:
    actions = set((CONFIG.get('thread_queue_mode') or {}).get('actions') or [])
    assert 'ig_media_metadata_collect' in actions
    for token in [
        'def media_request_status_fields(meta, asset_kind=""):',
        'def upsert_folder_metadata(folder, meta):',
        'def handle_ig_media_metadata_collect(self, req):',
        'if act == "ig_media_metadata_collect":',
        '"download_status": "metadata_collected"',
        '"direct_media_urls"',
    ]:
        assert token in SOURCE, token

    tree = ast.parse(SOURCE)
    wanted = {'media_request_status_fields', 'upsert_folder_metadata'}
    nodes = [node for node in tree.body if isinstance(node, ast.FunctionDef) and node.name in wanted]
    assert {node.name for node in nodes} == wanted

    def read_json_file(path: str, fallback: dict) -> dict:
        p = Path(path)
        return json.loads(p.read_text(encoding='utf-8')) if p.exists() else dict(fallback)

    def write_json_file(path: str, value: dict) -> None:
        Path(path).write_text(json.dumps(value, ensure_ascii=False, indent=2), encoding='utf-8')

    scope: dict[str, object] = {
        'os': os,
        'RUNTIME_USER': 'tester',
        'normalize_int': lambda value, default=0: int(value or default),
        'download_meta_sort': lambda meta: dict(meta.get('sort') or {}),
        'download_meta_metric': lambda meta, sort, name: (int(meta.get(name, sort.get(name, 0)) or 0), True),
        'canonical_instagram_post_url': lambda shortcode, *args, **kwargs: f'https://www.instagram.com/p/{shortcode}/',
        'now_text': lambda: '2026-08-01 20:00:00',
        'default_runtime_user': lambda: 'tester',
        'final_folder_index_path': lambda folder: str(Path(folder).parent / f'{Path(folder).name}.json'),
        'read_json_file': read_json_file,
        'write_json_file': write_json_file,
        'merge_download_item_metadata': lambda previous, current, meta: {**previous, **current},
        'safe_path_part': lambda value, fallback='instagram': str(value or fallback),
    }
    exec(compile(ast.Module(body=nodes, type_ignores=[]), str(ROOT / 'ws8771_main.py'), 'exec'), scope)
    folder = tmp_path / 'instagram' / 'tester'
    meta = {
        'shortcode': 'CODE277',
        'owner_username': 'tester',
        'caption': '본문 유지',
        'thumbnail_url': 'https://cdn.example/image.jpg',
        'video_url': 'https://cdn.example/video.mp4',
        'metadata_collected': 1,
        'video_requested': 0,
        'image_requested': 0,
        'metadata_only': True,
        'save_policy': 'metadata_only',
        'direct_media_urls': {
            'video': ['https://cdn.example/video.mp4'],
            'image': ['https://cdn.example/image.jpg'],
        },
        'views': 100,
        'likes': 20,
        'comments': 3,
        'reposts': 1,
    }
    json_path, item = scope['upsert_folder_metadata'](str(folder), meta)
    assert Path(json_path).exists()
    assert item['file_name'] == ''
    assert item['asset_kind'] == 'metadata'
    assert item['metadata_collected'] == 1
    assert item['video_requested'] == 0
    assert item['video_downloaded'] == 0
    assert item['image_requested'] == 0
    assert item['image_downloaded'] == 0
    assert item['caption'] == '본문 유지'
    assert item['direct_media_urls']['video'] == ['https://cdn.example/video.mp4']
    doc = json.loads(Path(json_path).read_text(encoding='utf-8'))
    assert 'CODE277::metadata::0' in doc['items']


def test_v310_metadata_only_html_report_has_direct_links_and_zero_local_files(tmp_path: Path) -> None:
    import importlib.util

    module_path = ROOT / 'ws8771_html_report.py'
    spec = importlib.util.spec_from_file_location('ws8771_html_report_v310_test', module_path)
    assert spec and spec.loader
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)

    root = tmp_path / 'instagram'
    media = root / 'tester'
    media.mkdir(parents=True)
    doc = {
        'folder_name': 'tester',
        'items': {
            'CODE277::metadata::0': {
                'shortcode': 'CODE277',
                'owner_username': 'tester',
                'asset_kind': 'metadata',
                'file_name': '',
                'metadata_collected': 1,
                'metadata_only': True,
                'video_requested': 0,
                'video_downloaded': 0,
                'image_requested': 0,
                'image_downloaded': 0,
                'post_url': 'https://www.instagram.com/p/CODE277/',
                'thumbnail_url': 'https://cdn.example/image.jpg',
                'video_url': 'https://cdn.example/video.mp4',
                'direct_media_urls': {
                    'video': ['https://cdn.example/video.mp4'],
                    'image': ['https://cdn.example/image.jpg'],
                },
            }
        },
    }
    json_path = root / 'tester.json'
    json_path.write_text(json.dumps(doc, ensure_ascii=False), encoding='utf-8')
    result = module.generate_account_report(str(json_path), str(media))
    assert result['ok'] is True
    assert result['item_count'] == 1
    html_text = (root / 'tester.html').read_text(encoding='utf-8')
    assert '파일 미저장 · metadata 수집' in html_text
    assert 'MP4 직접링크 1' in html_text
    assert 'JPG 직접링크 1' in html_text
    assert 'metadata 완료 · MP4 미선택 · JPG 미선택' in html_text
    assert '로컬 파일</dt><dd>0개' in html_text


def test_v310_ws_square_and_tray_ui_contract() -> None:
    for token in [
        'from Shared.windows_tray import WindowsTrayIcon',
        'text="."',
        'command=self.hide_to_tray',
        'self.fold_btn.config(text="ㅁ"',
        'def hide_to_tray(self):',
        'self.root.withdraw()',
        'def restore_from_tray(self):',
        'self.root.deiconify()',
        'def poll_tray_restore(self):',
    ]:
        assert token in SOURCE, token
    # Auto8700은 보라색 사각형, WS8771은 enter_mode 상태색을 사용해 서로 구분한다.
    assert 'self.fold_btn.config(text="ㅁ", width=3, bg=c["bg"]' in SOURCE


def test_edit_folder_copy_helper_copies_only_mp4_and_never_overwrites(tmp_path: Path) -> None:
    tree = ast.parse(SOURCE)
    names = {"normalize_edit_copy_sources", "copy_mp4_files_to_edit_folder"}
    nodes = [n for n in tree.body if isinstance(n, ast.FunctionDef) and n.name in names]
    assert {n.name for n in nodes} == names
    scope = {"os": os, "shutil": shutil}
    exec(compile(ast.Module(body=nodes, type_ignores=[]), str(ROOT / "ws8771_main.py"), "exec"), scope)
    copy_files = scope["copy_mp4_files_to_edit_folder"]

    src = tmp_path / "src"
    dst = tmp_path / "edit"
    src.mkdir()
    mp4 = src / "account_upload_unknown_No001_CODE.mp4"
    jpg = src / "account_upload_unknown_No001_CODE.jpg"
    mp4.write_bytes(b"video-v1")
    jpg.write_bytes(b"image")

    first = copy_files([str(mp4), str(jpg)], str(dst))
    assert first["ok"] is True
    assert first["requested_count"] == 1
    assert first["copied_count"] == 1
    assert (dst / mp4.name).read_bytes() == b"video-v1"
    assert not (dst / jpg.name).exists()

    (dst / mp4.name).write_bytes(b"edited-user-file")
    second = copy_files([str(mp4)], str(dst))
    assert second["ok"] is True
    assert second["copied_count"] == 0
    assert second["existing_count"] == 1
    assert (dst / mp4.name).read_bytes() == b"edited-user-file"


def test_edit_folder_copy_route_and_picker_are_wired() -> None:
    for token in [
        'if act == "ig_copy_to_edit_folder"',
        'def handle_ig_copy_to_edit_folder(self, req):',
        'def _pick_edit_folder_windows(self, initial_dir=""):',
        'filedialog.askdirectory(parent=self.root, **options)',
        'threading.current_thread() is not threading.main_thread()',
        'copy_mp4_files_to_edit_folder(paths, target)',
        'messagebox.askyesno("편집폴더", "폴더를 열까요?", parent=self.root)',
        'self.handle_open_local_path(',
        '"ig_open_folder",',
        'result["open_prompt_shown"] = False',
        'result["opened_folder"] = False',
        'if act == "ig_copy_to_edit_folder":\n                timeout_sec = 660.0',
    ]:
        assert token in SOURCE, token
    direct_block = SOURCE[SOURCE.index('def handle_thread_direct_request'):SOURCE.index('def normalize_enter_mode')]
    assert '"ig_auto_cycle_state", "ig_copy_to_edit_folder"' not in direct_block