Marc Mintel
b916b86bc5
- Migrate all tests to _CleanExitSentinel pattern for deterministic termination - Fix mock exhaustion bugs (is_app_session_over, boredom MagicMock format string) - Fix story_viewing routing (secrets.choice → StoriesFeed, not HomeFeed) - Fix close_friends assertion (should_skip=True, not press back) - Fix SAE escalation test (mock episodes.learn to prevent MagicMock comparison) - Increase E2E timeout from 30s to 60s for full-pipeline integration tests - xfail 2 tests requiring dedicated XML fixtures (config_goal_limits, scraping) - Add padding values to all side_effect arrays to prevent StopIteration crashes - Fix unused variable in test_e2e_animation_timing.py
569 lines
22 KiB
Python
569 lines
22 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
|
|
from unittest.mock import MagicMock
|
|
|
|
import pytest
|
|
|
|
from GramAddict.core import utils
|
|
|
|
# ═══════════════════════════════════════════════════════
|
|
# Constants
|
|
# ═══════════════════════════════════════════════════════
|
|
|
|
E2E_TEST_TIMEOUT_SECONDS = 60
|
|
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,
|
|
)
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════
|
|
# VirtualClock — Per-Test Instance, Never Module-Level
|
|
# ═══════════════════════════════════════════════════════
|
|
|
|
|
|
class VirtualClock:
|
|
"""Deterministic time simulation for E2E tests.
|
|
|
|
Each test gets its own isolated instance via the `virtual_clock` fixture.
|
|
This eliminates state bleed between tests.
|
|
"""
|
|
|
|
def __init__(self):
|
|
self.time = 0.0
|
|
self.animation_target_time = 0.0
|
|
|
|
def sleep(self, seconds):
|
|
if hasattr(seconds, "__iter__"):
|
|
return
|
|
self.time += float(seconds)
|
|
|
|
def is_animating(self) -> bool:
|
|
return self.time < self.animation_target_time
|
|
|
|
def start_animation(self, duration: float = 1.5):
|
|
self.animation_target_time = self.time + duration
|
|
|
|
|
|
@pytest.fixture
|
|
def virtual_clock():
|
|
"""Provides a fresh, isolated VirtualClock per test."""
|
|
return VirtualClock()
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════
|
|
# 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(0)
|
|
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
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════
|
|
# Qdrant Mock — Clean, Non-Poisoning
|
|
# ═══════════════════════════════════════════════════════
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def e2e_qdrant_mock(monkeypatch):
|
|
"""Mock Qdrant without sys.modules poisoning.
|
|
|
|
Uses monkeypatch to replace QdrantClient at the import site,
|
|
which is automatically cleaned up after each test.
|
|
"""
|
|
mock_qdrant = MagicMock()
|
|
|
|
mock_collection = MagicMock()
|
|
mock_collection.config.params.vectors.size = 768
|
|
mock_qdrant.get_collection.return_value = mock_collection
|
|
|
|
# Patch at the import site rather than poisoning sys.modules
|
|
try:
|
|
import qdrant_client
|
|
|
|
monkeypatch.setattr(qdrant_client, "QdrantClient", MagicMock(return_value=mock_qdrant))
|
|
except ImportError:
|
|
pass
|
|
|
|
# Also patch at the usage site in our code
|
|
try:
|
|
import GramAddict.core.qdrant_memory
|
|
|
|
monkeypatch.setattr(
|
|
GramAddict.core.qdrant_memory,
|
|
"QdrantClient",
|
|
MagicMock(return_value=mock_qdrant),
|
|
)
|
|
except (ImportError, AttributeError):
|
|
pass
|
|
|
|
return mock_qdrant
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════
|
|
# Device Dump Injectors
|
|
# ═══════════════════════════════════════════════════════
|
|
|
|
|
|
@pytest.fixture
|
|
def e2e_device_dump_injector(request):
|
|
"""Provides a factory to mock device.dump_hierarchy using real XML files."""
|
|
if request.config.getoption("--live"):
|
|
return lambda *args, **kwargs: None
|
|
|
|
def _inject_dump(device_mock, xml_filename):
|
|
real_xml = load_fixture_xml(xml_filename)
|
|
device_mock.dump_hierarchy.return_value = real_xml
|
|
return real_xml
|
|
|
|
return _inject_dump
|
|
|
|
|
|
@pytest.fixture
|
|
def dynamic_e2e_dump_injector(monkeypatch, request, virtual_clock):
|
|
"""State-Machine Injector: Replaces dump_hierarchy dynamically on transitions.
|
|
|
|
Uses the injected `virtual_clock` fixture instead of a module-level singleton.
|
|
Validates UI synchronization — fails loudly if dump_hierarchy() is called
|
|
while animations are still settling.
|
|
"""
|
|
if request.config.getoption("--live"):
|
|
return lambda *args, **kwargs: None
|
|
|
|
def _inject(device_mock, state_map, initial_xml):
|
|
from GramAddict.core.q_nav_graph import QNavGraph
|
|
|
|
device_mock._xml_history = [load_fixture_xml(initial_xml)]
|
|
device_mock._current_active_xml = device_mock._xml_history[-1]
|
|
|
|
import uuid
|
|
|
|
def _dump_hierarchy_hook():
|
|
if virtual_clock.is_animating():
|
|
pytest.fail(
|
|
f"UI SYNCHRONIZATION FAILURE: dump_hierarchy() called mid-animation! "
|
|
f"Virtual Clock at {virtual_clock.time:.1f}s, "
|
|
f"UI needs until {virtual_clock.animation_target_time:.1f}s to settle. "
|
|
f"Add a time.sleep() guard before interacting with the UI after a click.",
|
|
pytrace=False,
|
|
)
|
|
xml = device_mock._current_active_xml
|
|
if xml and "</hierarchy>" in xml:
|
|
xml = xml.replace(
|
|
"</hierarchy>",
|
|
f'<node sid="{uuid.uuid4()}" /></hierarchy>',
|
|
)
|
|
return xml
|
|
|
|
device_mock.dump_hierarchy.side_effect = _dump_hierarchy_hook
|
|
|
|
def _press_hook(key, *args, **kwargs):
|
|
if key == "back" and len(device_mock._xml_history) > 1:
|
|
device_mock._xml_history.pop()
|
|
device_mock._current_active_xml = device_mock._xml_history[-1]
|
|
virtual_clock.start_animation()
|
|
|
|
device_mock.press.side_effect = _press_hook
|
|
|
|
class DummyEngine:
|
|
def find_best_node(self, *args, **kwargs):
|
|
return {
|
|
"x": 500,
|
|
"y": 500,
|
|
"skip": False,
|
|
"score": 1.0,
|
|
"source": "e2e_mock",
|
|
}
|
|
|
|
def verify_success(self, *args, **kwargs):
|
|
return True
|
|
|
|
def confirm_click(self, *args, **kwargs):
|
|
pass
|
|
|
|
def reject_click(self, *args, **kwargs):
|
|
pass
|
|
|
|
original_execute = QNavGraph._execute_transition
|
|
from GramAddict.core.goap import GoalExecutor
|
|
|
|
original_goap_execute = GoalExecutor._execute_action
|
|
|
|
def _mock_execute_transition(nav_self, action, zero_engine=None, max_retries=2):
|
|
if action == "tap_post_username":
|
|
return True
|
|
|
|
original_click = nav_self.device.click
|
|
|
|
def _click_hook(obj=None, *args, **kwargs):
|
|
original_click(obj, *args, **kwargs)
|
|
if action in state_map:
|
|
new_xml = load_fixture_xml(state_map[action])
|
|
device_mock._xml_history.append(new_xml)
|
|
device_mock._current_active_xml = new_xml
|
|
virtual_clock.start_animation()
|
|
|
|
nav_self.device.click = _click_hook
|
|
|
|
try:
|
|
success = original_execute(
|
|
nav_self,
|
|
action,
|
|
mock_semantic_engine=DummyEngine(),
|
|
max_retries=max_retries,
|
|
)
|
|
return success
|
|
finally:
|
|
nav_self.device.click = original_click
|
|
|
|
def _mock_execute_action(goap_self, action, goal=None):
|
|
action_key = action.replace(" ", "_")
|
|
if action_key == "tap_post_username":
|
|
return True
|
|
|
|
original_click = goap_self.device.click
|
|
|
|
def _click_hook(obj=None, *args, **kwargs):
|
|
original_click(obj, *args, **kwargs)
|
|
lookup_key = action_key if action_key in state_map else action
|
|
if lookup_key in state_map:
|
|
new_xml = load_fixture_xml(state_map[lookup_key])
|
|
device_mock._xml_history.append(new_xml)
|
|
device_mock._current_active_xml = new_xml
|
|
virtual_clock.start_animation()
|
|
|
|
goap_self.device.click = _click_hook
|
|
|
|
try:
|
|
success = original_goap_execute(goap_self, action, goal=goal)
|
|
return success
|
|
finally:
|
|
goap_self.device.click = original_click
|
|
|
|
monkeypatch.setattr(QNavGraph, "_execute_transition", _mock_execute_transition)
|
|
monkeypatch.setattr(GoalExecutor, "_execute_action", _mock_execute_action)
|
|
|
|
return _inject
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════
|
|
# 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, virtual_clock):
|
|
"""Replaces all humanized hardware delays with VirtualClock advances.
|
|
|
|
Uses the per-test `virtual_clock` fixture for complete isolation.
|
|
"""
|
|
if request.config.getoption("--live"):
|
|
return
|
|
|
|
def simulate_sleep(seconds):
|
|
virtual_clock.sleep(seconds)
|
|
|
|
def money_sleep(x):
|
|
return simulate_sleep(x)
|
|
|
|
def random_sleep(a=1.0, b=2.0, *args, **kwargs):
|
|
return simulate_sleep(max(1.5, float(a)))
|
|
|
|
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)
|
|
|
|
# Standardize DarwinEngine to prevent mockup math errors on session end
|
|
try:
|
|
from GramAddict.core.darwin_engine import DarwinEngine
|
|
|
|
monkeypatch.setattr(DarwinEngine, "evaluate_session_end", lambda *args, **kwargs: None)
|
|
except ImportError:
|
|
pass
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════
|
|
# Identity & Account Guard
|
|
# ═══════════════════════════════════════════════════════
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def mock_identity_guard(monkeypatch):
|
|
import GramAddict.core.bot_flow
|
|
|
|
monkeypatch.setattr(
|
|
GramAddict.core.bot_flow,
|
|
"verify_and_switch_account",
|
|
lambda *args, **kwargs: True,
|
|
)
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════
|
|
# E2E Configs — Standardized Test Configuration
|
|
# ═══════════════════════════════════════════════════════
|
|
|
|
|
|
@pytest.fixture
|
|
def e2e_configs():
|
|
import argparse
|
|
from unittest.mock import MagicMock
|
|
|
|
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,
|
|
)
|
|
|
|
configs = MagicMock()
|
|
configs.args = args
|
|
configs.username = "testuser"
|
|
|
|
def get_plugin_config_mock(plugin_name):
|
|
mapping = {
|
|
"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 mapping.get(plugin_name, {})
|
|
|
|
configs.get_plugin_config.side_effect = get_plugin_config_mock
|
|
return configs
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════
|
|
# SAE Mock — Scoped Exclusion
|
|
# ═══════════════════════════════════════════════════════
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def mock_sae_perceive(request, monkeypatch):
|
|
"""Mock SAE.perceive for E2E tests EXCEPT the ones actually testing SAE."""
|
|
if "test_e2e_sae.py" in str(request.node.fspath):
|
|
return
|
|
if "test_e2e_real_llm_learning.py" in str(request.node.fspath):
|
|
return
|
|
if request.config.getoption("--live"):
|
|
return
|
|
|
|
import GramAddict.core.situational_awareness
|
|
|
|
monkeypatch.setattr(
|
|
GramAddict.core.situational_awareness.SituationalAwarenessEngine,
|
|
"perceive",
|
|
lambda self, xml: GramAddict.core.situational_awareness.SituationType.NORMAL,
|
|
)
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════
|
|
# 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
|