diff --git a/tests/e2e/test_e2e_dm_engine.py b/tests/e2e/test_e2e_dm_engine.py index 9ccbe22..24f1b35 100644 --- a/tests/e2e/test_e2e_dm_engine.py +++ b/tests/e2e/test_e2e_dm_engine.py @@ -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" diff --git a/tests/e2e/test_follow_verification_integrity.py b/tests/e2e/test_follow_verification_integrity.py index daffa11..d072173 100644 --- a/tests/e2e/test_follow_verification_integrity.py +++ b/tests/e2e/test_follow_verification_integrity.py @@ -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, "") - 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 = """