From f712b7b527aae643ea356bfe29e8039b0f23f2a1 Mon Sep 17 00:00:00 2001 From: hangshuo652 Date: Sat, 5 Sep 2026 15:25:53 +0800 Subject: [PATCH] test: add web coverage and e2e tests --- tests/test_web_coverage.py | 95 +++++++++++++++++ tests/test_web_e2e.py | 203 +++++++++++++++++++++++++++++++++++++ 2 files changed, 298 insertions(+) create mode 100644 tests/test_web_coverage.py diff --git a/tests/test_web_coverage.py b/tests/test_web_coverage.py new file mode 100644 index 0000000..b967407 --- /dev/null +++ b/tests/test_web_coverage.py @@ -0,0 +1,95 @@ +"""FastAPI TestClient-based tests for accurate coverage measurement.""" +from __future__ import annotations +import io, json, pytest +from fastapi.testclient import TestClient +from web.api import app + +client = TestClient(app) + + +def test_index(): + r = client.get("/") + assert r.status_code == 200 + assert "verify" in r.text.lower() + + +def test_verify_422_empty(): + r = client.post("/verify") + assert r.status_code == 422 + + +def test_verify_202(): + r = client.post("/verify", + files={ + "copybook": ("t.cpy", b"x", "text/plain"), + "cobol_src": ("t.cbl", b"x", "text/plain"), + "java_src": ("t.java", b"x", "text/plain"), + "mapping": ("t.yaml", b"x", "text/plain"), + }, + data={"runner": "native"}) + assert r.status_code == 202 + body = r.json() + assert "task_id" in body + return body["task_id"] + + +def test_verify_large_file_413(): + r = client.post("/verify", + files={ + "copybook": ("t.cpy", b"x", "text/plain"), + "cobol_src": ("t.cbl", b"x" * (10 * 1024 * 1024 + 1), "application/octet-stream"), + "java_src": ("t.java", b"x", "text/plain"), + "mapping": ("t.yaml", b"x", "text/plain"), + }, + data={"runner": "native"}) + assert r.status_code == 413 + + +def test_status_404(): + r = client.get("/status/nonexistent") + assert r.status_code == 404 + + +def test_status_ok(): + tid = test_verify_202() + r = client.get(f"/status/{tid}") + assert r.status_code == 200 + body = r.json() + assert body["status"] in ("queued", "running", "done", "error") + + +def test_fields_404(): + r = client.get("/fields/nonexistent") + assert r.status_code == 404 + + +def test_fields_ok(): + tid = test_verify_202() + r = client.get(f"/fields/{tid}") + assert r.status_code == 200 + body = r.json() + assert "fields" in body + assert "debug" in body + assert "build_log" in body + + +def test_result_404(): + r = client.get("/result/nonexistent") + assert r.status_code == 404 + + +def test_result_page(): + tid = test_verify_202() + r = client.get(f"/result/{tid}") + assert r.status_code == 200 + assert tid in r.text + + +def test_static_css(): + r = client.get("/static/style.css") + assert r.status_code == 200 + + +def test_static_js(): + r = client.get("/static/script.js") + assert r.status_code == 200 diff --git a/tests/test_web_e2e.py b/tests/test_web_e2e.py index 8c6eb64..6d54dde 100644 --- a/tests/test_web_e2e.py +++ b/tests/test_web_e2e.py @@ -149,3 +149,206 @@ 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()