Marc Mintel
39185593dd
- Updated e2e_workflow_ctx in conftest to structurally parse screen_type - Fixed StoryViewPlugin to gracefully handle pre-existing story screens - Resolved 'LIE DETECTED' failure in story test suite - Validated structural path transitions for story ring and story exit
886 lines
36 KiB
Python
886 lines
36 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.session_state import SessionState
|
|
|
|
# ═══════════════════════════════════════════════════════
|
|
# CLI Options
|
|
# ═══════════════════════════════════════════════════════
|
|
|
|
|
|
def pytest_addoption(parser):
|
|
# CLI options are now registered in the root tests/conftest.py
|
|
pass
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════
|
|
# 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="session", autouse=True)
|
|
def session_env(tmp_path_factory):
|
|
"""
|
|
Forces production parity by using real logic with local-only backends.
|
|
"""
|
|
# 1. Honest Qdrant: Use real QdrantClient with in-memory storage
|
|
os.environ["QDRANT_URL"] = ":memory:"
|
|
|
|
# 2. Honest Persistence: Use a temporary directory for accounts
|
|
accounts_dir = tmp_path_factory.mktemp("accounts")
|
|
os.environ["GRAMADDICT_ACCOUNTS_DIR"] = str(accounts_dir)
|
|
os.makedirs(os.path.join(str(accounts_dir), "testuser"), exist_ok=True)
|
|
|
|
|
|
@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
|
|
from GramAddict.core.situational_awareness import SituationalAwarenessEngine
|
|
from GramAddict.core.telepathic_engine import TelepathicEngine
|
|
|
|
PhysicsBody._instance = None
|
|
SendEventInjector.reset()
|
|
SituationalAwarenessEngine.reset()
|
|
TelepathicEngine.reset()
|
|
|
|
|
|
@pytest.fixture(scope="function", autouse=True)
|
|
def mock_vlm_visual_verification(monkeypatch):
|
|
"""
|
|
Prevents E2E tests from failing on toggle buttons (Like/Follow) which trigger
|
|
VLM visual verification (since E2E tests use black 1x1 screenshots).
|
|
"""
|
|
from GramAddict.core.perception.semantic_evaluator import SemanticEvaluator
|
|
|
|
def _mock_query_vlm(self, prompt, screenshot_b64=None):
|
|
return '{ "action_taken": "YES" }'
|
|
|
|
monkeypatch.setattr(SemanticEvaluator, "_query_vlm", _mock_query_vlm)
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════
|
|
# 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):
|
|
# We could transition here, but this is watcher click.
|
|
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)})
|
|
# Transition XML on successful tap
|
|
if isinstance(self.parent.xml, list) and len(self.parent.xml) > 1:
|
|
self.parent.xml.pop(0)
|
|
|
|
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):
|
|
if len(self.xml) > 0:
|
|
return self.xml[0]
|
|
return ""
|
|
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 _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)})
|
|
if isinstance(self.xml, list) and len(self.xml) > 1:
|
|
self.xml.pop(0)
|
|
except (ValueError, IndexError):
|
|
pass
|
|
elif isinstance(cmd, str) and cmd.startswith("input swipe"):
|
|
pass # We could log it if needed
|
|
elif isinstance(cmd, str) and "sendevent" in cmd:
|
|
# Very simplified way to catch biomechanical taps
|
|
if not hasattr(self, "_sendevent_tapped"):
|
|
self._sendevent_tapped = True
|
|
self.interaction_log.append({"action": "click", "coords": (0, 0), "type": "biomechanical"})
|
|
if isinstance(self.xml, list) and len(self.xml) > 1:
|
|
self.xml.pop(0)
|
|
|
|
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._validate_hitbox(x, y)
|
|
self.interaction_log.append({"action": "click", "coords": (x, y)})
|
|
if isinstance(self.xml, list) and len(self.xml) > 1:
|
|
self.xml.pop(0)
|
|
|
|
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)
|
|
monkeypatch.setattr(device_facade.DeviceFacade, "human_click", lambda self, x, y: self.deviceV2.click(x, y))
|
|
|
|
# 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
|
|
|
|
|
|
@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):
|
|
if len(self.xml) > 1:
|
|
return self.xml.pop(0)
|
|
elif len(self.xml) == 1:
|
|
return self.xml[0]
|
|
return ""
|
|
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
|
|
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
|
|
|
|
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._validate_hitbox(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)
|
|
|
|
# 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
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════
|
|
# 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))
|
|
|
|
|
|
# Note: mock_all_delays removed to favor production 'speed_multiplier' logic.
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════
|
|
# 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=100.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,
|
|
live=False,
|
|
)
|
|
|
|
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_starts = []
|
|
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))
|
|
self_._parent._advance_state()
|
|
|
|
def up(self_, x=None, y=None, **kwargs):
|
|
pass
|
|
|
|
self.deviceV2.touch = _Touch(self)
|
|
|
|
def _advance_state(self):
|
|
if self._dump_index + 1 < len(self._xml_sequence):
|
|
self._dump_index += 1
|
|
|
|
def dump_hierarchy(self):
|
|
if self._dump_index < len(self._xml_sequence):
|
|
return self._xml_sequence[self._dump_index]
|
|
return self._xml_sequence[-1]
|
|
|
|
def press(self, key):
|
|
self.pressed_keys.append(key)
|
|
self._advance_state()
|
|
|
|
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))
|
|
self._advance_state()
|
|
|
|
def swipe(self, sx, sy, ex, ey, **kwargs):
|
|
self.swipes.append({"start": (sx, sy), "end": (ex, ey)})
|
|
self._advance_state()
|
|
|
|
def app_start(self, pkg, use_monkey=False):
|
|
self.app_starts.append(pkg)
|
|
self._advance_state()
|
|
|
|
def app_stop(self, pkg):
|
|
self._advance_state()
|
|
|
|
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
|
|
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},
|
|
)
|
|
|
|
from GramAddict.core.goap import ScreenType
|
|
from GramAddict.core.perception.screen_identity import ScreenIdentity
|
|
|
|
identity = ScreenIdentity("testuser")
|
|
ctx.screen_type = identity.identify(xml).get("screen_type", ScreenType.UNKNOWN)
|
|
|
|
registry = setup_e2e_plugin_registry
|
|
results = registry.execute_all(ctx)
|
|
return results, ctx
|
|
|
|
return _run
|