cobol-java-v3/tests/test_web_e2e.py

"""
Playwright E2E tests for COBOL-Java Migration Platform Web UI.
Server must be running: python -m uvicorn web.api:app --host 127.0.0.1 --port $TEST_PORT
(端口经 TEST_PORT 环境变量配置, 默认 8000)

Run standalone: python -m pytest tests/test_web_e2e.py -v
"""
from __future__ import annotations
import os

import pytest

try:
    from playwright.sync_api import Page, expect, sync_playwright
    HAS_PLAYWRIGHT = True
except ImportError:
    HAS_PLAYWRIGHT = False

TEST_PORT = int(os.environ.get("TEST_PORT", "8000"))
BASE_URL = f"http://127.0.0.1:{TEST_PORT}"


@pytest.fixture(scope="module")
def browser(request):
    if not HAS_PLAYWRIGHT:
        pytest.skip("playwright not installed")
    try:
        import asyncio
        loop = asyncio.get_event_loop()
        if loop.is_running():
            pytest.skip("sync_playwright cannot run inside a running asyncio loop")
    except (RuntimeError, AttributeError):
        pass
    with sync_playwright() as p:
        browser = p.firefox.launch(headless=True)
        yield browser
        browser.close()


@pytest.fixture
def page(browser):
    page = browser.new_page()
    yield page
    page.close()


def test_upload_page_loads(page: Page):
    """验证上传页面正常加载"""
    page.goto(BASE_URL)
    expect(page).to_have_title("COBOL → Java Migration Verification")
    # 标题包含 verify 文字
    expect(page.locator("h1")).to_contain_text("verify")
    # 表单存在
    form = page.locator("#verify-form")
    expect(form).to_be_visible()


def test_form_elements_present(page: Page):
    """验证所有表单元素存在"""
    page.goto(BASE_URL)
    # 4 个文件输入
    expect(page.locator("input[name=copybook]")).to_be_visible()
    expect(page.locator("input[name=cobol_src]")).to_be_visible()
    expect(page.locator("input[name=java_src]")).to_be_visible()
    expect(page.locator("input[name=mapping]")).to_be_visible()
    # Runner 下拉框
    expect(page.locator("select[name=runner]")).to_be_visible()
    expect(page.locator("select[name=runner]")).to_have_value("native")
    # 提交按钮
    expect(page.locator("button[type=submit]")).to_be_visible()
    expect(page.locator("button[type=submit]")).to_contain_text("verify")


def test_submit_empty_form(page: Page):
    """验证空表单提交返回 422 (缺少必填字段)"""
    page.goto(BASE_URL)
    js = """
        (async () => {
            const fd = new FormData();
            const r = await fetch('__BASE_URL__/verify', { method: 'POST', body: fd });
            return r.status;
        })()
    """.replace("__BASE_URL__", BASE_URL)
    result = page.evaluate(js)
    assert result == 422


def test_submit_with_files(page: Page):
    """验证上传测试文件后表单正常响应"""
    page.goto(BASE_URL)
    page.set_input_files("input[name=copybook]",
        "tests/fixtures/simple.cpy")
    page.set_input_files("input[name=cobol_src]",
        "tests/fixtures/simple.cbl")
    page.set_input_files("input[name=mapping]",
        "tests/fixtures/simple.yaml")
    # 用 evaluate 直接调 API 绕过 webkitdirectory 限制
    js = """
        (async () => {
            const fd = new FormData();
            fd.append('copybook', new Blob(['test'], {type:'text/plain'}), 'test.cpy');
            fd.append('cobol_src', new Blob(['test'], {type:'text/plain'}), 'test.cbl');
            fd.append('java_src', new Blob(['test'], {type:'text/plain'}), 'test.java');
            fd.append('mapping', new Blob(['test'], {type:'text/plain'}), 'test.yaml');
            fd.append('runner', 'native');
            const r = await fetch('__BASE_URL__/verify', { method: 'POST', body: fd });
            return { status: r.status, body: await r.json() };
        })()
    """.replace("__BASE_URL__", BASE_URL)
    result = page.evaluate(js)
    assert result["status"] == 202
    assert "task_id" in result["body"]


def test_runner_selector_options(page: Page):
    """验证 Runner 下拉框有两个选项"""
    page.goto(BASE_URL)
    expect(page.locator("select[name=runner]")).to_be_visible()
    count = page.locator("select[name=runner] option").count()
    assert count == 2
    native_val = page.locator("select[name=runner] option").nth(0).get_attribute("value")
    spark_val = page.locator("select[name=runner] option").nth(1).get_attribute("value")
    assert native_val == "native"
    assert spark_val == "spark"


def test_status_endpoint(page: Page):
    """验证 /status/ 端点返回 JSON"""
    page.goto(f"{BASE_URL}/status/nonexistent")
    body = page.locator("body").inner_text()
    assert "404" in body or "not found" in body.lower()


