465 lines
18 KiB
Python
465 lines
18 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
|
|
|
|
# ═══════════════════════════════════════════════════════
|
|
# 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():
|
|
"""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
|
|
|
|
original_init = ScreenMemoryDB.__init__
|
|
|
|
def test_init(self, *args, **kwargs):
|
|
super(ScreenMemoryDB, self).__init__(collection_name="test_e2e_screens")
|
|
|
|
ScreenMemoryDB.__init__ = test_init
|
|
|
|
db = ScreenMemoryDB()
|
|
if db.is_connected:
|
|
db.wipe_collection()
|
|
|
|
yield db
|
|
|
|
# Restore original
|
|
ScreenMemoryDB.__init__ = original_init
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════
|
|
# 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 MockU2Device:
|
|
def __init__(self, xml):
|
|
self.xml = xml
|
|
self.info = {"sdkInt": 30, "displaySizeDpX": 400, "displayWidth": 1080, "screenOn": True}
|
|
self.settings = {}
|
|
|
|
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", (1080, 1920), color="black")
|
|
|
|
def app_current(self):
|
|
return {"package": "com.instagram.android"}
|
|
|
|
def shell(self, cmd):
|
|
pass
|
|
|
|
def press(self, key):
|
|
pass
|
|
|
|
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
|
|
|
|
if isinstance(img_path, str):
|
|
img = Image.open(img_path)
|
|
else:
|
|
img = img_path
|
|
|
|
class MockU2Watcher:
|
|
def when(self, xpath=None, **kwargs):
|
|
return self
|
|
|
|
def click(self):
|
|
return self
|
|
|
|
def start(self):
|
|
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 = {}
|
|
|
|
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):
|
|
return self.img
|
|
|
|
def app_current(self):
|
|
return {"package": "com.instagram.android"}
|
|
|
|
def shell(self, cmd):
|
|
pass
|
|
|
|
def press(self, key):
|
|
pass
|
|
|
|
def watcher(self, name):
|
|
return MockU2Watcher()
|
|
|
|
def app_start(self, package_name, use_monkey=False):
|
|
pass
|
|
|
|
def mock_connect(*args, **kwargs):
|
|
return MockU2Device(img, 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
|