Spaces:
Running
Running
| """Vérificateur : porte de compilation JIT, cache, puis pytest dans le bac à sable de M.O.U.H.""" | |
| from __future__ import annotations | |
| import re | |
| from typing import Optional | |
| from mouh.lab import with_lab | |
| from mouh.skills.sandbox import SandboxLimits, run_in_sandbox | |
| from ..contracts import Candidate | |
| from ..jev.compile_gate import SHARED_CACHE, VerificationCache, compile_files, fingerprint | |
| class PytestVerifier: | |
| """Comme count_passed, mais garde la sortie pytest : c'est le diagnostic de la réparation suivante.""" | |
| def __init__(self, timeout_s: float = 20.0, cache: Optional[VerificationCache] = SHARED_CACHE) -> None: | |
| self.timeout_s, self.cache = timeout_s, cache | |
| self.runs = self.compile_rejects = 0 | |
| def score(self, files, tests) -> Candidate: | |
| key = fingerprint(files, tests) | |
| if self.cache is not None and (hit := self.cache.get(key)) is not None: | |
| return Candidate(files=files, cached=True, **hit) | |
| report = compile_files(files) | |
| if not report.ok: | |
| self.compile_rejects += 1 | |
| result = {"passed": 0, "failed": max(1, sum(c.count("def test_") for c in tests.values())), | |
| "ok": False, "output": report.as_output()} | |
| else: | |
| self.runs += 1 | |
| run = run_in_sandbox(["-m", "pytest", "-q", "-p", "no:cacheprovider", "--tb=short", | |
| *sorted(k for k in tests if k.endswith(".py"))], # pas lab_spec.json | |
| {**with_lab(files, tests), **files, **tests}, timeout_s=self.timeout_s, | |
| limits=SandboxLimits(cpu_seconds=int(self.timeout_s))) | |
| out = run.stdout + run.stderr | |
| passed = int(m.group(1)) if (m := re.search(r"(\d+) passed", out)) else 0 | |
| failed = sum(int(x) for x in re.findall(r"(\d+) (?:failed|error)", out)) | |
| if run.timed_out: | |
| out += f"\n(temps limite de {self.timeout_s:.0f} s dépassé)" | |
| result = {"passed": passed, "failed": failed, "ok": run.returncode == 0 and not run.timed_out, | |
| "output": out[-1500:]} | |
| if self.cache is not None: | |
| self.cache.put(key, result) | |
| return Candidate(files=files, **result) | |