def test_result_endpoint_404(page: Page):
    """验证 /result/ 端点对不存在任务返回 404"""
    page.goto(f"{BASE_URL}/result/nonexistent")
    body = page.locator("body").inner_text()
    assert "404" in body or "not found" in body.lower()


def test_dark_theme_rendered(page: Page):
    """验证 Terminal Dark 主题渲染"""
    page.goto(BASE_URL)
    expect(page.locator(".badge")).to_be_visible()
    expect(page.locator("footer")).to_be_visible()


def test_page_title(page: Page):
    """验证页面标题"""
    page.goto(BASE_URL)
    expect(page).to_have_title("COBOL → Java Migration Verification")


def test_css_loaded(page: Page):
    """验证 CSS 静态文件正常加载"""
    page.goto(BASE_URL)
    css = page.locator("link[rel=stylesheet]")
    expect(css).to_have_attribute("href", "/static/style.css")
    resp = page.evaluate("async () => { const r = await fetch('/static/style.css'); return r.status; }")
    assert resp == 200


def test_js_loaded(page: Page):
    """验证 JS 静态文件正常加载"""
    page.goto(BASE_URL)
    js = page.locator('script[src="/static/script.js"]')
    expect(js).to_be_attached()
    resp = page.evaluate("async () => { const r = await fetch('/static/script.js'); return r.status; }")
    assert resp == 200


def test_file_size_limit(page: Page):
    """验证超过 10MB 的文件返回 413"""
    page.goto(BASE_URL)
    large_content = "x" * (10 * 1024 * 1024 + 1)
    js = f"""
        (async () => {{
            const fd = new FormData();
            fd.append('copybook', new Blob(['test'], {{type:'text/plain'}}), 'test.cpy');
            fd.append('cobol_src', new Blob([new ArrayBuffer(10*1024*1024+1)], {{type:'application/octet-stream'}}), 'large.cbl');
            fd.append('java_src', new Blob(['test'], {{type:'text/plain'}}), 'test.java');
            fd.append('mapping', new Blob(['test'], {{type:'text/plain'}}), 'test.yaml');
            fd.append('runner', 'native');
            const r = await fetch('{BASE_URL}/verify', {{ method: 'POST', body: fd }});
            return r.status;
        }})()
    """
    result = page.evaluate(js)
    assert result == 413


def test_fields_endpoint(page: Page):
    """验证 /fields/{task_id} 端点对不存在任务返回 404"""
    page.goto(f"{BASE_URL}/fields/nonexistent")
    body = page.locator("body").inner_text()
    assert "404" in body or "not found" in body.lower()


def test_fields_endpoint_with_task(page: Page):
    """验证 /fields/{task_id} 端点对已提交任务返回 JSON"""
    page.goto(BASE_URL)
    js = """
        (async () => {
            const fd = new FormData();
            fd.append('copybook', new Blob(['test'], {type:'text/plain'}), 'test.cpy');
            fd.append('cobol_src', new Blob(['test'], {type:'text/plain'}), 'test.cbl');
            fd.append('java_src', new Blob(['test'], {type:'text/plain'}), 'test.java');
            fd.append('mapping', new Blob(['test'], {type:'text/plain'}), 'test.yaml');
            fd.append('runner', 'native');
            const r = await fetch('__BASE_URL__/verify', { method: 'POST', body: fd });
            const d = await r.json();
            const r2 = await fetch('__BASE_URL__/fields/' + d.task_id);
            return { status: r2.status, body: await r2.json() };
        })()
    """.replace("__BASE_URL__", BASE_URL)
    result = page.evaluate(js)
    assert result["status"] == 200
    assert "task_id" in result["body"]
    assert "fields" in result["body"]


def test_result_page_for_existing_task(page: Page):
    """验证已提交任务的结果页面正常渲染"""
    page.goto(BASE_URL)
    js = """
        (async () => {
            const fd = new FormData();
            fd.append('copybook', new Blob(['test'], {type:'text/plain'}), 'test.cpy');
            fd.append('cobol_src', new Blob(['test'], {type:'text/plain'}), 'test.cbl');
            fd.append('java_src', new Blob(['test'], {type:'text/plain'}), 'test.java');
            fd.append('mapping', new Blob(['test'], {type:'text/plain'}), 'test.yaml');
            fd.append('runner', 'native');
            const r = await fetch('__BASE_URL__/verify', { method: 'POST', body: fd });
            return (await r.json()).task_id;
        })()
    """.replace("__BASE_URL__", BASE_URL)
    task_id = page.evaluate(js)
    page.goto(f"{BASE_URL}/result/{task_id}")
    expect(page.locator("h1")).to_contain_text(task_id)
    expect(page.locator(".badge")).to_contain_text("Verification Result")
    expect(page.locator("a.btn")).to_contain_text("New Verification")


