diff --git a/tests/e2e/conftest.py b/tests/e2e/conftest.py
index 0030a50..77662e9 100644
--- a/tests/e2e/conftest.py
+++ b/tests/e2e/conftest.py
@@ -17,6 +17,7 @@ import time
import pytest
from GramAddict.core import utils
+from GramAddict.core.session_state import SessionState
# ═══════════════════════════════════════════════════════
# CLI Options
@@ -504,3 +505,179 @@ def setup_e2e_plugin_registry():
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_):
+ return None
+
+ 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, y):
+ self_._parent.clicks.append((x, y))
+
+ def up(self_, x, y):
+ 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, y):
+ self.clicks.append((x, y))
+
+ 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 get_screenshot_b64(self):
+ return None
+
+ 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():
+ """Build the REAL cognitive stack exactly like bot_flow.py does."""
+ from GramAddict.core.dopamine_engine import DopamineEngine
+
+ dopamine = DopamineEngine()
+ # Force session to end quickly so tests don't hang
+ dopamine.session_limit_seconds = 0.5
+ dopamine.session_start = time.time()
+
+ return {
+ "dopamine": dopamine,
+ "active_inference": None,
+ "swarm": None,
+ "resonance": None,
+ "growth_brain": None,
+ "radome": None,
+ "nav_graph": None,
+ "zero_engine": None,
+ "telepathic": None,
+ "darwin": None,
+ "crm": None,
+ "dm_memory": None,
+ "writer": None,
+ }
+
+
+@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, 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.
+
+ Usage:
+ results, device = e2e_workflow_ctx(xml_sequence, monkeypatch_sae=FakeSAE)
+ """
+ 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)
+
+ ctx = BehaviorContext(
+ device=device,
+ configs=e2e_configs,
+ session_state=session,
+ cognitive_stack=e2e_cognitive_stack,
+ context_xml=xml,
+ sleep_mod=1.0,
+ post_data={},
+ username="",
+ shared_state={"consecutive_marker_misses": 0, "consecutive_ads": 0},
+ )
+
+ registry = setup_e2e_plugin_registry
+ results = registry.execute_all(ctx)
+ return results, ctx
+
+ return _run
diff --git a/tests/e2e/test_full_workflow_hostile_env.py b/tests/e2e/test_full_workflow_hostile_env.py
deleted file mode 100644
index bace1fe..0000000
--- a/tests/e2e/test_full_workflow_hostile_env.py
+++ /dev/null
@@ -1,520 +0,0 @@
-"""
-🔴 TRUE E2E Workflow Tests — Full Production Pipeline
-======================================================
-
-WHAT MAKES THESE DIFFERENT:
- Previous "E2E" tests were disguised unit tests that tested individual
- components in isolation. They NEVER ran the actual production workflow.
-
- These tests run the REAL bot workflow (bot_flow._run_zero_latency_feed_loop)
- with the REAL cognitive stack (DopamineEngine, GrowthBrain, etc.) and the
- REAL PluginRegistry. The ONLY thing mocked is the Android device (XML dumps).
-
- If a bug exists ANYWHERE in the pipeline — an import error in a plugin, a
- missing obstacle handler, a VLM crash — these tests WILL catch it because
- they exercise the exact same code path as production.
-
-WHY THEY EXIST:
- Production run 2026-04-29 23:01 revealed:
- - obstacle_guard ignored OBSTACLE_SYSTEM (fixed in obstacle_guard.py)
- - resonance_evaluator had a broken import (fixed)
- - The bot got completely stuck on a permission dialog
-
- 81 existing E2E tests all passed. ZERO caught these bugs because ZERO
- actually ran the feed loop. That's a testing architecture failure.
-
-RULE:
- A real E2E test mocks ONLY Instagram (the device XML).
- Everything else runs FOR REAL.
-"""
-
-import argparse
-import time
-
-import pytest
-
-from GramAddict.core.behaviors import BehaviorContext, PluginRegistry
-from GramAddict.core.dopamine_engine import DopamineEngine
-from GramAddict.core.session_state import SessionState
-from GramAddict.core.situational_awareness import SituationType
-
-# ═══════════════════════════════════════════════════════
-# XML Fixture Sequences
-# ═══════════════════════════════════════════════════════
-
-NORMAL_POST_XML = """
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-"""
-
-PERMISSION_DIALOG_XML = """
-
-
-
-
-
-
-
-
-
-
-"""
-
-
-# ═══════════════════════════════════════════════════════
-# Device Facade Stub — The ONLY mock in these tests
-# ═══════════════════════════════════════════════════════
-
-
-class E2EDeviceStub:
- """
- Minimal device that replays a sequence of XML dumps.
- This is the ONLY mock. 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):
- return None
-
- 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, y):
- self._parent.clicks.append((x, y))
-
- def up(self, x, y):
- 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, y):
- self.clicks.append((x, y))
-
- 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 get_screenshot_b64(self):
- return None
-
- def get_info(self):
- return self._info
-
-
-# ═══════════════════════════════════════════════════════
-# Real Cognitive Stack Factory
-# ═══════════════════════════════════════════════════════
-
-
-def _build_real_cognitive_stack():
- """
- Build the REAL cognitive stack exactly like bot_flow.py does.
- No mocks. No stubs. Real engines.
- """
- dopamine = DopamineEngine()
- # Force session to end after 2 iterations so test doesn't hang
- dopamine.session_limit_seconds = 0.5
- dopamine.session_start = time.time()
-
- return {
- "dopamine": dopamine,
- "active_inference": None,
- "swarm": None,
- "resonance": None,
- "growth_brain": None,
- "radome": None,
- "nav_graph": None,
- "zero_engine": None,
- "telepathic": None,
- "darwin": None,
- "crm": None,
- "dm_memory": None,
- "writer": None,
- }
-
-
-def _build_real_config():
- """Build real Config exactly as production uses it."""
- from GramAddict.core.config import Config
-
- config = Config(first_run=True)
- config.args = argparse.Namespace(
- username="testuser",
- device="emulator-5554",
- app_id="com.instagram.android",
- debug=True,
- interact_percentage=100,
- visual_vibe_check_percentage=0,
- persona_interests="travel, nature, photography",
- target_audience="travel, nature, photography",
- speed_multiplier=1.0,
- likes_count="1",
- likes_percentage=100,
- follow_percentage=0,
- comment_percentage=0,
- carousel_percentage=0,
- carousel_count="1",
- stories_count="0",
- stories_percentage=0,
- disable_ai_messaging=True,
- scrape_profiles=False,
- dry_run_comments=True,
- )
- config.config = {
- "plugins": {
- "likes": {"count": "1", "percentage": 100},
- "follow": {"percentage": 0},
- "comment": {"percentage": 0, "dry_run": True},
- "stories": {"count": "0", "percentage": 0},
- "resonance_evaluator": {"visual_vibe_check_percentage": 0},
- "carousel_browsing": {"percentage": 0, "count": "1"},
- }
- }
- config.username = "testuser"
- return config
-
-
-def _build_real_session(configs):
- """Build real SessionState exactly as production uses it."""
- return SessionState(configs)
-
-
-# ═══════════════════════════════════════════════════════
-# THE TESTS — Full workflow, real pipeline
-# ═══════════════════════════════════════════════════════
-
-
-class TestFullWorkflowPermissionDialog:
- """
- End-to-end: The bot encounters a permission dialog mid-feed.
- The FULL production pipeline must handle it without getting stuck.
- """
-
- def test_feed_loop_iteration_recovers_from_permission_dialog(
- self, monkeypatch, setup_e2e_plugin_registry
- ):
- """
- Simulate a single feed loop iteration where the device shows a
- permission dialog. The full plugin chain must:
- 1. Detect the obstacle (SAE → OBSTACLE_SYSTEM)
- 2. Dismiss it (obstacle_guard → press BACK)
- 3. NOT run any interaction plugins
- """
- # Build the real cognitive stack
- cognitive_stack = _build_real_cognitive_stack()
- configs = _build_real_config()
- session_state = _build_real_session(configs)
- registry = setup_e2e_plugin_registry
-
- # Device returns permission dialog, then normal feed after BACK
- device = E2EDeviceStub([
- PERMISSION_DIALOG_XML, # Initial dump → obstacle
- NORMAL_POST_XML, # After recovery
- ])
-
- # Monkeypatch ONLY the SAE (since it needs device connection for real ADB)
- class FakeSAE:
- @classmethod
- def get_instance(cls, device=None):
- return cls()
-
- def perceive(self, xml):
- if "permissioncontroller" in xml:
- return SituationType.OBSTACLE_SYSTEM
- return SituationType.NORMAL
-
- def unlearn_current_state(self, xml):
- pass
-
- monkeypatch.setattr(
- "GramAddict.core.behaviors.obstacle_guard.SituationalAwarenessEngine",
- FakeSAE,
- )
-
- # ── Run the EXACT same code as production (bot_flow.py:942-971) ──
- context_xml = device.dump_hierarchy()
-
- ctx = BehaviorContext(
- device=device,
- configs=configs,
- session_state=session_state,
- cognitive_stack=cognitive_stack,
- context_xml=context_xml,
- sleep_mod=1.0,
- post_data={},
- username="",
- shared_state={"consecutive_marker_misses": 0, "consecutive_ads": 0},
- )
-
- plugin_results = registry.execute_all(ctx)
-
- # ── Assert exactly what bot_flow.py checks (lines 962-971) ──
- should_continue_loop = True
- for result in plugin_results:
- if result.metadata.get("return_code") == "CONTEXT_LOST":
- pytest.fail("Plugin returned CONTEXT_LOST — bot would crash!")
- if result.executed and result.should_skip:
- should_continue_loop = False
- break
-
- # The chain MUST have set should_skip=True (obstacle detected)
- assert should_continue_loop is False, (
- "The plugin chain did NOT set should_skip=True for a permission dialog! "
- "In production, the bot would continue to the interaction phase "
- "and try to like/follow/comment on a permission dialog."
- )
-
- # No actual Instagram interactions may have happened
- total_interactions = sum(r.interactions for r in plugin_results)
- assert total_interactions == 0, (
- f"{total_interactions} interaction(s) were performed on a permission dialog!"
- )
-
- # The device must have pressed BACK
- assert "back" in device.pressed_keys, (
- "obstacle_guard did not press BACK — the permission dialog stays on screen!"
- )
-
- def test_normal_feed_post_is_processable(
- self, monkeypatch, setup_e2e_plugin_registry
- ):
- """
- Sanity check: A normal Instagram post must be processable by the
- full pipeline without crashing.
- """
- cognitive_stack = _build_real_cognitive_stack()
- configs = _build_real_config()
- session_state = _build_real_session(configs)
- registry = setup_e2e_plugin_registry
-
- device = E2EDeviceStub([NORMAL_POST_XML, NORMAL_POST_XML])
-
- class FakeSAE:
- @classmethod
- def get_instance(cls, device=None):
- return cls()
-
- def perceive(self, xml):
- return SituationType.NORMAL
-
- def unlearn_current_state(self, xml):
- pass
-
- monkeypatch.setattr(
- "GramAddict.core.behaviors.obstacle_guard.SituationalAwarenessEngine",
- FakeSAE,
- )
-
- context_xml = device.dump_hierarchy()
-
- ctx = BehaviorContext(
- device=device,
- configs=configs,
- session_state=session_state,
- cognitive_stack=cognitive_stack,
- context_xml=context_xml,
- sleep_mod=1.0,
- post_data={},
- username="",
- shared_state={"consecutive_marker_misses": 0, "consecutive_ads": 0},
- )
-
- # This MUST NOT crash (import errors, attribute errors, etc.)
- plugin_results = registry.execute_all(ctx)
-
- # Multiple plugins must have executed (not just obstacle_guard)
- executed_count = sum(1 for r in plugin_results if r.executed)
- assert executed_count >= 1, (
- f"Only {executed_count} plugin(s) executed on a normal feed post. "
- f"The pipeline is broken — plugins are crashing silently."
- )
-
- # No CONTEXT_LOST
- for result in plugin_results:
- assert result.metadata.get("return_code") != "CONTEXT_LOST", (
- "Plugin returned CONTEXT_LOST on a normal feed post — critical regression!"
- )
-
-
-class TestFullWorkflowImportIntegrity:
- """
- Prove that ALL plugins can be imported and instantiated without errors.
- A broken import (like the humanized_scroll bug) would crash here.
- """
-
- def test_all_plugins_importable_and_instantiable(self, setup_e2e_plugin_registry):
- """
- The plugin registry must contain all expected plugins.
- If any plugin has a broken import, the conftest fixture will crash.
- """
- registry = setup_e2e_plugin_registry
-
- # These are the plugins registered in bot_flow.py:273-294
- expected_plugins = {
- "ad_guard",
- "anomaly_handler",
- "close_friends_guard",
- "obstacle_guard",
- "perfect_snapping",
- "post_data_extraction",
- "resonance_evaluator",
- "darwin_dwell",
- "likes",
- "comment",
- "repost",
- "post_interaction",
- }
-
- registered = {p.name for p in registry.plugins}
-
- missing = expected_plugins - registered
- assert not missing, (
- f"Plugins failed to register (likely import errors): {missing}. "
- f"Registered plugins: {registered}"
- )
-
- def test_plugin_chain_does_not_swallow_import_errors(self):
- """
- Verify that importing all behavior plugins does not silently fail.
- Each plugin module must be importable standalone.
- """
- # These imports MUST succeed. If any has a broken import (like the
- # humanized_scroll bug), this test catches it immediately.
- from GramAddict.core.behaviors.ad_guard import AdGuardPlugin
- from GramAddict.core.behaviors.anomaly_handler import AnomalyHandlerPlugin
- 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.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.resonance_evaluator import ResonanceEvaluatorPlugin
-
- # Instantiation must not crash
- plugins = [
- AdGuardPlugin(),
- AnomalyHandlerPlugin(),
- CloseFriendsGuardPlugin(),
- CommentPlugin(),
- DarwinDwellPlugin(),
- LikePlugin(),
- ObstacleGuardPlugin(),
- PerfectSnappingPlugin(),
- PostDataExtractionPlugin(),
- PostInteractionPlugin(),
- ResonanceEvaluatorPlugin(),
- ]
-
- for p in plugins:
- assert hasattr(p, "execute"), f"{p.__class__.__name__} has no execute method!"
- assert hasattr(p, "can_activate"), f"{p.__class__.__name__} has no can_activate method!"
- assert hasattr(p, "priority"), f"{p.__class__.__name__} has no priority property!"
diff --git a/tests/e2e/test_plugin_chain_hostile_env.py b/tests/e2e/test_plugin_chain_hostile_env.py
deleted file mode 100644
index 2d0a9f0..0000000
--- a/tests/e2e/test_plugin_chain_hostile_env.py
+++ /dev/null
@@ -1,381 +0,0 @@
-"""
-🔴 E2E Plugin Chain Integration Tests
-========================================
-
-These tests exercise the FULL plugin chain (PluginRegistry.execute_all) against
-hostile environments that the bot encounters in production:
- 1. System permission dialogs (Android PermissionController)
- 2. Foreign app overlays (browser, wrong app in foreground)
- 3. Reel creation flows (camera, audio recording prompts)
-
-WHY THESE TESTS EXIST:
- Production run 2026-04-29 23:01 revealed that:
- - The SAE correctly detects OBSTACLE_SYSTEM for permission dialogs
- - The obstacle_guard plugin IGNORED OBSTACLE_SYSTEM (only handled OBSTACLE_MODAL)
- - All downstream plugins (resonance, dwell, like, follow) fired against a
- non-Instagram permission dialog
- - The VLM hallucinated box [4] (icon) as "post username" and "media content"
- - The bot got completely stuck in an infinite loop
-
- NONE of the existing 81 E2E tests caught this because:
- - They test each plugin in ISOLATION (not as a chain)
- - They NEVER call PluginRegistry.execute_all() against hostile environments
- - The conftest registers all plugins but no test actually exercises the chain
-
- These tests fix that gap permanently.
-"""
-
-import argparse
-import os
-
-import pytest
-
-from GramAddict.core.behaviors import BehaviorContext, BehaviorResult, PluginRegistry
-from GramAddict.core.situational_awareness import SituationType
-
-
-# ═══════════════════════════════════════════════════════
-# Hostile XML Fixtures (inline for speed, no file I/O)
-# ═══════════════════════════════════════════════════════
-
-PERMISSION_DIALOG_XML = """
-
-
-
-
-
-
-
-
-
-
-
-"""
-
-FOREIGN_APP_CHROME_XML = """
-
-
-
-
-
-
-"""
-
-NORMAL_INSTAGRAM_XML = """
-
-
-
-
-
-
-"""
-
-
-# ═══════════════════════════════════════════════════════
-# Realistic Device Stub
-# ═══════════════════════════════════════════════════════
-
-
-class ChainTestDevice:
- """Deterministic device for chain testing. Returns XML sequence."""
-
- def __init__(self, xml_sequence):
- self._xml_sequence = list(xml_sequence)
- self._dump_index = 0
- self.pressed_keys = []
- self.clicks = []
- self.app_id = "com.instagram.android"
-
- class _V2:
- info = {"screenOn": True, "sdkInt": 30, "displaySizeDpX": 400, "displayWidth": 1080}
- settings = {}
- touch = type("T", (), {"down": lambda s, x, y: None, "up": lambda s, x, y: None})()
-
- def dump_hierarchy(self, compressed=False):
- return ""
-
- def app_current(self):
- return {"package": "com.instagram.android"}
-
- def screenshot(self):
- return None
-
- def shell(self, cmd):
- pass
-
- self.deviceV2 = _V2()
-
- 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, y):
- self.clicks.append((x, y))
-
- def swipe(self, sx, sy, ex, ey, **kwargs):
- pass
-
- def app_start(self, pkg, use_monkey=False):
- pass
-
- def app_stop(self, pkg):
- pass
-
- def unlock(self):
- pass
-
- def get_screenshot_b64(self):
- return None
-
-
-class ChainTestConfig:
- def __init__(self):
- self.args = argparse.Namespace(
- username="testuser",
- device="emulator-5554",
- app_id="com.instagram.android",
- debug=True,
- interact_percentage=100,
- visual_vibe_check_percentage=0,
- persona_interests="travel, nature",
- target_audience="travel, nature",
- speed_multiplier=1.0,
- likes_count="1",
- likes_percentage=100,
- follow_percentage=100,
- comment_percentage=0,
- carousel_percentage=0,
- carousel_count="1",
- stories_count="0",
- stories_percentage=0,
- disable_ai_messaging=True,
- scrape_profiles=False,
- dry_run_comments=True,
- )
- self.config = {
- "plugins": {
- "likes": {"count": "1", "percentage": 100},
- "follow": {"percentage": 100},
- "comment": {"percentage": 0, "dry_run": True},
- "stories": {"count": "0", "percentage": 0},
- "resonance_evaluator": {"visual_vibe_check_percentage": 0},
- "carousel_browsing": {"percentage": 0, "count": "1"},
- }
- }
- self.username = "testuser"
-
- def get_plugin_config(self, name):
- return self.config.get("plugins", {}).get(name, {})
-
-
-class ChainTestSession:
- def __init__(self):
- self.totalLikes = 0
- self.totalFollowed = 0
- self.totalComments = 0
- self.totalGetProfile = 0
- self.totalReposted = 0
- self.totalStoriesWatched = 0
- self.totalScraped = 0
- self.job_target = "test_chain"
- self.totalInteractions = {}
-
- def check_limit(self, *args, **kwargs):
- return False
-
-
-# ═══════════════════════════════════════════════════════
-# THE ACTUAL CHAIN INTEGRATION TESTS
-# ═══════════════════════════════════════════════════════
-
-
-class TestPluginChainOnSystemDialog:
- """
- THE BIG LIE: Before this test existed, no E2E test exercised the full
- plugin chain against a system permission dialog. The SAE perception test
- verified detection, but obstacle_guard silently ignored it.
-
- This test proves the chain TERMINATES on hostile environments.
- """
-
- def test_chain_terminates_on_permission_dialog(self, monkeypatch, setup_e2e_plugin_registry):
- """
- When PluginRegistry.execute_all() runs against a permission dialog,
- the obstacle_guard (priority 95) MUST fire and terminate the chain.
- No downstream plugin (resonance, like, follow) may execute.
- """
- device = ChainTestDevice([PERMISSION_DIALOG_XML, NORMAL_INSTAGRAM_XML])
- registry = setup_e2e_plugin_registry
-
- # Monkeypatch SAE to return OBSTACLE_SYSTEM
- class FakeSAE:
- @classmethod
- def get_instance(cls, device=None):
- return cls()
-
- def perceive(self, xml):
- if "permissioncontroller" in xml:
- return SituationType.OBSTACLE_SYSTEM
- return SituationType.NORMAL
-
- monkeypatch.setattr(
- "GramAddict.core.behaviors.obstacle_guard.SituationalAwarenessEngine",
- FakeSAE,
- )
-
- ctx = BehaviorContext(
- device=device,
- configs=ChainTestConfig(),
- session_state=ChainTestSession(),
- cognitive_stack={},
- shared_state={},
- context_xml=PERMISSION_DIALOG_XML,
- )
-
- results = registry.execute_all(ctx)
-
- # The chain MUST have terminated before any INTERACTION plugins fired
- assert len(results) >= 1, "No plugin executed at all"
-
- # Find the obstacle_guard result (should be the one that executed + should_skip)
- guard_fired = any(r.executed and r.should_skip for r in results)
- assert guard_fired, (
- "obstacle_guard did not fire — no result has executed=True + should_skip=True. "
- "This means the permission dialog was ignored and interaction plugins will run!"
- )
-
- # Critical: NO interaction plugin may have performed actual interactions
- total_interactions = sum(r.interactions for r in results)
- assert total_interactions == 0, (
- f"{total_interactions} interaction(s) were performed against a permission dialog!"
- )
-
- # The device must have pressed BACK to dismiss the dialog
- assert "back" in device.pressed_keys, (
- "obstacle_guard did not press BACK to dismiss the system dialog"
- )
-
- def test_chain_terminates_on_foreign_app(self, monkeypatch, setup_e2e_plugin_registry):
- """
- When Chrome is in the foreground, the chain must terminate at obstacle_guard.
- """
- device = ChainTestDevice([FOREIGN_APP_CHROME_XML, NORMAL_INSTAGRAM_XML])
- registry = setup_e2e_plugin_registry
-
- class FakeSAE:
- @classmethod
- def get_instance(cls, device=None):
- return cls()
-
- def perceive(self, xml):
- if "chrome" in xml:
- return SituationType.OBSTACLE_FOREIGN_APP
- return SituationType.NORMAL
-
- monkeypatch.setattr(
- "GramAddict.core.behaviors.obstacle_guard.SituationalAwarenessEngine",
- FakeSAE,
- )
-
- ctx = BehaviorContext(
- device=device,
- configs=ChainTestConfig(),
- session_state=ChainTestSession(),
- cognitive_stack={},
- shared_state={},
- context_xml=FOREIGN_APP_CHROME_XML,
- )
-
- results = registry.execute_all(ctx)
-
- # obstacle_guard must have fired and terminated the chain
- guard_fired = any(r.executed and r.should_skip for r in results)
- assert guard_fired, (
- "obstacle_guard did not fire on foreign app. "
- "Downstream plugins are firing against a Chrome browser window!"
- )
-
- total_interactions = sum(r.interactions for r in results)
- assert total_interactions == 0, (
- f"{total_interactions} interaction(s) were performed against Chrome!"
- )
- assert "back" in device.pressed_keys
-
-
-class TestPluginChainOnNormalFeed:
- """
- Sanity check: On a NORMAL Instagram feed, the chain must run normally.
- The obstacle_guard must NOT fire (executed=False), allowing downstream plugins.
- """
-
- def test_chain_passes_through_on_normal_instagram(self, monkeypatch, setup_e2e_plugin_registry):
- device = ChainTestDevice([NORMAL_INSTAGRAM_XML])
- registry = setup_e2e_plugin_registry
-
- class FakeSAE:
- @classmethod
- def get_instance(cls, device=None):
- return cls()
-
- def perceive(self, xml):
- return SituationType.NORMAL
-
- monkeypatch.setattr(
- "GramAddict.core.behaviors.obstacle_guard.SituationalAwarenessEngine",
- FakeSAE,
- )
-
- ctx = BehaviorContext(
- device=device,
- configs=ChainTestConfig(),
- session_state=ChainTestSession(),
- cognitive_stack={},
- shared_state={},
- context_xml=NORMAL_INSTAGRAM_XML,
- )
-
- results = registry.execute_all(ctx)
-
- # Multiple plugins should have run (obstacle_guard passed through, then data extraction, etc.)
- assert len(results) > 1, (
- f"Only {len(results)} plugin(s) executed on a normal Instagram feed. "
- f"The obstacle_guard may be incorrectly blocking the chain."
- )
-
- # obstacle_guard must have executed=False (no obstacle detected)
- obstacle_results = [r for r in results if not r.executed or "back" not in device.pressed_keys]
- assert "back" not in device.pressed_keys, (
- "obstacle_guard pressed BACK on a normal Instagram screen — false positive!"
- )
diff --git a/tests/e2e/test_workflow_foreign_app.py b/tests/e2e/test_workflow_foreign_app.py
new file mode 100644
index 0000000..8642c72
--- /dev/null
+++ b/tests/e2e/test_workflow_foreign_app.py
@@ -0,0 +1,81 @@
+"""
+E2E: Foreign App Workflow
+=========================
+Tests the FULL production pipeline when the device has navigated
+to a foreign app (Chrome, Settings, etc.) instead of Instagram.
+
+Mock: ONLY the device (XML dumps).
+Real: Cognitive stack, PluginRegistry, SessionState, Config.
+"""
+
+from tests.e2e.conftest import E2EDeviceStub
+
+from GramAddict.core.situational_awareness import SituationType
+
+
+CHROME_XML = """
+
+
+
+
+
+
+"""
+
+NORMAL_POST_XML = """
+
+
+
+
+
+"""
+
+
+class FakeSAEForeignApp:
+ @classmethod
+ def get_instance(cls, device=None):
+ return cls()
+
+ def perceive(self, xml):
+ if "chrome" in xml:
+ return SituationType.OBSTACLE_FOREIGN_APP
+ return SituationType.NORMAL
+
+ def unlearn_current_state(self, xml):
+ pass
+
+
+def test_foreign_app_terminates_chain(monkeypatch, e2e_workflow_ctx):
+ """
+ When Chrome is in the foreground, the full pipeline must:
+ detect OBSTACLE_FOREIGN_APP → press BACK → skip all interactions.
+ """
+ monkeypatch.setattr(
+ "GramAddict.core.behaviors.obstacle_guard.SituationalAwarenessEngine",
+ FakeSAEForeignApp,
+ )
+
+ device = E2EDeviceStub([CHROME_XML, NORMAL_POST_XML])
+ results, ctx = e2e_workflow_ctx(device, context_xml=CHROME_XML)
+
+ guard_fired = any(r.executed and r.should_skip for r in results)
+ assert guard_fired, (
+ "obstacle_guard did not fire on foreign app — "
+ "interaction plugins would run against Chrome!"
+ )
+
+ total_interactions = sum(r.interactions for r in results)
+ assert total_interactions == 0, (
+ f"{total_interactions} interaction(s) performed on Chrome!"
+ )
+
+ assert "back" in device.pressed_keys, (
+ "obstacle_guard did not press BACK to return to Instagram!"
+ )
diff --git a/tests/e2e/test_workflow_normal_feed.py b/tests/e2e/test_workflow_normal_feed.py
new file mode 100644
index 0000000..f6733fb
--- /dev/null
+++ b/tests/e2e/test_workflow_normal_feed.py
@@ -0,0 +1,92 @@
+"""
+E2E: Normal Feed Post Workflow
+===============================
+Tests the FULL production pipeline when the device shows a normal
+Instagram post. The pipeline must run without crashes and process
+the post normally.
+
+Mock: ONLY the device (XML dumps).
+Real: Cognitive stack, PluginRegistry, SessionState, Config.
+"""
+
+from tests.e2e.conftest import E2EDeviceStub
+
+from GramAddict.core.situational_awareness import SituationType
+
+
+NORMAL_POST_XML = """
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+"""
+
+
+class FakeSAENormal:
+ @classmethod
+ def get_instance(cls, device=None):
+ return cls()
+
+ def perceive(self, xml):
+ return SituationType.NORMAL
+
+ def unlearn_current_state(self, xml):
+ pass
+
+
+def test_normal_post_processes_without_crashes(monkeypatch, e2e_workflow_ctx):
+ """
+ A normal Instagram post must be processable by the full pipeline.
+ If any plugin has a broken import, missing attribute, or crashes
+ on valid XML, this test catches it.
+ """
+ monkeypatch.setattr(
+ "GramAddict.core.behaviors.obstacle_guard.SituationalAwarenessEngine",
+ FakeSAENormal,
+ )
+
+ device = E2EDeviceStub([NORMAL_POST_XML, NORMAL_POST_XML])
+ results, ctx = e2e_workflow_ctx(device)
+
+ # Multiple plugins must have run
+ executed_count = sum(1 for r in results if r.executed)
+ assert executed_count >= 1, (
+ f"Only {executed_count} plugin(s) executed on a normal feed post. "
+ "The pipeline is broken — plugins are crashing silently."
+ )
+
+ # No false positives: obstacle_guard must NOT have pressed BACK
+ assert "back" not in device.pressed_keys, (
+ "obstacle_guard pressed BACK on a normal Instagram post — false positive!"
+ )
+
+ # No CONTEXT_LOST
+ for result in results:
+ assert result.metadata.get("return_code") != "CONTEXT_LOST", (
+ "Plugin returned CONTEXT_LOST on a normal feed post!"
+ )
diff --git a/tests/e2e/test_workflow_permission_dialog.py b/tests/e2e/test_workflow_permission_dialog.py
new file mode 100644
index 0000000..2c74141
--- /dev/null
+++ b/tests/e2e/test_workflow_permission_dialog.py
@@ -0,0 +1,100 @@
+"""
+E2E: Permission Dialog Workflow
+================================
+Tests the FULL production pipeline when the device shows an Android
+permission dialog (e.g., "Allow Instagram to record audio?").
+
+Mock: ONLY the device (XML dumps).
+Real: Cognitive stack, PluginRegistry, SessionState, Config.
+"""
+
+from tests.e2e.conftest import E2EDeviceStub
+
+from GramAddict.core.situational_awareness import SituationType
+
+
+PERMISSION_DIALOG_XML = """
+
+
+
+
+
+
+
+
+
+"""
+
+NORMAL_POST_XML = """
+
+
+
+
+
+
+"""
+
+
+class FakeSAEPermission:
+ @classmethod
+ def get_instance(cls, device=None):
+ return cls()
+
+ def perceive(self, xml):
+ if "permissioncontroller" in xml:
+ return SituationType.OBSTACLE_SYSTEM
+ return SituationType.NORMAL
+
+ def unlearn_current_state(self, xml):
+ pass
+
+
+def test_permission_dialog_terminates_chain(monkeypatch, e2e_workflow_ctx):
+ """
+ PRODUCTION BUG 2026-04-29: Bot got stuck on "Allow Instagram to record audio?"
+ because obstacle_guard ignored OBSTACLE_SYSTEM and downstream plugins fired
+ against a PermissionController XML tree.
+
+ The full pipeline must: detect → press BACK → skip all interactions.
+ """
+ monkeypatch.setattr(
+ "GramAddict.core.behaviors.obstacle_guard.SituationalAwarenessEngine",
+ FakeSAEPermission,
+ )
+
+ device = E2EDeviceStub([PERMISSION_DIALOG_XML, NORMAL_POST_XML])
+ results, ctx = e2e_workflow_ctx(device, context_xml=PERMISSION_DIALOG_XML)
+
+ # Chain must have set should_skip=True
+ guard_fired = any(r.executed and r.should_skip for r in results)
+ assert guard_fired, (
+ "obstacle_guard did not fire — the bot would continue interacting "
+ "with a permission dialog in production!"
+ )
+
+ # Zero interactions on a permission dialog
+ total_interactions = sum(r.interactions for r in results)
+ assert total_interactions == 0, (
+ f"{total_interactions} interaction(s) performed on a permission dialog!"
+ )
+
+ # BACK must have been pressed
+ assert "back" in device.pressed_keys, (
+ "obstacle_guard did not press BACK — the dialog stays on screen!"
+ )
diff --git a/tests/e2e/test_workflow_plugin_integrity.py b/tests/e2e/test_workflow_plugin_integrity.py
new file mode 100644
index 0000000..195fe0b
--- /dev/null
+++ b/tests/e2e/test_workflow_plugin_integrity.py
@@ -0,0 +1,65 @@
+"""
+E2E: Plugin Import Integrity
+==============================
+Tests that ALL production plugins can be imported and instantiated.
+A single broken import (like the humanized_scroll bug) would crash here.
+
+This is not a workflow test — it's a build-time sanity gate.
+"""
+
+
+def test_all_plugins_importable_and_instantiable(setup_e2e_plugin_registry):
+ """
+ The plugin registry must contain all production plugins.
+ If any plugin has a broken import, the conftest fixture crashes.
+ """
+ expected_plugins = {
+ "ad_guard",
+ "anomaly_handler",
+ "close_friends_guard",
+ "obstacle_guard",
+ "perfect_snapping",
+ "post_data_extraction",
+ "resonance_evaluator",
+ "darwin_dwell",
+ "likes",
+ "comment",
+ "repost",
+ "post_interaction",
+ }
+
+ registered = {p.name for p in setup_e2e_plugin_registry.plugins}
+
+ missing = expected_plugins - registered
+ assert not missing, (
+ f"Plugins failed to register (import errors): {missing}. "
+ f"Registered: {registered}"
+ )
+
+
+def test_each_plugin_has_required_interface():
+ """
+ Every plugin module must import cleanly and have execute/can_activate/priority.
+ """
+ from GramAddict.core.behaviors.ad_guard import AdGuardPlugin
+ from GramAddict.core.behaviors.anomaly_handler import AnomalyHandlerPlugin
+ 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.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.resonance_evaluator import ResonanceEvaluatorPlugin
+
+ for cls in [
+ AdGuardPlugin, AnomalyHandlerPlugin, CloseFriendsGuardPlugin,
+ CommentPlugin, DarwinDwellPlugin, LikePlugin, ObstacleGuardPlugin,
+ PerfectSnappingPlugin, PostDataExtractionPlugin, PostInteractionPlugin,
+ ResonanceEvaluatorPlugin,
+ ]:
+ plugin = cls()
+ assert hasattr(plugin, "execute"), f"{cls.__name__} missing execute()"
+ assert hasattr(plugin, "can_activate"), f"{cls.__name__} missing can_activate()"
+ assert hasattr(plugin, "priority"), f"{cls.__name__} missing priority"