From 54445a91b0540de28f42d529aed52542028aa1cc Mon Sep 17 00:00:00 2001 From: Yushu Li Date: Wed, 21 Oct 2026 21:24:11 -0700 Subject: [PATCH] feat(compat): export minimal Python AlphaEvolve bundles --- examples/alphaevolve_interop/README.md | 88 +++++- openevolve/compat/alphaevolve_bundle.py | 305 +++++++++++++++++++++ tests/test_alphaevolve_bundle.py | 350 ++++++++++++++++++++++++ 3 files changed, 740 insertions(+), 3 deletions(-) create mode 100644 openevolve/compat/alphaevolve_bundle.py create mode 100644 tests/test_alphaevolve_bundle.py diff --git a/examples/alphaevolve_interop/README.md b/examples/alphaevolve_interop/README.md index 46fb3475e..7b2c71af1 100644 --- a/examples/alphaevolve_interop/README.md +++ b/examples/alphaevolve_interop/README.md @@ -1,14 +1,15 @@ # AlphaEvolve interoperability -This walkthrough covers opt-in evaluator and configuration slices of +This walkthrough covers opt-in evaluator, configuration and Python bundle slices of [#468](https://github.com/algorithmicsuperintelligence/openevolve/issues/468). It runs the **unchanged** `program.py` and `evaluate.py` from Google's [AlphaEvolve examples](https://github.com/Google-Cloud-AI/alphaevolve-on-googlecloud) through an explicit adapter. A reverse adapter also lets a synchronous OpenEvolve evaluator score AlphaEvolve candidates. Separate opt-in helpers import a limited subset of experiment configuration for local Python runs and export a limited -configuration subset in the other direction. These helpers do not replace the -cloud SDK, export bundles, or implement all of #468. +configuration subset in the other direction. A minimal bundle exporter combines +these reverse helpers with explicitly selected Python sources and a local seed +check. These helpers do not replace the cloud SDK or implement all of #468. ## Evaluate an upstream seed locally @@ -208,6 +209,87 @@ The input config and mapping are not modified. Export followed by the importer prefixes literal problem text and requires a new local endpoint. This does not preserve search behavior or provide a runnable cloud export bundle. +## Export a minimal Python bundle + +From an installed OpenEvolve checkout, export an existing single-file Python +seed and synchronous `evaluate(program_path)` evaluator: + +```bash +printf '%s\n' 'Improve the search for the global minimum of the supplied function.' > problem.md + +python -m openevolve.compat.alphaevolve_bundle \ + examples/function_minimization/initial_program.py \ + examples/function_minimization/evaluator.py \ + --config examples/function_minimization/config.yaml \ + --output-dir ./alphaevolve-bundle \ + --title "Function minimization" \ + --instructions problem.md \ + --objective-metric combined_score \ + --requirements-file examples/function_minimization/requirements.txt + +# Review/install the evaluator's dependencies in your environment first. +python ./alphaevolve-bundle/check.py +``` + +The destination must not exist, and its parent must already exist. Exporting +does not execute either source file, install packages, import Google's SDK, +look up credentials, or contact a service. The CLI infers Python when the YAML +omits its language. `--instructions` supplies literal problem text; it does not +resolve prompt templates. Non-Gemini model names require explicit repeated +`--model-map SOURCE=GEMINI` options. The configuration mapping and warnings are +the same as `config_to_alphaevolve` above. + +The output contains: + +```text +alphaevolve-bundle/ + experiment.json # Limited experiment configuration + evaluation.json # Explicit objective metric and failure score + src/ + __init__.py + program.py # Original seed bytes + openevolve_evaluator.py # Original evaluator bytes + evaluate.py # Reverse adapter entry point + check.py + README.md + requirements.txt # Only when explicitly supplied +``` + +Use `from src.evaluate import evaluate` with the bundle root on Python's module +path and pass that callable as `evaluator_function` to the upstream SDK. Validate +`experiment.json` with `AlphaEvolveExperimentConfig.model_validate` as described +above. Read the generated README for installation and integration instructions. +The same OpenEvolve checkout containing the reverse adapter must be installed in +the target environment; no dependency or SDK is vendored into the bundle. + +Sources and the optional dependency file are copied byte for byte, without +secret removal or automatic discovery of local modules, included requirements +or data files. The exporter requires UTF-8 Python, a module-level synchronous +`evaluate` accepting one path, and rejects relative imports and discoverable +sibling module imports. Dynamic imports and working-directory dependencies +cannot all be detected. Use self-contained sources and review them before +sharing or running the bundle. + +`check.py` runs the copied seed through the exported evaluator, prints a JSON +evaluation to stdout and redirects Python-level evaluator/import stdout to +stderr. Evaluators must capture their own subprocess output to keep stdout +clean. The check also works by absolute path from another directory. +Import/contract errors and adapter-reported failures produce a nonzero exit +status. The check executes +trusted code with the current process's privileges; it adds no sandbox or +timeout and cannot recognize failures silently converted into numeric scores. + +This is an offline-checkable export slice. It does not create a cloud client, +start an experiment, enforce a cost limit or reproduce the OpenEvolve evaluation +pipeline and search behavior. Managed-service authentication and execution +remain separate operations that may incur charges. + +Run the offline bundle tests with: + +```bash +python -m unittest discover -s tests -p 'test_alphaevolve_bundle.py' -v +``` + ## Import a local Python experiment configuration `config_from_alphaevolve` imports an explicit experiment dictionary without diff --git a/openevolve/compat/alphaevolve_bundle.py b/openevolve/compat/alphaevolve_bundle.py new file mode 100644 index 000000000..7b7cc683a --- /dev/null +++ b/openevolve/compat/alphaevolve_bundle.py @@ -0,0 +1,305 @@ +"""Export a minimal, offline-checkable single-file Python AlphaEvolve bundle.""" + +import argparse +import ast +import io +import json +import shutil +import tokenize +from pathlib import Path +from typing import Optional, Union + +from yaml import YAMLError + +from openevolve.compat.alphaevolve_export import config_to_alphaevolve +from openevolve.compat.alphaevolve_reverse import adapt_openevolve_evaluator +from openevolve.config import Config + +_EVALUATOR = '''"""AlphaEvolve candidate adapter for the explicitly copied evaluator.""" + +import json +from pathlib import Path + +from openevolve.compat.alphaevolve_reverse import adapt_openevolve_evaluator + +from .openevolve_evaluator import evaluate as evaluate_program + +settings = json.loads( + (Path(__file__).resolve().parents[1] / "evaluation.json").read_text(encoding="utf-8") +) +evaluate = adapt_openevolve_evaluator(evaluate_program, **settings) +''' + +_CHECK = '''"""Explicitly execute the trusted evaluator against the copied seed.""" + +import contextlib +import json +import sys +from pathlib import Path + + +def main(): + root = Path(__file__).resolve().parent + candidate = { + "content": {"files": [{ + "path": "program.py", + "content": (root / "src" / "program.py").read_bytes().decode("utf-8"), + }]} + } + # Redirect Python stdout; evaluators must capture their subprocess output. + with contextlib.redirect_stdout(sys.stderr): + from src.evaluate import evaluate + + result = evaluate(candidate) + print(json.dumps(result, ensure_ascii=False, allow_nan=False, indent=2)) + failed = any( + entry.get("label") == "OpenEvolve Evaluation Error" + for entry in result.get("insights", {}).get("insights", []) + ) + return int(failed) + + +if __name__ == "__main__": + raise SystemExit(main()) +''' + +_README = """# Minimal AlphaEvolve evaluator bundle + +This directory packages a single Python seed and a synchronous +`evaluate(program_path)` evaluator. It contains: + +- `src/program.py`: the seed, copied byte for byte. +- `src/openevolve_evaluator.py`: the evaluator, copied byte for byte. +- `src/evaluate.py`: an AlphaEvolve candidate adapter. +- `evaluation.json`: the objective metric and finite failure score. +- `experiment.json`: the limited AlphaEvolve experiment configuration. +- `check.py`: an explicit local seed-evaluation check. +- `requirements.txt`, only if explicitly supplied during export. + +## Install dependencies + +Use Python 3.10 or newer and an OpenEvolve checkout containing +`openevolve.compat.alphaevolve_reverse`. Install that checkout in your chosen +environment, for example `python -m pip install /path/to/openevolve`. +The exporter does not infer, install or vendor dependencies. Review and install +your evaluator's packages; if `requirements.txt` is included, review it before +running `python -m pip install -r requirements.txt`. Included requirement files, +local paths and data files are not copied automatically. + +Source files are copied verbatim, without secret removal. Review them and any +explicit dependency file before sharing. The source OpenEvolve YAML, endpoints +and API keys are not serialized into the exported experiment configuration. + +## Check the seed locally + +From this directory, run `python check.py`. You can also run it by absolute path +from another working directory. The result is a JSON evaluation on stdout; +Python-level evaluator/import stdout is redirected to stderr. Evaluators must +capture their own subprocess or direct file-descriptor output to keep stdout +clean. Adapter-reported failures return a nonzero exit status; ordinary import +or contract errors also fail the command. + +This command executes trusted evaluator and candidate code with the current +process's privileges. It is not a sandbox or a timeout wrapper. Any file/network +access performed by that code remains its responsibility. Evaluators that turn +failures into ordinary numeric scores cannot be distinguished from success. + +## Use the real AlphaEvolve SDK + +With the official SDK installed and this directory on Python's module path: + +```python +import json +from pathlib import Path + +from alpha_evolve.models import AlphaEvolveExperimentConfig +from src.evaluate import evaluate + +config = json.loads(Path("experiment.json").read_text(encoding="utf-8")) +AlphaEvolveExperimentConfig.model_validate(config) # Offline schema check. +# Pass `evaluate` as evaluator_function to your configured AlphaEvolveExperiment. +``` + +Cloud authentication, client construction, experiment creation and controller +execution are separate, explicit operations and may incur charges. This bundle +does not perform them or claim a managed-service run. Generation limits are not +cost or evaluation limits. Configure evaluation workers separately from service +generation concurrency and check model availability/mixture rules yourself. + +## Scope + +The exporter checks UTF-8 Python syntax, requires a module-level synchronous +`evaluate` definition, and rejects relative imports and discoverable sibling +modules. It cannot detect all dynamic imports, external data or working-directory +assumptions. Resolve those dependencies before use and run the local check. +The evaluator must accept one path argument without additional required arguments. + +The reverse adapter retains numeric metrics, handles explicit error/timeout and +non-finite results, and removes temporary candidate files. It omits artifacts and +does not reproduce OpenEvolve's evaluator pipeline, retries, LLM feedback, +cascade stages, resource limits or search behavior. This is a limited export +slice of OpenEvolve #468, not a complete cloud exporter or SDK shim. +""" + + +def _read_python(path: Path, *, evaluator: bool = False) -> bytes: + if path.suffix.lower() != ".py": + raise ValueError("Bundle sources must be .py files") + source = path.read_bytes() + try: + encoding, _ = tokenize.detect_encoding(io.BytesIO(source).readline) + if encoding not in {"utf-8", "utf-8-sig"}: + raise ValueError("Bundle sources must use UTF-8") + source.decode("utf-8-sig") + tree = ast.parse(source) + compile(tree, "", "exec") # Validate scopes without executing code. + except (SyntaxError, UnicodeError): + raise ValueError("Bundle sources must be valid UTF-8 Python") from None + + if evaluator: + definitions = [ + node + for node in tree.body + if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) and node.name == "evaluate" + ] + if not definitions or not isinstance(definitions[-1], ast.FunctionDef): + raise ValueError("Evaluator must define a module-level synchronous evaluate function") + args = definitions[-1].args + positional = len(args.posonlyargs) + len(args.args) + if ( + positional - len(args.defaults) > 1 + or (not positional and args.vararg is None) + or any(default is None for default in args.kw_defaults) + ): + raise ValueError("Evaluator must accept one path without other required arguments") + for node in ast.walk(tree): + modules = [] + if isinstance(node, ast.ImportFrom): + if node.level: + raise ValueError("Relative imports are not supported in single-file bundles") + modules = [node.module] + elif isinstance(node, ast.Import): + modules = [alias.name for alias in node.names] + for module in modules: + name = module.split(".")[0] + if (path.parent / (name + ".py")).is_file() or (path.parent / name).is_dir(): + raise ValueError("Sibling module imports are not supported in single-file bundles") + return source + + +def export_to_alphaevolve( + initial_program: Union[str, Path], + evaluator: Union[str, Path], + config: Config, + out_dir: Union[str, Path], + *, + title: str, + problem_description: str, + objective_metric: str, + failure_score: float = -1e12, + model_mapping=None, + generation_concurrency: Optional[int] = None, + requirements_file: Optional[Union[str, Path]] = None, +) -> Path: + """Write a new minimal Python bundle without importing/executing source code. + + Reuse the explicit config export and reverse evaluator contracts. Copy only + the two named UTF-8 .py files and an optional explicit dependency file. Reject + existing destinations (including symlinks); their parent must already exist. + Validate inputs before creating output and clean up new output on write errors. + No dependencies, data, local modules, cloud runners or credentials are inferred. + Read the generated README and explicitly run check.py after installing deps. + """ + destination = Path(out_dir).absolute() + if destination.exists() or destination.is_symlink(): + raise FileExistsError("Bundle destination must not already exist") + sources = { + "src/program.py": _read_python(Path(initial_program)), + "src/openevolve_evaluator.py": _read_python(Path(evaluator), evaluator=True), + } + # Validate adapter options without loading the supplied evaluator. + adapt_openevolve_evaluator( + lambda path: {}, objective_metric=objective_metric, failure_score=failure_score + ) + experiment = config_to_alphaevolve( + config, + title=title, + problem_description=problem_description, + model_mapping=model_mapping, + generation_concurrency=generation_concurrency, + ) + settings = {"objective_metric": objective_metric, "failure_score": float(failure_score)} + for name, value in (("experiment.json", experiment), ("evaluation.json", settings)): + sources[name] = ( + json.dumps(value, ensure_ascii=False, allow_nan=False, indent=2) + "\n" + ).encode("utf-8") + sources.update( + { + "src/__init__.py": b"", + "src/evaluate.py": _EVALUATOR.encode("utf-8"), + "check.py": _CHECK.encode("utf-8"), + "README.md": _README.encode("utf-8"), + } + ) + if requirements_file is not None: + dependencies = Path(requirements_file).read_bytes() + dependencies.decode("utf-8-sig") + sources["requirements.txt"] = dependencies + + # mkdir is exclusive, so a pre-existing directory is never owned or removed. + destination.mkdir() + try: + (destination / "src").mkdir() + for name, data in sources.items(): + (destination / name).write_bytes(data) + except BaseException: + shutil.rmtree(destination, ignore_errors=True) + raise + return destination + + +def main(argv=None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("initial_program", type=Path) + parser.add_argument("evaluator", type=Path) + parser.add_argument("--config", type=Path, required=True) + parser.add_argument("--output-dir", type=Path, required=True) + parser.add_argument("--title", required=True) + parser.add_argument("--instructions", type=Path, required=True) + parser.add_argument("--objective-metric", required=True) + parser.add_argument("--failure-score", type=float, default=-1e12) + parser.add_argument("--model-map", action="append", default=[], metavar="SOURCE=GEMINI") + parser.add_argument("--generation-concurrency", type=int) + parser.add_argument("--requirements-file", type=Path) + args = parser.parse_args(argv) + try: + mapping = {} + for entry in args.model_map: + source, separator, target = entry.partition("=") + if not separator or not source or not target or source in mapping: + raise ValueError("--model-map requires unique SOURCE=GEMINI entries") + mapping[source] = target + config = Config.from_yaml(args.config) + if config.language is None: + config.language = "python" # Both source inputs must be validated .py files. + output = export_to_alphaevolve( + args.initial_program, + args.evaluator, + config, + args.output_dir, + title=args.title, + problem_description=args.instructions.read_text(encoding="utf-8"), + objective_metric=args.objective_metric, + failure_score=args.failure_score, + model_mapping=mapping, + generation_concurrency=args.generation_concurrency, + requirements_file=args.requirements_file, + ) + except (OSError, TypeError, ValueError, YAMLError) as exc: + parser.error(str(exc)) + print(output) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_alphaevolve_bundle.py b/tests/test_alphaevolve_bundle.py new file mode 100644 index 000000000..3995ff271 --- /dev/null +++ b/tests/test_alphaevolve_bundle.py @@ -0,0 +1,350 @@ +"""Offline bundle creation, relocation, execution and failure-path tests.""" + +import contextlib +import copy +import io +import json +import os +import shutil +import subprocess +import sys +import tempfile +import unittest +import warnings +from pathlib import Path +from unittest.mock import patch + +from openevolve.compat.alphaevolve_bundle import export_to_alphaevolve, main +from openevolve.config import Config + +EVALUATOR = """from dataclasses import dataclass +from pathlib import Path + +@dataclass +class Score: + value: int + +def evaluate(path): + print("evaluator diagnostic") + return {"score": Score(len(Path(path).read_bytes())).value} +""" + + +class TestAlphaEvolveBundle(unittest.TestCase): + def setUp(self): + self.temporary = tempfile.TemporaryDirectory() + self.addCleanup(self.temporary.cleanup) + self.root = Path(self.temporary.name) + self.inputs = self.root / "inputs" + self.inputs.mkdir() + self.program = self.inputs / "seed.py" + self.program.write_bytes("# 中文\r\n7\r\n".encode("utf-8")) + self.evaluator = self.inputs / "evaluator.py" + self.evaluator.write_text(EVALUATOR, encoding="utf-8") + self.config = Config.from_dict( + { + "language": "python", + "max_iterations": 2, + "llm": {"models": [{"name": "gemini-2.5-flash", "weight": 1}]}, + } + ) + self.output = self.root / "bundle" + + def export(self, **options): + arguments = { + "title": "Offline bundle", + "problem_description": "Count bytes in a seed. 中文。", + "objective_metric": "score", + **options, + } + with warnings.catch_warnings(): + warnings.simplefilter("ignore", UserWarning) + return export_to_alphaevolve( + self.program, self.evaluator, self.config, self.output, **arguments + ) + + def check(self, bundle=None): + environment = os.environ.copy() + environment.pop("PYTHONPATH", None) + elsewhere = self.root / "elsewhere" + elsewhere.mkdir(exist_ok=True) + return subprocess.run( + [sys.executable, str((bundle or self.output) / "check.py")], + cwd=elsewhere, + env=environment, + capture_output=True, + text=True, + timeout=20, + ) + + def cli_arguments(self): + self.yaml = self.inputs / "config.yaml" + # Deliberately omit language, as existing Python example YAMLs do. + self.yaml.write_text( + "max_iterations: 2\nllm:\n primary_model: local-model\n", encoding="utf-8" + ) + self.instructions = self.inputs / "instructions.md" + self.instructions.write_text("Count source bytes. 中文。\n", encoding="utf-8") + return [ + str(self.program), + str(self.evaluator), + "--config", + str(self.yaml), + "--output-dir", + str(self.output), + "--title", + "CLI example", + "--instructions", + str(self.instructions), + "--objective-metric", + "score", + "--model-map", + "local-model=gemini-2.5-flash", + ] + + def test_bundle_has_only_expected_files_and_preserves_source_bytes(self): + before = copy.deepcopy(self.config) + output = self.export() + self.assertEqual(output, self.output.absolute()) + self.assertEqual( + {p.relative_to(output).as_posix() for p in output.rglob("*") if p.is_file()}, + { + "src/program.py", + "src/openevolve_evaluator.py", + "src/evaluate.py", + "src/__init__.py", + "experiment.json", + "evaluation.json", + "check.py", + "README.md", + }, + ) + self.assertEqual((output / "src/program.py").read_bytes(), self.program.read_bytes()) + self.assertEqual( + (output / "src/openevolve_evaluator.py").read_bytes(), self.evaluator.read_bytes() + ) + experiment = json.loads((output / "experiment.json").read_text()) + self.assertEqual(experiment["run_settings"], {"max_programs": 2}) + self.assertEqual(experiment["problem_description"], "Count bytes in a seed. 中文。") + self.assertEqual(self.config, before) + self.assertNotIn(str(self.inputs), (output / "README.md").read_text()) + + def test_relocated_bundle_runs_without_original_files_or_original_working_directory(self): + expected_score = len(self.program.read_bytes()) + self.export() + moved = self.root / "relocated" + self.output.rename(moved) + shutil.rmtree(self.inputs) + completed = self.check(moved) + self.assertEqual(completed.returncode, 0, completed.stderr) + result = json.loads(completed.stdout) + self.assertEqual(result["scores"]["scores"], [{"metric": "score", "score": expected_score}]) + self.assertIn("evaluator diagnostic", completed.stderr) + self.assertNotIn("evaluator diagnostic", completed.stdout) + + def test_export_does_not_import_execute_sources_or_connect_to_network(self): + marker = self.root / "must-not-exist" + side_effect = f"from pathlib import Path\nPath({str(marker)!r}).touch()\n" + self.program.write_text(side_effect + "x = 1\n", encoding="utf-8") + self.evaluator.write_text(side_effect + EVALUATOR, encoding="utf-8") + with patch("socket.socket.connect", side_effect=AssertionError("network forbidden")): + self.export() + self.assertFalse(marker.exists()) + + def test_explicit_dependency_file_is_copied_without_discovering_other_files(self): + dependencies = self.inputs / "deps.txt" + dependencies.write_bytes(b"# reviewed dependencies\r\nnumpy>=1.22\r\n") + (self.inputs / "private-data.txt").write_text("not selected") + self.export(requirements_file=dependencies) + self.assertEqual((self.output / "requirements.txt").read_bytes(), dependencies.read_bytes()) + self.assertFalse((self.output / "private-data.txt").exists()) + + def test_existing_directory_file_and_symlinks_are_preserved(self): + for kind in ("directory", "file", "symlink", "dangling-symlink"): + with self.subTest(kind=kind): + destination = self.root / kind + if kind == "directory": + destination.mkdir() + (destination / "keep.txt").write_text("keep") + elif kind == "file": + destination.write_text("keep") + else: + destination.symlink_to( + self.root / ("inputs" if kind == "symlink" else "missing") + ) + with self.assertRaises(FileExistsError): + export_to_alphaevolve( + self.program, + self.evaluator, + self.config, + destination, + title="Example", + problem_description="Problem", + objective_metric="score", + ) + if kind == "directory": + self.assertEqual((destination / "keep.txt").read_text(), "keep") + elif kind == "file": + self.assertEqual(destination.read_text(), "keep") + else: + self.assertTrue(destination.is_symlink()) + + def test_write_error_and_interrupt_remove_only_new_output(self): + original_write = Path.write_bytes + for error in (OSError("test disk failure"), KeyboardInterrupt()): + calls = 0 + + def fail_write(path, data): + nonlocal calls + calls += 1 + if calls == 3: + raise error + return original_write(path, data) + + with self.subTest(error=type(error)), patch.object(Path, "write_bytes", fail_write): + with self.assertRaises(type(error)): + self.export() + self.assertFalse(self.output.exists()) + self.assertEqual(self.evaluator.read_text(), EVALUATOR) + + def test_destination_created_by_another_writer_is_not_removed(self): + original_mkdir = Path.mkdir + + def create_competing_directory(path, *args, **kwargs): + if path == self.output: + original_mkdir(path) + (path / "keep.txt").write_text("other writer") + raise FileExistsError("destination was claimed") + return original_mkdir(path, *args, **kwargs) + + with patch.object(Path, "mkdir", create_competing_directory): + with self.assertRaises(FileExistsError): + self.export() + self.assertEqual((self.output / "keep.txt").read_text(), "other writer") + + def test_invalid_python_encodings_syntax_and_scopes_are_rejected_before_writes(self): + for source in (b"def invalid(:\n", b"return 1\n", b"# coding: latin-1\nx=1\n", b"\xff"): + self.program.write_bytes(source) + with self.subTest(source=source), self.assertRaises(ValueError): + self.export() + self.assertFalse(self.output.exists()) + + def test_bom_utf8_is_preserved(self): + source = b"\xef\xbb\xbf7\r\n" + self.program.write_bytes(source) + self.export() + self.assertEqual((self.output / "src/program.py").read_bytes(), source) + self.assertEqual(self.check().returncode, 0) + + def test_relative_and_sibling_imports_are_rejected_before_writes(self): + (self.inputs / "helper.py").write_text("x = 1") + (self.inputs / "local_package").mkdir() + for imported in ( + "from .helper import x\n", + "from . import helper\n", + "import helper\n", + "from helper import x\n", + "import local_package.module\n", + ): + self.evaluator.write_text(imported + EVALUATOR, encoding="utf-8") + with self.subTest(imported=imported), self.assertRaisesRegex(ValueError, "imports"): + self.export() + self.assertFalse(self.output.exists()) + + def test_evaluator_must_accept_one_path_synchronously(self): + for source in ( + "async def evaluate(path): return {}\n", + "def other(path): return {}\n", + "def evaluate(): return {}\n", + "def evaluate(path, other): return {}\n", + "def evaluate(path, *, other): return {}\n", + ): + self.evaluator.write_text(source) + with self.subTest(source=source), self.assertRaisesRegex(ValueError, "Evaluator"): + self.export() + self.assertFalse(self.output.exists()) + + def test_optional_evaluator_arguments_are_supported(self): + self.evaluator.write_text("def evaluate(path, other=None): return {'score': 2}\n") + self.export() + self.assertEqual(self.check().returncode, 0) + + def test_invalid_config_and_adapter_options_leave_no_output(self): + for options in ({"failure_score": float("nan")}, {"objective_metric": "error"}): + with self.subTest(options=options), self.assertRaises(ValueError): + self.export(**options) + self.assertFalse(self.output.exists()) + self.config.llm.models[0].name = "local-model" + with self.assertRaises(ValueError): + self.export() + self.assertFalse(self.output.exists()) + + def test_missing_inputs_and_dependencies_leave_no_partial_output(self): + with self.assertRaises(FileNotFoundError): + self.export(requirements_file=self.inputs / "missing.txt") + self.assertFalse(self.output.exists()) + self.program.unlink() + with self.assertRaises(FileNotFoundError): + self.export() + self.assertFalse(self.output.exists()) + + def test_missing_destination_parent_does_not_create_it(self): + self.output = self.root / "missing-parent" / "bundle" + with self.assertRaises(FileNotFoundError): + self.export() + self.assertFalse(self.output.parent.exists()) + + def test_failed_seed_check_has_nonzero_status_and_sanitized_json(self): + self.evaluator.write_text("def evaluate(path): raise RuntimeError('private-detail')\n") + self.export(failure_score=-123.0) + result = self.check() + self.assertEqual(result.returncode, 1, result.stderr) + self.assertEqual(json.loads(result.stdout)["scores"]["scores"][0]["score"], -123.0) + self.assertNotIn("private-detail", result.stdout) + + def test_unicode_and_code_like_metric_names_remain_data(self): + metric = "score'); raise Exception('no'); #\n中文🚀" + self.evaluator.write_text( + f"def evaluate(path): return {{{metric!r}: 7}}\n", encoding="utf-8" + ) + self.export(objective_metric=metric) + self.assertNotIn(metric, (self.output / "src/evaluate.py").read_text()) + result = self.check() + self.assertEqual(result.returncode, 0, result.stderr) + self.assertEqual(json.loads(result.stdout)["scores"]["scores"][0]["metric"], metric) + + def test_cli_exports_with_explicit_model_mapping_and_preserves_yaml(self): + arguments = self.cli_arguments() + original = self.yaml.read_bytes() + with warnings.catch_warnings(), contextlib.redirect_stdout(io.StringIO()) as output: + warnings.simplefilter("ignore", UserWarning) + self.assertEqual(main(arguments), 0) + self.assertEqual(output.getvalue().strip(), str(self.output)) + self.assertEqual(self.yaml.read_bytes(), original) + self.assertEqual(self.check().returncode, 0) + experiment = json.loads((self.output / "experiment.json").read_text()) + self.assertEqual(experiment["program_language"], "python") + self.assertEqual(experiment["generation_settings"]["models"][0]["name"], "gemini-2.5-flash") + + def test_cli_invalid_or_duplicate_mappings_fail_without_output(self): + arguments = self.cli_arguments() + for entry in ("not-a-mapping", "local-model=gemini-other"): + with self.subTest(entry=entry), contextlib.redirect_stderr(io.StringIO()): + with self.assertRaises(SystemExit) as caught: + main(arguments + ["--model-map", entry]) + self.assertEqual(caught.exception.code, 2) + self.assertFalse(self.output.exists()) + + def test_cli_malformed_yaml_is_an_argument_error_without_partial_output(self): + arguments = self.cli_arguments() + self.yaml.write_text("llm: [\n") + with contextlib.redirect_stderr(io.StringIO()) as errors: + with self.assertRaises(SystemExit) as caught: + main(arguments) + self.assertEqual(caught.exception.code, 2) + self.assertNotIn("Traceback", errors.getvalue()) + self.assertFalse(self.output.exists()) + + +if __name__ == "__main__": + unittest.main()