""" 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(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): return None 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