test: finalize E2E timeout verification and purge global state leaks

This commit is contained in:
2026-05-02 18:59:35 +02:00
parent d2de5f91de
commit f32ee46d8c
44 changed files with 740 additions and 1387 deletions

View File

@@ -153,6 +153,16 @@ def isolated_screen_memory(monkeypatch):
yield db
@pytest.fixture(scope="function", autouse=True)
def reset_physics_singletons():
"""Resets the PhysicsBody and SendEventInjector singletons to prevent state pollution across tests."""
from GramAddict.core.physics.biomechanics import PhysicsBody
from GramAddict.core.physics.sendevent_injector import SendEventInjector
PhysicsBody._instance = None
SendEventInjector.reset()
# ═══════════════════════════════════════════════════════
# Device Dump Injectors
# ═══════════════════════════════════════════════════════
@@ -196,8 +206,11 @@ def make_real_device_with_xml(monkeypatch):
def dump_hierarchy(self, compressed=False):
if isinstance(self.xml, list):
res = self.xml.pop(0) if self.xml else ""
return res
if len(self.xml) > 1:
return self.xml.pop(0)
elif len(self.xml) == 1:
return self.xml[0]
return ""
return self.xml
def screenshot(self):
@@ -208,11 +221,42 @@ def make_real_device_with_xml(monkeypatch):
def app_current(self):
return {"package": "com.instagram.android"}
def _validate_hitbox(self, x, y):
import re
import xml.etree.ElementTree as ET
try:
current_xml = self.xml[0] if isinstance(self.xml, list) and len(self.xml) > 0 else self.xml
if not current_xml:
return
root = ET.fromstring(current_xml)
for node in root.iter():
bounds_str = node.attrib.get("bounds", "")
is_actionable = (
node.attrib.get("clickable", "false") == "true"
or node.attrib.get("long-clickable", "false") == "true"
or node.attrib.get("scrollable", "false") == "true"
or node.attrib.get("focusable", "false") == "true"
)
if bounds_str and is_actionable:
match = re.match(r"\[(\d+),(\d+)\]\[(\d+),(\d+)\]", bounds_str)
if match:
left, top, right, bottom = map(int, match.groups())
if left <= x <= right and top <= y <= bottom:
return
raise AssertionError(
f"🛑 ZERO-TRUST VIOLATION: Bot clicked ({x}, {y}) but there is NO actionable UI element at these coordinates! "
f"The VLM hallucinated or miscalculated bounds."
)
except ET.ParseError:
pass
def shell(self, cmd):
if isinstance(cmd, str) and cmd.startswith("input tap"):
parts = cmd.split()
try:
x, y = int(parts[-2]), int(parts[-1])
self._validate_hitbox(x, y)
self.interaction_log.append({"action": "click", "coords": (x, y)})
except (ValueError, IndexError):
pass
@@ -226,6 +270,7 @@ def make_real_device_with_xml(monkeypatch):
self.interaction_log.append({"action": "swipe", "start": (sx, sy), "end": (ex, ey)})
def click(self, x, y):
self._validate_hitbox(x, y)
self.interaction_log.append({"action": "click", "coords": (x, y)})
def watcher(self, name):
@@ -241,6 +286,20 @@ def make_real_device_with_xml(monkeypatch):
# Now we instantiate the REAL DeviceFacade!
device = DeviceFacade("test_device", "com.instagram.android", None)
# Expose legacy emulator properties mapping directly to the underlying MockU2Device interaction log
# so that existing test assertions keep working seamlessly.
type(device).interaction_log = property(lambda self: self.deviceV2.interaction_log)
type(device).pressed_keys = property(
lambda self: [log["key"] for log in self.deviceV2.interaction_log if log["action"] == "press"]
)
type(device).clicks = property(
lambda self: [log["coords"] for log in self.deviceV2.interaction_log if log["action"] == "click"]
)
type(device).swipes = property(
lambda self: [log["start"] for log in self.deviceV2.interaction_log if log["action"] == "swipe"]
)
return device
return _create
@@ -288,8 +347,11 @@ def make_real_device_with_image(monkeypatch):
def dump_hierarchy(self, compressed=False):
if self.xml:
if isinstance(self.xml, list):
res = self.xml.pop(0) if self.xml else ""
return res
if len(self.xml) > 1:
return self.xml.pop(0)
elif len(self.xml) == 1:
return self.xml[0]
return ""
return self.xml
return ""
@@ -299,16 +361,49 @@ def make_real_device_with_image(monkeypatch):
if res is None:
return Image.new("RGB", (1, 1), color="black")
return Image.open(res) if isinstance(res, str) else res
if self.img is None:
return Image.new("RGB", (1, 1), color="black")
return Image.open(self.img) if isinstance(self.img, str) else self.img
def app_current(self):
return {"package": "com.instagram.android"}
def _validate_hitbox(self, x, y):
import re
import xml.etree.ElementTree as ET
try:
current_xml = self.xml[0] if isinstance(self.xml, list) and len(self.xml) > 0 else self.xml
if not current_xml:
return
root = ET.fromstring(current_xml)
for node in root.iter():
bounds_str = node.attrib.get("bounds", "")
is_actionable = (
node.attrib.get("clickable", "false") == "true"
or node.attrib.get("long-clickable", "false") == "true"
or node.attrib.get("scrollable", "false") == "true"
or node.attrib.get("focusable", "false") == "true"
)
if bounds_str and is_actionable:
match = re.match(r"\[(\d+),(\d+)\]\[(\d+),(\d+)\]", bounds_str)
if match:
left, top, right, bottom = map(int, match.groups())
if left <= x <= right and top <= y <= bottom:
return
raise AssertionError(
f"🛑 ZERO-TRUST VIOLATION: Bot clicked ({x}, {y}) but there is NO actionable UI element at these coordinates! "
f"The VLM hallucinated or miscalculated bounds."
)
except ET.ParseError:
pass
def shell(self, cmd):
if isinstance(cmd, str) and cmd.startswith("input tap"):
parts = cmd.split()
try:
x, y = int(parts[-2]), int(parts[-1])
self._validate_hitbox(x, y)
self.interaction_log.append({"action": "click", "coords": (x, y)})
except (ValueError, IndexError):
pass
@@ -320,6 +415,7 @@ def make_real_device_with_image(monkeypatch):
self.interaction_log.append({"action": "swipe", "start": (sx, sy), "end": (ex, ey)})
def click(self, x, y):
self._validate_hitbox(x, y)
self.interaction_log.append({"action": "click", "coords": (x, y)})
def watcher(self, name):
@@ -333,6 +429,20 @@ def make_real_device_with_image(monkeypatch):
monkeypatch.setattr(device_facade.u2, "connect", mock_connect)
device = DeviceFacade("test_device", "com.instagram.android", None)
# Expose legacy emulator properties mapping directly to the underlying MockU2Device interaction log
# so that existing test assertions keep working seamlessly.
type(device).interaction_log = property(lambda self: self.deviceV2.interaction_log)
type(device).pressed_keys = property(
lambda self: [log["key"] for log in self.deviceV2.interaction_log if log["action"] == "press"]
)
type(device).clicks = property(
lambda self: [log["coords"] for log in self.deviceV2.interaction_log if log["action"] == "click"]
)
type(device).swipes = property(
lambda self: [log["start"] for log in self.deviceV2.interaction_log if log["action"] == "swipe"]
)
return device
return _create
@@ -674,170 +784,6 @@ def e2e_device():
return _create
@pytest.fixture(autouse=True)
def mock_llm_network_calls(monkeypatch, request):
"""Mocks ONLY the LLM network requests. The SAE structural logic remains 100% REAL."""
if request.config.getoption("--live"):
return
import json
from GramAddict.core import llm_provider
def fake_query_telepathic_llm(*args, **kwargs):
prompt = kwargs.get("user_prompt", args[3] if len(args) > 3 else "")
if not prompt and len(args) > 0:
prompt = args[0]
# Debug mock routing available at DEBUG log level if needed
if "OBSTACLE_LOCKED_SCREEN" in prompt and "FOREIGN_APP" in prompt:
return json.dumps({"situation": "OBSTACLE_FOREIGN_APP"})
if "MODAL, DIALOG, or POPUP" in prompt:
if "How are yo...Instagram?" in prompt or "Please lea... a rating!" in prompt or "Not Now" in prompt:
return json.dumps({"situation": "OBSTACLE_MODAL"})
return json.dumps({"situation": "NORMAL"})
import re
intent_match = re.search(r"intent: '([^']+)'", prompt)
intent = intent_match.group(1).lower() if intent_match else ""
if "selected_index" in prompt and "Find the exact box number" not in prompt:
if "profile tab" in intent:
match = re.search(r"\[(\d+)\][^\[]*profile_tab", prompt)
if match:
return json.dumps({"selected_index": int(match.group(1))})
if "following" in intent:
match = re.search(r"\[(\d+)\][^\[]*following", prompt.lower())
if match:
return json.dumps({"selected_index": int(match.group(1))})
if "first post" in intent:
# Find the first grid_card_layout_container or image_button
match = re.search(r"\[(\d+)\][^\[]*(?:grid_card_layout_container|image_button)", prompt)
if match:
return json.dumps({"selected_index": int(match.group(1))})
if "comment" in intent:
match = re.search(r"\[(\d+)\][^\[]*comment", prompt.lower())
if match:
return json.dumps({"selected_index": int(match.group(1))})
if "like" in intent:
match = re.search(r"\[(\d+)\][^\[]*like_button", prompt.lower())
if match:
return json.dumps({"selected_index": int(match.group(1))})
match = re.search(r"\[(\d+)\][^\[]*like", prompt.lower())
if match:
return json.dumps({"selected_index": int(match.group(1))})
if "message" in intent:
match = re.search(r"\[(\d+)\][^\[]*message", prompt.lower())
if match:
return json.dumps({"selected_index": int(match.group(1))})
if "search input" in intent or "search" in intent:
match = re.search(r"\[(\d+)\][^\[]*search", prompt.lower())
if match:
return json.dumps({"selected_index": int(match.group(1))})
if "story ring avatar" in intent:
match = re.search(r"\[(\d+)\][^\[]*story ring avatar", prompt.lower())
if match:
return json.dumps({"selected_index": int(match.group(1))})
if "author" in intent or "profile" in intent or "username" in intent:
match = re.search(r"\[(\d+)\][^\[]*row_feed_photo_profile", prompt.lower())
if match:
return json.dumps({"selected_index": int(match.group(1))})
match = re.search(r"\[(\d+)\][^\[]*row_search_user_container", prompt.lower())
if match:
return json.dumps({"selected_index": int(match.group(1))})
if "follow" in intent:
match = re.search(r"\[(\d+)\][^\[]*follow", prompt.lower())
if match:
return json.dumps({"selected_index": int(match.group(1))})
return json.dumps({"selected_index": None})
return json.dumps({"selected_index": 0})
if "Find the exact box number" in prompt:
if "tab" in intent:
# 🚨 VLM HALLUCINATION SIMULATION 🚨
# The real VLM often gets confused and picks a user profile or text instead of the bottom tab.
# If our intent_resolver didn't filter out non-tab items properly, the prompt will contain them.
# Let's actively try to pick a bad node first (e.g. 'desc='marisaundmarc'' or any non-tab keyword)
match_bad = re.search(r"\[(\d+)\][^\n]*?desc='marisaundmarc'", prompt.lower())
if match_bad:
return json.dumps({"box": int(match_bad.group(1))})
# If the bad node was successfully filtered, pick the correct tab
if "profile" in intent:
match = re.search(r"\[(\d+)\][^\n]*?desc='profile'", prompt.lower())
if match:
return json.dumps({"box": int(match.group(1))})
elif "explore" in intent or "search" in intent:
match = re.search(r"\[(\d+)\][^\n]*?search and explore", prompt.lower())
if match:
return json.dumps({"box": int(match.group(1))})
elif "reels" in intent:
match = re.search(r"\[(\d+)\][^\n]*?reels", prompt.lower())
if match:
return json.dumps({"box": int(match.group(1))})
elif "home" in intent:
match = re.search(r"\[(\d+)\][^\n]*?home", prompt.lower())
if match:
return json.dumps({"box": int(match.group(1))})
if "following" in intent:
match = re.search(r"\[(\d+)\][^\n]*?following", prompt.lower())
if match:
return json.dumps({"box": int(match.group(1))})
if "first post" in intent:
match = re.search(r"\[(\d+)\][^\n]*?(?:grid_card_layout_container|image_button)", prompt)
if match:
return json.dumps({"box": int(match.group(1))})
if "comment" in intent:
match = re.search(r"\[(\d+)\][^\n]*?comment", prompt.lower())
if match:
return json.dumps({"box": int(match.group(1))})
if "like" in intent:
match = re.search(r"\[(\d+)\][^\n]*?desc='like'", prompt.lower())
if match:
return json.dumps({"box": int(match.group(1))})
match = re.search(r"\[(\d+)\][^\n]*?like", prompt.lower())
if match:
return json.dumps({"box": int(match.group(1))})
if "message" in intent or "direct" in intent:
match = re.search(r"\[(\d+)\][^\n]*?message", prompt.lower())
if match:
return json.dumps({"box": int(match.group(1))})
if "search" in intent:
# We need to pick a search bar, not "Suchen" on keyboard
match = re.search(r"\[(\d+)\][^\n]*?search", prompt.lower())
if match:
return json.dumps({"box": int(match.group(1))})
if "story ring avatar" in intent:
match = re.search(r"\[(\d+)\][^\n]*?story ring avatar", prompt.lower())
if match:
return json.dumps({"box": int(match.group(1))})
if "author" in intent or "username" in intent:
# STRUCTURAL MATCH — match by resource_id pattern, not by hardcoded username.
# This is the fix for the 2026-05-01 lie where the mock searched for 'ninjatrader'
# but the real VLM picked 'Profile' tab instead of the actual author.
match = re.search(r"\[(\d+)\][^\n]*?row_feed_photo_profile", prompt.lower())
if match:
return json.dumps({"box": int(match.group(1))})
# Fallback for search feed usernames (since resource_id is stripped, we match the merged text)
match = re.search(r"\[(\d+)\][^\n]*?lokmatas", prompt.lower())
if match:
return json.dumps({"box": int(match.group(1))})
if "follow" in intent:
match = re.search(r"\[(\d+)\][^\n]*?follow", prompt.lower())
if match:
return json.dumps({"box": int(match.group(1))})
return json.dumps({"box": None})
return json.dumps({"box": None})
return json.dumps({"situation": "NORMAL"})
def fake_query_llm(*args, **kwargs):
return json.dumps({"action": "false_positive", "reason": "Test mock", "x": 0, "y": 0})
monkeypatch.setattr(llm_provider, "query_telepathic_llm", fake_query_telepathic_llm)
monkeypatch.setattr(llm_provider, "query_llm", fake_query_llm)
@pytest.fixture
def e2e_cognitive_stack_factory(e2e_configs):
"""Build the REAL cognitive stack exactly like bot_flow.py does."""