Spaces:

Erinaldordso
/

Minecraft-Friends-AI-ZeroGPU

Paused

Minecraft-Friends-AI-ZeroGPU / tests /test_python_names_resolve.py

Exercise the exit report instead of matching strings, and fix the name it needed

aa8772a verified about 1 month ago

6.42 kB

"""Every name the runtime touches must resolve.
Why this exists: a retry loop was added to voyager_runner.py that called
time.sleep() without importing time. The tests at the time only asserted that
certain strings appeared in the file, so they passed while the new code raised
NameError on its first use -- inside the except block, which turned a retryable
provider error into a hard crash. Gemini_Bot's restart count climbed 1 -> 4 ->
8 -> 9 and the cause was invisible.
pyflakes reports "undefined name", which is that exact failure class. Skipped
rather than failed when pyflakes is unavailable, so the suite still runs in a
bare environment.
"""
from __future__ import annotations

import json
import subprocess
import sys
from pathlib import Path

import pytest

ROOT = Path(__file__).resolve().parents[1]

# Modules this deployment maintains. Upstream files are excluded: they are
# pinned and must not be edited, so a warning there is not actionable.
MAINTAINED = (
    "app.py",
    "minecraft_stack.py",
    "voyager_stack.py",
    "voyager_runner.py",
    "voyager_verification.py",
    "voyager_learning.py",
    "voyager_metrics.py",
    "voyager_creative_critic.py",
    "voyager_runtime_contract.py",
    "gemini_bot_stack.py",
    "gemini_voyager_proxy.py",
    "gemini_bot_logic.js",
)


def _contract():
    sys.path.insert(0, str(ROOT))
    import voyager_runtime_contract

    return voyager_runtime_contract


def _tmp_home() -> str:
    import tempfile

    return tempfile.mkdtemp(prefix="exit-report-")


def _contract():
    sys.path.insert(0, str(ROOT))
    import voyager_runtime_contract

    return voyager_runtime_contract


def _tmp_home() -> str:
    import tempfile

    return tempfile.mkdtemp(prefix="exit-report-")


def _pyflakes() -> str | None:
    try:
        import pyflakes  # noqa: F401
    except ImportError:
        return None
    return sys.executable


def test_no_undefined_names_in_the_maintained_modules():
    interpreter = _pyflakes()
    if interpreter is None:
        pytest.skip("pyflakes is not installed; cannot check for undefined names")

    targets = [str(ROOT / name) for name in MAINTAINED if (ROOT / name).suffix == ".py"]
    assert targets, "no Python modules to check"

    result = subprocess.run(
        [interpreter, "-m", "pyflakes", *targets],
        capture_output=True,
        text=True,
    )
    undefined = [
        line
        for line in result.stdout.splitlines()
        if "undefined name" in line or "undefined local" in line
    ]
    assert not undefined, (
        "a name used at runtime does not resolve; this raises at execution time, "
        "not at import time, so only a real call would surface it:\n  "
        + "\n  ".join(undefined)
    )


def test_the_retry_loop_imports_what_it_uses():
    # The specific regression, asserted directly as well as by the check above.
    source = (ROOT / "voyager_runner.py").read_text(encoding="utf-8")
    assert "time.sleep(delay)" in source
    head = source[: source.index("VOYAGER_HOME = Path(")]
    assert "\nimport time\n" in head, "voyager_runner uses time.sleep but does not import time"


def test_the_retry_sleep_is_inside_the_except_block():
    # If the sleep sat outside the except, the loop would not back off.
    source = (ROOT / "voyager_runner.py").read_text(encoding="utf-8")
    start = source.index("while True:")
    end = source.index("finally:", start)
    body = source[start:end]
    assert "except Exception as exc:" in body
    assert body.index("except Exception as exc:") < body.index("time.sleep(delay)")
    assert body.index("time.sleep(delay)") < body.index("delay = min(delay * 2, delay_cap)")


