mmintel/instagram-bot
1
0
Fork 0
Code Issues Pull Requests 4 Actions Packages Projects Releases Wiki Activity

refactor(e2e): extract workflow infrastructure into conftest, one test file per workflow

Browse Source 
BEFORE: Giant monolith test files with duplicated device stubs, config
factories, and cognitive stack builders. The infrastructure was copy-pasted
into each file.

AFTER: Clean separation:
- conftest.py: E2EDeviceStub, e2e_device, e2e_cognitive_stack,
  e2e_session, e2e_workflow_ctx fixtures
- test_workflow_permission_dialog.py: Permission dialog recovery
- test_workflow_foreign_app.py: Foreign app (Chrome) recovery
- test_workflow_normal_feed.py: Normal post processing
- test_workflow_plugin_integrity.py: Import + interface sanity

Each workflow file is <100 lines. Zero duplication. Every test uses
the shared e2e_workflow_ctx fixture which runs the EXACT code path
from bot_flow.py:942-971.

DELETED: test_full_workflow_hostile_env.py, test_plugin_chain_hostile_env.py
(monolith files replaced by the above)
This commit is contained in:
Marc Mintel
2026-04-30 11:20:35 +02:00
parent b83b55e02b
commit 2c44331f03
7 changed files with 515 additions and 901 deletions
Download Patch File Download Diff File Expand all files Collapse all files

177
tests/e2e/conftest.py
View File

@@ -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

520
tests/e2e/test_full_workflow_hostile_env.py
View File

@@ -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 = """<?xml version="1.0" encoding="utf-8"?>
<hierarchy rotation="0">
<node class="android.widget.FrameLayout" package="com.instagram.android"
bounds="[0,0][1080,2400]">
<node resource-id="com.instagram.android:id/action_bar_container"
class="android.widget.FrameLayout" bounds="[0,0][1080,170]">
<node text="Instagram" class="android.widget.TextView" />
</node>
<node resource-id="com.instagram.android:id/row_feed_photo_profile_name"
class="android.widget.TextView" text="photographer_jane"
clickable="true" bounds="[160,300][600,340]" />
<node resource-id="com.instagram.android:id/row_feed_photo_imageview"
class="android.widget.ImageView"
content-desc="Photo by photographer_jane. A beautiful sunset over mountains."
bounds="[0,350][1080,1350]" />
<node resource-id="com.instagram.android:id/row_feed_button_like"
class="android.widget.ImageView" clickable="true"
content-desc="Like" bounds="[30,1360][120,1450]" />
<node resource-id="com.instagram.android:id/row_feed_button_comment"
class="android.widget.ImageView" clickable="true"
content-desc="Comment" bounds="[150,1360][240,1450]" />
<node resource-id="com.instagram.android:id/row_feed_button_share"
class="android.widget.ImageView" clickable="true"
content-desc="Share" bounds="[270,1360][360,1450]" />
<node resource-id="com.instagram.android:id/row_feed_textview_likes"
class="android.widget.TextView" text="1,234 likes"
bounds="[30,1460][400,1500]" />
<node resource-id="com.instagram.android:id/row_feed_comment_textview_layout"
class="android.widget.LinearLayout" bounds="[30,1510][1050,1600]">
<node text="photographer_jane" class="android.widget.TextView" />
<node text="Golden hour never disappoints 🌅" class="android.widget.TextView" />
</node>
<node resource-id="com.instagram.android:id/tab_bar" bounds="[0,2300][1080,2400]">
<node resource-id="com.instagram.android:id/feed_tab" content-desc="Home"
selected="true" clickable="true" bounds="[0,2300][216,2400]" />
<node resource-id="com.instagram.android:id/search_tab" content-desc="Search"
clickable="true" bounds="[216,2300][432,2400]" />
</node>
</node>
</hierarchy>
"""
PERMISSION_DIALOG_XML = """<?xml version="1.0" encoding="utf-8"?>
<hierarchy rotation="0">
<node class="android.widget.FrameLayout" package="com.android.permissioncontroller"
bounds="[0,0][1080,2400]">
<node resource-id="com.android.permissioncontroller:id/grant_dialog"
class="android.widget.LinearLayout" package="com.android.permissioncontroller"
clickable="false" bounds="[100,800][980,1600]">
<node text="Allow Instagram to record audio?"
class="android.widget.TextView"
resource-id="com.android.permissioncontroller:id/permission_message" />
<node text="While using the app"
class="android.widget.Button" clickable="true"
resource-id="com.android.permissioncontroller:id/permission_allow_foreground_only_button"
bounds="[100,1000][980,1100]" />
<node text="Don't allow"
class="android.widget.Button" clickable="true"
resource-id="com.android.permissioncontroller:id/permission_deny_button"
bounds="[100,1240][980,1340]" />
<node resource-id="com.android.permissioncontroller:id/permission_icon"
class="android.widget.ImageView"
content-desc=""
bounds="[440,850][640,950]" />
</node>
</node>
</hierarchy>
"""
# ═══════════════════════════════════════════════════════
# 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!"

