diff --git a/GramAddict/core/perception/intent_resolver.py b/GramAddict/core/perception/intent_resolver.py
index 7a18e58..960f196 100644
--- a/GramAddict/core/perception/intent_resolver.py
+++ b/GramAddict/core/perception/intent_resolver.py
@@ -40,9 +40,7 @@ class IntentResolver:
# Structural Guards
# ──────────────────────────────────────────────
- def filter_navigation_conflicts(
- self, candidates: List[SpatialNode], intent_description: str
- ) -> List[SpatialNode]:
+ def filter_navigation_conflicts(self, candidates: List[SpatialNode], intent_description: str) -> List[SpatialNode]:
"""
Prevents VLM from confusing navigation-bar buttons (Back, Close)
with bottom tab-bar buttons (Home, Profile, Search).
@@ -57,25 +55,34 @@ class IntentResolver:
"""
intent_lower = intent_description.lower()
- # Only apply for tab-related intents
- is_tab_intent = "tab" in intent_lower and "back" not in intent_lower
- if not is_tab_intent:
- return candidates
-
+ # Only apply for tab-related intents OR generic interaction intents
+ # We want to aggressively protect against clicking 'Create' / 'Camera' / 'Back' by accident
filtered = []
+ is_tab_intent = "tab" in intent_lower and "back" not in intent_lower
+ is_create_intent = "create" in intent_lower or "camera" in intent_lower or "story" in intent_lower
+
for node in candidates:
rid = (node.resource_id or "").lower()
desc = (node.content_desc or "").lower()
is_back = "back" in rid or desc == "back"
is_close = "close" in rid or desc == "close"
+ is_create = "camera" in rid or "create" in rid or desc == "camera" or desc == "create" or "creation" in rid
- if is_back or is_close:
+ if is_tab_intent and (is_back or is_close):
logger.debug(
f"🛡️ [Nav Conflict Guard] Excluded '{node.resource_id}' "
f"(desc='{node.content_desc}') for tab intent '{intent_description}'"
)
continue
+
+ if not is_create_intent and is_create:
+ logger.debug(
+ f"🛡️ [Creation Conflict Guard] Excluded '{node.resource_id}' "
+ f"(desc='{node.content_desc}') for intent '{intent_description}'"
+ )
+ continue
+
filtered.append(node)
return filtered
diff --git a/scripts/pre_commit_tests.sh b/scripts/pre_commit_tests.sh
index c910513..6a08122 100755
--- a/scripts/pre_commit_tests.sh
+++ b/scripts/pre_commit_tests.sh
@@ -26,10 +26,26 @@ else
elif [ -f "$core_test_file" ]; then
TEST_TARGETS="$TEST_TARGETS $core_test_file"
else
- # If no direct unit test, fallback to running all unit tests to be safe
- echo "⚠️ No direct unit test found for $file, falling back to all unit tests."
- TEST_TARGETS="tests/unit"
- break
+ # Try to find matching e2e tests by searching for each word in the module name
+ module_name="${filename%.py}"
+ e2e_matches=""
+ for word in $(echo "$module_name" | tr '_' '\n'); do
+ if [ ${#word} -ge 4 ]; then # Only search meaningful words (4+ chars)
+ found=$(find tests/e2e -name "test_*${word}*.py" 2>/dev/null | head -3)
+ if [ -n "$found" ]; then
+ e2e_matches="$e2e_matches $found"
+ fi
+ fi
+ done
+ e2e_matches=$(echo "$e2e_matches" | xargs -n1 2>/dev/null | sort -u | head -3 | xargs 2>/dev/null)
+ if [ -n "$e2e_matches" ]; then
+ echo "⚠️ No direct unit test for $file, using matching E2E tests: $e2e_matches"
+ TEST_TARGETS="$TEST_TARGETS $e2e_matches"
+ else
+ echo "⚠️ No direct unit test found for $file, falling back to all unit tests."
+ TEST_TARGETS="tests/unit"
+ break
+ fi
fi
fi
done
diff --git a/tests/e2e/conftest.py b/tests/e2e/conftest.py
index ad0ce8d..72d5ea7 100644
--- a/tests/e2e/conftest.py
+++ b/tests/e2e/conftest.py
@@ -202,6 +202,7 @@ def make_real_device_with_xml(monkeypatch):
def screenshot(self):
from PIL import Image
+
return Image.new("RGB", (1, 1), color="black")
def app_current(self):
@@ -426,6 +427,23 @@ def e2e_configs():
ai_condenser_url="http://localhost",
dry_run_comments=False,
visual_vibe_check_percentage=0,
+ current_likes_limit=10,
+ current_comments_limit=10,
+ current_follows_limit=10,
+ current_follow_limit=10,
+ current_unfollow_limit=10,
+ current_pm_limit=10,
+ current_scraped_limit=10,
+ current_watch_limit=10,
+ current_success_limit=10,
+ current_total_limit=10,
+ current_crashes_limit=10,
+ max_follows=10,
+ max_pm=10,
+ max_watch=10,
+ max_success=10,
+ max_total=10,
+ max_crashes=10,
)
from GramAddict.core.config import Config
@@ -547,7 +565,9 @@ class E2EDeviceStub:
return {"package": "com.instagram.android"}
def screenshot(self_):
- return None
+ from PIL import Image
+
+ return Image.new("RGB", (1, 1), color="black")
def shell(self_, cmd):
if isinstance(cmd, str) and cmd.startswith("input tap"):
@@ -564,10 +584,26 @@ class E2EDeviceStub:
def __init__(self_, parent):
self_._parent = parent
- def down(self_, x, y):
- self_._parent.clicks.append((x, y))
+ def down(self_, x=None, y=None, **kwargs):
+ if "obj" in kwargs:
+ obj = kwargs["obj"]
+ if isinstance(obj, dict) and "bounds" in obj:
+ import re
- def up(self_, x, y):
+ b = obj["bounds"]
+ if isinstance(b, str):
+ nums = [int(n) for n in re.findall(r"\d+", b)]
+ x, y = (nums[0] + nums[2]) // 2, (nums[1] + nums[3]) // 2
+ else:
+ x, y = (int(b[0]) + int(b[2])) // 2, (int(b[1]) + int(b[3])) // 2
+ else:
+ x, y = (
+ int(getattr(obj, "x", getattr(obj, "x1", 0))),
+ int(getattr(obj, "y", getattr(obj, "y1", 0))),
+ )
+ self_._parent.clicks.append((int(x) if x is not None else 0, int(y) if y is not None else 0))
+
+ def up(self_, x=None, y=None, **kwargs):
pass
self.deviceV2.touch = _Touch(self)
@@ -582,8 +618,21 @@ class E2EDeviceStub:
def press(self, key):
self.pressed_keys.append(key)
- def click(self, x, y):
- self.clicks.append((x, y))
+ def click(self, x=None, y=None, **kwargs):
+ if "obj" in kwargs:
+ obj = kwargs["obj"]
+ if isinstance(obj, dict) and "bounds" in obj:
+ import re
+
+ b = obj["bounds"]
+ if isinstance(b, str):
+ nums = [int(n) for n in re.findall(r"\d+", b)]
+ x, y = (nums[0] + nums[2]) // 2, (nums[1] + nums[3]) // 2
+ else:
+ x, y = (int(b[0]) + int(b[2])) // 2, (int(b[1]) + int(b[3])) // 2
+ else:
+ x, y = int(getattr(obj, "x", getattr(obj, "x1", 0))), int(getattr(obj, "y", getattr(obj, "y1", 0)))
+ self.clicks.append((int(x) if x is not None else 0, int(y) if y is not None else 0))
def swipe(self, sx, sy, ex, ey, **kwargs):
self.swipes.append({"start": (sx, sy), "end": (ex, ey)})
@@ -597,6 +646,12 @@ class E2EDeviceStub:
def unlock(self):
pass
+ def shell(self, cmd):
+ pass
+
+ def cm_to_pixels(self, cm):
+ return int(cm * 40)
+
def get_screenshot_b64(self):
return "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mNkYAAAAAYAAjCB0C8AAAAASUVORK5CYII="
@@ -612,8 +667,10 @@ class E2EDeviceStub:
@pytest.fixture
def e2e_device():
"""Factory to create an E2EDeviceStub from an XML sequence."""
+
def _create(xml_sequence):
return E2EDeviceStub(xml_sequence)
+
return _create
@@ -622,21 +679,121 @@ def mock_llm_network_calls(monkeypatch, request):
"""Mocks ONLY the LLM network requests. The SAE structural logic remains 100% REAL."""
if request.config.getoption("--live"):
return
-
- from GramAddict.core import llm_provider
+
import json
-
+
+ from GramAddict.core import llm_provider
+
def fake_query_telepathic_llm(*args, **kwargs):
- prompt = kwargs.get('user_prompt', args[3] if len(args) > 3 else "")
+ prompt = kwargs.get("user_prompt", args[3] if len(args) > 3 else "")
+ if not prompt and len(args) > 0:
+ prompt = args[0]
+
+ print(f"MOCK LLM PROMPT:\n{prompt}\n---------------------")
+
if "OBSTACLE_LOCKED_SCREEN" in prompt and "FOREIGN_APP" in prompt:
return json.dumps({"situation": "OBSTACLE_FOREIGN_APP"})
if "MODAL, DIALOG, or POPUP" in prompt:
+ if "How are yo...Instagram?" in prompt or "Please lea... a rating!" in prompt or "Not Now" in prompt:
+ return json.dumps({"situation": "OBSTACLE_MODAL"})
return json.dumps({"situation": "NORMAL"})
+ import re
+
+ intent_match = re.search(r"intent: '([^']+)'", prompt)
+ intent = intent_match.group(1).lower() if intent_match else ""
+
+ if "selected_index" in prompt and "Find the exact box number" not in prompt:
+ if "profile tab" in intent:
+ match = re.search(r"\[(\d+)\][^\[]*profile_tab", prompt)
+ if match:
+ return json.dumps({"selected_index": int(match.group(1))})
+ if "following" in intent:
+ match = re.search(r"\[(\d+)\][^\[]*following", prompt.lower())
+ if match:
+ return json.dumps({"selected_index": int(match.group(1))})
+ if "first post" in intent:
+ # Find the first grid_card_layout_container or image_button
+ match = re.search(r"\[(\d+)\][^\[]*(?:grid_card_layout_container|image_button)", prompt)
+ if match:
+ return json.dumps({"selected_index": int(match.group(1))})
+ if "comment" in intent:
+ match = re.search(r"\[(\d+)\][^\[]*comment", prompt.lower())
+ if match:
+ return json.dumps({"selected_index": int(match.group(1))})
+ if "like" in intent:
+ match = re.search(r"\[(\d+)\][^\[]*like_button", prompt.lower())
+ if match:
+ return json.dumps({"selected_index": int(match.group(1))})
+ match = re.search(r"\[(\d+)\][^\[]*like", prompt.lower())
+ if match:
+ return json.dumps({"selected_index": int(match.group(1))})
+ if "message" in intent:
+ match = re.search(r"\[(\d+)\][^\[]*message", prompt.lower())
+ if match:
+ return json.dumps({"selected_index": int(match.group(1))})
+ if "search input" in intent or "search" in intent:
+ match = re.search(r"\[(\d+)\][^\[]*search", prompt.lower())
+ if match:
+ return json.dumps({"selected_index": int(match.group(1))})
+ if "story ring avatar" in intent:
+ match = re.search(r"\[(\d+)\][^\[]*story ring avatar", prompt.lower())
+ if match:
+ return json.dumps({"selected_index": int(match.group(1))})
+ if "follow" in intent:
+ match = re.search(r"\[(\d+)\][^\[]*follow", prompt.lower())
+ if match:
+ return json.dumps({"selected_index": int(match.group(1))})
+ return json.dumps({"selected_index": None})
+ return json.dumps({"selected_index": 0})
+
+ if "Find the exact box number" in prompt:
+ if "profile tab" in intent:
+ match = re.search(r"\[(\d+)\][^\n]*?desc='profile'", prompt.lower())
+ if match:
+ return json.dumps({"box": int(match.group(1))})
+ if "following" in intent:
+ match = re.search(r"\[(\d+)\][^\n]*?following", prompt.lower())
+ if match:
+ return json.dumps({"box": int(match.group(1))})
+ if "first post" in intent:
+ match = re.search(r"\[(\d+)\][^\n]*?(?:grid_card_layout_container|image_button)", prompt)
+ if match:
+ return json.dumps({"box": int(match.group(1))})
+ if "comment" in intent:
+ match = re.search(r"\[(\d+)\][^\n]*?comment", prompt.lower())
+ if match:
+ return json.dumps({"box": int(match.group(1))})
+ if "like" in intent:
+ match = re.search(r"\[(\d+)\][^\n]*?desc='like'", prompt.lower())
+ if match:
+ return json.dumps({"box": int(match.group(1))})
+ match = re.search(r"\[(\d+)\][^\n]*?like", prompt.lower())
+ if match:
+ return json.dumps({"box": int(match.group(1))})
+ if "message" in intent or "direct" in intent:
+ match = re.search(r"\[(\d+)\][^\n]*?message", prompt.lower())
+ if match:
+ return json.dumps({"box": int(match.group(1))})
+ if "search" in intent:
+ # We need to pick a search bar, not "Suchen" on keyboard
+ match = re.search(r"\[(\d+)\][^\n]*?search", prompt.lower())
+ if match:
+ return json.dumps({"box": int(match.group(1))})
+ if "story ring avatar" in intent:
+ match = re.search(r"\[(\d+)\][^\n]*?story ring avatar", prompt.lower())
+ if match:
+ return json.dumps({"box": int(match.group(1))})
+ if "follow" in intent:
+ match = re.search(r"\[(\d+)\][^\n]*?follow", prompt.lower())
+ if match:
+ return json.dumps({"box": int(match.group(1))})
+ return json.dumps({"box": None})
+ return json.dumps({"box": None})
return json.dumps({"situation": "NORMAL"})
def fake_query_llm(*args, **kwargs):
return json.dumps({"action": "false_positive", "reason": "Test mock", "x": 0, "y": 0})
-
+
monkeypatch.setattr(llm_provider, "query_telepathic_llm", fake_query_telepathic_llm)
monkeypatch.setattr(llm_provider, "query_llm", fake_query_llm)
@@ -644,25 +801,26 @@ def mock_llm_network_calls(monkeypatch, request):
@pytest.fixture
def e2e_cognitive_stack_factory(e2e_configs):
"""Build the REAL cognitive stack exactly like bot_flow.py does."""
+
def _create(device, username="testuser"):
+ from GramAddict.core.active_inference import ActiveInferenceEngine
+ from GramAddict.core.darwin_engine import DarwinEngine
from GramAddict.core.dopamine_engine import DopamineEngine
+ from GramAddict.core.growth_brain import GrowthBrain
from GramAddict.core.interaction import LLMWriter
+ from GramAddict.core.q_nav_graph import QNavGraph
from GramAddict.core.qdrant_memory import DMMemoryDB, ParasocialCRMDB
from GramAddict.core.resonance_engine import ResonanceEngine
- from GramAddict.core.q_nav_graph import QNavGraph
- from GramAddict.core.telepathic_engine import TelepathicEngine
- from GramAddict.core.zero_latency_engine import ZeroLatencyEngine
- from GramAddict.core.darwin_engine import DarwinEngine
- from GramAddict.core.active_inference import ActiveInferenceEngine
from GramAddict.core.sensors.honeypot_radome import HoneypotRadome
from GramAddict.core.swarm_protocol import SwarmProtocol
- from GramAddict.core.growth_brain import GrowthBrain
+ from GramAddict.core.telepathic_engine import TelepathicEngine
+ from GramAddict.core.zero_latency_engine import ZeroLatencyEngine
dopamine = DopamineEngine()
dopamine.session_limit_seconds = 0.5
dopamine.session_start = time.time()
-
- info = device.get_info() if hasattr(device, 'get_info') else {"displayWidth": 1080, "displayHeight": 2400}
+
+ info = device.get_info() if hasattr(device, "get_info") else {"displayWidth": 1080, "displayHeight": 2400}
return {
"dopamine": dopamine,
@@ -679,6 +837,7 @@ def e2e_cognitive_stack_factory(e2e_configs):
"dm_memory": DMMemoryDB(),
"writer": LLMWriter(username, [], e2e_configs),
}
+
return _create
@@ -699,7 +858,7 @@ def e2e_workflow_ctx(e2e_configs, e2e_cognitive_stack_factory, setup_e2e_plugin_
def _run(device, context_xml=None):
xml = context_xml if context_xml else device.dump_hierarchy()
session = SessionState(e2e_configs)
-
+
cognitive_stack = e2e_cognitive_stack_factory(device)
ctx = BehaviorContext(
diff --git a/tests/e2e/test_production_bug_regression.py b/tests/e2e/test_production_bug_regression.py
index dd98e09..59f1553 100644
--- a/tests/e2e/test_production_bug_regression.py
+++ b/tests/e2e/test_production_bug_regression.py
@@ -292,7 +292,6 @@ class TestResonancePersonaInterestsEmpty:
The VLM prompt says "You are a user with the following interests: ." → blind eval.
"""
-
def test_persona_interests_are_not_empty_when_target_audience_set(self):
"""
When config has mission.target_audience set, the ResonanceEvaluator
@@ -402,7 +401,7 @@ class TestFollowBlockedOnReelsFeed:
class TestBug7FollowButtonGuard:
- def test_follow_button_blocked(self):
+ def test_follow_button_blocked(self, monkeypatch):
"""
When the intent is 'post media content', TelepathicEngine.find_best_node
must reject nodes that have 'follow' in their semantic string.
@@ -429,8 +428,8 @@ class TestBug7FollowButtonGuard:
def resolve(self, intent, candidates, device=None):
return candidates[0] if candidates else None
- tele._parser = DummyParser()
- tele._resolver = DummyResolver()
+ monkeypatch.setattr(tele, "_parser", DummyParser())
+ monkeypatch.setattr(tele, "_resolver", DummyResolver())
result = tele.find_best_node("", "post media content", track=False)
assert result is None, "TelepathicEngine should block 'Follow' button for 'post media content' intent!"
@@ -442,7 +441,7 @@ class TestBug7FollowButtonGuard:
class TestBug8PerfectSnappingBoundsExclusion:
- def test_exclude_bounds_filters_candidates(self):
+ def test_exclude_bounds_filters_candidates(self, monkeypatch):
"""
TelepathicEngine.find_best_node must filter out candidates whose bounds
match those in the `exclude_bounds` list.
@@ -487,8 +486,8 @@ class TestBug8PerfectSnappingBoundsExclusion:
# Just return the first available candidate to see which survived
return candidates[0] if candidates else None
- tele._parser = DummyParser()
- tele._resolver = DummyResolver()
+ monkeypatch.setattr(tele, "_parser", DummyParser())
+ monkeypatch.setattr(tele, "_resolver", DummyResolver())
# Without exclusion, Candidate 1 should be picked
result1 = tele.find_best_node("", "test intent", track=False)
diff --git a/tests/e2e/test_workflow_explore_feed.py b/tests/e2e/test_workflow_explore_feed.py
index 757597a..6fd548c 100644
--- a/tests/e2e/test_workflow_explore_feed.py
+++ b/tests/e2e/test_workflow_explore_feed.py
@@ -11,8 +11,6 @@ import os
from tests.e2e.conftest import E2EDeviceStub
-from GramAddict.core.situational_awareness import SituationType
-
FIXTURES_DIR = os.path.join(os.path.dirname(os.path.dirname(__file__)), "fixtures")
@@ -30,10 +28,10 @@ def test_explore_grid_processes_without_crashes(monkeypatch, e2e_workflow_ctx):
results, ctx = e2e_workflow_ctx(device)
executed_count = sum(1 for r in results if r.executed)
- assert executed_count >= 1, (
- f"Only {executed_count} plugin(s) executed on Explore grid."
- )
+ assert executed_count >= 1, f"Only {executed_count} plugin(s) executed on Explore grid."
+ # 🚨 BEHAVIORAL TRUTH ASSERTION 🚨
+ assert (
+ len(device.clicks) > 0 or len(device.swipes) > 0
+ ), "LIE DETECTED: The pipeline claimed success but did not tap any post on the explore grid!"
- assert "back" not in device.pressed_keys, (
- "obstacle_guard false positive on Explore feed!"
- )
+ assert "back" not in device.pressed_keys, "obstacle_guard false positive on Explore feed!"
diff --git a/tests/e2e/test_workflow_normal_feed.py b/tests/e2e/test_workflow_normal_feed.py
index ed67dae..ed17f2b 100644
--- a/tests/e2e/test_workflow_normal_feed.py
+++ b/tests/e2e/test_workflow_normal_feed.py
@@ -11,9 +11,6 @@ Real: Cognitive stack, PluginRegistry, SessionState, Config.
from tests.e2e.conftest import E2EDeviceStub
-from GramAddict.core.situational_awareness import SituationType
-
-
NORMAL_POST_XML = """
0 or len(device.swipes) > 0, (
+ "LIE DETECTED: The pipeline claimed success (executed=True) but did "
+ "not perform a single click or swipe on the device!"
+ )
+
+ # Check if the interaction was correctly targeted
+ # On a normal feed post, it should eventually like, comment, or swipe.
+ # The Like button is at bounds="[30,1360][120,1450]", center ~ (75, 1405)
+ liked = any(70 <= c[0] <= 80 and 1400 <= c[1] <= 1410 for c in device.clicks)
+ swiped = len(device.swipes) > 0
+
+ assert liked or swiped, (
+ f"LIE DETECTED: The bot interacted, but NOT with the like button or by swiping! "
+ f"Clicks: {device.clicks}, Swipes: {device.swipes}"
+ )
diff --git a/tests/e2e/test_workflow_profiles.py b/tests/e2e/test_workflow_profiles.py
index 370beec..050e742 100644
--- a/tests/e2e/test_workflow_profiles.py
+++ b/tests/e2e/test_workflow_profiles.py
@@ -14,8 +14,6 @@ import os
from tests.e2e.conftest import E2EDeviceStub
-from GramAddict.core.situational_awareness import SituationType
-
FIXTURES_DIR = os.path.join(os.path.dirname(os.path.dirname(__file__)), "fixtures")
@@ -33,13 +31,13 @@ def test_user_profile_processes_without_crashes(monkeypatch, e2e_workflow_ctx):
results, ctx = e2e_workflow_ctx(device)
executed_count = sum(1 for r in results if r.executed)
- assert executed_count >= 1, (
- f"Only {executed_count} plugin(s) executed on user profile."
- )
+ assert executed_count >= 1, f"Only {executed_count} plugin(s) executed on user profile."
+ # 🚨 BEHAVIORAL TRUTH ASSERTION 🚨
+ assert (
+ len(device.clicks) > 0 or len(device.swipes) > 0
+ ), "LIE DETECTED: The pipeline claimed success but did not interact with the Profile!"
- assert "back" not in device.pressed_keys, (
- "obstacle_guard false positive on user profile!"
- )
+ assert "back" not in device.pressed_keys, "obstacle_guard false positive on user profile!"
def test_scraping_profile_processes_without_crashes(monkeypatch, e2e_workflow_ctx):
@@ -51,9 +49,7 @@ def test_scraping_profile_processes_without_crashes(monkeypatch, e2e_workflow_ct
results, ctx = e2e_workflow_ctx(device)
executed_count = sum(1 for r in results if r.executed)
- assert executed_count >= 1, (
- f"Only {executed_count} plugin(s) executed on scraping profile."
- )
+ assert executed_count >= 1, f"Only {executed_count} plugin(s) executed on scraping profile."
def test_followers_list_processes_without_crashes(monkeypatch, e2e_workflow_ctx):
@@ -65,9 +61,7 @@ def test_followers_list_processes_without_crashes(monkeypatch, e2e_workflow_ctx)
results, ctx = e2e_workflow_ctx(device)
executed_count = sum(1 for r in results if r.executed)
- assert executed_count >= 1, (
- f"Only {executed_count} plugin(s) executed on followers list."
- )
+ assert executed_count >= 1, f"Only {executed_count} plugin(s) executed on followers list."
def test_unfollow_list_processes_without_crashes(monkeypatch, e2e_workflow_ctx):
@@ -79,6 +73,4 @@ def test_unfollow_list_processes_without_crashes(monkeypatch, e2e_workflow_ctx):
results, ctx = e2e_workflow_ctx(device)
executed_count = sum(1 for r in results if r.executed)
- assert executed_count >= 1, (
- f"Only {executed_count} plugin(s) executed on unfollow list."
- )
+ assert executed_count >= 1, f"Only {executed_count} plugin(s) executed on unfollow list."
diff --git a/tests/e2e/test_workflow_reels_feed.py b/tests/e2e/test_workflow_reels_feed.py
index a731fa0..73368ae 100644
--- a/tests/e2e/test_workflow_reels_feed.py
+++ b/tests/e2e/test_workflow_reels_feed.py
@@ -11,8 +11,6 @@ import os
from tests.e2e.conftest import E2EDeviceStub
-from GramAddict.core.situational_awareness import SituationType
-
FIXTURES_DIR = os.path.join(os.path.dirname(os.path.dirname(__file__)), "fixtures")
@@ -32,10 +30,11 @@ def test_reels_post_processes_without_crashes(monkeypatch, e2e_workflow_ctx):
executed_count = sum(1 for r in results if r.executed)
assert executed_count >= 1, (
- f"Only {executed_count} plugin(s) executed on a Reels post. "
- "The pipeline is crashing on Reels XML."
+ f"Only {executed_count} plugin(s) executed on a Reels post. " "The pipeline is crashing on Reels XML."
)
+ # 🚨 BEHAVIORAL TRUTH ASSERTION 🚨
+ assert (
+ len(device.clicks) > 0 or len(device.swipes) > 0
+ ), "LIE DETECTED: The pipeline claimed success but did not interact with the Reel!"
- assert "back" not in device.pressed_keys, (
- "obstacle_guard false positive on a Reels feed post!"
- )
+ assert "back" not in device.pressed_keys, "obstacle_guard false positive on a Reels feed post!"
diff --git a/tests/e2e/test_workflow_stories_feed.py b/tests/e2e/test_workflow_stories_feed.py
index 43df6d9..865b978 100644
--- a/tests/e2e/test_workflow_stories_feed.py
+++ b/tests/e2e/test_workflow_stories_feed.py
@@ -11,8 +11,6 @@ import os
from tests.e2e.conftest import E2EDeviceStub
-from GramAddict.core.situational_awareness import SituationType
-
FIXTURES_DIR = os.path.join(os.path.dirname(os.path.dirname(__file__)), "fixtures")
@@ -30,13 +28,13 @@ def test_stories_feed_processes_without_crashes(monkeypatch, e2e_workflow_ctx):
results, ctx = e2e_workflow_ctx(device)
executed_count = sum(1 for r in results if r.executed)
- assert executed_count >= 1, (
- f"Only {executed_count} plugin(s) executed on Stories feed."
- )
+ assert executed_count >= 1, f"Only {executed_count} plugin(s) executed on Stories feed."
+ # 🚨 BEHAVIORAL TRUTH ASSERTION 🚨
+ assert (
+ len(device.clicks) > 0 or len(device.swipes) > 0
+ ), "LIE DETECTED: The pipeline claimed success but did not interact with the Story!"
- assert "back" not in device.pressed_keys, (
- "obstacle_guard false positive on Stories feed!"
- )
+ assert "back" not in device.pressed_keys, "obstacle_guard false positive on Stories feed!"
def test_story_view_full_processes_without_crashes(monkeypatch, e2e_workflow_ctx):
@@ -48,6 +46,4 @@ def test_story_view_full_processes_without_crashes(monkeypatch, e2e_workflow_ctx
results, ctx = e2e_workflow_ctx(device)
executed_count = sum(1 for r in results if r.executed)
- assert executed_count >= 1, (
- f"Only {executed_count} plugin(s) executed on Story view."
- )
+ assert executed_count >= 1, f"Only {executed_count} plugin(s) executed on Story view."
diff --git a/tests/unit/test_device_connection.py b/tests/unit/test_device_connection.py
index 306b6b8..161ecca 100644
--- a/tests/unit/test_device_connection.py
+++ b/tests/unit/test_device_connection.py
@@ -16,7 +16,7 @@ def test_create_device_connection_failure(monkeypatch, caplog):
import subprocess
from collections import namedtuple
- from unittest.mock import MagicMock
+ from types import SimpleNamespace
import uiautomator2 as u2
@@ -26,7 +26,9 @@ def test_create_device_connection_failure(monkeypatch, caplog):
CompletedProcess = namedtuple("CompletedProcess", ["stdout", "stderr", "returncode"])
# Case 2: Proactive discovery with NO devices
monkeypatch.setattr(
- subprocess, "run", lambda *args, **kwargs: MagicMock(stdout="List of devices attached\n\n", return_code=0)
+ subprocess,
+ "run",
+ lambda *args, **kwargs: SimpleNamespace(stdout="List of devices attached\n\n", returncode=0),
)
with pytest.raises(SystemExit):
create_device("192.168.1.100:5555", "com.instagram.android", None)
@@ -35,7 +37,9 @@ def test_create_device_connection_failure(monkeypatch, caplog):
monkeypatch.setattr(
subprocess,
"run",
- lambda *args, **kwargs: MagicMock(stdout="List of devices attached\n10.0.0.5:5555\tdevice\n", return_code=0),
+ lambda *args, **kwargs: SimpleNamespace(
+ stdout="List of devices attached\n10.0.0.5:5555\tdevice\n", returncode=0
+ ),
)
with pytest.raises(SystemExit):
create_device("192.168.1.100:5555", "com.instagram.android", None)
diff --git a/tests/unit/test_is_ad_substring.py b/tests/unit/test_is_ad_substring.py
index d10c976..4eedf1e 100644
--- a/tests/unit/test_is_ad_substring.py
+++ b/tests/unit/test_is_ad_substring.py
@@ -5,6 +5,7 @@ def test_is_ad_false_positive_abroad():
# Simulate an IG node with 'abroad' in the text
xml_false_positive = """
+
"""
@@ -14,6 +15,7 @@ def test_is_ad_false_positive_abroad():
def test_is_ad_true_positive():
xml_true_positive = """
+
"""
@@ -23,6 +25,7 @@ def test_is_ad_true_positive():
def test_is_ad_true_positive_ad_word():
xml_ad = """
+
"""