def test_a_crash_is_recorded_so_the_next_restart_is_diagnosable():
    # The HF log stream served a frozen window and the Gemini status exposes no
    # log tail, so 'runner=1' was the only symptom available.
    contract = _contract()
    home = Path(_tmp_home()) / "Voyager"
    home.mkdir(parents=True, exist_ok=True)

    try:
        raise RuntimeError("provider rejected sk-abcdefghijklmnop with 401")
    except RuntimeError as exc:
        record = contract.record_exit(home, "Gemini_Bot", exc)

    path = home.parent / "last-exit-Gemini_Bot.json"
    assert path.exists(), "the report must land beside the checkout"
    on_disk = json.loads(path.read_text(encoding="utf-8"))

    assert on_disk["bot_username"] == "Gemini_Bot"
    assert on_disk["pid"] > 0
    assert on_disk["error"].startswith("RuntimeError:")
    assert "401" in on_disk["error"], "the message must survive"
    # It is published through a status endpoint, so it must not carry a credential.
    assert "sk-abcdefghijklmnop" not in json.dumps(on_disk)
    assert "<redacted>" in on_disk["error"]
    assert any("RuntimeError" in line for line in on_disk["traceback"]), (
        "the traceback must identify the failure"
    )
    # And it must be readable back through the contract.
    assert contract.read_exit_report(home, "Gemini_Bot")["error"] == record["error"]


def test_each_bot_gets_its_own_report():
    contract = _contract()
    root = Path(_tmp_home())
    (root / "Voyager").mkdir(parents=True, exist_ok=True)
    contract.record_exit(root / "Voyager", "Naldo_Bot", ValueError("a"))
    contract.record_exit(root / "Voyager", "Gemini_Bot", ValueError("b"))
    assert contract.read_exit_report(root / "Voyager", "Naldo_Bot")["error"].endswith("a")
    assert contract.read_exit_report(root / "Voyager", "Gemini_Bot")["error"].endswith("b")


def test_a_write_failure_does_not_mask_the_original_error(tmp_path):
    contract = _contract()
    # Point the report at a path that cannot be created.
    blocker = tmp_path / "blocked"
    blocker.write_text("not a directory", encoding="utf-8")
    assert contract.record_exit(blocker / "Voyager", "Naldo_Bot", ValueError("x")) == {}


