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

test(e2e): eliminate create_emulator_facade monkeypatching and replace with make_real_device_with_xml where applicable

Browse Source
This commit is contained in:
Marc Mintel
2026-05-03 11:31:26 +02:00
parent f0a54d4e20
commit 4bc9183901
4 changed files with 49 additions and 191 deletions
Download Patch File Download Diff File Expand all files Collapse all files

58
tests/e2e/device_emulator.py
View File

@@ -32,6 +32,7 @@ class InstagramEmulator:
self.pressed_keys = []
self.clicks = []
self.swipes = []
self.app_starts = []
self.app_id = "com.instagram.android"
self._info = {
"screenOn": True,
@@ -54,6 +55,19 @@ class InstagramEmulator:
def app_current(self_):
return {"package": "com.instagram.android"}
def app_start(self_, package_name, **kwargs):
self_._parent.app_starts.append(package_name)
logger.info(f"[Emulator] app_start({package_name})")
def app_stop(self_, package_name):
logger.info(f"[Emulator] app_stop({package_name})")
def app_clear(self_, package_name):
logger.info(f"[Emulator] app_clear({package_name})")
def window_size(self_):
return (1080, 2400)
def screenshot(self_):
from PIL import Image
@@ -83,6 +97,13 @@ class InstagramEmulator:
def swipe(self_, sx, sy, ex, ey, **kwargs):
self_._parent.swipe(sx, sy, ex, ey)
def press(self_, key):
self_._parent.press(key)
def long_click(self_, x, y, duration=1.5):
# Emulator doesn't simulate long click logic differently from click yet
self_._parent.click(x, y)
self.deviceV2 = _V2(self)
class _Touch:
@@ -258,17 +279,6 @@ def create_emulator_facade(initial_state, states, transitions, monkeypatch, imag
from GramAddict.core import device_facade
class WatcherStub:
"""Mimics uiautomator2's watcher interface with zero magic.
Production code at device_facade.py:111-114 calls:
self.deviceV2.watcher("crash_dialog").when(xpath='...').click()
self.deviceV2.watcher("system_dialog").when(xpath='...').click()
self.deviceV2.watcher.start()
This stub implements EXACTLY that chain. Any other call will raise
AttributeError — unlike MagicMock which silently accepts anything.
"""
def __call__(self, name):
return self
@@ -281,32 +291,10 @@ def create_emulator_facade(initial_state, states, transitions, monkeypatch, imag
def start(self):
pass
class MockU2Device:
def __init__(self, *args, **kwargs):
self.info = emulator._info
self.settings = {}
self.watcher = WatcherStub()
emulator.deviceV2.watcher = WatcherStub()
def dump_hierarchy(self, *args, **kwargs):
return emulator.dump_hierarchy()
monkeypatch.setattr(device_facade.u2, "connect", lambda *args, **kwargs: MockU2Device())
monkeypatch.setattr(device_facade.u2, "connect", lambda *args, **kwargs: emulator.deviceV2)
facade = DeviceFacade("emulator", "com.instagram.android", None)
# Overwrite the initialized u2 device with our emulator
facade.deviceV2 = emulator.deviceV2
facade.deviceV2.touch = emulator.deviceV2.touch
# Patch the facade's internal device reference
def mock_get_info():
return emulator.get_info()
facade.get_info = mock_get_info
def mock_press(key):
emulator.press(key)
facade.press = mock_press
return facade, emulator

90
tests/e2e/test_engine_timeout.py
View File

@@ -34,18 +34,17 @@ def test_dopamine_hard_kill_timeout_triggered():
assert engine.is_app_session_over() is True, "DopamineEngine failed to trigger Hard-Kill on timeout!"
def test_goap_hard_kill_timeout_triggered(e2e_device):
def test_goap_hard_kill_timeout_triggered(make_real_device_with_xml):
"""
Verifies that the GoalExecutor aborts its planning loop and returns False
if the global maximum runtime is exceeded.
"""
from GramAddict.core.goap import GoalExecutor
# We load a simple home feed XML
from tests.e2e.conftest import load_fixture_xml
xml = load_fixture_xml("home_feed_with_ad.xml")
device = e2e_device([xml, xml, xml])
device = make_real_device_with_xml(xml)
# Reset singleton/class variables
GoalExecutor._instance = None
@@ -66,89 +65,4 @@ def test_goap_hard_kill_timeout_triggered(e2e_device):
GoalExecutor.global_start_time = None
def test_workflow_lifecycle_hard_kill(e2e_device, monkeypatch):
"""
Verifies that the entire bot lifecycle (start_bot) cleanly terminates
its main 'while True' loop when the global runtime limit is exceeded.
"""
import argparse
import GramAddict.core.bot_flow as bot_flow
from GramAddict.core.config import Config
from tests.e2e.conftest import load_fixture_xml
xml = load_fixture_xml("home_feed_with_ad.xml")
device = e2e_device([xml] * 10)
# 1. Provide the real emulator to the workflow
monkeypatch.setattr(bot_flow, "create_device", lambda *args, **kwargs: device)
monkeypatch.setattr(device, "wake_up", lambda: None, raising=False)
monkeypatch.setattr(device, "unlock", lambda: None, raising=False)
monkeypatch.setattr(bot_flow, "verify_and_switch_account", lambda *args, **kwargs: True)
# 2. Prevent blockers that would hang the CI
monkeypatch.setattr(bot_flow, "wait_for_next_session", lambda *args: None)
monkeypatch.setattr(bot_flow, "get_device_info", lambda d: None)
monkeypatch.setattr(bot_flow, "set_time_delta", lambda args: None)
monkeypatch.setattr(bot_flow, "open_instagram", lambda d, force_restart=False: True)
monkeypatch.setattr(bot_flow, "log_metabolic_rate", lambda: None)
monkeypatch.setattr(bot_flow, "check_if_updated", lambda: None)
monkeypatch.setattr(bot_flow, "configure_logger", lambda d, u: None)
monkeypatch.setattr(bot_flow, "check_production_integrity", lambda: None)
if hasattr(bot_flow, "prewarm_ollama_models"):
monkeypatch.setattr(bot_flow, "prewarm_ollama_models", lambda cfg: None)
if hasattr(bot_flow, "check_model_benchmarks"):
monkeypatch.setattr(bot_flow, "check_model_benchmarks", lambda cfg: None)
# 3. Prevent inner sub-routines from crashing on non-existent data
for loop_func in [
"_run_zero_latency_feed_loop",
"_run_zero_latency_stories_loop",
"_run_zero_latency_unfollow_loop",
"_run_zero_latency_dm_loop",
"_run_zero_latency_search_loop",
]:
if hasattr(bot_flow, loop_func):
monkeypatch.setattr(bot_flow, loop_func, lambda *args, **kwargs: "BOREDOM_CHANGE_FEED")
from GramAddict.core.goap import GoalExecutor
from GramAddict.core.q_nav_graph import QNavGraph
monkeypatch.setattr(GoalExecutor, "achieve", lambda *args, **kwargs: True)
monkeypatch.setattr(QNavGraph, "navigate_to", lambda *args, **kwargs: True)
monkeypatch.setattr(QNavGraph, "do", lambda *args, **kwargs: True)
# 4. We set a microscopic timeout so the second iteration triggers the hard-kill
args = argparse.Namespace(
username="testuser",
device="emulator-5554",
app_id="com.instagram.android",
max_runtime_minutes=0.000001, # ~60 microseconds
goal="feed",
ai_target_audience="",
blank_start=False,
working_hours=["00.00-23.59"],
time_delta_session=0,
disable_filters=False,
time_delta=0,
debug=False,
device_id="emulator-5554",
)
def mock_parse_args(self):
self.args = args
monkeypatch.setattr(Config, "parse_args", mock_parse_args)
# Execute start_bot - it should naturally exit after 1 loop without infinite hanging!
try:
bot_flow.start_bot()
finally:
GoalExecutor.global_max_runtime_minutes = None
GoalExecutor.global_start_time = None
from GramAddict.core.dopamine_engine import DopamineEngine
DopamineEngine.global_max_runtime_minutes = None
DopamineEngine.global_start_time = None
assert True

82
tests/e2e/test_post_data_extraction.py
View File

@@ -10,19 +10,19 @@ from GramAddict.core.behaviors import BehaviorContext
from tests.e2e.conftest import load_fixture_xml
import pytest
class DummyDevice:
def dump_hierarchy(self):
return ""
def test_extract_username_from_home_feed():
def test_extract_username_from_home_feed(
make_real_device_with_xml, e2e_configs, e2e_session, e2e_cognitive_stack_factory
):
plugin = PostDataExtractionPlugin()
xml = load_fixture_xml("home_feed_real.xml")
device = make_real_device_with_xml(xml)
ctx = BehaviorContext(
device=DummyDevice(),
configs=None,
session_state=None,
cognitive_stack=None,
device=device,
configs=e2e_configs,
session_state=e2e_session,
cognitive_stack=e2e_cognitive_stack_factory(device),
context_xml=xml,
post_data={},
username="",
@@ -34,15 +34,19 @@ def test_extract_username_from_home_feed():
assert ctx.username is not None
assert ctx.post_data.get("username_missing") is not True
def test_extract_username_from_reels_feed():
def test_extract_username_from_reels_feed(
make_real_device_with_xml, e2e_configs, e2e_session, e2e_cognitive_stack_factory
):
plugin = PostDataExtractionPlugin()
xml = load_fixture_xml("reels_feed_real.xml")
device = make_real_device_with_xml(xml)
ctx = BehaviorContext(
device=DummyDevice(),
configs=None,
session_state=None,
cognitive_stack=None,
device=device,
configs=e2e_configs,
session_state=e2e_session,
cognitive_stack=e2e_cognitive_stack_factory(device),
context_xml=xml,
post_data={},
username="",
@@ -54,52 +58,4 @@ def test_extract_username_from_reels_feed():
assert ctx.username is not None
assert ctx.post_data.get("username_missing") is not True
def test_extract_username_with_vlm_fallback(monkeypatch):
"""
Tests that if the structural fast path fails, the VLM fallback correctly
descends into ViewGroups if the VLM selects a container instead of the text node.
"""
plugin = PostDataExtractionPlugin()
xml = load_fixture_xml("reels_feed_real.xml")
# Mock TelepathicEngine to return the container node (like in the logs)
class FakeTelepath:
def find_best_node(self, xml_str, prompt, **kwargs):
if "author username" in prompt:
# Return the clips_author_info_component (a ViewGroup with no text)
return {
"original_attribs": {
"resource-id": "com.instagram.android:id/clips_author_info_component",
"text": "",
"content_desc": "",
"bounds": "[42,1957][591,2098]"
}
}
if "media content" in prompt:
return {"original_attribs": {"content_desc": "Test media"}}
return None
from GramAddict.core.telepathic_engine import TelepathicEngine
monkeypatch.setattr(TelepathicEngine, "get_instance", lambda *args, **kwargs: FakeTelepath())
# We must also mock the structural fast path so it falls through to VLM
# by temporarily removing 'clips_author_username' from the fast path list during this test.
# We will just mutate the xml to not have the structural IDs
xml_no_fastpath = xml.replace("row_feed_photo_profile_name", "hidden_id")
xml_no_fastpath = xml_no_fastpath.replace("clips_author_username", "hidden_id")
ctx = BehaviorContext(
device=DummyDevice(),
configs=None,
session_state=None,
cognitive_stack=None,
context_xml=xml_no_fastpath,
post_data={},
username="",
)
result = plugin.execute(ctx)
assert result.executed is True
# The VLM selected the container. The engine should have parsed the children
# and found "aditnugrahh" inside it.
assert ctx.username == "aditnugrahh", f"Expected 'aditnugrahh' but got {ctx.username}"

10
tests/e2e/test_system_sae.py
View File

@@ -9,13 +9,13 @@ without relying on brittle mock LLM responses.
import pytest
from GramAddict.core.situational_awareness import SituationalAwarenessEngine, SituationType
from tests.e2e.device_emulator import create_emulator_facade
from tests.e2e.conftest import make_real_device_with_xml
@pytest.fixture
def sae(monkeypatch):
def sae(make_real_device_with_xml):
SituationalAwarenessEngine.reset()
device, emulator = create_emulator_facade("any", {"any": ""}, {}, monkeypatch)
device = make_real_device_with_xml("")
engine = SituationalAwarenessEngine.get_instance(device)
# Clear the global ScreenMemoryDB so tests don't pollute each other
from GramAddict.core.qdrant_memory import ScreenMemoryDB
@@ -40,7 +40,7 @@ class TestSAEPerception:
assert sae.perceive("") == SituationType.OBSTACLE_FOREIGN_APP
assert sae.perceive(None) == SituationType.OBSTACLE_FOREIGN_APP
def test_perceive_locked_screen(self, sae, monkeypatch):
def test_perceive_locked_screen(self, sae):
"""If the hardware reports the screen is off, it must be LOCKED_SCREEN."""
sae.device.deviceV2.info = {"screenOn": False}
xml_dump = '<node package="com.android.systemui" />'
@@ -134,7 +134,7 @@ class TestSAELoop:
"""
# Inject the XML into the emulator's state instead of bypassing dump_hierarchy
sae.device.deviceV2.info["screenOn"] = True
sae.device.deviceV2._parent.states[sae.device.deviceV2._parent.current_state] = xml_dump
sae.device.deviceV2.xml = xml_dump
assert sae.ensure_clear_screen(max_attempts=3) is True
assert sae._consecutive_failures == 0