diff --git a/tests/e2e/device_emulator.py b/tests/e2e/device_emulator.py index 674d464..63bc519 100644 --- a/tests/e2e/device_emulator.py +++ b/tests/e2e/device_emulator.py @@ -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 diff --git a/tests/e2e/test_engine_timeout.py b/tests/e2e/test_engine_timeout.py index 5d11ff8..c4dfca5 100644 --- a/tests/e2e/test_engine_timeout.py +++ b/tests/e2e/test_engine_timeout.py @@ -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 diff --git a/tests/e2e/test_post_data_extraction.py b/tests/e2e/test_post_data_extraction.py index 1e1494b..e6e8104 100644 --- a/tests/e2e/test_post_data_extraction.py +++ b/tests/e2e/test_post_data_extraction.py @@ -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}" + diff --git a/tests/e2e/test_system_sae.py b/tests/e2e/test_system_sae.py index 840e792..87f5688 100644 --- a/tests/e2e/test_system_sae.py +++ b/tests/e2e/test_system_sae.py @@ -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 = '' @@ -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