381
tests/e2e/test_plugin_chain_hostile_env.py
View File

@@ -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 = """<?xml version="1.0" encoding="utf-8"?>
<hierarchy rotation="0">
<node class="android.widget.FrameLayout" package="com.android.permissioncontroller"
bounds="[0,0][1080,2400]">
<node resource-id="com.android.permissioncontroller:id/grant_dialog"
class="android.widget.LinearLayout" package="com.android.permissioncontroller"
clickable="false" bounds="[100,800][980,1600]">
<node text="Allow Instagram to record audio?"
class="android.widget.TextView"
resource-id="com.android.permissioncontroller:id/permission_message" />
<node text="While using the app"
class="android.widget.Button" clickable="true"
resource-id="com.android.permissioncontroller:id/permission_allow_foreground_only_button"
bounds="[100,1000][980,1100]" />
<node text="Only this time"
class="android.widget.Button" clickable="true"
resource-id="com.android.permissioncontroller:id/permission_allow_one_time_button"
bounds="[100,1120][980,1220]" />
<node text="Don't allow"
class="android.widget.Button" clickable="true"
resource-id="com.android.permissioncontroller:id/permission_deny_button"
bounds="[100,1240][980,1340]" />
<node resource-id="com.android.permissioncontroller:id/permission_icon"
class="android.widget.ImageView"
content-desc=""
bounds="[440,850][640,950]" />
</node>
</node>
</hierarchy>
"""
FOREIGN_APP_CHROME_XML = """<?xml version="1.0" encoding="utf-8"?>
<hierarchy rotation="0">
<node class="android.widget.FrameLayout" package="com.android.chrome"
bounds="[0,0][1080,2400]">
<node text="https://www.example.com"
class="android.widget.EditText"
resource-id="com.android.chrome:id/url_bar" />
<node text="Example Domain"
class="android.widget.TextView" />
</node>
</hierarchy>
"""
NORMAL_INSTAGRAM_XML = """<?xml version="1.0" encoding="utf-8"?>
<hierarchy rotation="0">
<node class="android.widget.FrameLayout" package="com.instagram.android"
bounds="[0,0][1080,2400]">
<node resource-id="com.instagram.android:id/row_feed_button_like"
class="android.widget.ImageView" clickable="true"
content-desc="Like" bounds="[30,1500][120,1590]" />
<node resource-id="com.instagram.android:id/row_feed_photo_profile_name"
class="android.widget.TextView" text="photographer_jane"
bounds="[160,300][600,340]" />
</node>
</hierarchy>
"""
# ═══════════════════════════════════════════════════════
# 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!"
)

81
tests/e2e/test_workflow_foreign_app.py Normal file
View File

