"""E2E Tests for COBOL->Java Verification Platform Run: cd D:/cobol-java/v3-gstack-code-gen && python -m pytest tests/e2e/ -v Requires: web server on http://127.0.0.1:8000, WSL available """ import json, os, sys, time, uuid, shutil from datetime import datetime from pathlib import Path import pytest PROJECT = Path(__file__).parent.parent.parent.resolve() BASE_URL = "http://127.0.0.1:8000" TASKS_DIR = PROJECT / "tasks" UPLOADS_DIR = PROJECT / "uploads" FIXTURES = PROJECT / "tests" / "fixtures" TEST_FILES = PROJECT.parent / "test-files" def _wsl(cmd: str, timeout: int = 60) -> str: import subprocess r = subprocess.run(["wsl", "bash", "-c", cmd], capture_output=True, text=True, timeout=timeout) return r.stdout + r.stderr def create_task(copybook: str, cobol: str, java_dir: str, mapping: str, runner="native") -> str: tid = uuid.uuid4().hex[:8] task_dir = UPLOADS_DIR / tid task_dir.mkdir(parents=True, exist_ok=True) shutil.copy(copybook, task_dir / "copybook.cpy") shutil.copy(cobol, task_dir / "program.cbl") shutil.copy(mapping, task_dir / "mapping.yaml") java_dst = task_dir / "java" if Path(java_dir).is_dir(): if java_dst.exists(): shutil.rmtree(java_dst) shutil.copytree(java_dir, java_dst) else: shutil.copy(java_dir, java_dst) task = { "id": tid, "status": "queued", "copybook": f"uploads\\{tid}\\copybook.cpy", "cobol_src": f"uploads\\{tid}\\program.cbl", "java_src": f"uploads\\{tid}\\java", "mapping": f"uploads\\{tid}\\mapping.yaml", "runner": runner, "created": datetime.now().isoformat(), } (TASKS_DIR / f"{tid}.json").write_text(json.dumps(task)) return tid def run_worker_for_task(tid: str): script = ( "cd /mnt/d/cobol-java/v3-gstack-code-gen && " "export LLM_API_KEY=sk-ca4961087c7f4aefa8ed0fc6f3d02329 && " "export LLM_API_BASE=https://api.deepseek.com/v1 && " "export LLM_MODEL=deepseek-chat && " f"python3 -c \"exec(open('write_result.py').read().replace('ec17bf32','{tid}'))\"" ) out = _wsl(script, timeout=90) return out class TestPipelineE2E: """End-to-end pipeline tests with Playwright browser verification.""" @pytest.fixture(autouse=True) def browser(self, page): self.page = page yield self.page = None def test_result_page_summary(self): """Task processed in WSL → result page shows correct summary.""" tid = "75bf0dfe" # pre-processed PASS task self.page.goto(f"{BASE_URL}/result/{tid}") self.page.wait_for_load_state("networkidle") self.page.screenshot(path=str(PROJECT.parent / "screenshots" / "e2e-summary.png"), full_page=True) status = self.page.locator("dt:has-text('Status') + dd").first.text_content() matched = self.page.locator("dt:has-text('Matched') + dd").first.text_content() mismatched = self.page.locator("dt:has-text('Mismatched') + dd").first.text_content() assert status == "PASS", f"Expected PASS, got {status}" assert matched == "3", f"Expected 3 matched, got {matched}" assert mismatched == "0", f"Expected 0 mismatched, got {mismatched}" def test_result_page_field_table(self): """Field results table shows correct per-field status.""" tid = "75bf0dfe" self.page.goto(f"{BASE_URL}/result/{tid}") self.page.wait_for_load_state("networkidle") rows = self.page.locator("table tr").all() field_data = {} for row in rows: cells = row.locator("td").all() if len(cells) >= 3: name = cells[0].text_content().strip() status = cells[1].text_content().strip() cobol_val = cells[2].text_content().strip() java_val = cells[3].text_content().strip() if len(cells) > 3 else "" if name in ("BR-AMT", "BR-STATUS", "BR-DATE"): field_data[name] = (status, cobol_val, java_val) assert field_data["BR-AMT"][0] == "PASS" assert "1500" in field_data["BR-AMT"][1] or "1500" in field_data["BR-AMT"][2] assert field_data["BR-STATUS"][1] == field_data["BR-STATUS"][2] == "A" assert field_data["BR-DATE"][1] == field_data["BR-DATE"][2] == "20260522" self.page.screenshot(path=str(PROJECT.parent / "screenshots" / "e2e-field-table.png"), full_page=True) def test_result_page_fieldtree(self): """Pipeline details section shows COPYBOOK FieldTree.""" tid = "75bf0dfe" self.page.goto(f"{BASE_URL}/result/{tid}") self.page.wait_for_load_state("networkidle") tree_text = self.page.locator("h3:has-text('COPYBOOK FieldTree') + table").text_content() assert "CUST-ID" in tree_text assert "BR-AMT" in tree_text assert "COMP-3" in tree_text def test_status_api(self): """Status API returns correct JSON.""" tid = "75bf0dfe" self.page.goto(f"{BASE_URL}/status/{tid}") body = self.page.locator("body").text_content() data = json.loads(body) assert data["task_id"] == tid assert data["status"] == "done" assert data["result"]["status"] == "PASS" def test_fields_api(self): """Fields API returns per-field results.""" tid = "75bf0dfe" self.page.goto(f"{BASE_URL}/fields/{tid}") body = self.page.locator("body").text_content() data = json.loads(body) assert data["task_id"] == tid assert len(data["fields"]) >= 3 def test_home_page_loads(self): """Home page loads with all form elements.""" self.page.goto(BASE_URL) self.page.wait_for_load_state("networkidle") title = self.page.title() assert "COBOL" in title buttons = self.page.locator("button").all_text_contents() assert any("verify" in b.lower() for b in buttons) def test_result_navigation_loop(self): """Result page → New Verification → Home page.""" tid = "75bf0dfe" self.page.goto(f"{BASE_URL}/result/{tid}") self.page.wait_for_load_state("networkidle") self.page.locator("a:has-text('New Verification')").click() self.page.wait_for_load_state("networkidle") assert self.page.url == BASE_URL + "/" def test_new_task_full_pipeline(self): """Create task → WSL worker → verify result page.""" tid = create_task( str(FIXTURES / "simple.cpy"), str(FIXTURES / "simple.cbl"), str(TEST_FILES / "java"), str(FIXTURES / "simple.yaml"), ) out = run_worker_for_task(tid) self.page.goto(f"{BASE_URL}/result/{tid}") self.page.wait_for_load_state("networkidle") status = self.page.locator("dt:has-text('Status') + dd").first.text_content() assert status in ("PASS", "MISMATCH", "BLOCKED"), f"Unexpected status: {status}" if status == "PASS": matched = self.page.locator("dt:has-text('Matched') + dd").first.text_content() assert matched == "3" def test_create_task_and_verify(): """Non-browser test: create task, run worker, check status API.""" tid = create_task( str(FIXTURES / "simple.cpy"), str(FIXTURES / "simple.cbl"), str(TEST_FILES / "java"), str(FIXTURES / "simple.yaml"), ) out = run_worker_for_task(tid) assert "Task updated" in out or "PASS" in out or "MISMATCH" in out, f"Worker failed: {out[-300:]}" tf = TASKS_DIR / f"{tid}.json" data = json.loads(tf.read_text(encoding="utf-8-sig")) assert data["status"] == "done" assert data["result"]["status"] in ("PASS", "MISMATCH", "BLOCKED") def test_create_task_fails_with_invalid_cobol(): """Invalid COBOL → BLOCKED status.""" tid = create_task( str(FIXTURES / "simple.cpy"), str(FIXTURES / "simple.cpy"), # wrong: COPYBOOK, not COBOL source str(TEST_FILES / "java"), str(FIXTURES / "simple.yaml"), ) out = run_worker_for_task(tid) tf = TASKS_DIR / f"{tid}.json" data = json.loads(tf.read_text(encoding="utf-8-sig")) assert data["result"]["status"] in ("BLOCKED", "ERROR")