Marc Mintel
fddf14fd67
PRODUCTION FIX 2026-04-30 12:45: The entire E2E suite was systematically lying by testing a shell of the bot: 1. FakeSAENormal was masking the actual SituationalAwarenessEngine, meaning structural perception failures (like the 'stuck on Other Profile' bug) were completely invisible to tests. 2. The e2e_cognitive_stack was hardcoded to return None for 80% of engines (NavGraph, ZeroEngine, Telepathic, Darwin, ActiveInference), causing plugins to silently bypass their core logic during tests. 3. get_screenshot_b64 and screenshot() returned None, preventing the VLM stack from even running structural fallback code. Fix: - Systematically stripped FakeSAENormal from all 29 E2E workflow tests. - Built e2e_cognitive_stack_factory to instantiate the REAL Cognitive Stack (GrowthBrain, QNavGraph, ZeroLatencyEngine, etc.) just like in bot_flow.py. - Patched ONLY the external LLM network requests in conftest.py via mock_llm_network_calls so that structural execution stays 100% real but avoids non-deterministic LLM timeouts. - Injected a valid 1x1 black image for screenshots. All 29 E2E tests now execute the true production logic path and pass in 49 seconds.
722 lines
28 KiB
Python
722 lines
28 KiB
Python
"""
|
|
E2E Test Configuration — Hardened Test Infrastructure
|
|
======================================================
|
|
|
|
Design Principles:
|
|
1. No module-level mutable state (VirtualClock is a fixture, not a global)
|
|
2. No sys.modules poisoning (Qdrant mock via monkeypatch)
|
|
3. Unified fixture loading from a single source of truth
|
|
4. Global timeout to prevent infinite hangs in mocked loops
|
|
5. Deterministic loop termination via MaxIterationGuard
|
|
"""
|
|
|
|
import os
|
|
import signal
|
|
import time
|
|
|
|
import pytest
|
|
|
|
from GramAddict.core import utils
|
|
from GramAddict.core.session_state import SessionState
|
|
|
|
# ═══════════════════════════════════════════════════════
|
|
# CLI Options
|
|
# ═══════════════════════════════════════════════════════
|
|
|
|
|
|
def pytest_addoption(parser):
|
|
parser.addoption("--live", action="store_true", default=False, help="Run live tests")
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════
|
|
# Constants
|
|
# ═══════════════════════════════════════════════════════
|
|
|
|
E2E_TEST_TIMEOUT_SECONDS = 300
|
|
FIXTURES_DIR = os.path.join(os.path.dirname(os.path.dirname(__file__)), "fixtures")
|
|
E2E_FIXTURES_DIR = os.path.join(os.path.dirname(__file__), "fixtures")
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════
|
|
# Fixture Loading — Single Source of Truth
|
|
# ═══════════════════════════════════════════════════════
|
|
|
|
|
|
def load_fixture_xml(filename: str) -> str:
|
|
"""Load an XML fixture file. Checks e2e/fixtures first, then tests/fixtures.
|
|
|
|
Raises pytest.fail with a clear message if the fixture is missing.
|
|
"""
|
|
for fix_dir in (E2E_FIXTURES_DIR, FIXTURES_DIR):
|
|
path = os.path.join(fix_dir, filename)
|
|
if os.path.exists(path):
|
|
with open(path, "r", encoding="utf-8") as f:
|
|
return f.read()
|
|
|
|
pytest.fail(
|
|
f"MISSING REAL DUMP: '{filename}' not found in:\n"
|
|
f" - {E2E_FIXTURES_DIR}\n"
|
|
f" - {FIXTURES_DIR}\n"
|
|
f"Capture it using: python3 scripts/sync_fixtures.py --fixture {filename}",
|
|
pytrace=False,
|
|
)
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════
|
|
# Global Test Timeout — Prevents Infinite Hangs
|
|
# ═══════════════════════════════════════════════════════
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def e2e_test_timeout():
|
|
"""Hard timeout for every E2E test. Prevents mocked loops from hanging forever."""
|
|
|
|
def _timeout_handler(signum, frame):
|
|
pytest.fail(
|
|
f"E2E TEST TIMEOUT: Test exceeded {E2E_TEST_TIMEOUT_SECONDS}s. "
|
|
f"This almost certainly means the test entered an infinite loop "
|
|
f"due to exhausted mock side_effects or missing loop guards.",
|
|
pytrace=True,
|
|
)
|
|
|
|
old_handler = signal.signal(signal.SIGALRM, _timeout_handler)
|
|
signal.alarm(E2E_TEST_TIMEOUT_SECONDS)
|
|
yield
|
|
signal.alarm(E2E_TEST_TIMEOUT_SECONDS)
|
|
signal.signal(signal.SIGALRM, old_handler)
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════
|
|
# MaxIterationGuard — Deterministic Loop Termination
|
|
# ═══════════════════════════════════════════════════════
|
|
|
|
|
|
class MaxIterationGuard:
|
|
"""Prevents infinite loops in tests by counting iterations.
|
|
|
|
Usage:
|
|
guard = MaxIterationGuard(50, "feed loop")
|
|
while not done:
|
|
guard.tick() # Raises after 50 ticks
|
|
"""
|
|
|
|
def __init__(self, max_iterations: int, context: str = "unknown"):
|
|
self.max_iterations = max_iterations
|
|
self.context = context
|
|
self._count = 0
|
|
|
|
def tick(self):
|
|
self._count += 1
|
|
if self._count > self.max_iterations:
|
|
pytest.fail(
|
|
f"INFINITE LOOP DETECTED in '{self.context}': "
|
|
f"Exceeded {self.max_iterations} iterations. "
|
|
f"Fix the mock setup — the loop has no natural exit condition.",
|
|
pytrace=True,
|
|
)
|
|
|
|
@property
|
|
def count(self) -> int:
|
|
return self._count
|
|
|
|
|
|
@pytest.fixture
|
|
def iteration_guard():
|
|
"""Factory fixture for creating MaxIterationGuards."""
|
|
|
|
def _factory(max_iterations: int = 100, context: str = "e2e_loop"):
|
|
return MaxIterationGuard(max_iterations, context)
|
|
|
|
return _factory
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════
|
|
# Real Qdrant DB (Isolated Collection)
|
|
# ═══════════════════════════════════════════════════════
|
|
|
|
|
|
@pytest.fixture(scope="function", autouse=True)
|
|
def isolated_screen_memory(monkeypatch):
|
|
"""Ensures we use a separate Qdrant collection for E2E tests and clean it.
|
|
This replaces the old Qdrant mock so tests use the REAL database."""
|
|
from GramAddict.core.qdrant_memory import ScreenMemoryDB
|
|
|
|
def test_init(self, *args, **kwargs):
|
|
super(ScreenMemoryDB, self).__init__(collection_name="test_e2e_screens")
|
|
|
|
monkeypatch.setattr(ScreenMemoryDB, "__init__", test_init)
|
|
|
|
db = ScreenMemoryDB()
|
|
if db.is_connected:
|
|
db.wipe_collection()
|
|
|
|
yield db
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════
|
|
# Device Dump Injectors
|
|
# ═══════════════════════════════════════════════════════
|
|
|
|
|
|
@pytest.fixture
|
|
def make_real_device_with_xml(monkeypatch):
|
|
"""Provides a factory to create a REAL DeviceFacade but mocked uiautomator2."""
|
|
|
|
def _create(xml_content):
|
|
import GramAddict.core.device_facade as device_facade
|
|
from GramAddict.core.device_facade import DeviceFacade
|
|
|
|
class MockU2Watcher:
|
|
def when(self, xpath=None, **kwargs):
|
|
return self
|
|
|
|
def click(self):
|
|
return self
|
|
|
|
def start(self):
|
|
pass
|
|
|
|
class MockTouch:
|
|
def __init__(self, parent):
|
|
self.parent = parent
|
|
|
|
def down(self, x, y):
|
|
self.parent.interaction_log.append({"action": "click", "coords": (x, y)})
|
|
|
|
def up(self, x, y):
|
|
pass
|
|
|
|
class MockU2Device:
|
|
def __init__(self, xml):
|
|
self.xml = xml
|
|
self.info = {"sdkInt": 30, "displaySizeDpX": 400, "displayWidth": 1080, "screenOn": True}
|
|
self.settings = {}
|
|
self.interaction_log = []
|
|
self.touch = MockTouch(self)
|
|
|
|
def dump_hierarchy(self, compressed=False):
|
|
if isinstance(self.xml, list):
|
|
res = self.xml.pop(0) if self.xml else ""
|
|
return res
|
|
return self.xml
|
|
|
|
def screenshot(self):
|
|
from PIL import Image
|
|
return Image.new("RGB", (1, 1), color="black")
|
|
|
|
def app_current(self):
|
|
return {"package": "com.instagram.android"}
|
|
|
|
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.interaction_log.append({"action": "click", "coords": (x, y)})
|
|
except (ValueError, IndexError):
|
|
pass
|
|
elif isinstance(cmd, str) and cmd.startswith("input swipe"):
|
|
pass # We could log it if needed
|
|
|
|
def press(self, key):
|
|
self.interaction_log.append({"action": "press", "key": key})
|
|
|
|
def swipe(self, sx, sy, ex, ey, **kwargs):
|
|
self.interaction_log.append({"action": "swipe", "start": (sx, sy), "end": (ex, ey)})
|
|
|
|
def click(self, x, y):
|
|
self.interaction_log.append({"action": "click", "coords": (x, y)})
|
|
|
|
def watcher(self, name):
|
|
return MockU2Watcher()
|
|
|
|
def app_start(self, package_name, use_monkey=False):
|
|
pass
|
|
|
|
def mock_connect(*args, **kwargs):
|
|
return MockU2Device(xml_content)
|
|
|
|
monkeypatch.setattr(device_facade.u2, "connect", mock_connect)
|
|
|
|
# Now we instantiate the REAL DeviceFacade!
|
|
device = DeviceFacade("test_device", "com.instagram.android", None)
|
|
return device
|
|
|
|
return _create
|
|
|
|
|
|
@pytest.fixture
|
|
def make_real_device_with_image(monkeypatch):
|
|
"""Provides a factory to create a REAL DeviceFacade but mocked uiautomator2 returning a real image."""
|
|
|
|
def _create(img_path, xml_content=None):
|
|
from PIL import Image
|
|
|
|
import GramAddict.core.device_facade as device_facade
|
|
from GramAddict.core.device_facade import DeviceFacade
|
|
|
|
class MockU2Watcher:
|
|
def when(self, xpath=None, **kwargs):
|
|
return self
|
|
|
|
def click(self):
|
|
return self
|
|
|
|
def start(self):
|
|
pass
|
|
|
|
class MockTouch:
|
|
def __init__(self, parent):
|
|
self.parent = parent
|
|
|
|
def down(self, x, y):
|
|
self.parent.interaction_log.append({"action": "click", "coords": (x, y)})
|
|
|
|
def up(self, x, y):
|
|
pass
|
|
|
|
class MockU2Device:
|
|
def __init__(self, img, xml):
|
|
self.img = img
|
|
self.xml = xml
|
|
self.info = {"sdkInt": 30, "displaySizeDpX": 400, "displayWidth": 1080, "screenOn": True}
|
|
self.settings = {}
|
|
self.interaction_log = []
|
|
self.touch = MockTouch(self)
|
|
|
|
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
|
|
return self.xml
|
|
return ""
|
|
|
|
def screenshot(self):
|
|
if isinstance(self.img, list):
|
|
res = self.img.pop(0) if self.img else None
|
|
if res is None:
|
|
return Image.new("RGB", (1, 1), color="black")
|
|
return Image.open(res) if isinstance(res, str) else res
|
|
return Image.open(self.img) if isinstance(self.img, str) else self.img
|
|
|
|
def app_current(self):
|
|
return {"package": "com.instagram.android"}
|
|
|
|
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.interaction_log.append({"action": "click", "coords": (x, y)})
|
|
except (ValueError, IndexError):
|
|
pass
|
|
|
|
def press(self, key):
|
|
self.interaction_log.append({"action": "press", "key": key})
|
|
|
|
def swipe(self, sx, sy, ex, ey, **kwargs):
|
|
self.interaction_log.append({"action": "swipe", "start": (sx, sy), "end": (ex, ey)})
|
|
|
|
def click(self, x, y):
|
|
self.interaction_log.append({"action": "click", "coords": (x, y)})
|
|
|
|
def watcher(self, name):
|
|
return MockU2Watcher()
|
|
|
|
def app_start(self, package_name, use_monkey=False):
|
|
pass
|
|
|
|
def mock_connect(*args, **kwargs):
|
|
return MockU2Device(img_path, xml_content)
|
|
|
|
monkeypatch.setattr(device_facade.u2, "connect", mock_connect)
|
|
device = DeviceFacade("test_device", "com.instagram.android", None)
|
|
return device
|
|
|
|
return _create
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════
|
|
# Delay Mocking — Uses Fixture-Scoped Clock
|
|
# ═══════════════════════════════════════════════════════
|
|
|
|
|
|
def _patch_module_delays(monkeypatch, module_path: str, sleep_fn, random_sleep_fn):
|
|
"""Safely patch sleep/random in a single module. Missing attributes are skipped."""
|
|
import importlib
|
|
|
|
try:
|
|
mod = importlib.import_module(module_path)
|
|
except ImportError:
|
|
return # Module doesn't exist, nothing to patch
|
|
|
|
if hasattr(mod, "sleep"):
|
|
monkeypatch.setattr(mod, "sleep", sleep_fn)
|
|
if hasattr(mod, "random_sleep"):
|
|
monkeypatch.setattr(mod, "random_sleep", random_sleep_fn)
|
|
if hasattr(mod, "random") and hasattr(mod.random, "uniform"):
|
|
monkeypatch.setattr(mod.random, "uniform", lambda a, b: float(a))
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def mock_all_delays(monkeypatch, request):
|
|
"""Replaces all humanized hardware delays with no-ops."""
|
|
if request.config.getoption("--live"):
|
|
return
|
|
|
|
def money_sleep(*args, **kwargs):
|
|
pass
|
|
|
|
def random_sleep(*args, **kwargs):
|
|
pass
|
|
|
|
monkeypatch.setattr(time, "sleep", money_sleep)
|
|
monkeypatch.setattr(utils, "random_sleep", random_sleep)
|
|
monkeypatch.setattr(utils, "sleep", money_sleep)
|
|
if hasattr(utils, "random"):
|
|
monkeypatch.setattr(utils.random, "uniform", lambda a, b: float(a))
|
|
|
|
# Each module gets its own try-block so a missing attribute in one
|
|
# doesn't prevent patching the others.
|
|
_patch_module_delays(monkeypatch, "GramAddict.core.bot_flow", money_sleep, random_sleep)
|
|
_patch_module_delays(monkeypatch, "GramAddict.core.q_nav_graph", money_sleep, random_sleep)
|
|
_patch_module_delays(monkeypatch, "GramAddict.core.goap", money_sleep, random_sleep)
|
|
_patch_module_delays(monkeypatch, "GramAddict.core.device_facade", money_sleep, random_sleep)
|
|
_patch_module_delays(monkeypatch, "GramAddict.core.darwin_engine", money_sleep, random_sleep)
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════
|
|
# E2E Configs — Standardized Test Configuration
|
|
# ═══════════════════════════════════════════════════════
|
|
|
|
|
|
@pytest.fixture
|
|
def e2e_configs():
|
|
import argparse
|
|
|
|
args = argparse.Namespace(
|
|
username="testuser",
|
|
device="emulator-5554",
|
|
app_id="com.instagram.android",
|
|
debug=True,
|
|
feed=None,
|
|
carousel_percentage=0,
|
|
carousel_count="1",
|
|
explore=None,
|
|
reels=None,
|
|
stories=None,
|
|
interact_percentage=100,
|
|
likes_count="2-3",
|
|
likes_percentage=100,
|
|
follow_percentage=100,
|
|
comment_percentage=100,
|
|
stories_count="1-2",
|
|
stories_percentage=100,
|
|
working_hours=[0.0, 24.0],
|
|
time_delta_session=0,
|
|
speed_multiplier=1.0,
|
|
disable_filters=False,
|
|
interaction_users_amount="1",
|
|
scrape_profiles=False,
|
|
disable_ai_messaging=True,
|
|
total_unfollows_limit=0,
|
|
ai_telepathic_url="http://localhost",
|
|
ai_telepathic_model="llama3",
|
|
ai_condenser_url="http://localhost",
|
|
dry_run_comments=False,
|
|
visual_vibe_check_percentage=0,
|
|
)
|
|
|
|
from GramAddict.core.config import Config
|
|
|
|
config = Config(first_run=True)
|
|
config.args = args
|
|
config.username = "testuser"
|
|
config.config = {
|
|
"plugins": {
|
|
"likes": {"count": args.likes_count, "percentage": args.likes_percentage},
|
|
"comment": {
|
|
"percentage": args.comment_percentage,
|
|
"dry_run": args.dry_run_comments,
|
|
},
|
|
"follow": {"percentage": args.follow_percentage},
|
|
"stories": {
|
|
"count": args.stories_count,
|
|
"percentage": args.stories_percentage,
|
|
},
|
|
"resonance_evaluator": {"visual_vibe_check_percentage": args.visual_vibe_check_percentage},
|
|
"carousel_browsing": {
|
|
"percentage": getattr(args, "carousel_percentage", 0),
|
|
"count": getattr(args, "carousel_count", "1"),
|
|
},
|
|
}
|
|
}
|
|
return config
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════
|
|
# Plugin Registry — Standard Setup
|
|
# ═══════════════════════════════════════════════════════
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def setup_e2e_plugin_registry():
|
|
"""Ensures that all standard plugins are registered for E2E tests."""
|
|
from GramAddict.core.behaviors import PluginRegistry
|
|
from GramAddict.core.behaviors.ad_guard import AdGuardPlugin
|
|
from GramAddict.core.behaviors.anomaly_handler import AnomalyHandlerPlugin
|
|
from GramAddict.core.behaviors.carousel_browsing import CarouselBrowsingPlugin
|
|
from GramAddict.core.behaviors.close_friends_guard import CloseFriendsGuardPlugin
|
|
from GramAddict.core.behaviors.comment import CommentPlugin
|
|
from GramAddict.core.behaviors.darwin_dwell import DarwinDwellPlugin
|
|
from GramAddict.core.behaviors.follow import FollowPlugin
|
|
from GramAddict.core.behaviors.grid_like import GridLikePlugin
|
|
from GramAddict.core.behaviors.like import LikePlugin
|
|
from GramAddict.core.behaviors.obstacle_guard import ObstacleGuardPlugin
|
|
from GramAddict.core.behaviors.perfect_snapping import PerfectSnappingPlugin
|
|
from GramAddict.core.behaviors.post_data_extraction import PostDataExtractionPlugin
|
|
from GramAddict.core.behaviors.post_interaction import PostInteractionPlugin
|
|
from GramAddict.core.behaviors.profile_guard import ProfileGuardPlugin
|
|
from GramAddict.core.behaviors.profile_visit import ProfileVisitPlugin
|
|
from GramAddict.core.behaviors.rabbit_hole import RabbitHolePlugin
|
|
from GramAddict.core.behaviors.repost import RepostPlugin
|
|
from GramAddict.core.behaviors.resonance_evaluator import ResonanceEvaluatorPlugin
|
|
from GramAddict.core.behaviors.story_view import StoryViewPlugin
|
|
|
|
PluginRegistry.reset()
|
|
plugin_registry = PluginRegistry.get_instance()
|
|
plugin_registry.register(ProfileGuardPlugin())
|
|
plugin_registry.register(StoryViewPlugin())
|
|
plugin_registry.register(FollowPlugin())
|
|
plugin_registry.register(GridLikePlugin())
|
|
plugin_registry.register(CarouselBrowsingPlugin())
|
|
plugin_registry.register(AdGuardPlugin())
|
|
plugin_registry.register(CloseFriendsGuardPlugin())
|
|
plugin_registry.register(AnomalyHandlerPlugin())
|
|
plugin_registry.register(ObstacleGuardPlugin())
|
|
plugin_registry.register(PerfectSnappingPlugin())
|
|
plugin_registry.register(PostDataExtractionPlugin())
|
|
plugin_registry.register(ResonanceEvaluatorPlugin())
|
|
plugin_registry.register(RabbitHolePlugin())
|
|
plugin_registry.register(DarwinDwellPlugin())
|
|
plugin_registry.register(ProfileVisitPlugin())
|
|
plugin_registry.register(LikePlugin())
|
|
plugin_registry.register(CommentPlugin())
|
|
plugin_registry.register(RepostPlugin())
|
|
plugin_registry.register(PostInteractionPlugin())
|
|
yield plugin_registry
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════
|
|
# E2E Device Stub — The ONLY mock in true E2E tests
|
|
# ═══════════════════════════════════════════════════════
|
|
|
|
|
|
class E2EDeviceStub:
|
|
"""
|
|
Replays a sequence of XML dumps simulating the Android device.
|
|
This is the ONLY thing mocked in E2E tests. Everything else is real.
|
|
"""
|
|
|
|
def __init__(self, xml_sequence):
|
|
self._xml_sequence = list(xml_sequence)
|
|
self._dump_index = 0
|
|
self.pressed_keys = []
|
|
self.clicks = []
|
|
self.swipes = []
|
|
self.app_id = "com.instagram.android"
|
|
self._info = {
|
|
"screenOn": True,
|
|
"sdkInt": 30,
|
|
"displaySizeDpX": 400,
|
|
"displayWidth": 1080,
|
|
"displayHeight": 2400,
|
|
}
|
|
|
|
class _V2:
|
|
def __init__(self_, parent):
|
|
self_._parent = parent
|
|
self_.info = parent._info
|
|
self_.settings = {}
|
|
|
|
def dump_hierarchy(self_, compressed=False):
|
|
return self_._parent.dump_hierarchy()
|
|
|
|
def app_current(self_):
|
|
return {"package": "com.instagram.android"}
|
|
|
|
def screenshot(self_):
|
|
return None
|
|
|
|
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_._parent.clicks.append((x, y))
|
|
except (ValueError, IndexError):
|
|
pass
|
|
|
|
self.deviceV2 = _V2(self)
|
|
|
|
class _Touch:
|
|
def __init__(self_, parent):
|
|
self_._parent = parent
|
|
|
|
def down(self_, x, y):
|
|
self_._parent.clicks.append((x, y))
|
|
|
|
def up(self_, x, y):
|
|
pass
|
|
|
|
self.deviceV2.touch = _Touch(self)
|
|
|
|
def dump_hierarchy(self):
|
|
if self._dump_index < len(self._xml_sequence):
|
|
xml = self._xml_sequence[self._dump_index]
|
|
self._dump_index += 1
|
|
return xml
|
|
return self._xml_sequence[-1]
|
|
|
|
def press(self, key):
|
|
self.pressed_keys.append(key)
|
|
|
|
def click(self, x, y):
|
|
self.clicks.append((x, y))
|
|
|
|
def swipe(self, sx, sy, ex, ey, **kwargs):
|
|
self.swipes.append({"start": (sx, sy), "end": (ex, ey)})
|
|
|
|
def app_start(self, pkg, use_monkey=False):
|
|
pass
|
|
|
|
def app_stop(self, pkg):
|
|
pass
|
|
|
|
def unlock(self):
|
|
pass
|
|
|
|
def get_screenshot_b64(self):
|
|
return "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mNkYAAAAAYAAjCB0C8AAAAASUVORK5CYII="
|
|
|
|
def get_info(self):
|
|
return self._info
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════
|
|
# E2E Workflow Fixtures — Real cognitive stack, real config
|
|
# ═══════════════════════════════════════════════════════
|
|
|
|
|
|
@pytest.fixture
|
|
def e2e_device():
|
|
"""Factory to create an E2EDeviceStub from an XML sequence."""
|
|
def _create(xml_sequence):
|
|
return E2EDeviceStub(xml_sequence)
|
|
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
|
|
|
|
from GramAddict.core import llm_provider
|
|
import json
|
|
|
|
def fake_query_telepathic_llm(*args, **kwargs):
|
|
prompt = kwargs.get('user_prompt', args[3] if len(args) > 3 else "")
|
|
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:
|
|
return json.dumps({"situation": "NORMAL"})
|
|
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."""
|
|
def _create(device, username="testuser"):
|
|
from GramAddict.core.dopamine_engine import DopamineEngine
|
|
from GramAddict.core.interaction import LLMWriter
|
|
from GramAddict.core.qdrant_memory import DMMemoryDB, ParasocialCRMDB
|
|
from GramAddict.core.resonance_engine import ResonanceEngine
|
|
from GramAddict.core.q_nav_graph import QNavGraph
|
|
from GramAddict.core.telepathic_engine import TelepathicEngine
|
|
from GramAddict.core.zero_latency_engine import ZeroLatencyEngine
|
|
from GramAddict.core.darwin_engine import DarwinEngine
|
|
from GramAddict.core.active_inference import ActiveInferenceEngine
|
|
from GramAddict.core.sensors.honeypot_radome import HoneypotRadome
|
|
from GramAddict.core.swarm_protocol import SwarmProtocol
|
|
from GramAddict.core.growth_brain import GrowthBrain
|
|
|
|
dopamine = DopamineEngine()
|
|
dopamine.session_limit_seconds = 0.5
|
|
dopamine.session_start = time.time()
|
|
|
|
info = device.get_info() if hasattr(device, 'get_info') else {"displayWidth": 1080, "displayHeight": 2400}
|
|
|
|
return {
|
|
"dopamine": dopamine,
|
|
"active_inference": ActiveInferenceEngine(username),
|
|
"swarm": SwarmProtocol(username),
|
|
"resonance": ResonanceEngine(username, persona_interests=[], crm=ParasocialCRMDB()),
|
|
"growth_brain": GrowthBrain(username, persona_interests=[]),
|
|
"radome": HoneypotRadome(info.get("displayWidth", 1080), info.get("displayHeight", 2400)),
|
|
"nav_graph": QNavGraph(device),
|
|
"zero_engine": ZeroLatencyEngine(device),
|
|
"telepathic": TelepathicEngine.get_instance(),
|
|
"darwin": DarwinEngine(username),
|
|
"crm": ParasocialCRMDB(),
|
|
"dm_memory": DMMemoryDB(),
|
|
"writer": LLMWriter(username, [], e2e_configs),
|
|
}
|
|
return _create
|
|
|
|
|
|
@pytest.fixture
|
|
def e2e_session(e2e_configs):
|
|
"""Build a real SessionState."""
|
|
return SessionState(e2e_configs)
|
|
|
|
|
|
@pytest.fixture
|
|
def e2e_workflow_ctx(e2e_configs, e2e_cognitive_stack_factory, setup_e2e_plugin_registry):
|
|
"""
|
|
Factory that builds a REAL BehaviorContext + runs execute_all().
|
|
This is the EXACT code path from bot_flow.py:942-971.
|
|
"""
|
|
from GramAddict.core.behaviors import BehaviorContext
|
|
|
|
def _run(device, context_xml=None):
|
|
xml = context_xml if context_xml else device.dump_hierarchy()
|
|
session = SessionState(e2e_configs)
|
|
|
|
cognitive_stack = e2e_cognitive_stack_factory(device)
|
|
|
|
ctx = BehaviorContext(
|
|
device=device,
|
|
configs=e2e_configs,
|
|
session_state=session,
|
|
cognitive_stack=cognitive_stack,
|
|
context_xml=xml,
|
|
sleep_mod=1.0,
|
|
post_data={},
|
|
username="testuser",
|
|
shared_state={"consecutive_marker_misses": 0, "consecutive_ads": 0},
|
|
)
|
|
|
|
registry = setup_e2e_plugin_registry
|
|
results = registry.execute_all(ctx)
|
|
return results, ctx
|
|
|
|
return _run
|