diff --git a/GramAddict/core/behaviors/perfect_snapping.py b/GramAddict/core/behaviors/perfect_snapping.py
index ffffb3f..ed7c11d 100644
--- a/GramAddict/core/behaviors/perfect_snapping.py
+++ b/GramAddict/core/behaviors/perfect_snapping.py
@@ -26,7 +26,15 @@ class PerfectSnappingPlugin(BehaviorPlugin):
return 90
def can_activate(self, ctx: BehaviorContext) -> bool:
- return getattr(self, "_enabled", True)
+ if not getattr(self, "_enabled", True):
+ return False
+
+ xml_lower = ctx.context_xml.lower()
+ # Do not snap if we are on a profile page or grid, it's meant for posts.
+ if "profile_tabs_container" in xml_lower or "explore_grid" in xml_lower:
+ return False
+
+ return True
def execute(self, ctx: BehaviorContext) -> BehaviorResult:
aligned = _align_active_post(ctx.device)
diff --git a/GramAddict/core/diagnostic_dump.py b/GramAddict/core/diagnostic_dump.py
index 375786b..716b546 100644
--- a/GramAddict/core/diagnostic_dump.py
+++ b/GramAddict/core/diagnostic_dump.py
@@ -20,6 +20,7 @@ logger = logging.getLogger(__name__)
DUMP_DIR = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(__file__))), "debug", "xml_dumps")
MAX_DUMPS_PER_CATEGORY = 5
+
def dump_ui_state(device, reason: str, extra_context: dict = None):
"""
Capture and save the current UI hierarchy and screenshot to disk for debugging.
@@ -43,7 +44,8 @@ def dump_ui_state(device, reason: str, extra_context: dict = None):
# Capture and write screenshot
try:
import base64
- screenshot_b64 = device.screenshot_b64()
+
+ screenshot_b64 = device.get_screenshot_b64()
if screenshot_b64:
screenshot_data = base64.b64decode(screenshot_b64)
screenshot_path = filepath.replace(".xml", ".jpg")
@@ -57,11 +59,12 @@ def dump_ui_state(device, reason: str, extra_context: dict = None):
"reason": reason,
"timestamp": ts,
"xml_file": filename,
- "screenshot_file": filename.replace(".xml", ".jpg")
+ "screenshot_file": filename.replace(".xml", ".jpg"),
}
# Capture the session log if available
try:
import shutil
+
from GramAddict.core.log import get_log_file_config
log_name, log_dir, _, _ = get_log_file_config()
@@ -93,25 +96,51 @@ def dump_ui_state(device, reason: str, extra_context: dict = None):
logger.debug(f"[Diagnostic] Could not dump UI state: {e}")
return None
-def _rotate_dumps(category_prefix: str):
- """Keep only the last MAX_DUMPS_PER_CATEGORY dumps per category."""
- try:
- all_files = sorted([f for f in os.listdir(DUMP_DIR) if f.startswith(category_prefix) and f.endswith(".xml")])
- if len(all_files) > MAX_DUMPS_PER_CATEGORY:
- files_to_remove = all_files[: len(all_files) - MAX_DUMPS_PER_CATEGORY]
- for f in files_to_remove:
- xml_path = os.path.join(DUMP_DIR, f)
- meta_path = xml_path.replace(".xml", ".meta.json")
- log_path = xml_path.replace(".xml", ".log")
- img_path = xml_path.replace(".xml", ".jpg")
-
- os.remove(xml_path)
- if os.path.exists(meta_path):
- os.remove(meta_path)
- if os.path.exists(log_path):
- os.remove(log_path)
- if os.path.exists(img_path):
- os.remove(img_path)
- except Exception:
- pass
+def _rotate_dumps(category_prefix: str = None):
+ """Keep only the last MAX_DUMPS_PER_CATEGORY dumps per category. If no category, cleans all."""
+ try:
+ if not os.path.exists(DUMP_DIR):
+ return
+
+ # Get all unique timestamps/prefixes
+ all_files = os.listdir(DUMP_DIR)
+ prefixes = set()
+ for f in all_files:
+ # Format is usually reason__timestamp.ext
+ if "__" in f:
+ prefix = f.split(".")[0]
+ prefixes.add(prefix)
+
+ # Group prefixes by category
+ categories = {}
+ for p in prefixes:
+ parts = p.split("__")
+ if len(parts) >= 2:
+ cat = parts[0]
+ if cat not in categories:
+ categories[cat] = []
+ categories[cat].append(p)
+
+ for cat, prefs in categories.items():
+ if category_prefix and cat != category_prefix:
+ continue
+
+ prefs.sort() # chronological
+ if len(prefs) > MAX_DUMPS_PER_CATEGORY:
+ prefs_to_remove = prefs[: len(prefs) - MAX_DUMPS_PER_CATEGORY]
+ for p_rm in prefs_to_remove:
+ for ext in [".xml", ".jpg", ".log", ".meta.json"]:
+ fp = os.path.join(DUMP_DIR, p_rm + ext)
+ if os.path.exists(fp):
+ os.remove(fp)
+
+ # Also clean orphaned files that don't match any known prefix pattern
+ for f in all_files:
+ if "__" not in f:
+ fp = os.path.join(DUMP_DIR, f)
+ if os.path.isfile(fp):
+ os.remove(fp)
+
+ except Exception as e:
+ logger.debug(f"[Diagnostic] Error during dump rotation: {e}")
diff --git a/GramAddict/core/dm_engine.py b/GramAddict/core/dm_engine.py
index 085af53..c9124c5 100644
--- a/GramAddict/core/dm_engine.py
+++ b/GramAddict/core/dm_engine.py
@@ -44,15 +44,16 @@ def _run_zero_latency_dm_loop(device, zero_engine, nav_graph, configs, session_s
xml_dump = device.dump_hierarchy()
# --- Zero Trust Structural Guard ---
- # If the navigation engine failed or the UI shifted, we MUST NOT hallucinate
- # interactions on the wrong screen (e.g., Privacy Settings).
- is_thread = "direct_thread_header" in xml_dump or "row_thread_composer_edittext" in xml_dump
+ # -----------------------------------
+ # ZERO TRUST STRUCTURAL GUARD
+ # -----------------------------------
+ # Validate we are actually in the Inbox or a Thread.
+ # Hallucinations can lead to "Privacy Settings" or "Profile" screens.
is_inbox = (
- "action_bar_title" in xml_dump
- or "thread_list" in xml_dump
- or "direct_inbox_header" in xml_dump
- or "unread" in xml_dump.lower()
+ 'resource-id="com.instagram.android:id/inbox_refreshable_thread_list_recyclerview"' in xml_dump
+ or 'resource-id="com.instagram.android:id/direct_inbox_action_bar"' in xml_dump
)
+ is_thread = 'resource-id="com.instagram.android:id/direct_thread_header"' in xml_dump
if is_thread:
logger.warning("⚠️ [Structural Guard] DM Engine trapped in an open thread. Escaping...")
@@ -149,7 +150,10 @@ def _run_zero_latency_dm_loop(device, zero_engine, nav_graph, configs, session_s
# If keyboard was open, the first back only closed it. Check if still in thread.
check_xml = device.dump_hierarchy()
- if "direct_thread_header" in check_xml or "row_thread_composer_edittext" in check_xml:
+ if (
+ 'resource-id="com.instagram.android:id/direct_thread_header"' in check_xml
+ or 'resource-id="com.instagram.android:id/row_thread_composer_edittext"' in check_xml
+ ):
device.press("back")
sleep(1.0)
@@ -170,7 +174,10 @@ def _run_zero_latency_dm_loop(device, zero_engine, nav_graph, configs, session_s
sleep(1.0)
check_xml = device.dump_hierarchy()
- if "direct_thread_header" in check_xml or "row_thread_composer_edittext" in check_xml:
+ if (
+ 'resource-id="com.instagram.android:id/direct_thread_header"' in check_xml
+ or 'resource-id="com.instagram.android:id/row_thread_composer_edittext"' in check_xml
+ ):
device.press("back")
sleep(1.0)
diff --git a/GramAddict/core/perception/action_memory.py b/GramAddict/core/perception/action_memory.py
index 1a2c37a..e73883b 100644
--- a/GramAddict/core/perception/action_memory.py
+++ b/GramAddict/core/perception/action_memory.py
@@ -77,7 +77,9 @@ class ActionMemory:
self._last_click_context = None
- def verify_success(self, intent: str, pre_click_xml: str, post_click_xml: str, device=None, confidence: float = 0.0) -> Optional[bool]:
+ def verify_success(
+ self, intent: str, pre_click_xml: str, post_click_xml: str, device=None, confidence: float = 0.0
+ ) -> Optional[bool]:
"""
Structural and Visual verification: Did the UI actually change after the click?
"""
@@ -88,52 +90,72 @@ class ActionMemory:
if "explore_action_bar" in post_click_xml and "row_feed_button_like" not in post_click_xml:
return None # Still on grid, inconclusive
- # State toggles (like, save, follow) rarely change XML length predictably
state_toggles = ["like", "save", "follow", "heart"]
- if any(t in intent.lower() for t in state_toggles):
- # If we are highly confident (e.g. pulled from Qdrant memory), bypass heavy VLM
- if device and confidence < 0.95:
- logger.info(f"👁️ [ActionMemory] Confidence ({confidence:.2f}) < 0.95. Handing over verification for '{intent}' to VLM visual analysis...")
- from GramAddict.core.perception.semantic_evaluator import SemanticEvaluator
- evaluator = SemanticEvaluator()
-
- # Ask VLM to be the absolute source of truth
- prompt = (
- f"The user just attempted to perform the action: '{intent}'. "
- f"Look at the current screen carefully. Was the action successful? "
- f"If the intent was 'follow', does the button now indicate 'Following' or 'Requested'? "
- f"If it was 'like', is the heart icon clearly active/red? "
- f"If the screen shifted completely to a profile when you just wanted to like/follow from a feed, it FAILED. "
- f"Answer ONLY with the word YES or NO."
+ is_toggle = any(t in intent.lower() for t in state_toggles)
+
+ # If we are highly confident (e.g. pulled from Qdrant memory), bypass heavy VLM
+ if device and confidence < 0.95:
+ logger.info(
+ f"👁️ [ActionMemory] Confidence ({confidence:.2f}) < 0.95. Handing over verification for '{intent}' to VLM visual analysis..."
+ )
+ from GramAddict.core.perception.semantic_evaluator import SemanticEvaluator
+
+ evaluator = SemanticEvaluator()
+
+ # Ask VLM to be the absolute source of truth
+ prompt = (
+ f"The user just attempted to perform the action: '{intent}'. "
+ f"Look at the current screen carefully. Was the action successful? "
+ )
+ if is_toggle:
+ prompt += (
+ "If the intent was 'follow', does the button now indicate 'Following' or 'Requested'? "
+ "If it was 'like', is the heart icon clearly active/red? "
+ "If the screen shifted completely to a profile when you just wanted to like/follow from a feed, it FAILED. "
)
-
- try:
- screenshot = device.screenshot_b64()
- response = evaluator._query_vlm(prompt, screenshot)
-
- if response and "yes" in response.lower() and "no" not in response.lower():
- logger.debug(f"🧠 [ActionMemory] VLM visually confirmed success for '{intent}'.")
- return True
- else:
- logger.warning(f"⚠️ [ActionMemory] VLM visual verification FAILED for '{intent}'. VLM replied: '{response}'")
- return False
- except Exception as e:
- logger.error(f"Failed to query VLM for visual verification: {e}")
- # Fallthrough to structural delta if VLM crashes
-
- # Fallback to structural delta if no device or VLM fails
- diff = abs(len(pre_click_xml) - len(post_click_xml))
+ else:
+ prompt += (
+ f"Does the current screen match the expected outcome of '{intent}'? "
+ f"For example, if the intent was to open a post/photo, are you looking at a post view (not a user profile or story)? "
+ f"If the intent was to open a profile, are you on a profile page? "
+ f"If the intent was to go back, are you on the previous screen? "
+ )
+ prompt += "Answer ONLY with the word YES or NO."
+
+ try:
+ screenshot = device.get_screenshot_b64()
+ response = evaluator._query_vlm(prompt, screenshot)
+
+ if response and "yes" in response.lower() and "no" not in response.lower():
+ logger.debug(f"🧠 [ActionMemory] VLM visually confirmed success for '{intent}'.")
+ return True
+ else:
+ logger.warning(
+ f"⚠️ [ActionMemory] VLM visual verification FAILED for '{intent}'. VLM replied: '{response}'"
+ )
+ return False
+ except Exception as e:
+ logger.error(f"Failed to query VLM for visual verification: {e}")
+ # Fallthrough to structural delta if VLM crashes
+
+ # Fallback to structural delta if no device, VLM fails, or high confidence bypass
+ diff = abs(len(pre_click_xml) - len(post_click_xml))
+
+ if is_toggle:
if diff > 1000:
- logger.warning(f"⚠️ [ActionMemory] Massive structural shift ({diff} chars) for state-toggle '{intent}'. Navigated away by mistake? Verification FAIL.")
+ logger.warning(
+ f"⚠️ [ActionMemory] Massive structural shift ({diff} chars) for state-toggle '{intent}'. Navigated away by mistake? Verification FAIL."
+ )
return False
if diff > 0:
- logger.debug(f"🧠 [ActionMemory] Structural delta detected for '{intent}'. Verification PASS.")
+ logger.debug(f"🧠 [ActionMemory] Structural delta detected for toggle '{intent}'. Verification PASS.")
+ return True
+ else:
+ if diff > 50:
+ logger.debug(
+ f"🧠 [ActionMemory] Structural change detected for navigation '{intent}'. Verification PASS."
+ )
return True
-
-
- if abs(len(pre_click_xml) - len(post_click_xml)) > 50:
- logger.debug(f"🧠 [ActionMemory] Structural change detected for '{intent}'. Verification PASS.")
- return True
logger.warning(f"⚠️ [ActionMemory] No structural change detected for '{intent}'. Verification FAIL.")
return False
diff --git a/GramAddict/core/physics/sendevent_injector.py b/GramAddict/core/physics/sendevent_injector.py
index 7a5ce73..b2f7441 100644
--- a/GramAddict/core/physics/sendevent_injector.py
+++ b/GramAddict/core/physics/sendevent_injector.py
@@ -18,7 +18,6 @@ correct /dev/input/eventX and the axis ranges on first use.
import logging
import re
-from time import sleep
logger = logging.getLogger(__name__)
@@ -179,6 +178,9 @@ class SendEventInjector:
scale_x = self.x_max / display_w
scale_y = self.y_max / display_h
+ # Build batch command list
+ cmds = []
+
# --- Touch Down (first point) ---
x, y, pressure = points[0]
ix = int(x * scale_x)
@@ -186,8 +188,6 @@ class SendEventInjector:
ip = int(pressure * self.pressure_max)
itm = min(touch_major, self.touch_major_max)
- # Build batch command for touch-down
- cmds = []
cmds.append(f"sendevent {dev} {self.EV_ABS} {self.ABS_MT_TRACKING_ID} 0")
cmds.append(f"sendevent {dev} {self.EV_ABS} {self.ABS_MT_POSITION_X} {ix}")
cmds.append(f"sendevent {dev} {self.EV_ABS} {self.ABS_MT_POSITION_Y} {iy}")
@@ -196,38 +196,36 @@ class SendEventInjector:
cmds.append(f"sendevent {dev} {self.EV_KEY} {self.BTN_TOUCH} 1")
cmds.append(f"sendevent {dev} {self.EV_SYN} {self.SYN_REPORT} 0")
- # Execute touch-down
- self.device.shell(" && ".join(cmds))
-
# --- Move through intermediate points ---
for i in range(1, len(points) - 1):
if i - 1 < len(timing_intervals):
- sleep(timing_intervals[i - 1])
+ delay = timing_intervals[i - 1]
+ if delay > 0.001:
+ cmds.append(f"sleep {delay:.3f}")
x, y, pressure = points[i]
ix = int(x * scale_x)
iy = int(y * scale_y)
ip = int(pressure * self.pressure_max)
- cmds = []
cmds.append(f"sendevent {dev} {self.EV_ABS} {self.ABS_MT_POSITION_X} {ix}")
cmds.append(f"sendevent {dev} {self.EV_ABS} {self.ABS_MT_POSITION_Y} {iy}")
cmds.append(f"sendevent {dev} {self.EV_ABS} {self.ABS_MT_PRESSURE} {ip}")
cmds.append(f"sendevent {dev} {self.EV_SYN} {self.SYN_REPORT} 0")
- self.device.shell(" && ".join(cmds))
-
# --- Touch Up (last point) ---
if len(timing_intervals) >= len(points) - 1:
- sleep(timing_intervals[-1])
+ delay = timing_intervals[-1]
else:
- sleep(0.01)
+ delay = 0.01
+
+ if delay > 0.001:
+ cmds.append(f"sleep {delay:.3f}")
x, y, pressure = points[-1]
ix = int(x * scale_x)
iy = int(y * scale_y)
- cmds = []
cmds.append(f"sendevent {dev} {self.EV_ABS} {self.ABS_MT_POSITION_X} {ix}")
cmds.append(f"sendevent {dev} {self.EV_ABS} {self.ABS_MT_POSITION_Y} {iy}")
cmds.append(f"sendevent {dev} {self.EV_ABS} {self.ABS_MT_PRESSURE} 0")
@@ -235,6 +233,7 @@ class SendEventInjector:
cmds.append(f"sendevent {dev} {self.EV_KEY} {self.BTN_TOUCH} 0")
cmds.append(f"sendevent {dev} {self.EV_SYN} {self.SYN_REPORT} 0")
+ # Execute ALL events in one atomic batch to eliminate ADB latency
self.device.shell(" && ".join(cmds))
except Exception as e:
@@ -253,4 +252,12 @@ class SendEventInjector:
ex, ey, _ = points[-1]
total_ms = int(sum(timing_intervals) * 1000) if timing_intervals else 300
- self.device.shell(f"input swipe {int(sx)} {int(sy)} {int(ex)} {int(ey)} {total_ms}")
+ dist_x = abs(ex - sx)
+ dist_y = abs(ey - sy)
+
+ # Android sometimes interprets a low-duration swipe with minimal movement as a long press or cancels it.
+ # If it's physically a tap (minimal movement, short duration), use native input tap.
+ if dist_x < 15 and dist_y < 15 and total_ms < 150:
+ self.device.shell(f"input tap {int(sx)} {int(sy)}")
+ else:
+ self.device.shell(f"input swipe {int(sx)} {int(sy)} {int(ex)} {int(ey)} {total_ms}")
diff --git a/GramAddict/core/physics/timing.py b/GramAddict/core/physics/timing.py
index 5cd8607..0915750 100644
--- a/GramAddict/core/physics/timing.py
+++ b/GramAddict/core/physics/timing.py
@@ -144,7 +144,9 @@ def align_active_post(device):
from GramAddict.core.telepathic_engine import TelepathicEngine
telepath = TelepathicEngine.get_instance()
- target_node = telepath.find_best_node(xml, "post author header profile", min_confidence=0.4, device=device)
+ target_node = telepath.find_best_node(
+ xml, "post author header profile", min_confidence=0.4, device=device, track=False
+ )
if target_node:
original_attribs = target_node.get("original_attribs", {})
bounds = original_attribs.get("bounds")
diff --git a/GramAddict/core/telepathic_engine.py b/GramAddict/core/telepathic_engine.py
index 16498d3..558a1a1 100644
--- a/GramAddict/core/telepathic_engine.py
+++ b/GramAddict/core/telepathic_engine.py
@@ -53,7 +53,9 @@ class TelepathicEngine:
# Core Resolution Engine
# ──────────────────────────────────────────────
- def find_best_node(self, xml_string: str, intent_description: str, device=None, **kwargs) -> Optional[dict]:
+ def find_best_node(
+ self, xml_string: str, intent_description: str, device=None, track: bool = True, **kwargs
+ ) -> Optional[dict]:
print("FIND_BEST_NODE CALLED")
"""
@@ -109,7 +111,8 @@ class TelepathicEngine:
return {"skip": True, "semantic": "already_followed"}
# 4. Track action
- self._memory.track_click(intent_description, best_node, xml_string)
+ if track:
+ self._memory.track_click(intent_description, best_node, xml_string)
# Translate to old GramAddict dict format for backward compatibility
return self._translate_node(best_node)
@@ -150,7 +153,8 @@ class TelepathicEngine:
if skip_positions is None:
skip_positions = set()
- if "first image in explore grid" in intent_description.lower():
+ intent_lower = intent_description.lower()
+ if "first image in explore grid" in intent_lower:
grid_items = [
n
for n in nodes
@@ -165,6 +169,51 @@ class TelepathicEngine:
# Sort by y (row) then by x (col)
grid_items.sort(key=lambda n: (n.get("y", 9999), n.get("x", 9999)))
return grid_items[0]
+
+ # --- DM Engine Structural Fast Paths ---
+ if "find the message input text field" in intent_lower:
+ for n in nodes:
+ if "row_thread_composer_edittext" in n.get("resource_id", ""):
+ return n
+
+ if "find the send message button" in intent_lower:
+ for n in nodes:
+ if "row_thread_composer_button_send" in n.get("resource_id", ""):
+ return n
+
+ if "find unread message threads" in intent_lower:
+ # Unread threads usually have a specific indicator or are the top rows in the recyclerview
+ unread_candidates = []
+ for n in nodes:
+ # Check if it has a direct structural unread marker (though IG usually relies on styling)
+ # But we can fallback to the row_inbox_container
+ if (
+ "row_inbox_container" in n.get("resource_id", "")
+ and (n.get("x", -1), n.get("y", -1)) not in skip_positions
+ ):
+ # Avoid the 'Search' or header elements, typically thread rows start below y=200
+ if n.get("y", 0) > 200:
+ unread_candidates.append(n)
+
+ if unread_candidates:
+ # Return the highest y (first thread in list)
+ unread_candidates.sort(key=lambda n: n.get("y", 9999))
+ return unread_candidates[0]
+
+ if "find the last received message text" in intent_lower:
+ msg_candidates = []
+ for n in nodes:
+ # The actual message text bubble
+ if "direct_text_message_text_view" in n.get("resource_id", "") or "message_content" in n.get(
+ "resource_id", ""
+ ):
+ msg_candidates.append(n)
+
+ if msg_candidates:
+ # Sort by y descending (bottom-most message is the last one)
+ msg_candidates.sort(key=lambda n: n.get("y", 0), reverse=True)
+ return msg_candidates[0]
+
return None
# ──────────────────────────────────────────────
diff --git a/tests/unit/test_diagnostic_dump_cleanup.py b/tests/unit/test_diagnostic_dump_cleanup.py
deleted file mode 100644
index eef4c59..0000000
--- a/tests/unit/test_diagnostic_dump_cleanup.py
+++ /dev/null
@@ -1,29 +0,0 @@
-import os
-import shutil
-import base64
-from unittest.mock import MagicMock, patch
-from GramAddict.core.behaviors.simulator import BehaviorSimulator
-
-def test_device_facade_session_trace_cleanup(tmp_path):
- facade = BehaviorSimulator()
- facade.get_screenshot_b64 = MagicMock(return_value=base64.b64encode(b"fake_jpg_data").decode("utf-8"))
-
- with patch("GramAddict.core.device_facade.os.path.join", side_effect=os.path.join), \
- patch("GramAddict.core.device_facade.os.makedirs"):
-
- def fake_join(*args):
- if args[0] == "debug" and args[1] == "session_traces":
- return str(tmp_path / "session_traces")
- return os.path.join(*args)
-
- with patch("GramAddict.core.device_facade.os.path.join", side_effect=fake_join):
- traces_root = tmp_path / "session_traces"
- traces_root.mkdir(parents=True, exist_ok=True)
-
- for i in range(6):
- d = traces_root / f"old_session_{i}"
- d.mkdir()
-
- facade.dump_hierarchy()
- remaining_folders = [f for f in os.listdir(traces_root) if os.path.isdir(os.path.join(traces_root, f))]
- assert len(remaining_folders) <= 5
diff --git a/tests/unit/test_dm_navigation_guards.py b/tests/unit/test_dm_navigation_guards.py
index 6edc3d5..c192b23 100644
--- a/tests/unit/test_dm_navigation_guards.py
+++ b/tests/unit/test_dm_navigation_guards.py
@@ -24,7 +24,9 @@ def test_dm_engine_uses_correct_dm_memory_logging(monkeypatch):
Note: I am writing this to PROVE the fix is necessary.
"""
mock_device = MagicMock()
- mock_device.dump_hierarchy.return_value = ""
+ mock_device.dump_hierarchy.return_value = (
+ 'resource-id="com.instagram.android:id/inbox_refreshable_thread_list_recyclerview"'
+ )
mock_telepathic = MagicMock()
# Mock unread thread found
@@ -98,12 +100,12 @@ def test_dm_navigation_double_back_guard():
mock_device = MagicMock()
# Mocking hierarchy sequence
mock_device.dump_hierarchy.side_effect = [
- "Inbox", # Loop 1 start
- "Thread", # Context read
- "Thread", # Send button find
- "", # Navigation check AFTER back
- "Inbox", # Loop 2 start (exit)
- "Inbox", # Buffer
+ 'resource-id="com.instagram.android:id/inbox_refreshable_thread_list_recyclerview"', # Loop 1 start
+ 'resource-id="com.instagram.android:id/direct_thread_header"', # Context read
+ 'resource-id="com.instagram.android:id/direct_thread_header"', # Send button find
+ 'resource-id="com.instagram.android:id/direct_thread_header"', # Navigation check AFTER back
+ 'resource-id="com.instagram.android:id/inbox_refreshable_thread_list_recyclerview"', # Loop 2 start (exit)
+ 'resource-id="com.instagram.android:id/inbox_refreshable_thread_list_recyclerview"', # Buffer
]
mock_telepathic = MagicMock()