def test_clear_button(page: Page):
    """验证 Clear 按钮清空表单"""
    page.goto(BASE_URL)
    page.set_input_files("input[name=copybook]", "tests/fixtures/simple.cpy")
    page.locator("button[type=reset]").click()
    file_input = page.locator("input[name=copybook]")
    expect(file_input).to_have_value("")


def test_runner_spark_option(page: Page):
    """验证 Runner 可切换为 Spark"""
    page.goto(BASE_URL)
    page.select_option("select[name=runner]", "spark")
    expect(page.locator("select[name=runner]")).to_have_value("spark")


def test_concurrent_submissions(page: Page):
    """验证并发提交多个任务"""
    page.goto(BASE_URL)
    js = """
        (async () => {
            const results = [];
            for (let i = 0; i < 3; i++) {
                const fd = new FormData();
                fd.append('copybook', new Blob(['test'], {type:'text/plain'}), 'test.cpy');
                fd.append('cobol_src', new Blob(['test'], {type:'text/plain'}), 'test.cbl');
                fd.append('java_src', new Blob(['test'], {type:'text/plain'}), 'test.java');
                fd.append('mapping', new Blob(['test'], {type:'text/plain'}), 'test.yaml');
                fd.append('runner', 'native');
                const r = await fetch('__BASE_URL__/verify', { method: 'POST', body: fd });
                results.push({ status: r.status, body: await r.json() });
            }
            return results;
        })()
    """.replace("__BASE_URL__", BASE_URL)
    results = page.evaluate(js)
    assert len(results) == 3
    task_ids = set()
    for r in results:
        assert r["status"] == 202
        task_ids.add(r["body"]["task_id"])
    assert len(task_ids) == 3, "每个任务应有唯一 task_id"


def test_status_api_json(page: Page):
    """验证 /status/ API 返回正确的 JSON 结构"""
    page.goto(BASE_URL)
    js = """
        (async () => {
            const fd = new FormData();
            fd.append('copybook', new Blob(['test'], {type:'text/plain'}), 'test.cpy');
            fd.append('cobol_src', new Blob(['test'], {type:'text/plain'}), 'test.cbl');
            fd.append('java_src', new Blob(['test'], {type:'text/plain'}), 'test.java');
            fd.append('mapping', new Blob(['test'], {type:'text/plain'}), 'test.yaml');
            fd.append('runner', 'native');
            const r1 = await fetch('__BASE_URL__/verify', { method: 'POST', body: fd });
            const d = await r1.json();
            const r2 = await fetch('__BASE_URL__/status/' + d.task_id);
            return await r2.json();
        })()
    """.replace("__BASE_URL__", BASE_URL)
    result = page.evaluate(js)
    assert "task_id" in result
    assert "status" in result
    assert result["status"] in ("queued", "running", "done", "error")


def test_file_accept_attributes(page: Page):
    """验证文件输入的 accept 属性正确"""
    page.goto(BASE_URL)
    expect(page.locator("input[name=copybook]")).to_have_attribute("accept", ".cpy,.cbl,.copy")
    expect(page.locator("input[name=cobol_src]")).to_have_attribute("accept", ".cbl")
    expect(page.locator("input[name=mapping]")).to_have_attribute("accept", ".yaml,.yml")


def test_footer_version(page: Page):
    """验证 footer 显示版本号"""
    page.goto(BASE_URL)
    footer = page.locator("footer")
    expect(footer).to_contain_text("v0.2.0")


def test_result_page_back_link(page: Page):
    """验证结果页面有返回首页链接"""
    page.goto(BASE_URL)
    js = """
        (async () => {
            const fd = new FormData();
            fd.append('copybook', new Blob(['test'], {type:'text/plain'}), 'test.cpy');
            fd.append('cobol_src', new Blob(['test'], {type:'text/plain'}), 'test.cbl');
            fd.append('java_src', new Blob(['test'], {type:'text/plain'}), 'test.java');
            fd.append('mapping', new Blob(['test'], {type:'text/plain'}), 'test.yaml');
            fd.append('runner', 'native');
            const r = await fetch('__BASE_URL__/verify', { method: 'POST', body: fd });
            return (await r.json()).task_id;
        })()
    """.replace("__BASE_URL__", BASE_URL)
    task_id = page.evaluate(js)
    page.goto(f"{BASE_URL}/result/{task_id}")
    back_link = page.locator("a.btn.btn-secondary")
    expect(back_link).to_be_visible()
    expect(back_link).to_have_attribute("href", "/")
    back_link.click()
    expect(page).to_have_url(f"{BASE_URL}/")


def test_index_returns_html(page: Page):
    """验证根路径返回 HTML"""
    page.goto(BASE_URL)
    expect(page.locator("html")).to_be_visible()
    expect(page.locator("head meta[charset]")).to_be_attached()