test: add web coverage and e2e tests
This commit is contained in:
hangshuo652
@@ -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
|
||||
@@ -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()
|
||||
|
||||
Reference in New Issue
Block a user