purge: eliminate last MagicMock and all lambda dump_hierarchy overrides
- device_emulator.py: Replace MagicMock(watcher) with explicit WatcherStub that implements exact u2 watcher chain (when/click/start). Any API drift now causes AttributeError instead of silent acceptance. - test_system_sae.py: Inject XML into emulator state instead of bypassing dump_hierarchy with lambda - test_behavior_ad_guard.py: Pass XML to fixture, remove lambda override - test_behavior_scrape_profile.py: Pass XML to fixture, remove lambda override Result: ZERO unittest.mock imports, ZERO MagicMock instances, ZERO dump_hierarchy lambdas across entire E2E suite (50 files).
This commit is contained in:
@@ -255,15 +255,37 @@ def create_emulator_facade(initial_state, states, transitions, monkeypatch, imag
|
||||
emulator = InstagramEmulator(initial_state, states, transitions, images=images)
|
||||
|
||||
# We must patch u2.connect to avoid ConnectError during DeviceFacade init
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
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
|
||||
|
||||
def when(self, xpath=None, **kwargs):
|
||||
return self
|
||||
|
||||
def click(self):
|
||||
return self
|
||||
|
||||
def start(self):
|
||||
pass
|
||||
|
||||
class MockU2Device:
|
||||
def __init__(self, *args, **kwargs):
|
||||
self.info = emulator._info
|
||||
self.settings = {}
|
||||
self.watcher = MagicMock()
|
||||
self.watcher = WatcherStub()
|
||||
|
||||
def dump_hierarchy(self, *args, **kwargs):
|
||||
return emulator.dump_hierarchy()
|
||||
|
||||
@@ -22,8 +22,7 @@ def test_ad_guard_detects_sponsored_post(make_real_device_with_image):
|
||||
with open(xml_path, "r", encoding="utf-8") as f:
|
||||
xml = f.read()
|
||||
|
||||
device = make_real_device_with_image(jpg_path)
|
||||
device.dump_hierarchy = lambda: xml
|
||||
device = make_real_device_with_image(jpg_path, xml)
|
||||
|
||||
import types
|
||||
|
||||
|
||||
@@ -23,9 +23,7 @@ def test_scrape_profile_extracts_data_correctly(make_real_device_with_image):
|
||||
with open(xml_path, "r", encoding="utf-8") as f:
|
||||
xml = f.read()
|
||||
|
||||
device = make_real_device_with_image(jpg_path)
|
||||
# mock dump_hierarchy so the plugin uses the static XML
|
||||
device.dump_hierarchy = lambda: xml
|
||||
device = make_real_device_with_image(jpg_path, xml)
|
||||
|
||||
# Create dummy config and session state
|
||||
import types
|
||||
|
||||
@@ -9,37 +9,23 @@ without relying on brittle mock LLM responses.
|
||||
import pytest
|
||||
|
||||
from GramAddict.core.situational_awareness import SituationalAwarenessEngine, SituationType
|
||||
|
||||
|
||||
class DummyDeviceInfo:
|
||||
def __init__(self, screen_on=True):
|
||||
self.info = {"screenOn": screen_on}
|
||||
|
||||
|
||||
class DummyDeviceV2:
|
||||
def __init__(self, screen_on=True):
|
||||
self.info = {"screenOn": screen_on}
|
||||
|
||||
|
||||
class DummyDevice:
|
||||
def __init__(self, screen_on=True):
|
||||
self.app_id = "com.instagram.android"
|
||||
self.deviceV2 = DummyDeviceV2(screen_on)
|
||||
from tests.e2e.device_emulator import create_emulator_facade
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def sae():
|
||||
def sae(monkeypatch):
|
||||
SituationalAwarenessEngine.reset()
|
||||
device = DummyDevice()
|
||||
device, emulator = create_emulator_facade("any", {"any": ""}, {}, monkeypatch)
|
||||
engine = SituationalAwarenessEngine.get_instance(device)
|
||||
# Clear the global ScreenMemoryDB so tests don't pollute each other
|
||||
from GramAddict.core.qdrant_memory import ScreenMemoryDB
|
||||
|
||||
db = ScreenMemoryDB()
|
||||
if db.is_connected:
|
||||
try:
|
||||
db.client.delete(
|
||||
collection_name=db.collection_name,
|
||||
points_selector={"filter": {}} # Delete all points
|
||||
collection_name=db.collection_name,
|
||||
points_selector={"filter": {}}, # Delete all points
|
||||
)
|
||||
except Exception:
|
||||
pass
|
||||
@@ -54,30 +40,30 @@ class TestSAEPerception:
|
||||
assert sae.perceive("") == SituationType.OBSTACLE_FOREIGN_APP
|
||||
assert sae.perceive(None) == SituationType.OBSTACLE_FOREIGN_APP
|
||||
|
||||
def test_perceive_locked_screen(self, sae):
|
||||
def test_perceive_locked_screen(self, sae, monkeypatch):
|
||||
"""If the hardware reports the screen is off, it must be LOCKED_SCREEN."""
|
||||
sae.device.deviceV2.info["screenOn"] = False
|
||||
sae.device.deviceV2.info = {"screenOn": False}
|
||||
xml_dump = '<node package="com.android.systemui" />'
|
||||
assert sae.perceive(xml_dump) == SituationType.OBSTACLE_LOCKED_SCREEN
|
||||
|
||||
def test_perceive_action_blocked(self, sae):
|
||||
"""Critical account safety check: 'Action Blocked' must return DANGER_ACTION_BLOCKED."""
|
||||
xml_dump = '''
|
||||
xml_dump = """
|
||||
<node package="com.instagram.android">
|
||||
<node resource-id="android:id/dialog" />
|
||||
<node text="Action Blocked" />
|
||||
<node text="Try again later to protect our community." />
|
||||
</node>
|
||||
'''
|
||||
"""
|
||||
assert sae.perceive(xml_dump) == SituationType.DANGER_ACTION_BLOCKED
|
||||
|
||||
def test_perceive_system_dialog(self, sae):
|
||||
"""System permission dialogs must be identified structurally."""
|
||||
xml_dump = '''
|
||||
xml_dump = """
|
||||
<node package="com.google.android.permissioncontroller">
|
||||
<node text="Allow Instagram to record audio?" />
|
||||
</node>
|
||||
'''
|
||||
"""
|
||||
assert sae.perceive(xml_dump) == SituationType.OBSTACLE_SYSTEM
|
||||
|
||||
def test_perceive_creation_flow_modal(self, sae):
|
||||
@@ -85,20 +71,20 @@ class TestSAEPerception:
|
||||
Creation flows (camera, story gallery) are inside Instagram but block navigation.
|
||||
They must be caught by structural markers.
|
||||
"""
|
||||
xml_dump = '''
|
||||
xml_dump = """
|
||||
<node package="com.instagram.android">
|
||||
<node resource-id="com.instagram.android:id/quick_capture_button" />
|
||||
</node>
|
||||
'''
|
||||
"""
|
||||
assert sae.perceive(xml_dump) == SituationType.OBSTACLE_MODAL
|
||||
|
||||
def test_perceive_story_gallery_cancel_modal(self, sae):
|
||||
"""Story gallery cancel button blocks navigation."""
|
||||
xml_dump = '''
|
||||
xml_dump = """
|
||||
<node package="com.instagram.android">
|
||||
<node resource-id="com.instagram.android:id/gallery_cancel_button" />
|
||||
</node>
|
||||
'''
|
||||
"""
|
||||
assert sae.perceive(xml_dump) == SituationType.OBSTACLE_MODAL
|
||||
|
||||
|
||||
@@ -106,7 +92,7 @@ class TestSAECompression:
|
||||
"""Tests the XML compression logic to ensure signatures are compact but retain identity."""
|
||||
|
||||
def test_compress_strips_noise_keeps_important_data(self, sae):
|
||||
xml_dump = '''
|
||||
xml_dump = """
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<hierarchy rotation="0">
|
||||
<node index="0" text="" resource-id="" class="android.widget.FrameLayout" package="com.instagram.android" content-desc="" checkable="false" checked="false" clickable="false" enabled="true" focusable="false" focused="false" scrollable="false" long-clickable="false" password="false" selected="false" bounds="[0,0][1080,2400]">
|
||||
@@ -114,9 +100,9 @@ class TestSAECompression:
|
||||
<node content-desc="Profile Picture" clickable="true" />
|
||||
</node>
|
||||
</hierarchy>
|
||||
'''
|
||||
"""
|
||||
compressed = sae._compress_xml(xml_dump)
|
||||
|
||||
|
||||
# Must extract packages
|
||||
assert "PACKAGES: com.instagram.android" in compressed
|
||||
# Must extract resource-ids without the package prefix
|
||||
@@ -141,22 +127,23 @@ class TestSAELoop:
|
||||
|
||||
def test_ensure_clear_screen_returns_true_immediately_if_normal(self, sae):
|
||||
"""If the screen is NORMAL, the SAE should return True without taking actions."""
|
||||
xml_dump = '''
|
||||
xml_dump = """
|
||||
<node package="com.instagram.android">
|
||||
<node resource-id="com.instagram.android:id/row_feed_button_like" />
|
||||
</node>
|
||||
'''
|
||||
# Bypass dump_hierarchy
|
||||
sae.device.dump_hierarchy = lambda: xml_dump
|
||||
"""
|
||||
# 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
|
||||
assert sae.ensure_clear_screen(max_attempts=3) is True
|
||||
assert sae._consecutive_failures == 0
|
||||
|
||||
def test_is_instagram_foreground(self, sae):
|
||||
xml_dump = '''
|
||||
xml_dump = """
|
||||
<node package="com.instagram.android">
|
||||
<node resource-id="com.instagram.android:id/row_feed_button_like" />
|
||||
</node>
|
||||
'''
|
||||
"""
|
||||
assert sae.is_instagram_foreground(xml_dump=xml_dump) is True
|
||||
|
||||
foreign_xml = '<node package="com.apple.ios" />'
|
||||
|
||||
Reference in New Issue
Block a user