@@ -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 = """<?xml version="1.0" encoding="utf-8"?>
<hierarchy rotation="0">
<node class="android.widget.FrameLayout" package="com.android.chrome"
bounds="[0,0][1080,2400]">
<node text="https://www.example.com"
class="android.widget.EditText"
resource-id="com.android.chrome:id/url_bar" />
<node text="Example Domain"
class="android.widget.TextView" />
</node>
</hierarchy>
"""
NORMAL_POST_XML = """<?xml version="1.0" encoding="utf-8"?>
<hierarchy rotation="0">
<node class="android.widget.FrameLayout" package="com.instagram.android"
bounds="[0,0][1080,2400]">
<node resource-id="com.instagram.android:id/row_feed_button_like"
class="android.widget.ImageView" clickable="true"
content-desc="Like" bounds="[30,1360][120,1450]" />
</node>
</hierarchy>
"""
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!"
)

92
tests/e2e/test_workflow_normal_feed.py Normal file
View File

@@ -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 = """<?xml version="1.0" encoding="utf-8"?>
<hierarchy rotation="0">
<node class="android.widget.FrameLayout" package="com.instagram.android"
bounds="[0,0][1080,2400]">
<node resource-id="com.instagram.android:id/action_bar_container"
class="android.widget.FrameLayout" bounds="[0,0][1080,170]">
<node text="Instagram" class="android.widget.TextView" />
</node>
<node resource-id="com.instagram.android:id/row_feed_photo_profile_name"
class="android.widget.TextView" text="photographer_jane"
clickable="true" bounds="[160,300][600,340]" />
<node resource-id="com.instagram.android:id/row_feed_photo_imageview"
class="android.widget.ImageView"
content-desc="Photo by photographer_jane. A beautiful sunset."
bounds="[0,350][1080,1350]" />
<node resource-id="com.instagram.android:id/row_feed_button_like"
class="android.widget.ImageView" clickable="true"
content-desc="Like" bounds="[30,1360][120,1450]" />
<node resource-id="com.instagram.android:id/row_feed_button_comment"
class="android.widget.ImageView" clickable="true"
content-desc="Comment" bounds="[150,1360][240,1450]" />
<node resource-id="com.instagram.android:id/row_feed_textview_likes"
class="android.widget.TextView" text="1,234 likes"
bounds="[30,1460][400,1500]" />
<node resource-id="com.instagram.android:id/tab_bar" bounds="[0,2300][1080,2400]">
<node resource-id="com.instagram.android:id/feed_tab" content-desc="Home"
selected="true" clickable="true" bounds="[0,2300][216,2400]" />
</node>
</node>
</hierarchy>
"""
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!"
)

100
tests/e2e/test_workflow_permission_dialog.py Normal file
View File

@@ -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 = """<?xml version="1.0" encoding="utf-8"?>
<hierarchy rotation="0">
<node class="android.widget.FrameLayout" package="com.android.permissioncontroller"
bounds="[0,0][1080,2400]">
<node resource-id="com.android.permissioncontroller:id/grant_dialog"
class="android.widget.LinearLayout" package="com.android.permissioncontroller"
clickable="false" bounds="[100,800][980,1600]">
<node text="Allow Instagram to record audio?"
class="android.widget.TextView"
resource-id="com.android.permissioncontroller:id/permission_message" />
<node text="While using the app"
class="android.widget.Button" clickable="true"
resource-id="com.android.permissioncontroller:id/permission_allow_foreground_only_button"
bounds="[100,1000][980,1100]" />
<node text="Don't allow"
class="android.widget.Button" clickable="true"
resource-id="com.android.permissioncontroller:id/permission_deny_button"
bounds="[100,1240][980,1340]" />
</node>
</node>
</hierarchy>
"""
NORMAL_POST_XML = """<?xml version="1.0" encoding="utf-8"?>
<hierarchy rotation="0">
<node class="android.widget.FrameLayout" package="com.instagram.android"
bounds="[0,0][1080,2400]">
<node resource-id="com.instagram.android:id/row_feed_photo_profile_name"
class="android.widget.TextView" text="photographer_jane"
clickable="true" bounds="[160,300][600,340]" />
<node resource-id="com.instagram.android:id/row_feed_button_like"
class="android.widget.ImageView" clickable="true"
content-desc="Like" bounds="[30,1360][120,1450]" />
</node>
</hierarchy>
"""
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!"
)

65
tests/e2e/test_workflow_plugin_integrity.py Normal file
View File

@@ -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"