def test_the_exit_report_is_written_outside_the_pinned_checkout():
    # A runtime file inside the checkout would look like an undeclared addition
    # to the divergence check and fail the next boot.
    import sys as _sys

    _sys.path.insert(0, str(ROOT))
    import voyager_runtime_contract as contract

    home = ROOT / "work" / "patched3" / "Voyager"
    path = contract.exit_report_path(home, "Gemini_Bot")
    assert path.parent == home.parent, "the report must sit beside the checkout, not inside it"
    assert path.name == "last-exit-Gemini_Bot.json"

	"""Every name the runtime touches must resolve.

	Why this exists: a retry loop was added to voyager_runner.py that called
	time.sleep() without importing time. The tests at the time only asserted that
	certain strings appeared in the file, so they passed while the new code raised
	NameError on its first use -- inside the except block, which turned a retryable
	provider error into a hard crash. Gemini_Bot's restart count climbed 1 -> 4 ->
	8 -> 9 and the cause was invisible.

	pyflakes reports "undefined name", which is that exact failure class. Skipped
	rather than failed when pyflakes is unavailable, so the suite still runs in a
	bare environment.
	"""
	from __future__ import annotations

	import json
	import subprocess
	import sys
	from pathlib import Path

	import pytest

	ROOT = Path(__file__).resolve().parents[1]

	# Modules this deployment maintains. Upstream files are excluded: they are
	# pinned and must not be edited, so a warning there is not actionable.
	MAINTAINED = (
	"app.py",
	"minecraft_stack.py",
	"voyager_stack.py",
	"voyager_runner.py",
	"voyager_verification.py",
	"voyager_learning.py",
	"voyager_metrics.py",
	"voyager_creative_critic.py",
	"voyager_runtime_contract.py",
	"gemini_bot_stack.py",
	"gemini_voyager_proxy.py",
	"gemini_bot_logic.js",
	)


	def _contract():
	sys.path.insert(0, str(ROOT))
	import voyager_runtime_contract

	return voyager_runtime_contract


	def _tmp_home() -> str:
	import tempfile

	return tempfile.mkdtemp(prefix="exit-report-")


	def _contract():
	sys.path.insert(0, str(ROOT))
	import voyager_runtime_contract

	return voyager_runtime_contract


	def _tmp_home() -> str:
	import tempfile

	return tempfile.mkdtemp(prefix="exit-report-")


	def _pyflakes() -> str \| None:
	try:
	import pyflakes # noqa: F401
	except ImportError:
	return None
	return sys.executable


	def test_no_undefined_names_in_the_maintained_modules():
	interpreter = _pyflakes()
	if interpreter is None:
	pytest.skip("pyflakes is not installed; cannot check for undefined names")

	targets = [str(ROOT / name) for name in MAINTAINED if (ROOT / name).suffix == ".py"]
	assert targets, "no Python modules to check"

	result = subprocess.run(
	[interpreter, "-m", "pyflakes", *targets],
	capture_output=True,
	text=True,
	)
	undefined = [
	line
	for line in result.stdout.splitlines()
	if "undefined name" in line or "undefined local" in line
	]
	assert not undefined, (
	"a name used at runtime does not resolve; this raises at execution time, "
	"not at import time, so only a real call would surface it:\n "
	+ "\n ".join(undefined)
	)


	def test_the_retry_loop_imports_what_it_uses():
	# The specific regression, asserted directly as well as by the check above.
	source = (ROOT / "voyager_runner.py").read_text(encoding="utf-8")
	assert "time.sleep(delay)" in source
	head = source[: source.index("VOYAGER_HOME = Path(")]
	assert "\nimport time\n" in head, "voyager_runner uses time.sleep but does not import time"


	def test_the_retry_sleep_is_inside_the_except_block():
	# If the sleep sat outside the except, the loop would not back off.
	source = (ROOT / "voyager_runner.py").read_text(encoding="utf-8")
	start = source.index("while True:")
	end = source.index("finally:", start)
	body = source[start:end]
	assert "except Exception as exc:" in body
	assert body.index("except Exception as exc:") < body.index("time.sleep(delay)")
	assert body.index("time.sleep(delay)") < body.index("delay = min(delay * 2, delay_cap)")


	def test_a_crash_is_recorded_so_the_next_restart_is_diagnosable():
	# The HF log stream served a frozen window and the Gemini status exposes no
	# log tail, so 'runner=1' was the only symptom available.
	contract = _contract()
	home = Path(_tmp_home()) / "Voyager"
	home.mkdir(parents=True, exist_ok=True)

	try:
	raise RuntimeError("provider rejected sk-abcdefghijklmnop with 401")
	except RuntimeError as exc:
	record = contract.record_exit(home, "Gemini_Bot", exc)

	path = home.parent / "last-exit-Gemini_Bot.json"
	assert path.exists(), "the report must land beside the checkout"
	on_disk = json.loads(path.read_text(encoding="utf-8"))

	assert on_disk["bot_username"] == "Gemini_Bot"
	assert on_disk["pid"] > 0
	assert on_disk["error"].startswith("RuntimeError:")
	assert "401" in on_disk["error"], "the message must survive"
	# It is published through a status endpoint, so it must not carry a credential.
	assert "sk-abcdefghijklmnop" not in json.dumps(on_disk)
	assert "<redacted>" in on_disk["error"]
	assert any("RuntimeError" in line for line in on_disk["traceback"]), (
	"the traceback must identify the failure"
	)
	# And it must be readable back through the contract.
	assert contract.read_exit_report(home, "Gemini_Bot")["error"] == record["error"]


	def test_each_bot_gets_its_own_report():
	contract = _contract()
	root = Path(_tmp_home())
	(root / "Voyager").mkdir(parents=True, exist_ok=True)
	contract.record_exit(root / "Voyager", "Naldo_Bot", ValueError("a"))
	contract.record_exit(root / "Voyager", "Gemini_Bot", ValueError("b"))
	assert contract.read_exit_report(root / "Voyager", "Naldo_Bot")["error"].endswith("a")
	assert contract.read_exit_report(root / "Voyager", "Gemini_Bot")["error"].endswith("b")


	def test_a_write_failure_does_not_mask_the_original_error(tmp_path):
	contract = _contract()
	# Point the report at a path that cannot be created.
	blocker = tmp_path / "blocked"
	blocker.write_text("not a directory", encoding="utf-8")
	assert contract.record_exit(blocker / "Voyager", "Naldo_Bot", ValueError("x")) == {}


	def test_the_exit_report_is_written_outside_the_pinned_checkout():
	# A runtime file inside the checkout would look like an undeclared addition
	# to the divergence check and fail the next boot.
	import sys as _sys

	_sys.path.insert(0, str(ROOT))
	import voyager_runtime_contract as contract

	home = ROOT / "work" / "patched3" / "Voyager"
	path = contract.exit_report_path(home, "Gemini_Bot")
	assert path.parent == home.parent, "the report must sit beside the checkout, not inside it"
	assert path.name == "last-exit-Gemini_Bot.json"