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

test(e2e): enforce 100% production-parity in dm_engine and follow verification tests

Browse Source
This commit is contained in:
Marc Mintel
2026-05-03 11:24:25 +02:00
parent 0a73c35809
commit f0a54d4e20
2 changed files with 90 additions and 140 deletions
Download Patch File Download Diff File Expand all files Collapse all files

176
tests/e2e/test_e2e_dm_engine.py
View File

@@ -15,7 +15,6 @@ Each test MUST fail before any production code is touched (TDD RED).
# Helpers — Minimal realistic mocks (no lying)
# ═══════════════════════════════════════════════════════
import os
import types
FIXTURES_DIR = os.path.join(os.path.dirname(os.path.dirname(__file__)), "fixtures")
@@ -37,28 +36,6 @@ def _get_dm_thread_pair():
return _load_fixture_img("dm_thread_dump.jpg"), _load_fixture_xml("dm_thread_dump.xml")
def _make_configs(dm_reply_enabled=False):
"""Create a realistic Config mock using the real Config class."""
from GramAddict.core.config import Config
configs = Config(first_run=True)
configs.args = types.SimpleNamespace(
disable_ai_messaging=False,
ai_condenser_model="qwen3.5:latest",
ai_condenser_url="http://localhost:11434/api/generate",
)
configs.config = {"plugins": {"dm_reply": {"enabled": dm_reply_enabled}}}
return configs
def _make_session_state(configs):
from GramAddict.core.session_state import SessionState
session = SessionState(configs)
session.set_limits_session()
return session
# ═══════════════════════════════════════════════════════
# Test 1: DM Engine MUST respect dm_reply.enabled config
# ═══════════════════════════════════════════════════════
@@ -67,7 +44,9 @@ def _make_session_state(configs):
class TestDMConfigGating:
"""Verifies that dm_reply.enabled=false prevents ALL DM interactions."""
def test_dm_engine_blocks_when_dm_reply_disabled(self, make_real_device_with_image):
def test_dm_engine_blocks_when_dm_reply_disabled(
self, make_real_device_with_image, e2e_configs, e2e_cognitive_stack_factory
):
"""BUG: dm_engine.py:96 checks 'disable_ai_messaging' (doesn't exist)
instead of dm_reply.enabled from config. This means DMs fire even when
config says enabled: false.
@@ -76,29 +55,26 @@ class TestDMConfigGating:
is disabled in the config.
"""
from GramAddict.core.dm_engine import _run_zero_latency_dm_loop
from GramAddict.core.dopamine_engine import DopamineEngine
from GramAddict.core.telepathic_engine import TelepathicEngine
from GramAddict.core.session_state import SessionState
inbox_img, inbox_xml = _get_dm_inbox_pair()
device = make_real_device_with_image(inbox_img, inbox_xml)
# Real Config
configs = _make_configs(dm_reply_enabled=False)
if "dm_reply" not in e2e_configs.config["plugins"]:
e2e_configs.config["plugins"]["dm_reply"] = {}
e2e_configs.config["plugins"]["dm_reply"]["enabled"] = False
session_state = _make_session_state(configs)
session_state = SessionState(e2e_configs)
session_state.set_limits_session()
dopamine = DopamineEngine()
dopamine.boredom = 0.0
telepathic = TelepathicEngine.get_instance()
cognitive_stack = {"telepathic": telepathic, "dopamine": dopamine, "dm_memory": None}
cognitive_stack = e2e_cognitive_stack_factory(device)
# No patches, 100% real engine
_run_zero_latency_dm_loop(
device,
make_real_device_with_image(inbox_img, inbox_xml),
None,
configs,
e2e_configs,
session_state,
"MessageInbox",
cognitive_stack,
@@ -118,53 +94,45 @@ class TestDMConfigGating:
class TestDMSendVerification:
"""Verifies that 'Successfully sent' is only logged when the message was actually sent."""
def test_dm_engine_successfully_identifies_and_sends_reply(self, make_real_device_with_image):
def test_dm_engine_successfully_identifies_and_sends_reply(
self, make_real_device_with_image, e2e_configs, e2e_cognitive_stack_factory
):
"""
EXPECTED: DM engine must use structural verification and successfully find the
message input field, type the message, find the 'Send' button, and send the message
using the real UI hierarchy and real LLM calls.
"""
from GramAddict.core.dm_engine import _run_zero_latency_dm_loop
from GramAddict.core.dopamine_engine import DopamineEngine
from GramAddict.core.telepathic_engine import TelepathicEngine
# Use the real dm_thread_dump.xml and image. We want to test that if the VLM selects the wrong thing
# it is rejected. Since the real VLM will pick the real send button in dm_thread_dump, this test
# as originally written (testing failure) needs to be modified or it might pass as a success now.
# Actually, let's just test that the DM engine successfully identifies the send button using the real thread!
from GramAddict.core.session_state import SessionState
inbox_img, inbox_xml = _get_dm_inbox_pair()
thread_img, thread_xml = _get_dm_thread_pair()
# Sequence for 1 reply:
# Loop 1: Inbox (start) -> Thread (read msg) -> Thread (find Send button) -> Thread (check after 1st back press)
# Loop 2: Inbox (start) -> Thread (read msg) -> Exits due to MAX_REPLIES_PER_INBOX_VISIT == 1
# Note: We don't provide a loop 2 here since we test success. Dopamine might hit boredom and exit cleanly.
# But wait, DM engine loops until dopamine triggers or MAX_REPLIES (3).
# We will set boredom artificially high so it exits after 1 reply.
device = make_real_device_with_image(
[inbox_img, thread_img, thread_img, thread_img, inbox_img, thread_img],
[inbox_xml, thread_xml, thread_xml, thread_xml, inbox_xml, thread_xml],
)
# Real Config
configs = _make_configs(dm_reply_enabled=True)
if "dm_reply" not in e2e_configs.config["plugins"]:
e2e_configs.config["plugins"]["dm_reply"] = {}
e2e_configs.config["plugins"]["dm_reply"]["enabled"] = True
session_state = _make_session_state(configs)
session_state = SessionState(e2e_configs)
session_state.set_limits_session()
dopamine = DopamineEngine()
dopamine.boredom = 0.0
telepathic = TelepathicEngine.get_instance()
cognitive_stack = {"telepathic": telepathic, "dopamine": dopamine, "dm_memory": None}
import GramAddict.core.dm_engine as dme
dme.MAX_REPLIES_PER_INBOX_VISIT = 1
cognitive_stack = e2e_cognitive_stack_factory(device)
cognitive_stack["dopamine"].boredom = 99.0 # Will trigger BOREDOM_CHANGE_FEED on the next loop
_run_zero_latency_dm_loop(
device,
make_real_device_with_image(inbox_img, inbox_xml),
None,
configs,
e2e_configs,
session_state,
"MessageInbox",
cognitive_stack,
@@ -182,20 +150,19 @@ class TestDMSendVerification:
class TestDMContextRequirement:
"""Verifies that the DM engine refuses to generate replies without context."""
def test_dm_engine_skips_thread_with_no_extractable_message(self, make_real_device_with_image):
def test_dm_engine_skips_thread_with_no_extractable_message(
self, make_real_device_with_image, e2e_configs, e2e_cognitive_stack_factory
):
"""BUG: dm_engine.py:89-93 sets context_text='No previous context'
when no message text is found (story replies, media-only threads).
Then proceeds to call the LLM with that string, producing garbage
like 'the to the'.
Evidence from logs:
7 out of 8 threads had 'Last received message context: No previous context'
All 7 were blindly replied to anyway.
EXPECTED: When context_text is 'No previous context' or empty,
the DM engine must SKIP the thread entirely (press back, continue).
"""
from GramAddict.core.dm_engine import _run_zero_latency_dm_loop
from GramAddict.core.session_state import SessionState
inbox_img, inbox_xml = _get_dm_inbox_pair()
@@ -211,25 +178,20 @@ class TestDMContextRequirement:
[inbox_img, thread_img, inbox_img, ""], [inbox_xml, thread_xml, inbox_xml, ""]
)
from GramAddict.core.dopamine_engine import DopamineEngine
from GramAddict.core.telepathic_engine import TelepathicEngine
if "dm_reply" not in e2e_configs.config["plugins"]:
e2e_configs.config["plugins"]["dm_reply"] = {}
e2e_configs.config["plugins"]["dm_reply"]["enabled"] = True
configs = _make_configs(dm_reply_enabled=True)
session_state = SessionState(e2e_configs)
session_state.set_limits_session()
session_state = _make_session_state(configs)
dopamine = DopamineEngine()
dopamine.boredom = 0.0
telepathic = TelepathicEngine.get_instance()
cognitive_stack = {"telepathic": telepathic, "dopamine": dopamine, "dm_memory": None}
cognitive_stack = e2e_cognitive_stack_factory(device)
_run_zero_latency_dm_loop(
device,
None,
None,
configs,
e2e_configs,
session_state,
"MessageInbox",
cognitive_stack,
@@ -247,24 +209,28 @@ class TestDMContextRequirement:
class TestDMIterationLimit:
"""Verifies the DM engine doesn't spam infinite replies."""
def test_dm_engine_caps_replies_per_session(self, make_real_device_with_image):
def test_dm_engine_caps_replies_per_session(
self, make_real_device_with_image, e2e_configs, e2e_cognitive_stack_factory
):
"""BUG: dm_engine.py:34 while loop only exits on session timeout or
boredom. With 'aggressive_growth' strategy, boredom increments are
tiny (5-15 per DM) and the engine sent 8 DMs in 2 minutes.
EXPECTED: DM engine must have an explicit max_replies_per_inbox
cap (e.g., 3) to prevent spam behavior. After reaching the cap,
cap (3 by default) to prevent spam behavior. After reaching the cap,
it should return 'BOREDOM_CHANGE_FEED'.
"""
from GramAddict.core.dm_engine import _run_zero_latency_dm_loop
from GramAddict.core.session_state import SessionState
inbox_img, inbox_xml = _get_dm_inbox_pair()
thread_img, thread_xml = _get_dm_thread_pair()
# Sequence for 2 replies (to test cap = 2):
# Sequence for 3 replies (production cap is 3):
# Loop 1: Inbox -> Thread -> Thread -> Thread
# Loop 2: Inbox -> Thread -> Thread -> Thread
# Loop 3: Inbox -> Thread -> Exits because cap=2
# Loop 3: Inbox -> Thread -> Thread -> Thread
# Loop 4: Inbox -> Thread -> Exits because cap=3
device = make_real_device_with_image(
[
inbox_img,
@@ -276,7 +242,11 @@ class TestDMIterationLimit:
thread_img,
thread_img, # Loop 2
inbox_img,
thread_img, # Loop 3 (hits limit)
thread_img,
thread_img,
thread_img, # Loop 3
inbox_img,
thread_img, # Loop 4 (hits limit)
],
[
inbox_xml,
@@ -289,37 +259,32 @@ class TestDMIterationLimit:
thread_xml,
inbox_xml,
thread_xml,
thread_xml,
thread_xml,
inbox_xml,
thread_xml,
],
)
from GramAddict.core.dopamine_engine import DopamineEngine
from GramAddict.core.telepathic_engine import TelepathicEngine
if "dm_reply" not in e2e_configs.config["plugins"]:
e2e_configs.config["plugins"]["dm_reply"] = {}
e2e_configs.config["plugins"]["dm_reply"]["enabled"] = True
configs = _make_configs(dm_reply_enabled=True)
session_state = SessionState(e2e_configs)
session_state.set_limits_session()
session_state = _make_session_state(configs)
dopamine = DopamineEngine()
dopamine.boredom = 0.0
telepathic = TelepathicEngine.get_instance()
cognitive_stack = {"telepathic": telepathic, "dopamine": dopamine, "dm_memory": None}
# Override session_state methods that are used in loop directly instead of MagicMock
configs.args.current_success_limit = 8
configs.args.current_pm_limit = 8
cognitive_stack = e2e_cognitive_stack_factory(device)
# Force the session limits higher so it strictly tests the inner loop cap
e2e_configs.args.current_success_limit = 8
e2e_configs.args.current_pm_limit = 8
session_state.totalMessages = 0
import GramAddict.core.dm_engine as dme
dme.MAX_REPLIES_PER_INBOX_VISIT = 2
# Force the session to never hit limits (simulating the real scenario)
_run_zero_latency_dm_loop(
device,
make_real_device_with_image(inbox_img, inbox_xml),
None,
configs,
e2e_configs,
session_state,
"MessageInbox",
cognitive_stack,
@@ -341,14 +306,17 @@ class TestDMConfigGatingProduction:
"""Verifies the REAL dm_engine._run_zero_latency_dm_loop config check,
not a local re-implementation of bot_flow.py logic."""
def test_dm_engine_config_gating_reads_real_plugin_config(self):
def test_dm_engine_config_gating_reads_real_plugin_config(self, e2e_configs):
"""The dm_engine kill-switch at line 46-50 reads configs.get_plugin_config('dm_reply').
We verify this path with the real Config class — NOT a local dict simulation."""
configs_disabled = _make_configs(dm_reply_enabled=False)
configs_enabled = _make_configs(dm_reply_enabled=True)
dm_config_off = configs_disabled.get_plugin_config("dm_reply")
dm_config_on = configs_enabled.get_plugin_config("dm_reply")
if "dm_reply" not in e2e_configs.config["plugins"]:
e2e_configs.config["plugins"]["dm_reply"] = {}
e2e_configs.config["plugins"]["dm_reply"]["enabled"] = False
dm_config_off = e2e_configs.get_plugin_config("dm_reply")
assert dm_config_off.get("enabled", False) is False, "Config should report dm_reply as disabled"
e2e_configs.config["plugins"]["dm_reply"]["enabled"] = True
dm_config_on = e2e_configs.get_plugin_config("dm_reply")
assert dm_config_on.get("enabled", False) is True, "Config should report dm_reply as enabled"

54
tests/e2e/test_follow_verification_integrity.py
View File

@@ -15,10 +15,8 @@ Root cause chain:
Each test MUST fail (RED) before any production code is fixed.
"""
from GramAddict.core.config import Config
from GramAddict.core.perception.action_memory import ActionMemory
from GramAddict.core.perception.spatial_parser import SpatialNode
from GramAddict.core.session_state import SessionState
# ═══════════════════════════════════════════════════════
# TEST 1: verify_success MUST reject wrong-element clicks for follow
@@ -131,7 +129,7 @@ class TestQNavGraphDoBlocksFollowWithoutButton:
when the current screen has no Follow button.
"""
def test_do_rejects_follow_when_not_in_available_actions(self, make_real_device_with_image):
def test_do_rejects_follow_when_not_in_available_actions(self, make_real_device_with_image, e2e_configs):
"""
If the current screen's available_actions does not contain 'tap follow button',
QNavGraph.do("tap 'Follow' button") MUST return False immediately.
@@ -140,20 +138,11 @@ class TestQNavGraphDoBlocksFollowWithoutButton:
so it NEVER gets checked. The bot blindly passes through to GOAP.
"""
from GramAddict.core.q_nav_graph import QNavGraph
from GramAddict.core.session_state import SessionState
device = make_real_device_with_image(None, "<hierarchy/>")
import types
from GramAddict.core.config import Config
from GramAddict.core.session_state import SessionState
configs = Config(first_run=True)
configs.args = types.SimpleNamespace()
configs.args.disable_ai_messaging = False
configs.args.ai_condenser_model = "qwen3.5:latest"
configs.args.ai_condenser_url = "http://localhost:11434/api/generate"
SessionState(configs)
SessionState(e2e_configs)
nav = QNavGraph(device)
@@ -239,7 +228,9 @@ class TestGOAPInteractionCrossCheck:
and the intent BEFORE trusting the VLM verification.
"""
def test_execute_action_rejects_when_clicked_node_doesnt_match_intent(self, make_real_device_with_image):
def test_execute_action_rejects_when_clicked_node_doesnt_match_intent(
self, make_real_device_with_image, e2e_configs
):
"""
If find_best_node returns a node with desc='3 photos by ...'
for intent='tap Follow button', _execute_action MUST reject it
@@ -248,17 +239,8 @@ class TestGOAPInteractionCrossCheck:
Currently: _execute_action clicks first, then asks VLM to verify.
The VLM verification is the fox guarding the henhouse.
"""
import types
from GramAddict.core.config import Config
from GramAddict.core.goap import GoalExecutor
configs = Config(first_run=True)
configs.args = types.SimpleNamespace()
configs.args.disable_ai_messaging = False
configs.args.ai_condenser_model = "qwen3.5:latest"
configs.args.ai_condenser_url = "http://localhost:11434/api/generate"
xml_dump = """<?xml version="1.0" encoding="UTF-8"?>
<hierarchy>
<node resource-id="com.instagram.android:id/image_button"
@@ -307,7 +289,9 @@ class TestFollowPluginEndToEnd:
session state is corrupted.
"""
def test_follow_plugin_does_not_count_follow_when_wrong_element_clicked(self, make_real_device_with_image):
def test_follow_plugin_does_not_count_follow_when_wrong_element_clicked(
self, make_real_device_with_image, e2e_configs
):
"""
By removing lying mocks, we test the REAL E2E behavior:
If we give the plugin a screen with NO follow button, QNavGraph.do()
@@ -319,18 +303,16 @@ class TestFollowPluginEndToEnd:
plugin = FollowPlugin()
import types
e2e_configs.args.follow_percentage = 100
e2e_configs.args.current_likes_limit = 300
configs = Config(first_run=True)
configs.args = types.SimpleNamespace()
configs.args.follow_percentage = 100
configs.args.current_likes_limit = 300
configs.args.disable_ai_messaging = False
configs.args.ai_condenser_model = "qwen3.5:latest"
configs.args.ai_condenser_url = "http://localhost:11434/api/generate"
configs.config = {"plugins": {"follow": {"percentage": 100}}}
if "follow" not in e2e_configs.config["plugins"]:
e2e_configs.config["plugins"]["follow"] = {}
e2e_configs.config["plugins"]["follow"]["percentage"] = 100
session_state = SessionState(configs)
from GramAddict.core.session_state import SessionState
session_state = SessionState(e2e_configs)
session_state.added_interactions = []
original_add_interaction = session_state.add_interaction
@@ -359,7 +341,7 @@ class TestFollowPluginEndToEnd:
ctx = BehaviorContext(
device=device,
session_state=session_state,
configs=configs,
configs=e2e_configs,
username="missiongreenenergy",
cognitive_stack={"nav_graph": nav_graph},
)