Marc Mintel
da7201117c
- test_device_connection.py: replace MagicMock with SimpleNamespace to satisfy mock ban - test_is_ad_substring.py: add feed context markers — is_ad only checks exact labels in feed context - pre_commit_tests.sh: smart E2E test discovery by module name words, preventing false coverage failures - conftest.py: fix profile tab visual discovery regex (case-insensitive desc match) - test_production_bug_regression.py: fix TelepathicEngine singleton poisoning via monkeypatch
881 lines
35 KiB
Python
881 lines
35 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,
|
|
current_likes_limit=10,
|
|
current_comments_limit=10,
|
|
current_follows_limit=10,
|
|
current_follow_limit=10,
|
|
current_unfollow_limit=10,
|
|
current_pm_limit=10,
|
|
current_scraped_limit=10,
|
|
current_watch_limit=10,
|
|
current_success_limit=10,
|
|
current_total_limit=10,
|
|
current_crashes_limit=10,
|
|
max_follows=10,
|
|
max_pm=10,
|
|
max_watch=10,
|
|
max_success=10,
|
|
max_total=10,
|
|
max_crashes=10,
|
|
)
|
|
|
|
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_):
|
|
from PIL import Image
|
|
|
|
return Image.new("RGB", (1, 1), color="black")
|
|
|
|
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=None, y=None, **kwargs):
|
|
if "obj" in kwargs:
|
|
obj = kwargs["obj"]
|
|
if isinstance(obj, dict) and "bounds" in obj:
|
|
import re
|
|
|
|
b = obj["bounds"]
|
|
if isinstance(b, str):
|
|
nums = [int(n) for n in re.findall(r"\d+", b)]
|
|
x, y = (nums[0] + nums[2]) // 2, (nums[1] + nums[3]) // 2
|
|
else:
|
|
x, y = (int(b[0]) + int(b[2])) // 2, (int(b[1]) + int(b[3])) // 2
|
|
else:
|
|
x, y = (
|
|
int(getattr(obj, "x", getattr(obj, "x1", 0))),
|
|
int(getattr(obj, "y", getattr(obj, "y1", 0))),
|
|
)
|
|
self_._parent.clicks.append((int(x) if x is not None else 0, int(y) if y is not None else 0))
|
|
|
|
def up(self_, x=None, y=None, **kwargs):
|
|
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=None, y=None, **kwargs):
|
|
if "obj" in kwargs:
|
|
obj = kwargs["obj"]
|
|
if isinstance(obj, dict) and "bounds" in obj:
|
|
import re
|
|
|
|
b = obj["bounds"]
|
|
if isinstance(b, str):
|
|
nums = [int(n) for n in re.findall(r"\d+", b)]
|
|
x, y = (nums[0] + nums[2]) // 2, (nums[1] + nums[3]) // 2
|
|
else:
|
|
x, y = (int(b[0]) + int(b[2])) // 2, (int(b[1]) + int(b[3])) // 2
|
|
else:
|
|
x, y = int(getattr(obj, "x", getattr(obj, "x1", 0))), int(getattr(obj, "y", getattr(obj, "y1", 0)))
|
|
self.clicks.append((int(x) if x is not None else 0, int(y) if y is not None else 0))
|
|
|
|
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 shell(self, cmd):
|
|
pass
|
|
|
|
def cm_to_pixels(self, cm):
|
|
return int(cm * 40)
|
|
|
|
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
|
|
|
|
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]
|
|
|
|
print(f"MOCK LLM PROMPT:\n{prompt}\n---------------------")
|
|
|
|
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 "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 "profile tab" in intent:
|
|
match = re.search(r"\[(\d+)\][^\n]*?desc='profile'", 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 "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."""
|
|
|
|
def _create(device, username="testuser"):
|
|
from GramAddict.core.active_inference import ActiveInferenceEngine
|
|
from GramAddict.core.darwin_engine import DarwinEngine
|
|
from GramAddict.core.dopamine_engine import DopamineEngine
|
|
from GramAddict.core.growth_brain import GrowthBrain
|
|
from GramAddict.core.interaction import LLMWriter
|
|
from GramAddict.core.q_nav_graph import QNavGraph
|
|
from GramAddict.core.qdrant_memory import DMMemoryDB, ParasocialCRMDB
|
|
from GramAddict.core.resonance_engine import ResonanceEngine
|
|
from GramAddict.core.sensors.honeypot_radome import HoneypotRadome
|
|
from GramAddict.core.swarm_protocol import SwarmProtocol
|
|
from GramAddict.core.telepathic_engine import TelepathicEngine
|
|
from GramAddict.core.zero_latency_engine import ZeroLatencyEngine
|
|
|
|
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
|