Spaces:
Running
Running
| import asyncio | |
| import json | |
| from pathlib import Path | |
| import re | |
| import pytest | |
| import httpx | |
| import verifier | |
| from urllib.parse import quote | |
| from demo.examples import (EXAMPLE_NAMES, EXAMPLE_LABELS, ORIGINAL_EXAMPLE_NAMES, | |
| GHIDRA_EXAMPLE_NAMES, ROOT, ghidra_metadata) | |
| from demo.faultless import NAMES as COMPLETENESS_NAMES | |
| from demo.presentation import json_detail, mismatch_summary | |
| from tests.support import check_report | |
| pytestmark = pytest.mark.demo | |
| def test_mismatch_summary_preserves_byte_arguments(value): | |
| detail = {"seed": "0", "difference": {"location": "call[0].argument[0]", | |
| "original": "76616c75653a20257500", "candidate": value, | |
| "width": "80", "xor": None, "bit_ranges": []}} | |
| summary = mismatch_summary(detail) | |
| assert "76616c75653a20257500" in summary | |
| assert value in summary | |
| def test_example_reference_pane_events(tmp_path): | |
| from gradio.state_holder import SessionState | |
| from demo.app import create_app | |
| app = create_app(tmp_path / "examples") | |
| loads = [fn for fn in app.fns.values() if getattr(fn.fn, "__name__", None) == "load_reference"] | |
| load = loads[0] | |
| from demo.examples import example_groups | |
| groups = example_groups(EXAMPLE_NAMES) | |
| assert len(loads) == len(groups) == 6 | |
| datasets = [fn.inputs[0] for fn in loads] | |
| reset = next(fn for fn in app.fns.values() if fn.api_name == "reset_outputs") | |
| assert set(reset.targets) == {(dataset._id, "click") for dataset in datasets} | |
| assert sum(fn.api_name == "reset_outputs" for fn in app.fns.values()) == 1 | |
| accordions = [component["props"] for component in app.config["components"] | |
| if component["type"] == "accordion"] | |
| for label, indices in groups: | |
| accordion = next(props for props in accordions if props["label"] == f"{label} ({len(indices)})") | |
| assert accordion["open"] == (label == "Basics") | |
| assert all(fn.api_visibility == "private" for fn in loads) | |
| clear = next(fn for fn in app.fns.values() if getattr(fn.fn, "__name__", None) == "clear_reference") | |
| blur = next(fn for fn in app.fns.values() | |
| if getattr(fn.fn, "__name__", None) == "clear_changed_reference") | |
| reference, workspace, reference_function = load.outputs | |
| binary = next(block for block in app.blocks.values() if block.elem_id == "binary") | |
| function = next(block for block in app.blocks.values() | |
| if getattr(block, "label", None) == "Function start address") | |
| assert not reference.interactive and reference.language == "c" | |
| assert not reference.visible and not workspace.elem_classes | |
| assert not any(component["type"] in ("tabs", "tabitem") | |
| for component in app.config["components"]) | |
| assert load.api_visibility == clear.api_visibility == "private" | |
| assert set(clear.targets) == {(binary._id, "upload"), (binary._id, "delete"), | |
| (binary._id, "clear"), (function._id, "input")} | |
| assert clear.outputs == load.outputs | |
| assert blur.targets == [(function._id, "blur")] | |
| assert blur.outputs == load.outputs | |
| async def exercise_events(): | |
| state = SessionState(app) | |
| selections = [(loader, local_index, EXAMPLE_NAMES[global_index]) | |
| for loader, (_, indices) in zip(loads, groups) | |
| for local_index, global_index in enumerate(indices)] | |
| for loader, index, name in selections: | |
| result = await app.process_api(loader, [index], state=state) | |
| dataset = loader.inputs[0] | |
| assert state[reference_function._id] == dataset.samples[index][1] | |
| pane, layout, _ = result["data"] | |
| # Gradio Code strips surrounding whitespace during postprocessing. | |
| from demo.examples import example_metadata | |
| expected = example_metadata(name)["reference"].strip() | |
| assert pane["value"] == expected | |
| assert pane["visible"] == bool(expected) | |
| assert layout["elem_classes"] == (["has-reference"] if expected else []) | |
| original_function = state[reference_function._id] | |
| # Merely focusing and leaving the original range keeps its reference. | |
| result = await app.process_api(blur, [original_function, None], state=state) | |
| assert all(update == {"__type__": "update"} for update in result["data"][:2]) | |
| assert state[reference_function._id] == original_function | |
| # Pasting or deleting text also clears the reference when leaving the field. | |
| result = await app.process_api(blur, ["", None], state=state) | |
| pane, layout, _ = result["data"] | |
| assert pane["value"] == "" and not pane["visible"] and layout["elem_classes"] == [] | |
| assert state[reference_function._id] is None | |
| await app.process_api(loader, [index], state=state) | |
| result = await app.process_api(clear, [], state=state) | |
| pane, layout, _ = result["data"] | |
| assert pane["value"] == "" and not pane["visible"] and layout["elem_classes"] == [] | |
| try: | |
| asyncio.run(exercise_events()) | |
| finally: | |
| app.close() | |
| def test_queued_ui_upload_streaming_and_download(tmp_path, monkeypatch, names): | |
| from gradio_client import Client, handle_file | |
| from gradio_client.exceptions import AppError | |
| from demo import app as demo_app, examples as demo_examples | |
| labels = dict(zip(EXAMPLE_NAMES, EXAMPLE_LABELS)) | |
| monkeypatch.setattr(demo_examples, "EXAMPLE_NAMES", names) | |
| monkeypatch.setattr(demo_app, "EXAMPLE_LABELS", tuple(labels[name] for name in names)) | |
| app = demo_app.create_app(tmp_path / "examples") | |
| try: | |
| _, url, _ = app.launch(server_name="127.0.0.1", server_port=None, prevent_thread_lock=True, quiet=True) | |
| private_source = quote(str(Path(verifier.__file__).resolve()), safe="/") | |
| assert httpx.get(url + "gradio_api/file=" + private_source).status_code == 403 | |
| import shutil | |
| helper = shutil.which("decompile-verify-declarations") | |
| assert helper is not None | |
| private_helper = quote(str(Path(helper).resolve()), safe="/") | |
| assert httpx.get(url + "gradio_api/file=" + private_helper).status_code == 403 | |
| client = Client(url, verbose=False) | |
| examples = json.loads((tmp_path / "examples" / "examples.json").read_text()) | |
| assert len(examples) == len(names) | |
| for row, name in zip(examples, names): | |
| row[0] = handle_file(row[0]) | |
| row[5] = 5 | |
| metadata = demo_examples.example_metadata(name) | |
| if metadata and metadata.get("expected_error"): | |
| for endpoint in ("/assembly", "/verify"): | |
| with pytest.raises(AppError, match=re.escape(metadata["expected_error"])): | |
| client.predict(*row, api_name=endpoint) | |
| continue | |
| preview = client.predict(*row, api_name="/assembly") | |
| assert "Assembly preview" in preview[0] | |
| assert "Arguments" in preview[0] and "Return" in preview[0] | |
| assert preview[1]["data"] == [] | |
| assert any(edge in preview[4] for edge in ("ret", "jmp")) | |
| assert any(edge in preview[5] for edge in ("ret", "jmp")) | |
| assert f'{row[1].split(":")[0]}:' in preview[4] | |
| job = client.submit(*row, api_name="/verify") | |
| updates = list(job) | |
| final = job.result() | |
| assert len(updates) >= 2 | |
| assert len(updates[0][1]["data"]) < len(final[1]["data"]) | |
| data = Path(final[3]).read_text() | |
| check_report(name, data, 5) | |
| report = json.loads(data) | |
| assert report["completed"] == 5 | |
| assert f'({report["task"]["output"]["width"]} bits)' in preview[0] | |
| assert final[4] == preview[4] == report["original_assembly"] | |
| assert final[5] == preview[5] == report["candidate_assembly"] | |
| if name == "tail-call": | |
| assert report["results"][0]["original_calls"][0]["kind"] == "tail" | |
| assert report["results"][0]["candidate_calls"][0]["kind"] == "call" | |
| if report["verdict"] == "mismatch": | |
| first = next(result for result in report["results"] if result["verdict"] == "mismatch") | |
| assert final[2] == json_detail(first) | |
| summary = mismatch_summary(final[2]) | |
| for side in ("original", "candidate"): | |
| value = first["difference"][side] | |
| assert (hex(value) if isinstance(value, int) else str(value)) in summary | |
| reset = client.predict(api_name="/reset_outputs") | |
| assert reset == ("", {"headers": preview[1]["headers"], "data": [], "metadata": None}, | |
| None, None, "", "") | |
| if "tail-call" not in names: | |
| return | |
| # Exercise actual file upload, edited C, compiler choice and start seed. | |
| row = examples[names.index("tail-call")].copy() | |
| row[2] = "unsigned char edited(unsigned x) { return 0; }" | |
| row[3] = "globals: {}\ncallees: {}\n" | |
| row[4], row[6], row[8] = "edited", "42", "clang" | |
| row[10] = "original_prototype: unsigned char original(unsigned x);" | |
| final = client.predict(*row, api_name="/verify") | |
| report = json.loads(Path(final[3]).read_text()) | |
| assert report["task"]["candidate_symbol"] == "edited" | |
| assert report["task"]["compiler"] == "clang" | |
| assert report["task"]["return_width"] == 8 | |
| assert "al" in final[0] | |
| assert [r["seed"] for r in report["results"]] == list(range(42, 47)) | |
| finally: | |
| app.close() | |