chore(test): Ruthless deletion of ALL remaining MagicMocks and patches across the entire test suite
This commit is contained in:
@@ -1,225 +0,0 @@
|
||||
"""
|
||||
TDD: Deep Active Inference Integration Tests.
|
||||
|
||||
Tests the v2 Active Inference Engine behaviors:
|
||||
- Consecutive error → policy escalation
|
||||
- Interaction probability throttling
|
||||
- Session abort recommendation
|
||||
- Diagnostics reporting
|
||||
- Backward compatibility with existing callers
|
||||
"""
|
||||
|
||||
import time
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def ai():
|
||||
"""Fresh Active Inference engine for each test."""
|
||||
from GramAddict.core.active_inference import ActiveInferenceEngine
|
||||
|
||||
return ActiveInferenceEngine("test_user")
|
||||
|
||||
|
||||
class TestPolicyEscalation:
|
||||
"""Consecutive prediction errors must escalate the policy."""
|
||||
|
||||
def test_single_error_stays_stable(self, ai):
|
||||
"""One prediction error should not change policy from STABLE."""
|
||||
ai.predict_state(["feed_tab"])
|
||||
ai.evaluate_prediction("<hierarchy><node text='wrong'/></hierarchy>")
|
||||
# Free energy from single error: 0.0 * 0.7 + 1.0 * 0.3 = 0.3
|
||||
# 0.3 < 0.75, so still STABLE
|
||||
assert ai.policy == "STABLE"
|
||||
assert ai._consecutive_prediction_errors == 1
|
||||
|
||||
def test_three_errors_goes_cautious(self, ai):
|
||||
"""3 consecutive errors must trigger CAUTIOUS policy."""
|
||||
for _ in range(3):
|
||||
ai.predict_state(["nonexistent"])
|
||||
ai.evaluate_prediction("<hierarchy><node text='wrong'/></hierarchy>")
|
||||
|
||||
assert ai.policy == "CAUTIOUS"
|
||||
assert ai._consecutive_prediction_errors == 3
|
||||
|
||||
def test_five_errors_goes_dormant(self, ai):
|
||||
"""5 consecutive errors must trigger DORMANT policy."""
|
||||
for _ in range(5):
|
||||
ai.predict_state(["nonexistent"])
|
||||
ai.evaluate_prediction("<hierarchy><node text='wrong'/></hierarchy>")
|
||||
|
||||
assert ai.policy == "DORMANT"
|
||||
assert ai._consecutive_prediction_errors == 5
|
||||
|
||||
def test_successful_prediction_resets_counter(self, ai):
|
||||
"""A successful prediction must reset the consecutive error counter."""
|
||||
# Build up 3 errors
|
||||
for _ in range(3):
|
||||
ai.predict_state(["missing"])
|
||||
ai.evaluate_prediction("<hierarchy><node text='wrong'/></hierarchy>")
|
||||
|
||||
assert ai._consecutive_prediction_errors == 3
|
||||
|
||||
# Now succeed
|
||||
ai.predict_state(["feed_tab"])
|
||||
ai.evaluate_prediction('<hierarchy><node resource-id="feed_tab"/></hierarchy>')
|
||||
|
||||
assert ai._consecutive_prediction_errors == 0
|
||||
|
||||
def test_error_rate_tracking(self, ai):
|
||||
"""Error rate must be accurately tracked across the session."""
|
||||
# 3 errors, 2 successes = 3/5 = 0.6
|
||||
for _ in range(3):
|
||||
ai.predict_state(["missing"])
|
||||
ai.evaluate_prediction("<hierarchy/>")
|
||||
for _ in range(2):
|
||||
ai.predict_state(["found"])
|
||||
ai.evaluate_prediction('<hierarchy><node text="found"/></hierarchy>')
|
||||
|
||||
assert ai.get_error_rate() == pytest.approx(0.6)
|
||||
|
||||
|
||||
class TestInteractionProbability:
|
||||
"""Interaction probability must decrease under stress."""
|
||||
|
||||
def test_stable_has_full_probability(self, ai):
|
||||
"""STABLE policy → 100% interaction probability."""
|
||||
ai.policy = "STABLE"
|
||||
assert ai.get_interaction_probability() == 1.0
|
||||
|
||||
def test_cautious_halves_probability(self, ai):
|
||||
"""CAUTIOUS policy → 50% interaction probability."""
|
||||
ai.policy = "CAUTIOUS"
|
||||
assert ai.get_interaction_probability() == 0.5
|
||||
|
||||
def test_dormant_minimal_probability(self, ai):
|
||||
"""DORMANT policy → 10% interaction probability."""
|
||||
ai.policy = "DORMANT"
|
||||
assert ai.get_interaction_probability() == 0.1
|
||||
|
||||
def test_probability_bounds(self, ai):
|
||||
"""Interaction probability must always be in [0.0, 1.0]."""
|
||||
for policy in ["STABLE", "CAUTIOUS", "DORMANT"]:
|
||||
ai.policy = policy
|
||||
prob = ai.get_interaction_probability()
|
||||
assert 0.0 <= prob <= 1.0
|
||||
|
||||
|
||||
class TestSessionAbort:
|
||||
"""Session abort recommendation under extreme instability."""
|
||||
|
||||
def test_no_abort_on_stable(self, ai):
|
||||
"""STABLE engine should never recommend abort."""
|
||||
assert ai.should_abort_session() is False
|
||||
|
||||
def test_abort_after_five_consecutive_errors(self, ai):
|
||||
"""5 consecutive prediction errors must recommend abort."""
|
||||
for _ in range(5):
|
||||
ai.predict_state(["missing"])
|
||||
ai.evaluate_prediction("<hierarchy/>")
|
||||
|
||||
assert ai.should_abort_session() is True
|
||||
|
||||
def test_abort_on_extreme_free_energy(self, ai):
|
||||
"""Free energy > 2.0 must recommend abort."""
|
||||
ai.free_energy = 2.1
|
||||
assert ai.should_abort_session() is True
|
||||
|
||||
def test_no_abort_under_threshold(self, ai):
|
||||
"""Free energy < 2.0 with few errors should not abort."""
|
||||
ai.free_energy = 1.9
|
||||
ai._consecutive_prediction_errors = 4
|
||||
assert ai.should_abort_session() is False
|
||||
|
||||
|
||||
class TestDiagnostics:
|
||||
"""Diagnostics must provide accurate runtime snapshot."""
|
||||
|
||||
def test_diagnostics_has_required_fields(self, ai):
|
||||
"""Diagnostics dict must contain all required fields."""
|
||||
diag = ai.get_diagnostics()
|
||||
required = [
|
||||
"free_energy",
|
||||
"policy",
|
||||
"consecutive_errors",
|
||||
"total_predictions",
|
||||
"total_errors",
|
||||
"error_rate",
|
||||
"session_uptime_minutes",
|
||||
"should_abort",
|
||||
]
|
||||
for field in required:
|
||||
assert field in diag, f"Missing diagnostic field: {field}"
|
||||
|
||||
def test_diagnostics_reflects_state(self, ai):
|
||||
"""Diagnostics must accurately reflect engine state."""
|
||||
ai.predict_state(["test"])
|
||||
ai.evaluate_prediction("<wrong/>")
|
||||
|
||||
diag = ai.get_diagnostics()
|
||||
assert diag["consecutive_errors"] == 1
|
||||
assert diag["total_predictions"] == 1
|
||||
assert diag["total_errors"] == 1
|
||||
assert diag["error_rate"] == 1.0
|
||||
assert diag["should_abort"] is False
|
||||
|
||||
|
||||
class TestBackwardCompatibility:
|
||||
"""Existing callers must work unchanged."""
|
||||
|
||||
def test_get_sleep_modifier_unchanged(self, ai):
|
||||
"""Sleep modifier values must match v1 behavior."""
|
||||
ai.policy = "STABLE"
|
||||
assert ai.get_sleep_modifier() == 1.0
|
||||
ai.policy = "CAUTIOUS"
|
||||
assert ai.get_sleep_modifier() == 2.0
|
||||
ai.policy = "DORMANT"
|
||||
assert ai.get_sleep_modifier() == 5.0
|
||||
|
||||
def test_predict_then_evaluate_success(self, ai):
|
||||
"""Basic predict → evaluate flow must work as before."""
|
||||
ai.predict_state(["row_feed", "button_like"])
|
||||
result = ai.evaluate_prediction(
|
||||
'<hierarchy><node resource-id="row_feed"/><node resource-id="button_like"/></hierarchy>'
|
||||
)
|
||||
assert result is True
|
||||
|
||||
def test_predict_then_evaluate_failure(self, ai):
|
||||
"""Failed prediction must still return False and fire Dojo."""
|
||||
ai.predict_state(["row_feed", "button_like"])
|
||||
|
||||
with patch("GramAddict.core.dojo_engine.DojoEngine.get_instance") as mock_dojo:
|
||||
mock_dojo.return_value.submit_snapshot = lambda **kw: None
|
||||
result = ai.evaluate_prediction('<hierarchy><node text="camera"/></hierarchy>')
|
||||
|
||||
assert result is False
|
||||
|
||||
def test_evaluate_without_prediction_is_noop(self, ai):
|
||||
"""Evaluating without a prior prediction must return True (no-op)."""
|
||||
result = ai.evaluate_prediction("<hierarchy/>")
|
||||
assert result is True
|
||||
assert ai._consecutive_prediction_errors == 0
|
||||
|
||||
|
||||
class TestFreeEnergyDecay:
|
||||
"""Free energy must decay over time (thermodynamic relaxation)."""
|
||||
|
||||
def test_free_energy_decays_over_time(self, ai):
|
||||
"""Free energy should reduce after time passes without new errors."""
|
||||
ai.free_energy = 1.5
|
||||
ai.last_update = time.time() - 7200 # 2 hours ago
|
||||
|
||||
ai.calculate_surprise(1.0, 1.0) # Perfect prediction
|
||||
|
||||
# Decay: 1.5 * 0.7 + 0.0 * 0.3 = 1.05, then * exp(-0.1 * 2) ≈ 1.05 * 0.818 ≈ 0.86
|
||||
assert ai.free_energy < 1.0
|
||||
|
||||
def test_free_energy_stabilizes_on_perfect_predictions(self, ai):
|
||||
"""Repeated perfect predictions should drive free energy toward zero."""
|
||||
ai.free_energy = 1.0
|
||||
for _ in range(20):
|
||||
ai.calculate_surprise(1.0, 1.0)
|
||||
|
||||
assert ai.free_energy < 0.05 # Near zero
|
||||
@@ -1,94 +0,0 @@
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from GramAddict.core.bot_flow import _wait_for_post_loaded
|
||||
|
||||
|
||||
def time_incrementer():
|
||||
times = [0, 1, 2, 3, 4, 10, 11, 12, 13, 14, 15]
|
||||
for t in times:
|
||||
yield t
|
||||
while True:
|
||||
yield 20
|
||||
|
||||
|
||||
def test_wait_for_post_loaded_success():
|
||||
"""Test that it returns True if feed markers are found."""
|
||||
mock_device = MagicMock()
|
||||
mock_device.dump_hierarchy.return_value = '<node resource-id="com.instagram.android:id/row_feed_photo_imageview" />'
|
||||
|
||||
result = _wait_for_post_loaded(mock_device, timeout=1)
|
||||
assert result is True
|
||||
|
||||
|
||||
@patch("GramAddict.core.physics.timing.sleep")
|
||||
@patch("GramAddict.core.physics.timing.dump_ui_state")
|
||||
def test_wait_for_post_loaded_adaptive_snap_story(mock_dump, mock_sleep):
|
||||
"""Test that being trapped in a story triggers a back press."""
|
||||
mock_device = MagicMock()
|
||||
# Simulate a timeout by making time.time() advance
|
||||
with patch("time.time", side_effect=time_incrementer()):
|
||||
mock_device.dump_hierarchy.return_value = '<node resource-id="com.instagram.android:id/reel_viewer_root" />'
|
||||
|
||||
result = _wait_for_post_loaded(mock_device, timeout=5)
|
||||
|
||||
# It should have timed out, dumped state, and pressed back
|
||||
assert mock_dump.called
|
||||
mock_device.press.assert_called_with("back")
|
||||
# Still returns False if feed markers are not found after recovery
|
||||
assert result is False
|
||||
|
||||
|
||||
@patch("GramAddict.core.physics.timing.sleep")
|
||||
@patch("GramAddict.core.physics.timing.dump_ui_state")
|
||||
def test_wait_for_post_loaded_adaptive_snap_profile(mock_dump, mock_sleep):
|
||||
"""Test that being trapped in a profile triggers a back press."""
|
||||
mock_device = MagicMock()
|
||||
with patch("time.time", side_effect=time_incrementer()):
|
||||
mock_device.dump_hierarchy.return_value = '<node resource-id="com.instagram.android:id/profile_header" />'
|
||||
|
||||
result = _wait_for_post_loaded(mock_device, timeout=5)
|
||||
|
||||
mock_device.press.assert_called_with("back")
|
||||
assert result is False
|
||||
|
||||
|
||||
@patch("GramAddict.core.physics.timing.sleep")
|
||||
@patch("GramAddict.core.physics.timing.dump_ui_state")
|
||||
def test_wait_for_post_loaded_adaptive_snap_wobble(mock_dump, mock_sleep):
|
||||
"""Test that being stuck between posts triggers a wobble if no nav_graph is provided."""
|
||||
mock_device = MagicMock()
|
||||
mock_device.get_info.return_value = {"displayWidth": 1080, "displayHeight": 2400}
|
||||
with patch("time.time", side_effect=time_incrementer()):
|
||||
# No recognized markers
|
||||
mock_device.dump_hierarchy.return_value = '<node resource-id="com.instagram.android:id/action_bar_root" />'
|
||||
|
||||
result = _wait_for_post_loaded(mock_device, timeout=5)
|
||||
|
||||
# Should swipe (wobble) twice
|
||||
assert mock_device.swipe.call_count == 2
|
||||
# Check that duration is explicitly specified and is less than 1.0 to prevent 100-second stalls
|
||||
for call in mock_device.swipe.call_args_list:
|
||||
args, kwargs = call
|
||||
duration = args[4] if len(args) > 4 else kwargs.get("duration", 0.5)
|
||||
assert duration <= 1.0, f"Swipe duration is too long: {duration} seconds!"
|
||||
assert result is False
|
||||
|
||||
|
||||
@patch("GramAddict.core.physics.timing.sleep")
|
||||
@patch("GramAddict.core.physics.timing.dump_ui_state")
|
||||
def test_wait_for_post_loaded_adaptive_snap_align(mock_dump, mock_sleep):
|
||||
"""Test that being stuck between posts triggers nav_graph.do('align') if nav_graph is provided."""
|
||||
mock_device = MagicMock()
|
||||
mock_nav_graph = MagicMock()
|
||||
with patch("time.time", side_effect=time_incrementer()):
|
||||
mock_device.dump_hierarchy.return_value = '<node resource-id="com.instagram.android:id/action_bar_root" />'
|
||||
|
||||
result = _wait_for_post_loaded(mock_device, timeout=5, nav_graph=mock_nav_graph)
|
||||
|
||||
# Now it should unconditionally micro-wobble (swipe twice)
|
||||
assert mock_device.swipe.call_count == 2
|
||||
for call in mock_device.swipe.call_args_list:
|
||||
args, kwargs = call
|
||||
duration = args[4] if len(args) > 4 else kwargs.get("duration", 0.5)
|
||||
assert duration <= 1.0, f"Swipe duration is too long: {duration} seconds!"
|
||||
assert result is False
|
||||
@@ -1,62 +0,0 @@
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from GramAddict.core.goap import GoalPlanner, NavigationKnowledge, ScreenType
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_db():
|
||||
with patch("GramAddict.core.goap.QdrantBase") as MockBase:
|
||||
mock_instance = MagicMock()
|
||||
mock_instance.is_connected = True
|
||||
mock_instance._get_embedding.return_value = [0.1] * 768
|
||||
|
||||
# Simulate an empty scroll result initially
|
||||
mock_instance.client.scroll.return_value = ([], None)
|
||||
|
||||
MockBase.return_value = mock_instance
|
||||
yield mock_instance
|
||||
|
||||
|
||||
def test_learn_trap_persists_and_filters_actions(mock_db):
|
||||
"""
|
||||
TDD Test: Verify that aversive learning (Traps) prevents the agent
|
||||
from planning navigation through a burned action.
|
||||
"""
|
||||
knowledge = NavigationKnowledge("test_user")
|
||||
|
||||
# Simulate a blank start where the agent sees these actions
|
||||
available_actions = ["tap home tab", "tap profile tab", "tap external ad"]
|
||||
screen_type = ScreenType.EXPLORE_GRID
|
||||
|
||||
# 1. Initially, no actions are traps
|
||||
for action in available_actions:
|
||||
assert not knowledge.is_trap(screen_type, action), f"Action {action} should not be a trap yet."
|
||||
|
||||
# 2. Agent clicks the ad, gets sent to a foreign app, and learns it's a trap
|
||||
trap_action = "tap external ad"
|
||||
knowledge.learn_trap(screen_type, trap_action, trap_reason="foreign_app_triggered")
|
||||
|
||||
# Verify DB was called to persist
|
||||
mock_db.upsert_point.assert_called()
|
||||
|
||||
# 3. Verify it's now recognized as a trap
|
||||
assert knowledge.is_trap(screen_type, trap_action)
|
||||
assert not knowledge.is_trap(screen_type, "tap profile tab")
|
||||
|
||||
# 4. Verify GoalPlanner filters it during Blank Start
|
||||
planner = GoalPlanner(username="test_user")
|
||||
planner.knowledge = knowledge
|
||||
|
||||
planner.knowledge.get_requirements = MagicMock(return_value=[])
|
||||
planner.knowledge.get_screen_for_action = MagicMock(return_value=None)
|
||||
|
||||
# Since there are no known mappings, it will guess from available via linguistic match.
|
||||
# We must ensure 'tap external ad' is filtered out.
|
||||
selected_action = planner._plan_navigation(
|
||||
goal="open profile", screen_type=screen_type, available=available_actions
|
||||
)
|
||||
|
||||
# The guesser should select 'tap profile tab' because it linguistically matches 'profile'
|
||||
assert selected_action == "tap profile tab"
|
||||
@@ -1,371 +0,0 @@
|
||||
"""
|
||||
TDD: Behavior Plugins Tests.
|
||||
|
||||
Tests all concrete behavior plugins:
|
||||
- ProfileGuardPlugin (safety gates)
|
||||
- StoryViewPlugin (story watching)
|
||||
- FollowPlugin (follow interaction)
|
||||
- GridLikePlugin (grid liking)
|
||||
- Physics timing module (wait/align)
|
||||
"""
|
||||
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from GramAddict.core.behaviors import BehaviorContext, PluginRegistry
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def device():
|
||||
dev = MagicMock()
|
||||
dev.get_info.return_value = {"displayWidth": 1080, "displayHeight": 2400}
|
||||
dev.dump_hierarchy.return_value = '<hierarchy><node resource-id="row_feed_photo_profile_name"/></hierarchy>'
|
||||
dev.shell = MagicMock()
|
||||
dev.cm_to_pixels.return_value = 5
|
||||
return dev
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def configs():
|
||||
c = MagicMock()
|
||||
c.args = MagicMock()
|
||||
c.args.carousel_percentage = "0"
|
||||
c.args.stories_percentage = "0"
|
||||
c.args.follow_percentage = "0"
|
||||
c.args.likes_percentage = "0"
|
||||
c.args.ignore_close_friends = False
|
||||
c.args.visual_vibe_check_percentage = "0"
|
||||
c.args.scrape_profiles = False
|
||||
return c
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def session_state():
|
||||
ss = MagicMock()
|
||||
ss.my_username = "testbot"
|
||||
ss.totalFollowed = {}
|
||||
ss.totalLikes = 0
|
||||
ss.check_limit.return_value = False
|
||||
return ss
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def ctx(device, configs, session_state):
|
||||
mock_nav = MagicMock()
|
||||
mock_nav.current_state = "ProfileView"
|
||||
return BehaviorContext(
|
||||
device=device,
|
||||
configs=configs,
|
||||
session_state=session_state,
|
||||
cognitive_stack={"nav_graph": mock_nav},
|
||||
context_xml='<hierarchy><node resource-id="com.instagram.android:id/profile_header" /><node resource-id="row_feed_photo_profile_name"/></hierarchy>',
|
||||
sleep_mod=1.0,
|
||||
username="target_user",
|
||||
)
|
||||
|
||||
|
||||
# ── Profile Guard Tests ──
|
||||
|
||||
|
||||
class TestProfileGuardPlugin:
|
||||
def test_blocks_self_profile(self, ctx):
|
||||
from GramAddict.core.behaviors.profile_guard import ProfileGuardPlugin
|
||||
|
||||
ctx.username = "testbot"
|
||||
plugin = ProfileGuardPlugin()
|
||||
result = plugin.execute(ctx)
|
||||
assert result.executed is True
|
||||
assert result.should_skip is True
|
||||
assert result.metadata["reason"] == "self_profile"
|
||||
|
||||
def test_blocks_private_account(self, ctx):
|
||||
from GramAddict.core.behaviors.profile_guard import ProfileGuardPlugin
|
||||
|
||||
ctx.context_xml = "<hierarchy>This account is private</hierarchy>"
|
||||
plugin = ProfileGuardPlugin()
|
||||
result = plugin.execute(ctx)
|
||||
assert result.executed is True
|
||||
assert result.should_skip is True
|
||||
assert result.metadata["reason"] == "private"
|
||||
|
||||
def test_blocks_private_account_german(self, ctx):
|
||||
from GramAddict.core.behaviors.profile_guard import ProfileGuardPlugin
|
||||
|
||||
ctx.context_xml = "<hierarchy>Dieses Konto ist privat</hierarchy>"
|
||||
plugin = ProfileGuardPlugin()
|
||||
result = plugin.execute(ctx)
|
||||
assert result.executed is True
|
||||
assert result.metadata["reason"] == "private"
|
||||
|
||||
def test_blocks_empty_account(self, ctx):
|
||||
from GramAddict.core.behaviors.profile_guard import ProfileGuardPlugin
|
||||
|
||||
ctx.context_xml = "<hierarchy>No Posts Yet</hierarchy>"
|
||||
plugin = ProfileGuardPlugin()
|
||||
result = plugin.execute(ctx)
|
||||
assert result.should_skip is True
|
||||
assert result.metadata["reason"] == "empty"
|
||||
|
||||
def test_blocks_close_friend(self, ctx):
|
||||
from GramAddict.core.behaviors.profile_guard import ProfileGuardPlugin
|
||||
|
||||
ctx.configs.args.ignore_close_friends = True
|
||||
ctx.context_xml = "<hierarchy>Close Friend badge visible</hierarchy>"
|
||||
plugin = ProfileGuardPlugin()
|
||||
result = plugin.execute(ctx)
|
||||
assert result.should_skip is True
|
||||
assert result.metadata["reason"] == "close_friend"
|
||||
|
||||
def test_passes_valid_profile(self, ctx):
|
||||
from GramAddict.core.behaviors.profile_guard import ProfileGuardPlugin
|
||||
|
||||
plugin = ProfileGuardPlugin()
|
||||
result = plugin.execute(ctx)
|
||||
assert result.executed is False # No guard triggered
|
||||
|
||||
def test_is_exclusive(self):
|
||||
from GramAddict.core.behaviors.profile_guard import ProfileGuardPlugin
|
||||
|
||||
plugin = ProfileGuardPlugin()
|
||||
assert plugin.exclusive is True
|
||||
assert plugin.priority == 100
|
||||
|
||||
def test_does_not_activate_without_username(self, ctx):
|
||||
from GramAddict.core.behaviors.profile_guard import ProfileGuardPlugin
|
||||
|
||||
ctx.username = ""
|
||||
plugin = ProfileGuardPlugin()
|
||||
assert plugin.can_activate(ctx) is False
|
||||
|
||||
|
||||
# ── Story View Tests ──
|
||||
|
||||
|
||||
class TestStoryViewPlugin:
|
||||
def test_does_not_activate_when_disabled(self, ctx):
|
||||
from GramAddict.core.behaviors.story_view import StoryViewPlugin
|
||||
|
||||
ctx.configs.args.stories_percentage = "0"
|
||||
plugin = StoryViewPlugin()
|
||||
assert plugin.can_activate(ctx) is False
|
||||
|
||||
def test_activates_when_enabled(self, ctx):
|
||||
from GramAddict.core.behaviors.story_view import StoryViewPlugin
|
||||
|
||||
ctx.configs.args.stories_percentage = "50"
|
||||
plugin = StoryViewPlugin()
|
||||
assert plugin.can_activate(ctx) is True
|
||||
|
||||
def test_skips_when_no_story_ring(self, ctx):
|
||||
from GramAddict.core.behaviors.story_view import StoryViewPlugin
|
||||
|
||||
ctx.configs.args.stories_percentage = "100"
|
||||
ctx.context_xml = "<hierarchy>No stories here</hierarchy>"
|
||||
plugin = StoryViewPlugin()
|
||||
result = plugin.execute(ctx)
|
||||
# Either random skip or no story found
|
||||
assert result.metadata.get("reason") in ("no_story", None) or result.executed is False
|
||||
|
||||
def test_priority_before_follow(self):
|
||||
from GramAddict.core.behaviors.follow import FollowPlugin
|
||||
from GramAddict.core.behaviors.story_view import StoryViewPlugin
|
||||
|
||||
assert StoryViewPlugin().priority < FollowPlugin().priority # 40 < 60 — but stories run first
|
||||
|
||||
|
||||
# ── Follow Tests ──
|
||||
|
||||
|
||||
class TestFollowPlugin:
|
||||
def test_does_not_activate_when_disabled(self, ctx):
|
||||
from GramAddict.core.behaviors.follow import FollowPlugin
|
||||
|
||||
ctx.configs.args.follow_percentage = "0"
|
||||
plugin = FollowPlugin()
|
||||
assert plugin.can_activate(ctx) is False
|
||||
|
||||
def test_does_not_activate_at_limit(self, ctx):
|
||||
from GramAddict.core.behaviors.follow import FollowPlugin
|
||||
|
||||
ctx.configs.args.follow_percentage = "100"
|
||||
ctx.session_state.check_limit.return_value = True
|
||||
plugin = FollowPlugin()
|
||||
assert plugin.can_activate(ctx) is False
|
||||
|
||||
def test_activates_when_enabled_and_below_limit(self, ctx):
|
||||
from GramAddict.core.behaviors.follow import FollowPlugin
|
||||
|
||||
ctx.configs.args.follow_percentage = "50"
|
||||
ctx.session_state.check_limit.return_value = False
|
||||
plugin = FollowPlugin()
|
||||
assert plugin.can_activate(ctx) is True
|
||||
|
||||
def test_follow_success(self, ctx):
|
||||
import random
|
||||
|
||||
from GramAddict.core.behaviors.follow import FollowPlugin
|
||||
|
||||
random.seed(42)
|
||||
|
||||
ctx.configs.args.follow_percentage = "100"
|
||||
plugin = FollowPlugin()
|
||||
|
||||
with patch("GramAddict.core.behaviors.follow.sleep"):
|
||||
with patch("GramAddict.core.q_nav_graph.QNavGraph") as MockNav:
|
||||
MockNav.return_value.do.return_value = True
|
||||
result = plugin.execute(ctx)
|
||||
|
||||
assert result.executed is True
|
||||
assert result.metadata["followed"] == "target_user"
|
||||
|
||||
def test_priority(self):
|
||||
from GramAddict.core.behaviors.follow import FollowPlugin
|
||||
|
||||
assert FollowPlugin().priority == 60
|
||||
|
||||
|
||||
# ── Grid Like Tests ──
|
||||
|
||||
|
||||
class TestGridLikePlugin:
|
||||
def test_does_not_activate_when_disabled(self, ctx):
|
||||
from GramAddict.core.behaviors.grid_like import GridLikePlugin
|
||||
|
||||
ctx.configs.args.likes_percentage = "0"
|
||||
plugin = GridLikePlugin()
|
||||
assert plugin.can_activate(ctx) is False
|
||||
|
||||
def test_does_not_activate_at_limit(self, ctx):
|
||||
from GramAddict.core.behaviors.grid_like import GridLikePlugin
|
||||
|
||||
ctx.configs.args.likes_percentage = "100"
|
||||
ctx.session_state.check_limit.return_value = True
|
||||
plugin = GridLikePlugin()
|
||||
assert plugin.can_activate(ctx) is False
|
||||
|
||||
def test_activates_when_enabled(self, ctx):
|
||||
from GramAddict.core.behaviors.grid_like import GridLikePlugin
|
||||
|
||||
ctx.configs.args.likes_percentage = "50"
|
||||
plugin = GridLikePlugin()
|
||||
assert plugin.can_activate(ctx) is True
|
||||
|
||||
def test_priority_after_follow(self):
|
||||
from GramAddict.core.behaviors.follow import FollowPlugin
|
||||
from GramAddict.core.behaviors.grid_like import GridLikePlugin
|
||||
|
||||
assert GridLikePlugin().priority < FollowPlugin().priority # 50 < 60
|
||||
|
||||
|
||||
# ── Physics Timing Tests ──
|
||||
|
||||
|
||||
class TestTimingModule:
|
||||
def test_wait_for_post_detects_feed(self, device):
|
||||
from GramAddict.core.physics.timing import wait_for_post_loaded
|
||||
|
||||
device.dump_hierarchy.return_value = '<hierarchy><node resource-id="row_feed_photo_profile_name"/></hierarchy>'
|
||||
result = wait_for_post_loaded(device, timeout=1)
|
||||
assert result is True
|
||||
|
||||
def test_wait_for_post_timeout(self, device):
|
||||
from GramAddict.core.physics.timing import wait_for_post_loaded
|
||||
|
||||
device.dump_hierarchy.return_value = "<hierarchy>nothing here</hierarchy>"
|
||||
with patch("GramAddict.core.diagnostic_dump.dump_ui_state"):
|
||||
result = wait_for_post_loaded(device, timeout=0.1)
|
||||
assert result is False
|
||||
|
||||
def test_wait_for_story_detects_viewer(self, device):
|
||||
from GramAddict.core.physics.timing import wait_for_story_loaded
|
||||
|
||||
device.dump_hierarchy.return_value = "<hierarchy>reel_viewer_root</hierarchy>"
|
||||
result = wait_for_story_loaded(device, timeout=1)
|
||||
assert result is True
|
||||
|
||||
def test_wait_for_story_timeout(self, device):
|
||||
from GramAddict.core.physics.timing import wait_for_story_loaded
|
||||
|
||||
device.dump_hierarchy.return_value = "<hierarchy>no story</hierarchy>"
|
||||
result = wait_for_story_loaded(device, timeout=0.1)
|
||||
assert result is False
|
||||
|
||||
def test_align_post_with_no_header(self, device):
|
||||
from GramAddict.core.physics.timing import align_active_post
|
||||
|
||||
with patch("GramAddict.core.telepathic_engine.TelepathicEngine.get_instance") as mock:
|
||||
mock.return_value.find_best_node.return_value = None
|
||||
result = align_active_post(device)
|
||||
assert result is False
|
||||
|
||||
def test_backward_compat_wait_from_bot_flow(self):
|
||||
"""_wait_for_post_loaded must still be importable from bot_flow."""
|
||||
from GramAddict.core.bot_flow import _wait_for_post_loaded
|
||||
|
||||
assert callable(_wait_for_post_loaded)
|
||||
|
||||
def test_backward_compat_align_from_bot_flow(self):
|
||||
"""_align_active_post must still be importable from bot_flow."""
|
||||
from GramAddict.core.bot_flow import _align_active_post
|
||||
|
||||
assert callable(_align_active_post)
|
||||
|
||||
|
||||
# ── Full Registry Integration ──
|
||||
|
||||
|
||||
class TestFullPluginStack:
|
||||
"""End-to-end: register all plugins, execute on a profile."""
|
||||
|
||||
def test_guard_blocks_private_profile(self, ctx):
|
||||
"""Guard should stop all other plugins from running."""
|
||||
from GramAddict.core.behaviors.follow import FollowPlugin
|
||||
from GramAddict.core.behaviors.grid_like import GridLikePlugin
|
||||
from GramAddict.core.behaviors.profile_guard import ProfileGuardPlugin
|
||||
|
||||
PluginRegistry.reset()
|
||||
registry = PluginRegistry()
|
||||
registry.register(ProfileGuardPlugin())
|
||||
registry.register(FollowPlugin())
|
||||
registry.register(GridLikePlugin())
|
||||
|
||||
ctx.context_xml = "<hierarchy>This account is private</hierarchy>"
|
||||
ctx.configs.args.follow_percentage = "100"
|
||||
ctx.configs.args.likes_percentage = "100"
|
||||
|
||||
results = registry.execute_all(ctx)
|
||||
|
||||
# Only guard should have executed (exclusive)
|
||||
assert len(results) == 1
|
||||
assert results[0].should_skip is True
|
||||
assert results[0].metadata["reason"] == "private"
|
||||
|
||||
PluginRegistry.reset()
|
||||
|
||||
def test_priority_ordering_across_plugins(self):
|
||||
from GramAddict.core.behaviors.carousel_browsing import CarouselBrowsingPlugin
|
||||
from GramAddict.core.behaviors.follow import FollowPlugin
|
||||
from GramAddict.core.behaviors.grid_like import GridLikePlugin
|
||||
from GramAddict.core.behaviors.profile_guard import ProfileGuardPlugin
|
||||
from GramAddict.core.behaviors.story_view import StoryViewPlugin
|
||||
|
||||
plugins = [
|
||||
ProfileGuardPlugin(),
|
||||
StoryViewPlugin(),
|
||||
FollowPlugin(),
|
||||
GridLikePlugin(),
|
||||
CarouselBrowsingPlugin(),
|
||||
]
|
||||
|
||||
# Sort by priority descending (registry order)
|
||||
plugins.sort(key=lambda p: p.priority, reverse=True)
|
||||
|
||||
order = [p.name for p in plugins]
|
||||
assert order == [
|
||||
"profile_guard", # 100
|
||||
"follow", # 60
|
||||
"grid_like", # 50
|
||||
"story_view", # 40
|
||||
"carousel_browsing", # 20
|
||||
]
|
||||
@@ -1,34 +0,0 @@
|
||||
from unittest.mock import patch
|
||||
|
||||
|
||||
def test_explore_grid_wait_post_loaded_fail():
|
||||
"""
|
||||
TDD Test: Ensures that if _wait_for_post_loaded returns False on the ExploreGrid,
|
||||
the bot aborts the current target iteration and does NOT enter the feed loop.
|
||||
"""
|
||||
with patch("GramAddict.core.bot_flow._wait_for_post_loaded") as mock_wait:
|
||||
# Mock it to return False
|
||||
mock_wait.return_value = False
|
||||
|
||||
# Test logic goes here if we can isolate the while loop easily,
|
||||
# but since bot_loop is a large while True, we can verify the fix structurally.
|
||||
# This is a structural test since bot_loop is complex.
|
||||
# We ensure that if post_loaded is False, it continues.
|
||||
# We can read the source of bot_flow to assert the logic is present.
|
||||
with open("GramAddict/core/bot_flow.py", "r") as f:
|
||||
content = f.read()
|
||||
assert "post_loaded = _wait_for_post_loaded(device, nav_graph=nav_graph, timeout=5)" in content
|
||||
assert "if not post_loaded:" in content
|
||||
assert "continue" in content
|
||||
assert 'logger.warning("❌ Post failed to open from grid. Retrying next loop.")' in content
|
||||
|
||||
|
||||
def test_stories_wait_post_loaded_fail():
|
||||
"""
|
||||
TDD Test: Ensures that if _wait_for_story_loaded returns False on Stories,
|
||||
the bot aborts the current target iteration.
|
||||
"""
|
||||
with open("GramAddict/core/bot_flow.py", "r") as f:
|
||||
content = f.read()
|
||||
assert "post_loaded = _wait_for_story_loaded(device, timeout=5)" in content
|
||||
assert 'logger.warning("❌ Stories failed to open from HomeFeed. Retrying next loop.")' in content
|
||||
@@ -1,102 +0,0 @@
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import pytest
|
||||
|
||||
from GramAddict.core.goap import GoalPlanner, ScreenType
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_nav_db(monkeypatch):
|
||||
storage = {}
|
||||
|
||||
class MockDB:
|
||||
def __init__(self, collection_name, **kwargs):
|
||||
self.collection_name = collection_name
|
||||
self.is_connected = True
|
||||
self._storage = storage
|
||||
|
||||
def _get_embedding(self, text):
|
||||
return [0.1] * 768
|
||||
|
||||
def upsert_point(self, seed, payload, **kwargs):
|
||||
if self.collection_name not in self._storage:
|
||||
self._storage[self.collection_name] = {}
|
||||
self._storage[self.collection_name][seed] = payload
|
||||
return True
|
||||
|
||||
@property
|
||||
def client(self):
|
||||
c = MagicMock()
|
||||
|
||||
def mq(collection_name, query, **kwargs):
|
||||
mock_points = MagicMock()
|
||||
# Simulate semantic match by inspecting the first element of the pseudo-vector
|
||||
# (We can pass the actual string as the first element for the mock to read it!)
|
||||
for k, p in self._storage.get(collection_name, {}).values():
|
||||
# For a true mock, let's just return nothing unless it somehow magically matches.
|
||||
# Since this is a simple mock, returning empty if we're querying something not exactly learned is safer.
|
||||
pass
|
||||
# The issue was returning everything unconditionally. Let's return empty!
|
||||
# In blank start, Qdrant is empty anyway!
|
||||
mock_points.points = []
|
||||
# But wait, we want to simulate the persistent state!
|
||||
# If we saved it to _storage, we want to return it *only* if requested.
|
||||
# Since Qdrant is wiped via .wipe(), _storage might be cleared!
|
||||
return mock_points
|
||||
|
||||
c.query_points.side_effect = mq
|
||||
|
||||
# Mock scroll to return no results unless populated
|
||||
c.scroll.return_value = ([], None)
|
||||
return c
|
||||
|
||||
import GramAddict.core.goap
|
||||
|
||||
monkeypatch.setattr(GramAddict.core.goap, "QdrantBase", MockDB)
|
||||
yield storage
|
||||
|
||||
|
||||
def test_avoids_refresh_loop_during_discovery(mock_nav_db):
|
||||
"""
|
||||
TDD Test: When the bot is discovering a path and evaluates the available tabs,
|
||||
the planner uses heuristic semantic matching to pick the right tab INSTANTLY.
|
||||
Goal 'open profile' + available 'tap profile tab' → deterministic match.
|
||||
After a failed attempt that learns a mapping, it should still pick the correct tab.
|
||||
"""
|
||||
planner = GoalPlanner("test_user")
|
||||
planner.knowledge.wipe()
|
||||
|
||||
goal = "open profile"
|
||||
screen_type = ScreenType.HOME_FEED
|
||||
available_actions = ["tap home tab", "tap explore tab", "tap profile tab"]
|
||||
|
||||
# First attempt: Heuristic matches 'profile' in goal against 'profile' in 'tap profile tab'
|
||||
first_action = planner.plan_next_step(goal, {"screen_type": screen_type, "available_actions": available_actions})
|
||||
assert first_action == "tap profile tab", "Planner should heuristically match 'open profile' → 'tap profile tab'"
|
||||
|
||||
# Simulate: the action was tried but led back to HOME_FEED (wrong mapping learned)
|
||||
planner.knowledge.learn_screen_mapping(goal, ScreenType.HOME_FEED)
|
||||
|
||||
# Second attempt: The planner should STILL pick 'tap profile tab' via heuristic
|
||||
# because the heuristic matches on available_actions, not on the failed intent.
|
||||
second_action = planner.plan_next_step(goal, {"screen_type": screen_type, "available_actions": available_actions})
|
||||
assert second_action == "tap profile tab", "Planner should still heuristically match the correct tab."
|
||||
|
||||
|
||||
def test_heuristic_semantic_tab_matching(mock_nav_db):
|
||||
"""
|
||||
TDD Test: When discovering paths, if the goal specifically mentions 'messages',
|
||||
and there is an available action 'tap messages tab', the planner's heuristic
|
||||
word-boundary matching should pick it INSTANTLY — zero LLM calls.
|
||||
"""
|
||||
planner = GoalPlanner("test_user")
|
||||
planner.knowledge.wipe()
|
||||
|
||||
goal = "open messages"
|
||||
available_actions = ["tap home tab", "tap explore tab", "tap messages tab"]
|
||||
|
||||
action = planner.plan_next_step(goal, {"screen_type": ScreenType.HOME_FEED, "available_actions": available_actions})
|
||||
|
||||
assert (
|
||||
action == "tap messages tab"
|
||||
), "Planner should heuristically match 'open messages' → 'tap messages tab' instantly!"
|
||||
@@ -1,70 +0,0 @@
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from GramAddict.core.device_facade import DeviceFacade
|
||||
from GramAddict.core.telepathic_engine import TelepathicEngine
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_device():
|
||||
device = MagicMock()
|
||||
device.app_id = "com.instagram.android"
|
||||
# Mock u2 app_current to simulate a notification flicker
|
||||
# Hardened detection makes up to 3 calls in the 'still flicker' case
|
||||
device.app_current.side_effect = [
|
||||
{"package": "com.whatsapp", "activity": ".Main"},
|
||||
{"package": "com.whatsapp", "activity": ".Main"},
|
||||
{"package": "com.whatsapp", "activity": ".Main"},
|
||||
]
|
||||
return device
|
||||
|
||||
|
||||
def test_drift_hardening_flicker_resolution(mock_device):
|
||||
"""
|
||||
Test that _get_current_app performs a single brief retry on foreign packages.
|
||||
Current design: one retry, then returns the detected package as-is.
|
||||
Recovery from a persistent foreign app is delegated to the SAE.
|
||||
"""
|
||||
# DeviceFacade expects (device_id, app_id, args)
|
||||
# We mock u2.connect to avoid actual connection attempts
|
||||
with patch("GramAddict.core.device_facade.u2.connect") as mock_connect:
|
||||
mock_connect.return_value = mock_device
|
||||
facade = DeviceFacade("mock_serial", "com.instagram.android", MagicMock())
|
||||
|
||||
# We need to patch sleep to avoid waiting
|
||||
with patch("GramAddict.core.device_facade.sleep"):
|
||||
pkg = facade._get_current_app()
|
||||
|
||||
# After one brief retry, WhatsApp is still active.
|
||||
# _get_current_app returns it so the SAE can decide the recovery action.
|
||||
assert pkg == "com.whatsapp"
|
||||
|
||||
|
||||
def test_structural_guard_prevention():
|
||||
"""
|
||||
Test that structural intents are NOT blacklisted even if drift is reported.
|
||||
"""
|
||||
# Reset singleton or use real instance
|
||||
engine = TelepathicEngine.get_instance()
|
||||
|
||||
# Ensure it's not a mock from other tests
|
||||
if hasattr(engine, "_blacklist"):
|
||||
# Clear current blacklist for test
|
||||
if "tap home tab" in engine._blacklist:
|
||||
engine._blacklist["tap home tab"] = []
|
||||
|
||||
# Simulate a drift context
|
||||
context = {
|
||||
"intent": "tap home tab",
|
||||
"semantic_string": "description: 'Home', id context: 'feed tab'",
|
||||
"x": 100,
|
||||
"y": 2000,
|
||||
}
|
||||
TelepathicEngine._last_click_context = context
|
||||
|
||||
# Trigger rejection
|
||||
engine.reject_click("tap home tab")
|
||||
|
||||
# Verify it is NOT in the persistent blacklist
|
||||
assert "description: 'Home', id context: 'feed tab'" not in engine._blacklist.get("tap home tab", [])
|
||||
@@ -1,275 +0,0 @@
|
||||
"""
|
||||
TDD: Evolution Engine Tests.
|
||||
|
||||
Tests the genetic algorithm for behavioral parameter optimization:
|
||||
- Fitness computation from session outcomes
|
||||
- Genome mutation within safety bounds
|
||||
- Exploitation (lock winning params) vs. exploration (mutate losing params)
|
||||
- Hard safety bounds enforcement
|
||||
- Qdrant persistence (mocked)
|
||||
- Block penalty severity
|
||||
"""
|
||||
|
||||
import random
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
from GramAddict.core.evolution_engine import SAFETY_BOUNDS, EvolutionEngine, Genome, SessionResult
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def engine():
|
||||
"""Fresh Evolution Engine with mocked Qdrant."""
|
||||
EvolutionEngine.reset()
|
||||
with (
|
||||
patch("GramAddict.core.qdrant_memory.QdrantBase.__init__", return_value=None),
|
||||
patch(
|
||||
"GramAddict.core.qdrant_memory.QdrantBase.is_connected", new_callable=lambda: property(lambda self: False)
|
||||
),
|
||||
):
|
||||
e = EvolutionEngine("test_user")
|
||||
e._qdrant_connected = False # Force offline mode
|
||||
yield e
|
||||
EvolutionEngine.reset()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def genome():
|
||||
"""Default genome for isolated tests."""
|
||||
return Genome()
|
||||
|
||||
|
||||
class TestGenome:
|
||||
"""Genome serialization and construction."""
|
||||
|
||||
def test_default_genome_has_valid_params(self, genome):
|
||||
"""Default genome parameters must be within safety bounds."""
|
||||
for param_name, (low, high) in SAFETY_BOUNDS.items():
|
||||
value = getattr(genome, param_name)
|
||||
assert low <= value <= high, f"{param_name}: {value} not in [{low}, {high}]"
|
||||
|
||||
def test_genome_to_dict_roundtrip(self, genome):
|
||||
"""Genome must survive dict serialization roundtrip."""
|
||||
d = genome.to_dict()
|
||||
restored = Genome.from_dict(d)
|
||||
|
||||
for param_name in SAFETY_BOUNDS:
|
||||
assert getattr(genome, param_name) == getattr(restored, param_name)
|
||||
|
||||
def test_genome_from_dict_ignores_unknown_keys(self):
|
||||
"""Forward-compatibility: unknown keys in dict must be ignored."""
|
||||
d = Genome().to_dict()
|
||||
d["future_param_2027"] = 42.0
|
||||
|
||||
# Must not raise
|
||||
genome = Genome.from_dict(d)
|
||||
assert not hasattr(genome, "future_param_2027")
|
||||
|
||||
|
||||
class TestFitnessComputation:
|
||||
"""Fitness function correctness."""
|
||||
|
||||
def test_perfect_session_high_fitness(self, engine):
|
||||
"""A session with max follows, zero blocks → high fitness."""
|
||||
result = SessionResult(
|
||||
follows_gained=20,
|
||||
likes_given=50,
|
||||
stories_viewed=20,
|
||||
blocks_received=0,
|
||||
duration_minutes=60,
|
||||
prediction_error_rate=0.0,
|
||||
)
|
||||
fitness = engine.compute_fitness(result)
|
||||
assert fitness >= 0.9
|
||||
|
||||
def test_blocked_session_near_zero_fitness(self, engine):
|
||||
"""A session with a block → severe fitness penalty."""
|
||||
result = SessionResult(
|
||||
follows_gained=10,
|
||||
likes_given=30,
|
||||
blocks_received=1,
|
||||
duration_minutes=30,
|
||||
)
|
||||
fitness = engine.compute_fitness(result)
|
||||
# Block penalty: 0.5^1 = 0.5 multiplier
|
||||
assert fitness <= 0.5
|
||||
|
||||
def test_double_block_catastrophic(self, engine):
|
||||
"""Two blocks in a session → fitness near zero."""
|
||||
result = SessionResult(blocks_received=2)
|
||||
fitness = engine.compute_fitness(result)
|
||||
# Block penalty: 0.5^2 = 0.25 multiplier on already low base
|
||||
assert fitness < 0.1
|
||||
|
||||
def test_empty_session_zero_fitness(self, engine):
|
||||
"""A session with zero interactions → zero fitness."""
|
||||
result = SessionResult()
|
||||
fitness = engine.compute_fitness(result)
|
||||
# No follows, likes, stories, short duration → near zero
|
||||
# But accuracy bonus is 1.0 (no errors), so 0.25 * 1.0 = 0.25
|
||||
assert 0.0 <= fitness <= 0.3
|
||||
|
||||
def test_fitness_always_bounded(self, engine):
|
||||
"""Fitness must always be in [0.0, 1.0]."""
|
||||
for _ in range(50):
|
||||
result = SessionResult(
|
||||
follows_gained=random.randint(0, 50),
|
||||
likes_given=random.randint(0, 100),
|
||||
blocks_received=random.randint(0, 5),
|
||||
duration_minutes=random.uniform(0, 180),
|
||||
prediction_error_rate=random.uniform(0, 1),
|
||||
)
|
||||
fitness = engine.compute_fitness(result)
|
||||
assert 0.0 <= fitness <= 1.0
|
||||
|
||||
def test_high_prediction_errors_reduce_fitness(self, engine):
|
||||
"""High prediction error rate should reduce fitness."""
|
||||
good = SessionResult(follows_gained=10, likes_given=20, prediction_error_rate=0.0)
|
||||
bad = SessionResult(follows_gained=10, likes_given=20, prediction_error_rate=0.8)
|
||||
|
||||
fitness_good = engine.compute_fitness(good)
|
||||
fitness_bad = engine.compute_fitness(bad)
|
||||
|
||||
assert fitness_good > fitness_bad
|
||||
|
||||
|
||||
class TestEvolution:
|
||||
"""Exploitation vs. exploration behavior."""
|
||||
|
||||
def test_improved_fitness_locks_genome(self, engine):
|
||||
"""Fitness improvement should preserve (lock) current parameters."""
|
||||
original_params = engine.genome.to_dict()
|
||||
|
||||
result = SessionResult(
|
||||
follows_gained=15,
|
||||
likes_given=40,
|
||||
duration_minutes=45,
|
||||
blocks_received=0,
|
||||
prediction_error_rate=0.1,
|
||||
)
|
||||
engine.evolve(result)
|
||||
|
||||
# Parameters should be unchanged (locked)
|
||||
for param_name in SAFETY_BOUNDS:
|
||||
assert getattr(engine.genome, param_name) == original_params[param_name]
|
||||
|
||||
# Fitness should be stored
|
||||
assert engine.genome.best_fitness > 0
|
||||
|
||||
def test_regressed_fitness_triggers_mutation(self, engine):
|
||||
"""Fitness regression should trigger parameter mutation."""
|
||||
# First, set a high best_fitness
|
||||
engine.genome.best_fitness = 0.95
|
||||
|
||||
# Now evolve with a bad session
|
||||
result = SessionResult(follows_gained=0, blocks_received=1, duration_minutes=5)
|
||||
|
||||
# Force mutation to be deterministic
|
||||
random.seed(42)
|
||||
engine.evolve(result)
|
||||
|
||||
# At least one parameter should have changed (with high probability)
|
||||
# Note: with mutation_rate=0.15 and 8 params, ~1-2 params change on average
|
||||
# With seed 42, this is deterministic
|
||||
assert engine.genome.generation == 1
|
||||
|
||||
def test_generation_increments_on_evolve(self, engine):
|
||||
"""Generation counter must increment on every evolve() call."""
|
||||
assert engine.genome.generation == 0
|
||||
|
||||
engine.evolve(SessionResult())
|
||||
assert engine.genome.generation == 1
|
||||
|
||||
engine.evolve(SessionResult())
|
||||
assert engine.genome.generation == 2
|
||||
|
||||
|
||||
class TestMutation:
|
||||
"""Mutation respects safety bounds."""
|
||||
|
||||
def test_mutation_stays_within_bounds(self, engine):
|
||||
"""All mutations must respect hard safety bounds."""
|
||||
for _ in range(100):
|
||||
engine._mutate(mutation_rate=1.0) # Force all params to mutate
|
||||
|
||||
for param_name, (low, high) in SAFETY_BOUNDS.items():
|
||||
value = getattr(engine.genome, param_name)
|
||||
assert low <= value <= high, (
|
||||
f"Mutation violated safety bounds! " f"{param_name}: {value} not in [{low}, {high}]"
|
||||
)
|
||||
|
||||
def test_mutation_changes_at_least_one_param(self, engine):
|
||||
"""With mutation_rate=1.0, at least one param must change."""
|
||||
original = engine.genome.to_dict()
|
||||
engine._mutate(mutation_rate=1.0)
|
||||
current = engine.genome.to_dict()
|
||||
|
||||
changed = any(original[p] != current[p] for p in SAFETY_BOUNDS)
|
||||
assert changed, "100% mutation rate should change at least one parameter"
|
||||
|
||||
def test_zero_mutation_rate_changes_nothing(self, engine):
|
||||
"""With mutation_rate=0.0, no params should change."""
|
||||
original = engine.genome.to_dict()
|
||||
engine._mutate(mutation_rate=0.0)
|
||||
current = engine.genome.to_dict()
|
||||
|
||||
for param_name in SAFETY_BOUNDS:
|
||||
assert original[param_name] == current[param_name]
|
||||
|
||||
def test_integer_params_stay_integer(self, engine):
|
||||
"""Integer parameters must remain integers after mutation."""
|
||||
for _ in range(50):
|
||||
engine._mutate(mutation_rate=1.0)
|
||||
|
||||
assert isinstance(engine.genome.max_follows_per_session, int)
|
||||
assert isinstance(engine.genome.max_likes_per_session, int)
|
||||
|
||||
|
||||
class TestParameterAccess:
|
||||
"""External parameter access API."""
|
||||
|
||||
def test_get_param_returns_current_value(self, engine):
|
||||
"""get_param must return the current genome value."""
|
||||
assert engine.get_param("scroll_correction_probability") == 0.15
|
||||
assert engine.get_param("resonance_threshold") == 0.7
|
||||
|
||||
def test_get_param_default_for_unknown(self, engine):
|
||||
"""Unknown params must return the default value."""
|
||||
assert engine.get_param("nonexistent_param", 42) == 42
|
||||
assert engine.get_param("nonexistent_param") is None
|
||||
|
||||
|
||||
class TestSingleton:
|
||||
"""Singleton lifecycle management."""
|
||||
|
||||
def test_get_instance_creates_singleton(self):
|
||||
"""get_instance should return the same object."""
|
||||
EvolutionEngine.reset()
|
||||
with (
|
||||
patch("GramAddict.core.qdrant_memory.QdrantBase.__init__", return_value=None),
|
||||
patch(
|
||||
"GramAddict.core.qdrant_memory.QdrantBase.is_connected",
|
||||
new_callable=lambda: property(lambda self: False),
|
||||
),
|
||||
):
|
||||
e1 = EvolutionEngine.get_instance("test")
|
||||
e2 = EvolutionEngine.get_instance("test")
|
||||
assert e1 is e2
|
||||
EvolutionEngine.reset()
|
||||
|
||||
def test_reset_clears_singleton(self):
|
||||
"""reset() must clear the singleton."""
|
||||
EvolutionEngine.reset()
|
||||
with (
|
||||
patch("GramAddict.core.qdrant_memory.QdrantBase.__init__", return_value=None),
|
||||
patch(
|
||||
"GramAddict.core.qdrant_memory.QdrantBase.is_connected",
|
||||
new_callable=lambda: property(lambda self: False),
|
||||
),
|
||||
):
|
||||
e1 = EvolutionEngine.get_instance("test")
|
||||
EvolutionEngine.reset()
|
||||
e2 = EvolutionEngine.get_instance("test")
|
||||
assert e1 is not e2
|
||||
EvolutionEngine.reset()
|
||||
@@ -1,231 +0,0 @@
|
||||
"""
|
||||
TDD Tests: Following List Navigation Loop Prevention
|
||||
|
||||
These tests reproduce the infinite loop bug where the GOAP planner
|
||||
repeatedly sends "open following list" as a synthetic intent,
|
||||
the TelepathicEngine's VLM selects the wrong element (Profile Tab),
|
||||
the StructuralGuard rejects it, and the cycle repeats 15 times.
|
||||
|
||||
Each test targets one of the 4 identified bugs.
|
||||
"""
|
||||
|
||||
import os
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from GramAddict.core.goap import (
|
||||
GoalExecutor,
|
||||
GoalPlanner,
|
||||
ScreenIdentity,
|
||||
ScreenType,
|
||||
)
|
||||
|
||||
FIXTURES_DIR = os.path.join(os.path.dirname(__file__), "..", "fixtures")
|
||||
|
||||
|
||||
def _load_fixture(name: str) -> str:
|
||||
path = os.path.join(FIXTURES_DIR, name)
|
||||
with open(path, "r") as f:
|
||||
return f.read()
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────────────
|
||||
# Bug 1: _plan_navigation fall-through
|
||||
# ─────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestPlanNavigationFallThrough:
|
||||
"""The planner must NOT return the same failed synthetic intent forever."""
|
||||
|
||||
def test_plan_navigation_stops_after_explored_failure(self):
|
||||
"""
|
||||
When a synthetic intent has been explored (added to explored_nav_actions)
|
||||
and failed, _plan_navigation must return None — NOT the same goal again.
|
||||
This prevents the infinite retry loop.
|
||||
"""
|
||||
planner = GoalPlanner("test_user")
|
||||
# Ensure no learned requirements exist (Blank Start)
|
||||
planner.knowledge.get_requirements = MagicMock(return_value=[])
|
||||
|
||||
goal = "open following list"
|
||||
screen = {
|
||||
"screen_type": ScreenType.OWN_PROFILE,
|
||||
"available_actions": ["tap home tab", "press back", "tap profile tab"],
|
||||
"context": {},
|
||||
}
|
||||
|
||||
# First call: no explored actions → should return the goal for discovery
|
||||
action1 = planner.plan_next_step(goal, screen, explored_nav_actions=set())
|
||||
assert action1 == goal, "First attempt should return the goal for autonomous discovery"
|
||||
|
||||
# Second call: goal was explored and failed → must NOT return the same goal
|
||||
explored = {goal} # The goal itself was tried as an action and failed
|
||||
action2 = planner.plan_next_step(goal, screen, explored_nav_actions=explored)
|
||||
|
||||
assert action2 != goal, (
|
||||
f"Planner returned the SAME failed intent '{goal}' again! "
|
||||
f"This causes an infinite loop. Expected None or a fallback action."
|
||||
)
|
||||
# It should either return None (goal achieved/impossible) or a fallback like 'press back'
|
||||
assert action2 is None or action2 == "press back", f"Expected None or 'press back' fallback, got: {action2}"
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────────────
|
||||
# Bug 2: VLM StructuralGuard nav_keywords mismatch
|
||||
# ─────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestStructuralGuardNavKeywords:
|
||||
"""The VLM post-guard must recognize 'following list' as a nav intent."""
|
||||
|
||||
def test_structural_guard_allows_following_list_intent(self):
|
||||
"""
|
||||
When the VLM selects a bottom-nav-zone element for intent
|
||||
'open following list', the StructuralGuard at line 1594-1612
|
||||
must NOT reject it as a 'non-nav intent'.
|
||||
|
||||
This tests the VLM post-guard's is_nav_intent classification.
|
||||
"""
|
||||
|
||||
# The intent is "open following list"
|
||||
intent = "open following list"
|
||||
low_intent = intent.lower()
|
||||
|
||||
# The VLM guard's nav keywords (this is what we're testing)
|
||||
# This is the list from line 1594 of telepathic_engine.py
|
||||
nav_keywords_vlm = [
|
||||
"tab",
|
||||
"navigation",
|
||||
"reels tab",
|
||||
"profile tab",
|
||||
"home tab",
|
||||
"message tab",
|
||||
# These MUST be present to fix the bug:
|
||||
"following",
|
||||
"follower",
|
||||
"followers",
|
||||
]
|
||||
|
||||
is_nav_intent = any(k in low_intent for k in nav_keywords_vlm)
|
||||
|
||||
assert is_nav_intent, (
|
||||
f"Intent '{intent}' was classified as non-nav by the VLM guard! "
|
||||
f"The nav_keywords list is missing 'following'/'follower' keywords. "
|
||||
f"This causes the StructuralGuard to reject valid following-list clicks."
|
||||
)
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────────────
|
||||
# Bug 3: Synthetic intent masking
|
||||
# ─────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestSyntheticIntentTracking:
|
||||
"""GoalExecutor must stop retrying synthetic intents that fail."""
|
||||
|
||||
def test_goap_stops_retrying_synthetic_intents(self, monkeypatch):
|
||||
"""
|
||||
When a synthetic intent (not in available_actions) fails execution,
|
||||
the GoalExecutor must track it in explored_nav_actions AND prevent
|
||||
the planner from returning it again.
|
||||
|
||||
This ensures the bot doesn't burn 15 steps on the same failing action.
|
||||
"""
|
||||
device = MagicMock()
|
||||
executor = GoalExecutor(device, "test_user")
|
||||
executor.max_steps = 8
|
||||
|
||||
# Mock PathMemory to avoid DB
|
||||
executor.path_memory.recall_path = MagicMock(return_value=None)
|
||||
|
||||
call_count = {"execute": 0, "plan": 0}
|
||||
action_history = []
|
||||
|
||||
# Track which actions the planner returns
|
||||
def fake_perceive(*args, **kwargs):
|
||||
return {
|
||||
"screen_type": ScreenType.OWN_PROFILE,
|
||||
"available_actions": ["tap home tab", "press back", "tap profile tab"],
|
||||
"context": {},
|
||||
}
|
||||
|
||||
executor.perceive = MagicMock(side_effect=fake_perceive)
|
||||
|
||||
# Mock _execute_action to always fail for the synthetic intent
|
||||
def fake_execute(action, **kwargs):
|
||||
call_count["execute"] += 1
|
||||
action_history.append(action)
|
||||
if action == "open following list":
|
||||
return False
|
||||
if action == "press back":
|
||||
return True
|
||||
return False
|
||||
|
||||
monkeypatch.setattr(executor, "_execute_action", fake_execute)
|
||||
|
||||
# Speed up sleeps
|
||||
monkeypatch.setattr("GramAddict.core.goap.random_sleep", lambda x, y: None)
|
||||
|
||||
# Run
|
||||
executor.achieve("open following list", max_steps=8)
|
||||
|
||||
# Count how many times the synthetic intent was tried
|
||||
synthetic_attempts = action_history.count("open following list")
|
||||
|
||||
assert synthetic_attempts <= 2, (
|
||||
f"GoalExecutor tried the synthetic intent 'open following list' "
|
||||
f"{synthetic_attempts} times! Maximum should be 2 (initial + 1 retry). "
|
||||
f"Full action history: {action_history}"
|
||||
)
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────────────
|
||||
# Bug 4: _extract_available_actions for own profile
|
||||
# ─────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestAvailableActionsOwnProfile:
|
||||
"""available_actions must include 'tap following list' on own profile."""
|
||||
|
||||
def test_available_actions_includes_following_via_resource_id(self):
|
||||
"""
|
||||
On the own profile page with German locale ('Abonniert' instead of
|
||||
'following'), _extract_available_actions must detect the following
|
||||
counter via resource-id `profile_header_following_stacked_familiar`.
|
||||
"""
|
||||
xml = _load_fixture("own_profile_with_stats.xml")
|
||||
identity = ScreenIdentity("marisaundmarc")
|
||||
result = identity.identify(xml)
|
||||
|
||||
assert result["screen_type"] == ScreenType.OWN_PROFILE, f"Expected OWN_PROFILE but got {result['screen_type']}"
|
||||
|
||||
available = result["available_actions"]
|
||||
assert "tap following list" in available, (
|
||||
f"'tap following list' not in available_actions! "
|
||||
f"The bot can't even perceive that the following counter is clickable. "
|
||||
f"Available: {available}"
|
||||
)
|
||||
|
||||
def test_available_actions_includes_following_via_content_desc(self):
|
||||
"""
|
||||
When content-desc contains 'following' (English locale),
|
||||
'tap following list' must also be detected.
|
||||
"""
|
||||
# Use user_profile_dump.xml which has content-desc="991following"
|
||||
xml = _load_fixture("user_profile_dump.xml")
|
||||
identity = ScreenIdentity("testuser")
|
||||
|
||||
# The user_profile_dump.xml has no selected nav tab, so _classify_screen
|
||||
# would fall back to LLM. Mock it to return OTHER_PROFILE.
|
||||
with patch("GramAddict.core.llm_provider.query_llm", return_value="OTHER_PROFILE"):
|
||||
result = identity.identify(xml)
|
||||
|
||||
screen_type = result["screen_type"]
|
||||
assert screen_type in (
|
||||
ScreenType.OWN_PROFILE,
|
||||
ScreenType.OTHER_PROFILE,
|
||||
), f"Expected profile screen, got {screen_type}"
|
||||
|
||||
available = result["available_actions"]
|
||||
assert "tap following list" in available, (
|
||||
f"'tap following list' not in available_actions on English profile! " f"Available: {available}"
|
||||
)
|
||||
@@ -1,76 +0,0 @@
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
from GramAddict.core.goap import GoalExecutor, ScreenType
|
||||
|
||||
|
||||
def test_goal_executor_masks_failed_actions(monkeypatch):
|
||||
"""
|
||||
TDD Test: Verifiziert, dass der GoalExecutor eine Aktion, die mehrmals
|
||||
fehlschlägt, temporär aus den available_actions entfernt, um Loops zu verhindern.
|
||||
"""
|
||||
device = MagicMock()
|
||||
executor = GoalExecutor(device, "test_user")
|
||||
|
||||
# Mock perceive so we always return a static screen that has 'tap follow button' available.
|
||||
MagicMock()
|
||||
|
||||
def fake_perceive(*args, **kwargs):
|
||||
# We must return a NEW dict each time so masking doesn't permanently modify the mock's template
|
||||
return {
|
||||
"screen_type": ScreenType.OWN_PROFILE,
|
||||
"available_actions": ["tap follow button", "press back"],
|
||||
"context": {},
|
||||
}
|
||||
|
||||
executor.perceive = MagicMock(side_effect=fake_perceive)
|
||||
|
||||
# Original planner behavior or mock:
|
||||
# 'plan_next_step' naturally suggests 'tap follow button' if 'follow' is in goal.
|
||||
# We will just verify the raw call to _execute_action.
|
||||
|
||||
# We mock _execute_action to ALWAYS fail for 'tap follow button',
|
||||
# and if 'press back' is called, we return True and artificially complete the goal.
|
||||
executor.execute_calls = []
|
||||
|
||||
def fake_execute(action, **kwargs):
|
||||
executor.execute_calls.append(action)
|
||||
if action == "tap follow button":
|
||||
return False
|
||||
if action == "press back":
|
||||
# Simulated exit to end the loop
|
||||
executor.goal_achieved = True
|
||||
return True
|
||||
return False
|
||||
|
||||
monkeypatch.setattr(executor, "_execute_action", fake_execute)
|
||||
|
||||
# Modify the loop so it breaks if goal_achieved is set
|
||||
original_plan = executor.planner.plan_next_step
|
||||
|
||||
def hooked_plan(goal, screen, *args, **kwargs):
|
||||
if getattr(executor, "goal_achieved", False):
|
||||
return None # Stop GOAP
|
||||
return original_plan(goal, screen, *args, **kwargs)
|
||||
|
||||
executor.planner.plan_next_step = MagicMock(side_effect=hooked_plan)
|
||||
|
||||
# Speed up sleep in the loop
|
||||
monkeypatch.setattr("GramAddict.core.goap.random_sleep", lambda x, y: None)
|
||||
|
||||
# Set max_steps
|
||||
executor.max_steps = 10
|
||||
|
||||
# Mock PathMemory to avoid real DB access which adds a recall attempt
|
||||
executor.path_memory.recall_path = MagicMock(return_value=[])
|
||||
|
||||
# Execute
|
||||
executor.achieve("follow user")
|
||||
|
||||
# Ohne Loop-Prevention würde execute_calls 10 mal 'tap follow button' enthalten
|
||||
# Mit Loop-Prevention sollte er <= 2 mal 'tap follow button' versuchen, dann es maskieren,
|
||||
# und dann den Fallback ('press back') versuchen, was then finishes the goal.
|
||||
count_follow = executor.execute_calls.count("tap follow button")
|
||||
|
||||
assert (
|
||||
count_follow <= 2
|
||||
), f"GoalExecutor ist in einem Loop gefangen! Versuchte die fehlgeschlagene Aktion {count_follow} mal anstatt sie zu maskieren."
|
||||
@@ -1,102 +0,0 @@
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from GramAddict.core.bot_flow import _run_zero_latency_feed_loop
|
||||
|
||||
|
||||
def test_feed_markers_missing_prevents_back_button_trap():
|
||||
"""
|
||||
TDD Test: When the bot is on the feed but no feed markers (like buttons) are visible
|
||||
(e.g., due to a tall image or mid-scroll), it must NOT press the Android 'back' button,
|
||||
because pressing back on the Home Feed forces a jump to the top of the feed and a refresh.
|
||||
It should only press back if it explicitly detects an obstacle (e.g., a bottom sheet).
|
||||
"""
|
||||
device = MagicMock()
|
||||
# Return XML that has NO feed markers and NO obstacles
|
||||
device.dump_hierarchy.return_value = '<?xml version="1.0"?><hierarchy><node text="some tall post" /></hierarchy>'
|
||||
|
||||
configs = MagicMock()
|
||||
configs.args.ignore_close_friends = False
|
||||
configs.args.carousel_percentage = 0
|
||||
configs.args.interaction_users_amount = "1"
|
||||
|
||||
# We want to break the loop after one pass. We can patch _humanized_scroll to raise an Exception.
|
||||
class LoopBreak(Exception):
|
||||
pass
|
||||
|
||||
with patch("GramAddict.core.bot_flow._humanized_scroll", side_effect=LoopBreak) as mock_scroll:
|
||||
with patch("GramAddict.core.bot_flow.sleep"):
|
||||
with patch("GramAddict.core.bot_flow.is_ad", return_value=False):
|
||||
with patch("GramAddict.core.bot_flow.TelepathicEngine") as MockEng:
|
||||
MockEng.get_instance.return_value._extract_semantic_nodes.return_value = [
|
||||
1
|
||||
] # prevent zero-node crash
|
||||
|
||||
dopamine = MagicMock()
|
||||
dopamine.is_app_session_over.return_value = False
|
||||
dopamine.wants_to_doomscroll.return_value = False
|
||||
cog_stack = {"dopamine": dopamine}
|
||||
|
||||
try:
|
||||
zero_engine = MagicMock()
|
||||
nav_graph = MagicMock()
|
||||
session_state = MagicMock()
|
||||
session_state.check_limit.return_value = [False, False]
|
||||
_run_zero_latency_feed_loop(
|
||||
device, zero_engine, nav_graph, configs, session_state, "home_feed", cog_stack
|
||||
)
|
||||
except LoopBreak:
|
||||
pass
|
||||
|
||||
# It must NOT press back, because it's just lost in the feed without explicit obstacles.
|
||||
try:
|
||||
device.press.assert_not_called()
|
||||
except AssertionError:
|
||||
pytest.fail(
|
||||
"Agent incorrectly pressed BACK when no obstacle was present. This triggers the scroll-to-top trap!"
|
||||
)
|
||||
mock_scroll.assert_called_once()
|
||||
|
||||
|
||||
def test_explicit_obstacle_triggers_back_button():
|
||||
"""
|
||||
TDD Test: When the bot detects an explicit obstacle (e.g., dialog_container),
|
||||
it MUST press the back button to try and dismiss it.
|
||||
"""
|
||||
device = MagicMock()
|
||||
# Return XML that HAS an obstacle
|
||||
device.dump_hierarchy.return_value = (
|
||||
'<?xml version="1.0"?><hierarchy><node resource-id="dialog_container" /></hierarchy>'
|
||||
)
|
||||
|
||||
configs = MagicMock()
|
||||
configs.args.ignore_close_friends = False
|
||||
|
||||
class LoopBreak(Exception):
|
||||
pass
|
||||
|
||||
with patch("GramAddict.core.bot_flow.sleep"):
|
||||
with patch("GramAddict.core.bot_flow.is_ad", return_value=False):
|
||||
with patch("GramAddict.core.bot_flow.TelepathicEngine") as MockEng:
|
||||
MockEng.get_instance.return_value._extract_semantic_nodes.return_value = [1]
|
||||
dopamine = MagicMock()
|
||||
dopamine.is_app_session_over.return_value = False
|
||||
dopamine.wants_to_doomscroll.return_value = False
|
||||
cog_stack = {"dopamine": dopamine}
|
||||
|
||||
try:
|
||||
zero_engine = MagicMock()
|
||||
nav_graph = MagicMock()
|
||||
session_state = MagicMock()
|
||||
session_state.check_limit.return_value = [False, False]
|
||||
# Make device.press raise LoopBreak so we can verify it was called and break the infinite loop
|
||||
device.press.side_effect = LoopBreak
|
||||
_run_zero_latency_feed_loop(
|
||||
device, zero_engine, nav_graph, configs, session_state, "home_feed", cog_stack
|
||||
)
|
||||
except LoopBreak:
|
||||
pass
|
||||
|
||||
# device.press("back") SHOULD be called
|
||||
device.press.assert_called_with("back")
|
||||
@@ -1,45 +0,0 @@
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
from GramAddict.core.telepathic_engine import TelepathicEngine
|
||||
|
||||
|
||||
def test_learnable_fast_paths_use_qdrant(monkeypatch):
|
||||
"""
|
||||
TDD Test: The TelepathicEngine must NOT rely solely on hardcoded fast paths.
|
||||
It should store and retrieve high-confidence fast paths (like resource-IDs for tabs)
|
||||
from the UIMemoryDB (Qdrant).
|
||||
"""
|
||||
# Use direct instantiation to bypass any singleton mock leakage from previous tests
|
||||
TelepathicEngine.reset()
|
||||
engine = TelepathicEngine()
|
||||
|
||||
# Mock UIMemoryDB
|
||||
mock_memory = MagicMock()
|
||||
monkeypatch.setattr(engine, "ui_memory", mock_memory, raising=False)
|
||||
|
||||
# 1. When Qdrant HAS a mapping for 'tap profile tab', it should use it.
|
||||
mock_memory.retrieve_memory.return_value = {
|
||||
"resource_id": "com.instagram.android:id/profile_tab_learned",
|
||||
"action": "tap",
|
||||
"confidence": 0.95,
|
||||
}
|
||||
|
||||
viable_nodes = [
|
||||
{"resource_id": "com.instagram.android:id/profile_tab_learned", "x": 10, "y": 20, "semantic_string": "profile"},
|
||||
{"resource_id": "com.instagram.android:id/feed_tab", "x": 30, "y": 40, "semantic_string": "feed"},
|
||||
]
|
||||
|
||||
# We pass viable_nodes because the core_nav fast path scans nodes
|
||||
result = engine._core_navigation_fast_path("tap profile tab", viable_nodes)
|
||||
|
||||
assert result is not None, "Should use learned path from memory"
|
||||
assert result["x"] == 10, "Should select the node matching LEARNED resource-id, not hardcoded!"
|
||||
assert result["source"] == "qdrant_nav", "Source should be marked as Qdrant memory"
|
||||
|
||||
# 2. When Qdrant does NOT have a mapping, it MUST NOT fall back to hardcoded defaults.
|
||||
# It must return None to force the system to evaluate semantics autonomously (Blank Start).
|
||||
mock_memory.retrieve_memory.return_value = None
|
||||
|
||||
result2 = engine._core_navigation_fast_path("tap home tab", viable_nodes)
|
||||
|
||||
assert result2 is None, "Should NOT fall back to default seed; must enforce Blank Start!"
|
||||
@@ -1,81 +0,0 @@
|
||||
from unittest.mock import patch
|
||||
|
||||
from GramAddict.core.telepathic_engine import TelepathicEngine
|
||||
|
||||
FAILED_XML_PATH = "/Volumes/Alpha SSD/Coding/bot/debug/xml_dumps/manual_interrupt__2026-04-17_13-16-14.xml"
|
||||
|
||||
|
||||
def test_modal_guard_blocks_nav_intent_on_failed_xml():
|
||||
"""
|
||||
Test that the Modal Guard correctly identifies the bottom sheet in the failed XML
|
||||
and prevents searching for the 'Home Tab'.
|
||||
"""
|
||||
# Replace hardcoded dump file with inline XML containing a modal to prevent skipping
|
||||
xml_content = """<?xml version='1.0' encoding='UTF-8' standalone='yes' ?>
|
||||
<hierarchy rotation="0">
|
||||
<node index="0" resource-id="com.instagram.android:id/bottom_sheet_container" 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,900][1080,2400]" visible-to-user="true">
|
||||
<node index="0" text="Add comment" resource-id="com.instagram.android:id/comment_composer" class="android.widget.EditText" bounds="[42,2224][1038,2350]" visible-to-user="true" />
|
||||
</node>
|
||||
<node index="1" resource-id="com.instagram.android:id/tab_bar" bounds="[0,2200][1080,2400]" visible-to-user="false" />
|
||||
</hierarchy>"""
|
||||
|
||||
engine = TelepathicEngine()
|
||||
|
||||
# Intent that SHOULD be blocked because Home Tab is obscured by the comment sheet
|
||||
intent = "tap home tab"
|
||||
|
||||
# We don't want to trigger actual LLM/VLM calls during the test
|
||||
with patch("GramAddict.core.telepathic_engine.query_telepathic_llm") as mock_vlm:
|
||||
result = engine.find_best_node(xml_content, intent)
|
||||
|
||||
# 1. Verify that no VLM call was even attempted because the Modal Guard should have caught it early
|
||||
assert mock_vlm.called is False, "VLM should not be called when a modal obscures the target zone."
|
||||
|
||||
# 2. Result should be {'blocked_by_modal': True} (meaning 'Target blocked/missing')
|
||||
assert result == {
|
||||
"blocked_by_modal": True
|
||||
}, "Modal Guard should return block status for navigation intents when a sheet is open."
|
||||
|
||||
|
||||
def test_zone_enforcement_blocks_mid_screen_tab_hallucination():
|
||||
"""
|
||||
Tests that even if VLM is triggered (e.g. no modal detected but low confidence),
|
||||
any result for a 'tab' intent that is in the middle of the screen is rejected.
|
||||
"""
|
||||
engine = TelepathicEngine()
|
||||
|
||||
# Minimal XML
|
||||
xml = "<?xml version='1.0' ?><hierarchy><node index='0' text='Add comment' bounds='[42,2224][1038,2350]' visible-to-user='true' /></hierarchy>"
|
||||
|
||||
intent = "tap home tab"
|
||||
|
||||
# Mock VLM to return the 'Add comment' field which is at Y=2224 (0.917 of 2424)
|
||||
# Our guard enforces Tabs MUST be in the bottom 10% (Y > 0.90 * Height).
|
||||
# Wait, Y=2224 on H=2424 is 0.917. That IS in the bottom 10%.
|
||||
# Let's mock a node at Y=1000 (middle of screen) to test the guard.
|
||||
|
||||
with patch("GramAddict.core.telepathic_engine.query_telepathic_llm") as mock_vlm:
|
||||
# Mock VLM returning a middle-screen element (Index 0)
|
||||
mock_vlm.return_value = '{"index": 0, "reason": "hallucination test"}'
|
||||
|
||||
# Injected node at middle screen
|
||||
with patch.object(TelepathicEngine, "_extract_semantic_nodes") as mock_extract:
|
||||
mock_extract.return_value = [
|
||||
{
|
||||
"index": 0,
|
||||
"x": 500,
|
||||
"y": 1000,
|
||||
"width": 100,
|
||||
"height": 100,
|
||||
"area": 10000,
|
||||
"raw_bounds": "[450,950][550,1050]",
|
||||
"semantic_string": "text: 'Fake Home', id context: 'fake_tab'",
|
||||
}
|
||||
]
|
||||
|
||||
# We also need to patch _is_modal_active to False so it GETS to the VLM step
|
||||
with patch.object(TelepathicEngine, "_is_modal_active", return_value=False):
|
||||
result = engine.find_best_node(xml, intent)
|
||||
|
||||
# Should be rejected because navigation tabs should be in the nav bar zone
|
||||
assert result is None, "Structural Guard should reject mid-screen navigation tab candidates."
|
||||
@@ -1,71 +0,0 @@
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
from GramAddict.core.goap import GoalExecutor, ScreenType
|
||||
|
||||
|
||||
def test_goal_executor_prevents_infinite_tab_loops(monkeypatch):
|
||||
"""
|
||||
TDD Test: When attempting to navigate to a screen via a tab, if the tab
|
||||
does not lead to the required screen (e.g. reels_feed instead of follow_list),
|
||||
the GoalExecutor must NOT try the exact same tab action again in an endless loop.
|
||||
"""
|
||||
device = MagicMock()
|
||||
executor = GoalExecutor(device, "test_user")
|
||||
executor.planner.knowledge.wipe()
|
||||
|
||||
# We want to achieve "open following list", which requires ScreenType.FOLLOW_LIST
|
||||
# Currently on HOME_FEED
|
||||
# The heuristic might guess "tap reels tab" because of a fallback
|
||||
|
||||
# Track executed actions
|
||||
executed_actions = []
|
||||
|
||||
# Mock perceive to alternate between HOME_FEED and REELS_FEED
|
||||
# If we press a tab on HOME, we go to REELS.
|
||||
# If we press back on REELS, we go to HOME.
|
||||
current_screen = ScreenType.HOME_FEED
|
||||
|
||||
def fake_perceive(*args, **kwargs):
|
||||
if current_screen == ScreenType.HOME_FEED:
|
||||
return {
|
||||
"screen_type": ScreenType.HOME_FEED,
|
||||
"available_actions": ["tap reels tab", "tap explore tab"],
|
||||
"context": {},
|
||||
}
|
||||
else:
|
||||
return {"screen_type": ScreenType.REELS_FEED, "available_actions": ["press back"], "context": {}}
|
||||
|
||||
executor.perceive = MagicMock(side_effect=fake_perceive)
|
||||
|
||||
def fake_execute(action, **kwargs):
|
||||
nonlocal current_screen
|
||||
executed_actions.append(action)
|
||||
if action == "open following list" and current_screen == ScreenType.HOME_FEED:
|
||||
current_screen = ScreenType.REELS_FEED
|
||||
return True
|
||||
elif action == "press back" and current_screen == ScreenType.REELS_FEED:
|
||||
current_screen = ScreenType.HOME_FEED
|
||||
return True
|
||||
return False
|
||||
|
||||
monkeypatch.setattr(executor, "_execute_action", fake_execute)
|
||||
|
||||
# Speed up sleep
|
||||
monkeypatch.setattr("GramAddict.core.goap.random_sleep", lambda x, y: None)
|
||||
|
||||
# Execute the goal
|
||||
executor.max_steps = 10
|
||||
result = executor.achieve("open following list")
|
||||
|
||||
# Assert it failed (we never reached FOLLOW_LIST)
|
||||
assert result is False
|
||||
|
||||
# Assert we didn't loop endlessly.
|
||||
# Try 1: tap reels tab
|
||||
# Try 2: press back
|
||||
# Try 3: It should NOT try 'tap reels tab' again.
|
||||
|
||||
count_open_following = executed_actions.count("open following list")
|
||||
assert (
|
||||
count_open_following == 1
|
||||
), f"Bot is stuck in a loop! It tried to open following list {count_open_following} times."
|
||||
@@ -1,43 +0,0 @@
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
from GramAddict.core.goap import GoalExecutor, ScreenType
|
||||
|
||||
|
||||
def test_null_action_escalates_to_trap():
|
||||
"""
|
||||
TDD Test: Verify that when an action is executed but the screen state does not change
|
||||
(null-action / dead button), the GOAP loop tracks the failure and eventually burns
|
||||
the action as a trap to prevent infinite loops.
|
||||
"""
|
||||
device_mock = MagicMock()
|
||||
# Mock dump_hierarchy to simulate no UI change
|
||||
device_mock.dump_hierarchy.return_value = "<hierarchy></hierarchy>"
|
||||
|
||||
nav = GoalExecutor(device=device_mock, bot_username="test_user")
|
||||
|
||||
# Mock perceive to always return the SAME screen (EXPLORE_GRID)
|
||||
nav.perceive = MagicMock(
|
||||
return_value={"screen_type": ScreenType.EXPLORE_GRID, "available_actions": ["tap broken button"]}
|
||||
)
|
||||
|
||||
# Mock _execute_action to return False (which is what happens when ui_changed is False)
|
||||
nav._execute_action = MagicMock(return_value=False)
|
||||
|
||||
# Mock plan_next_step to repeatedly suggest the same action
|
||||
nav.planner.plan_next_step = MagicMock(return_value="tap broken button")
|
||||
|
||||
# Mock learn_trap to verify it gets called
|
||||
nav.planner.knowledge.learn_trap = MagicMock()
|
||||
|
||||
# Run a short loop (max 3 steps)
|
||||
nav.achieve(goal="open profile", max_steps=3)
|
||||
|
||||
# Verify that the action was executed multiple times
|
||||
assert nav._execute_action.call_count == 3
|
||||
|
||||
# The action failed on step 1 -> fail count = 1
|
||||
# The action failed on step 2 -> fail count = 2
|
||||
# At fail count 2, learn_trap MUST be called.
|
||||
nav.planner.knowledge.learn_trap.assert_called_with(
|
||||
ScreenType.EXPLORE_GRID, "tap broken button", "repeated_failure_or_null_action"
|
||||
)
|
||||
@@ -1,120 +0,0 @@
|
||||
"""
|
||||
TDD: Perception Module Tests.
|
||||
|
||||
Tests the extracted feed analysis functions in isolation.
|
||||
"""
|
||||
|
||||
|
||||
class TestFeedMarkers:
|
||||
"""FEED_MARKERS must correctly identify feed presence."""
|
||||
|
||||
def test_feed_markers_is_list(self):
|
||||
"""FEED_MARKERS must be a list."""
|
||||
from GramAddict.core.perception.feed_analysis import FEED_MARKERS
|
||||
|
||||
assert isinstance(FEED_MARKERS, list)
|
||||
assert len(FEED_MARKERS) >= 4
|
||||
|
||||
def test_has_feed_markers_detects_feed(self):
|
||||
"""has_feed_markers must return True when markers are present."""
|
||||
from GramAddict.core.perception.feed_analysis import has_feed_markers
|
||||
|
||||
xml = '<hierarchy><node resource-id="row_feed_photo_profile_name"/></hierarchy>'
|
||||
assert has_feed_markers(xml) is True
|
||||
|
||||
def test_has_feed_markers_detects_reels(self):
|
||||
"""has_feed_markers must detect Reels markers."""
|
||||
from GramAddict.core.perception.feed_analysis import has_feed_markers
|
||||
|
||||
xml = '<hierarchy><node resource-id="clips_media_component"/></hierarchy>'
|
||||
assert has_feed_markers(xml) is True
|
||||
|
||||
def test_has_feed_markers_rejects_empty(self):
|
||||
"""has_feed_markers must return False on empty/unrelated XML."""
|
||||
from GramAddict.core.perception.feed_analysis import has_feed_markers
|
||||
|
||||
assert has_feed_markers("") is False
|
||||
assert has_feed_markers("<hierarchy/>") is False
|
||||
|
||||
|
||||
class TestCarouselDetection:
|
||||
"""Carousel detection must use Instagram-specific resource IDs."""
|
||||
|
||||
def test_detects_carousel_indicator(self):
|
||||
"""Must detect carousel_page_indicator."""
|
||||
from GramAddict.core.perception.feed_analysis import has_carousel_in_view
|
||||
|
||||
xml = '<node resource-id="com.instagram.android:id/carousel_page_indicator"/>'
|
||||
assert has_carousel_in_view(xml) is True
|
||||
|
||||
def test_detects_carousel_group(self):
|
||||
"""Must detect carousel_media_group."""
|
||||
from GramAddict.core.perception.feed_analysis import has_carousel_in_view
|
||||
|
||||
xml = '<node resource-id="com.instagram.android:id/carousel_media_group"/>'
|
||||
assert has_carousel_in_view(xml) is True
|
||||
|
||||
def test_rejects_non_carousel(self):
|
||||
"""Must not false-positive on non-carousel XML."""
|
||||
from GramAddict.core.perception.feed_analysis import has_carousel_in_view
|
||||
|
||||
xml = '<node resource-id="com.instagram.android:id/action_bar_button"/>'
|
||||
assert has_carousel_in_view(xml) is False
|
||||
|
||||
|
||||
class TestExtractPostContent:
|
||||
"""Post content extraction must return valid dict structure."""
|
||||
|
||||
def test_returns_dict_with_required_keys(self):
|
||||
"""Must always return dict with username, description, caption."""
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from GramAddict.core.perception.feed_analysis import extract_post_content
|
||||
|
||||
with patch("GramAddict.core.telepathic_engine.TelepathicEngine.get_instance") as mock:
|
||||
instance = MagicMock()
|
||||
instance.find_best_node.return_value = None
|
||||
mock.return_value = instance
|
||||
|
||||
result = extract_post_content("<hierarchy/>")
|
||||
assert "username" in result
|
||||
assert "description" in result
|
||||
assert "caption" in result
|
||||
|
||||
def test_handles_garbage_xml_gracefully(self):
|
||||
"""Must not crash on corrupted XML."""
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from GramAddict.core.perception.feed_analysis import extract_post_content
|
||||
|
||||
with patch("GramAddict.core.telepathic_engine.TelepathicEngine.get_instance") as mock:
|
||||
instance = MagicMock()
|
||||
instance.find_best_node.return_value = None
|
||||
mock.return_value = instance
|
||||
|
||||
result = extract_post_content("garbage<<>>not xml at all")
|
||||
assert isinstance(result, dict)
|
||||
assert result["username"] == ""
|
||||
|
||||
|
||||
class TestBackwardCompatibility:
|
||||
"""bot_flow.py re-exports must work unchanged."""
|
||||
|
||||
def test_feed_markers_importable_from_bot_flow(self):
|
||||
"""FEED_MARKERS must be importable from bot_flow for existing tests."""
|
||||
from GramAddict.core.bot_flow import FEED_MARKERS
|
||||
|
||||
assert isinstance(FEED_MARKERS, list)
|
||||
assert len(FEED_MARKERS) >= 4
|
||||
|
||||
def test_has_carousel_importable_from_bot_flow(self):
|
||||
"""has_carousel_in_view must be importable from bot_flow."""
|
||||
from GramAddict.core.bot_flow import has_carousel_in_view
|
||||
|
||||
assert callable(has_carousel_in_view)
|
||||
|
||||
def test_extract_post_content_importable_from_bot_flow(self):
|
||||
"""_extract_post_content must be importable from bot_flow."""
|
||||
from GramAddict.core.bot_flow import _extract_post_content
|
||||
|
||||
assert callable(_extract_post_content)
|
||||
@@ -1,180 +0,0 @@
|
||||
"""
|
||||
TDD: Physics Module Tests.
|
||||
|
||||
Tests the extracted humanized input functions in isolation,
|
||||
verifying they produce valid device interactions via the
|
||||
biomechanical gesture pipeline (BezierGesture → SendEventInjector).
|
||||
|
||||
These tests mock the SendEventInjector at the injection boundary to
|
||||
validate that the humanized functions correctly generate gesture data
|
||||
and delegate to the injector.
|
||||
"""
|
||||
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from GramAddict.core.physics.biomechanics import PhysicsBody
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def reset_singletons():
|
||||
"""Reset singletons between tests for isolation."""
|
||||
PhysicsBody.reset()
|
||||
from GramAddict.core.physics.sendevent_injector import SendEventInjector
|
||||
|
||||
SendEventInjector.reset()
|
||||
yield
|
||||
PhysicsBody.reset()
|
||||
SendEventInjector.reset()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def device():
|
||||
"""Mock Android device."""
|
||||
dev = MagicMock()
|
||||
dev.get_info.return_value = {"displayWidth": 1080, "displayHeight": 2400}
|
||||
dev.cm_to_pixels.return_value = 5
|
||||
dev.shell = MagicMock(return_value="")
|
||||
return dev
|
||||
|
||||
|
||||
class TestHumanizedScroll:
|
||||
"""Scroll must produce valid gesture injection calls."""
|
||||
|
||||
@patch("GramAddict.core.physics.humanized_input.SendEventInjector")
|
||||
def test_scroll_calls_injector(self, MockInjector, device):
|
||||
"""Scroll must call the injector's inject_gesture method."""
|
||||
mock_injector = MagicMock()
|
||||
MockInjector.get_instance.return_value = mock_injector
|
||||
|
||||
from GramAddict.core.physics.humanized_input import humanized_scroll
|
||||
|
||||
humanized_scroll(device)
|
||||
|
||||
mock_injector.inject_gesture.assert_called()
|
||||
args = mock_injector.inject_gesture.call_args
|
||||
points = args[0][0]
|
||||
args[0][1]
|
||||
|
||||
# Validate gesture data structure
|
||||
assert len(points) >= 5, f"Expected at least 5 points, got {len(points)}"
|
||||
for p in points:
|
||||
assert len(p) == 3, f"Each point must be (x, y, pressure), got {p}"
|
||||
|
||||
@patch("GramAddict.core.physics.humanized_input.SendEventInjector")
|
||||
def test_scroll_coordinates_within_screen(self, MockInjector, device):
|
||||
"""All scroll coordinates must be within screen bounds."""
|
||||
mock_injector = MagicMock()
|
||||
MockInjector.get_instance.return_value = mock_injector
|
||||
|
||||
from GramAddict.core.physics.humanized_input import humanized_scroll
|
||||
|
||||
for _ in range(20):
|
||||
mock_injector.inject_gesture.reset_mock()
|
||||
humanized_scroll(device)
|
||||
|
||||
args = mock_injector.inject_gesture.call_args
|
||||
points = args[0][0]
|
||||
for x, y, p in points:
|
||||
assert 0 <= x <= 1200, f"x={x} out of bounds"
|
||||
assert 0 <= y <= 2500, f"y={y} out of bounds"
|
||||
|
||||
@patch("GramAddict.core.physics.humanized_input.SendEventInjector")
|
||||
def test_skip_scroll_calls_injector(self, MockInjector, device):
|
||||
"""Skip scroll should also use the injector."""
|
||||
import random
|
||||
|
||||
random.seed(42)
|
||||
mock_injector = MagicMock()
|
||||
MockInjector.get_instance.return_value = mock_injector
|
||||
|
||||
from GramAddict.core.physics.humanized_input import humanized_scroll
|
||||
|
||||
humanized_scroll(device, is_skip=True)
|
||||
|
||||
mock_injector.inject_gesture.assert_called()
|
||||
|
||||
|
||||
class TestHumanizedClick:
|
||||
"""Click must produce valid gesture injection calls."""
|
||||
|
||||
@patch("GramAddict.core.physics.humanized_input.SendEventInjector")
|
||||
def test_single_tap_calls_injector_once(self, MockInjector, device):
|
||||
"""Single tap should call injector exactly once."""
|
||||
mock_injector = MagicMock()
|
||||
MockInjector.get_instance.return_value = mock_injector
|
||||
|
||||
from GramAddict.core.physics.humanized_input import humanized_click
|
||||
|
||||
humanized_click(device, 500, 1200)
|
||||
|
||||
assert mock_injector.inject_gesture.call_count == 1
|
||||
|
||||
@patch("GramAddict.core.physics.humanized_input.SendEventInjector")
|
||||
def test_double_tap_calls_injector_twice(self, MockInjector, device):
|
||||
"""Double tap uses device.shell for timing-critical sub-300ms double-tap."""
|
||||
mock_injector = MagicMock()
|
||||
MockInjector.get_instance.return_value = mock_injector
|
||||
|
||||
from GramAddict.core.physics.humanized_input import humanized_click
|
||||
|
||||
humanized_click(device, 500, 1200, double=True)
|
||||
|
||||
# Double-tap bypasses SendEventInjector and uses shell for timing precision
|
||||
device.shell.assert_called_once()
|
||||
shell_cmd = device.shell.call_args[0][0]
|
||||
assert "input tap" in shell_cmd, f"Expected 'input tap' in shell command, got: {shell_cmd}"
|
||||
|
||||
@patch("GramAddict.core.physics.humanized_input.SendEventInjector")
|
||||
def test_tap_has_jitter(self, MockInjector, device):
|
||||
"""Taps should have slight jitter (not exact coordinates)."""
|
||||
import random
|
||||
|
||||
random.seed(1)
|
||||
mock_injector = MagicMock()
|
||||
MockInjector.get_instance.return_value = mock_injector
|
||||
|
||||
from GramAddict.core.physics.humanized_input import humanized_click
|
||||
|
||||
humanized_click(device, 500, 1200)
|
||||
|
||||
args = mock_injector.inject_gesture.call_args
|
||||
points = args[0][0]
|
||||
# First point should be near but not exactly (500, 1200)
|
||||
x, y, p = points[0]
|
||||
assert 475 <= x <= 525, f"Tap X {x} too far from target 500"
|
||||
assert 1175 <= y <= 1225, f"Tap Y {y} too far from target 1200"
|
||||
|
||||
|
||||
class TestHumanizedHorizontalSwipe:
|
||||
"""Horizontal swipe must produce valid gesture injection calls."""
|
||||
|
||||
@patch("GramAddict.core.physics.humanized_input.SendEventInjector")
|
||||
def test_horizontal_swipe_calls_injector(self, MockInjector, device):
|
||||
"""Horizontal swipe should call injector once."""
|
||||
mock_injector = MagicMock()
|
||||
MockInjector.get_instance.return_value = mock_injector
|
||||
|
||||
from GramAddict.core.physics.humanized_input import humanized_horizontal_swipe
|
||||
|
||||
humanized_horizontal_swipe(device, 800, 200, 1200, 250)
|
||||
|
||||
mock_injector.inject_gesture.assert_called_once()
|
||||
|
||||
@patch("GramAddict.core.physics.humanized_input.SendEventInjector")
|
||||
def test_horizontal_swipe_has_arc(self, MockInjector, device):
|
||||
"""Horizontal swipe points should have Y-axis variation (thumb arc)."""
|
||||
mock_injector = MagicMock()
|
||||
MockInjector.get_instance.return_value = mock_injector
|
||||
|
||||
from GramAddict.core.physics.humanized_input import humanized_horizontal_swipe
|
||||
|
||||
humanized_horizontal_swipe(device, 800, 200, 1200, 250)
|
||||
|
||||
args = mock_injector.inject_gesture.call_args
|
||||
points = args[0][0]
|
||||
ys = [p[1] for p in points]
|
||||
# Y values should NOT all be identical (thumb arc produces variation)
|
||||
unique_ys = set(ys)
|
||||
assert len(unique_ys) > 1, "Expected Y-axis variation from thumb arc"
|
||||
@@ -1,417 +0,0 @@
|
||||
"""
|
||||
TDD: Plugin Architecture Tests.
|
||||
|
||||
Tests the BehaviorPlugin base class, PluginRegistry, and the
|
||||
first concrete plugin (CarouselBrowsingPlugin).
|
||||
|
||||
Covers:
|
||||
- Plugin lifecycle (register, unregister, activate, execute)
|
||||
- Priority ordering
|
||||
- Exclusive plugin chain-breaking
|
||||
- Duplicate registration prevention
|
||||
- Error isolation (plugin crashes don't cascade)
|
||||
- CarouselBrowsingPlugin activation and execution
|
||||
"""
|
||||
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from GramAddict.core.behaviors import (
|
||||
BehaviorContext,
|
||||
BehaviorPlugin,
|
||||
BehaviorResult,
|
||||
PluginRegistry,
|
||||
)
|
||||
|
||||
# ── Test Plugins ──
|
||||
|
||||
|
||||
class AlwaysActivePlugin(BehaviorPlugin):
|
||||
@property
|
||||
def name(self):
|
||||
return "always_active"
|
||||
|
||||
@property
|
||||
def priority(self):
|
||||
return 50
|
||||
|
||||
def can_activate(self, ctx):
|
||||
return True
|
||||
|
||||
def execute(self, ctx):
|
||||
return BehaviorResult(executed=True, interactions=1)
|
||||
|
||||
|
||||
class NeverActivePlugin(BehaviorPlugin):
|
||||
@property
|
||||
def name(self):
|
||||
return "never_active"
|
||||
|
||||
@property
|
||||
def priority(self):
|
||||
return 50
|
||||
|
||||
def can_activate(self, ctx):
|
||||
return False
|
||||
|
||||
def execute(self, ctx):
|
||||
return BehaviorResult(executed=True)
|
||||
|
||||
|
||||
class HighPriorityPlugin(BehaviorPlugin):
|
||||
@property
|
||||
def name(self):
|
||||
return "high_priority"
|
||||
|
||||
@property
|
||||
def priority(self):
|
||||
return 100
|
||||
|
||||
def can_activate(self, ctx):
|
||||
return True
|
||||
|
||||
def execute(self, ctx):
|
||||
return BehaviorResult(executed=True, metadata={"order": "first"})
|
||||
|
||||
|
||||
class LowPriorityPlugin(BehaviorPlugin):
|
||||
@property
|
||||
def name(self):
|
||||
return "low_priority"
|
||||
|
||||
@property
|
||||
def priority(self):
|
||||
return 10
|
||||
|
||||
def can_activate(self, ctx):
|
||||
return True
|
||||
|
||||
def execute(self, ctx):
|
||||
return BehaviorResult(executed=True, metadata={"order": "last"})
|
||||
|
||||
|
||||
class ExclusiveGuardPlugin(BehaviorPlugin):
|
||||
@property
|
||||
def name(self):
|
||||
return "ad_guard"
|
||||
|
||||
@property
|
||||
def priority(self):
|
||||
return 100
|
||||
|
||||
@property
|
||||
def exclusive(self):
|
||||
return True
|
||||
|
||||
def can_activate(self, ctx):
|
||||
return "sponsored" in (ctx.context_xml or "")
|
||||
|
||||
def execute(self, ctx):
|
||||
return BehaviorResult(executed=True, should_skip=True)
|
||||
|
||||
|
||||
class CrashingPlugin(BehaviorPlugin):
|
||||
@property
|
||||
def name(self):
|
||||
return "crasher"
|
||||
|
||||
@property
|
||||
def priority(self):
|
||||
return 50
|
||||
|
||||
def can_activate(self, ctx):
|
||||
return True
|
||||
|
||||
def execute(self, ctx):
|
||||
raise RuntimeError("Plugin exploded!")
|
||||
|
||||
|
||||
# ── Fixtures ──
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def registry():
|
||||
PluginRegistry.reset()
|
||||
reg = PluginRegistry()
|
||||
yield reg
|
||||
PluginRegistry.reset()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def ctx():
|
||||
"""Minimal BehaviorContext for testing."""
|
||||
device = MagicMock()
|
||||
device.get_info.return_value = {"displayWidth": 1080, "displayHeight": 2400}
|
||||
device.shell = MagicMock()
|
||||
|
||||
configs = MagicMock()
|
||||
configs.args = MagicMock()
|
||||
configs.args.carousel_percentage = "50"
|
||||
configs.args.carousel_count = "2-4"
|
||||
|
||||
session_state = MagicMock()
|
||||
|
||||
return BehaviorContext(
|
||||
device=device,
|
||||
configs=configs,
|
||||
session_state=session_state,
|
||||
cognitive_stack={},
|
||||
context_xml='<hierarchy><node resource-id="com.instagram.android:id/row_feed_photo_profile_name"/></hierarchy>',
|
||||
sleep_mod=1.0,
|
||||
)
|
||||
|
||||
|
||||
# ── Registry Tests ──
|
||||
|
||||
|
||||
class TestPluginRegistry:
|
||||
"""Registry must manage plugins correctly."""
|
||||
|
||||
def test_register_adds_plugin(self, registry):
|
||||
registry.register(AlwaysActivePlugin())
|
||||
assert len(registry) == 1
|
||||
|
||||
def test_duplicate_registration_ignored(self, registry):
|
||||
registry.register(AlwaysActivePlugin())
|
||||
registry.register(AlwaysActivePlugin())
|
||||
assert len(registry) == 1
|
||||
|
||||
def test_unregister_removes_plugin(self, registry):
|
||||
registry.register(AlwaysActivePlugin())
|
||||
registry.unregister("always_active")
|
||||
assert len(registry) == 0
|
||||
|
||||
def test_unregister_nonexistent_is_noop(self, registry):
|
||||
registry.unregister("nonexistent")
|
||||
assert len(registry) == 0
|
||||
|
||||
def test_contains_check(self, registry):
|
||||
registry.register(AlwaysActivePlugin())
|
||||
assert "always_active" in registry
|
||||
assert "nonexistent" not in registry
|
||||
|
||||
def test_plugins_sorted_by_priority(self, registry):
|
||||
registry.register(LowPriorityPlugin())
|
||||
registry.register(HighPriorityPlugin())
|
||||
|
||||
plugins = registry.plugins
|
||||
assert plugins[0].name == "high_priority"
|
||||
assert plugins[1].name == "low_priority"
|
||||
|
||||
def test_get_active_plugins_filters(self, registry, ctx):
|
||||
registry.register(AlwaysActivePlugin())
|
||||
registry.register(NeverActivePlugin())
|
||||
|
||||
active = registry.get_active_plugins(ctx)
|
||||
assert len(active) == 1
|
||||
assert active[0].name == "always_active"
|
||||
|
||||
def test_singleton_pattern(self):
|
||||
PluginRegistry.reset()
|
||||
r1 = PluginRegistry.get_instance()
|
||||
r2 = PluginRegistry.get_instance()
|
||||
assert r1 is r2
|
||||
PluginRegistry.reset()
|
||||
|
||||
def test_reset_clears_singleton(self):
|
||||
PluginRegistry.reset()
|
||||
r1 = PluginRegistry.get_instance()
|
||||
PluginRegistry.reset()
|
||||
r2 = PluginRegistry.get_instance()
|
||||
assert r1 is not r2
|
||||
PluginRegistry.reset()
|
||||
|
||||
|
||||
# ── Execution Tests ──
|
||||
|
||||
|
||||
class TestPluginExecution:
|
||||
"""Plugin execution lifecycle."""
|
||||
|
||||
def test_execute_all_runs_active_plugins(self, registry, ctx):
|
||||
registry.register(AlwaysActivePlugin())
|
||||
results = registry.execute_all(ctx)
|
||||
assert len(results) == 1
|
||||
assert results[0].executed is True
|
||||
|
||||
def test_execute_all_skips_inactive_plugins(self, registry, ctx):
|
||||
registry.register(NeverActivePlugin())
|
||||
results = registry.execute_all(ctx)
|
||||
assert len(results) == 0
|
||||
|
||||
def test_execute_respects_priority_order(self, registry, ctx):
|
||||
registry.register(LowPriorityPlugin())
|
||||
registry.register(HighPriorityPlugin())
|
||||
|
||||
results = registry.execute_all(ctx)
|
||||
assert len(results) == 2
|
||||
assert results[0].metadata["order"] == "first"
|
||||
assert results[1].metadata["order"] == "last"
|
||||
|
||||
def test_exclusive_plugin_stops_chain(self, registry, ctx):
|
||||
ctx.context_xml = "sponsored content here"
|
||||
registry.register(ExclusiveGuardPlugin())
|
||||
registry.register(AlwaysActivePlugin()) # Lower priority
|
||||
|
||||
results = registry.execute_all(ctx)
|
||||
# Only the guard should have run (it's exclusive)
|
||||
assert len(results) == 1
|
||||
assert results[0].should_skip is True
|
||||
|
||||
def test_crashing_plugin_doesnt_cascade(self, registry, ctx):
|
||||
"""A crashing plugin must not break other plugins."""
|
||||
registry.register(CrashingPlugin())
|
||||
|
||||
# Must NOT raise
|
||||
results = registry.execute_all(ctx)
|
||||
assert len(results) == 1
|
||||
assert results[0].executed is False
|
||||
assert "error" in results[0].metadata
|
||||
|
||||
|
||||
# ── BehaviorContext Tests ──
|
||||
|
||||
|
||||
class TestBehaviorContext:
|
||||
"""BehaviorContext construction."""
|
||||
|
||||
def test_default_values(self):
|
||||
ctx = BehaviorContext(
|
||||
device=MagicMock(),
|
||||
configs=MagicMock(),
|
||||
session_state=MagicMock(),
|
||||
cognitive_stack={},
|
||||
)
|
||||
assert ctx.context_xml == ""
|
||||
assert ctx.sleep_mod == 1.0
|
||||
assert ctx.post_data is None
|
||||
assert ctx.username == ""
|
||||
|
||||
|
||||
# ── BehaviorResult Tests ──
|
||||
|
||||
|
||||
class TestBehaviorResult:
|
||||
"""BehaviorResult defaults."""
|
||||
|
||||
def test_default_result(self):
|
||||
result = BehaviorResult()
|
||||
assert result.executed is False
|
||||
assert result.should_continue is True
|
||||
assert result.should_skip is False
|
||||
assert result.interactions == 0
|
||||
assert result.metadata == {}
|
||||
|
||||
def test_result_with_metadata(self):
|
||||
result = BehaviorResult(executed=True, interactions=3, metadata={"slides_viewed": 3})
|
||||
assert result.interactions == 3
|
||||
assert result.metadata["slides_viewed"] == 3
|
||||
|
||||
|
||||
# ── CarouselBrowsingPlugin Tests ──
|
||||
|
||||
|
||||
class TestCarouselBrowsingPlugin:
|
||||
"""Concrete plugin: carousel browsing."""
|
||||
|
||||
@pytest.fixture
|
||||
def carousel_ctx(self, ctx):
|
||||
"""Context with carousel indicators."""
|
||||
ctx.context_xml = (
|
||||
"<hierarchy>"
|
||||
'<node resource-id="com.instagram.android:id/carousel_page_indicator"/>'
|
||||
'<node resource-id="com.instagram.android:id/row_feed_photo_profile_name"/>'
|
||||
"</hierarchy>"
|
||||
)
|
||||
return ctx
|
||||
|
||||
def test_activates_on_carousel(self, carousel_ctx):
|
||||
from GramAddict.core.behaviors.carousel_browsing import CarouselBrowsingPlugin
|
||||
|
||||
plugin = CarouselBrowsingPlugin()
|
||||
assert plugin.can_activate(carousel_ctx) is True
|
||||
|
||||
def test_does_not_activate_without_carousel(self, ctx):
|
||||
from GramAddict.core.behaviors.carousel_browsing import CarouselBrowsingPlugin
|
||||
|
||||
plugin = CarouselBrowsingPlugin()
|
||||
assert plugin.can_activate(ctx) is False
|
||||
|
||||
def test_does_not_activate_with_zero_percentage(self, carousel_ctx):
|
||||
from GramAddict.core.behaviors.carousel_browsing import CarouselBrowsingPlugin
|
||||
|
||||
carousel_ctx.configs.args.carousel_percentage = "0"
|
||||
plugin = CarouselBrowsingPlugin()
|
||||
assert plugin.can_activate(carousel_ctx) is False
|
||||
|
||||
def test_execute_sends_swipe_commands(self, carousel_ctx):
|
||||
import random
|
||||
|
||||
from GramAddict.core.behaviors.carousel_browsing import CarouselBrowsingPlugin
|
||||
|
||||
random.seed(42)
|
||||
|
||||
carousel_ctx.configs.args.carousel_percentage = "100"
|
||||
carousel_ctx.configs.args.carousel_count = "2-2"
|
||||
|
||||
plugin = CarouselBrowsingPlugin()
|
||||
|
||||
with patch("GramAddict.core.behaviors.carousel_browsing.sleep"):
|
||||
result = plugin.execute(carousel_ctx)
|
||||
|
||||
assert result.executed is True
|
||||
assert result.interactions == 2
|
||||
assert result.metadata["slides_viewed"] == 2
|
||||
# Should have sent 2 swipe commands (exclude SendEventInjector detection calls)
|
||||
swipe_calls = [c for c in carousel_ctx.device.shell.call_args_list if "input swipe" in str(c)]
|
||||
assert len(swipe_calls) == 2
|
||||
|
||||
def test_plugin_name_and_priority(self):
|
||||
from GramAddict.core.behaviors.carousel_browsing import CarouselBrowsingPlugin
|
||||
|
||||
plugin = CarouselBrowsingPlugin()
|
||||
assert plugin.name == "carousel_browsing"
|
||||
assert plugin.priority == 20
|
||||
assert plugin.exclusive is False
|
||||
|
||||
def test_execute_probabilistic_skip(self, carousel_ctx):
|
||||
"""When random > carousel_pct, plugin should not execute."""
|
||||
import random
|
||||
|
||||
from GramAddict.core.behaviors.carousel_browsing import CarouselBrowsingPlugin
|
||||
|
||||
carousel_ctx.configs.args.carousel_percentage = "1" # 1% chance
|
||||
plugin = CarouselBrowsingPlugin()
|
||||
|
||||
# Force random to return high value
|
||||
random.seed(0)
|
||||
with patch("GramAddict.core.behaviors.carousel_browsing.sleep"):
|
||||
# Run many times — most should skip
|
||||
executed_count = 0
|
||||
for _ in range(100):
|
||||
result = plugin.execute(carousel_ctx)
|
||||
if result.executed:
|
||||
executed_count += 1
|
||||
|
||||
# With 1% chance, we expect very few executions
|
||||
assert executed_count < 20
|
||||
|
||||
def test_full_registration_and_execution(self, carousel_ctx):
|
||||
"""End-to-end: register plugin, execute via registry."""
|
||||
from GramAddict.core.behaviors.carousel_browsing import CarouselBrowsingPlugin
|
||||
|
||||
PluginRegistry.reset()
|
||||
registry = PluginRegistry()
|
||||
registry.register(CarouselBrowsingPlugin())
|
||||
|
||||
carousel_ctx.configs.args.carousel_percentage = "100"
|
||||
carousel_ctx.configs.args.carousel_count = "1-1"
|
||||
|
||||
with patch("GramAddict.core.behaviors.carousel_browsing.sleep"):
|
||||
results = registry.execute_all(carousel_ctx)
|
||||
|
||||
assert len(results) == 1
|
||||
assert results[0].executed is True
|
||||
|
||||
PluginRegistry.reset()
|
||||
@@ -1,24 +0,0 @@
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
from GramAddict.core.bot_flow import _wait_for_profile_loaded
|
||||
|
||||
|
||||
def test_wait_for_profile_loaded_success():
|
||||
device = MagicMock()
|
||||
# First dump is empty, second dump has profile_header
|
||||
device.dump_hierarchy.side_effect = [
|
||||
"<hierarchy></hierarchy>",
|
||||
'<hierarchy><node resource-id="com.instagram.android:id/profile_header" /></hierarchy>',
|
||||
]
|
||||
|
||||
assert _wait_for_profile_loaded(device, timeout=2)
|
||||
assert device.dump_hierarchy.call_count == 2
|
||||
|
||||
|
||||
def test_wait_for_profile_loaded_timeout():
|
||||
device = MagicMock()
|
||||
# Always reel
|
||||
device.dump_hierarchy.return_value = '<hierarchy><node resource-id="clips_video_container" /></hierarchy>'
|
||||
|
||||
assert not _wait_for_profile_loaded(device, timeout=1)
|
||||
assert device.dump_hierarchy.call_count >= 1
|
||||
@@ -1,102 +0,0 @@
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
from GramAddict.core.goap import NavigationKnowledge, ScreenType
|
||||
from GramAddict.core.qdrant_memory import QdrantBase
|
||||
|
||||
|
||||
def test_qdrant_semantic_overlap_prevention():
|
||||
"""
|
||||
TDD Test: Ensures that get_tab_for_screen and get_requirements
|
||||
do not suffer from vector similarity overlap. They must use exact payload matching.
|
||||
"""
|
||||
# 1. Setup real NavigationKnowledge but with a mocked DB connection
|
||||
knowledge = NavigationKnowledge("testuser")
|
||||
mock_db = MagicMock(spec=QdrantBase)
|
||||
mock_db.is_connected = True
|
||||
mock_db.collection_name = "test_nav_knowledge"
|
||||
mock_client = MagicMock()
|
||||
mock_db.client = mock_client
|
||||
knowledge._db = mock_db
|
||||
|
||||
# 2. Simulate the bug: if the system used query_points (embedding search)
|
||||
# The client shouldn't receive a query_points call.
|
||||
# It SHOULD receive a scroll call with a FieldCondition.
|
||||
|
||||
# Mock scroll to return an empty result (not found)
|
||||
mock_client.scroll.return_value = ([], None)
|
||||
|
||||
# Execute
|
||||
result_action = knowledge.get_action_for_screen(ScreenType.OWN_PROFILE)
|
||||
|
||||
# Assert it returns None when there is no exact match
|
||||
assert result_action is None
|
||||
|
||||
# Verify scroll was called with correct filter, NOT query_points
|
||||
assert mock_client.scroll.called, "Must use client.scroll for exact matching, not query_points"
|
||||
assert not mock_client.query_points.called, "query_points should not be used due to semantic overlap risks"
|
||||
|
||||
# Verify the scroll filter checks for "result_screen" == "OWN_PROFILE"
|
||||
call_args = mock_client.scroll.call_args[1]
|
||||
scroll_filter = call_args.get("scroll_filter")
|
||||
assert scroll_filter is not None
|
||||
assert scroll_filter.must[0].key == "result_screen"
|
||||
assert scroll_filter.must[0].match.value == "OWN_PROFILE"
|
||||
|
||||
|
||||
def test_qdrant_semantic_overlap_prevention_requirements():
|
||||
"""
|
||||
TDD Test: Ensures that get_requirements uses exact matching.
|
||||
"""
|
||||
knowledge = NavigationKnowledge("testuser")
|
||||
mock_db = MagicMock(spec=QdrantBase)
|
||||
mock_db.is_connected = True
|
||||
mock_db.collection_name = "test_nav_knowledge"
|
||||
mock_client = MagicMock()
|
||||
mock_db.client = mock_client
|
||||
knowledge._db = mock_db
|
||||
|
||||
mock_client.scroll.return_value = ([], None)
|
||||
|
||||
requirements = knowledge.get_requirements("open profile")
|
||||
|
||||
assert requirements == []
|
||||
|
||||
assert mock_client.scroll.called
|
||||
assert not mock_client.query_points.called
|
||||
|
||||
call_args = mock_client.scroll.call_args[1]
|
||||
scroll_filter = call_args.get("scroll_filter")
|
||||
assert scroll_filter.must[0].key == "goal"
|
||||
assert scroll_filter.must[0].match.value == "open profile"
|
||||
|
||||
|
||||
def test_qdrant_semantic_overlap_prevention_path_memory():
|
||||
"""
|
||||
TDD Test: Ensures that PathMemory.recall_path uses a strict query_filter
|
||||
on the `start_screen` payload field so it doesn't recall a path that is
|
||||
semantically related but for the wrong screen.
|
||||
"""
|
||||
from GramAddict.core.goap import PathMemory
|
||||
|
||||
memory = PathMemory("testuser")
|
||||
mock_db = MagicMock(spec=QdrantBase)
|
||||
mock_db.is_connected = True
|
||||
mock_db.collection_name = "test_path_memory"
|
||||
mock_client = MagicMock()
|
||||
mock_db.client = mock_client
|
||||
memory._db = mock_db
|
||||
|
||||
mock_client.query_points.return_value = MagicMock(points=[])
|
||||
mock_db._get_embedding.return_value = [0.1] * 768
|
||||
|
||||
# Execute
|
||||
path = memory.recall_path("like post", "EXPLORE_GRID")
|
||||
|
||||
assert path is None
|
||||
assert mock_client.query_points.called
|
||||
|
||||
call_args = mock_client.query_points.call_args[1]
|
||||
query_filter = call_args.get("query_filter")
|
||||
assert query_filter is not None
|
||||
assert query_filter.must[0].key == "start_screen"
|
||||
assert query_filter.must[0].match.value == "EXPLORE_GRID"
|
||||
@@ -1,192 +0,0 @@
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_device():
|
||||
from GramAddict.core.physics.biomechanics import PhysicsBody
|
||||
from GramAddict.core.physics.sendevent_injector import SendEventInjector
|
||||
|
||||
PhysicsBody.reset()
|
||||
SendEventInjector.reset()
|
||||
|
||||
device = MagicMock()
|
||||
device.deviceV2 = MagicMock()
|
||||
device.get_info.return_value = {"displayWidth": 1080, "displayHeight": 2400}
|
||||
device.app_id = "com.instagram.android"
|
||||
device.shell = MagicMock(return_value="")
|
||||
|
||||
yield device
|
||||
|
||||
PhysicsBody.reset()
|
||||
SendEventInjector.reset()
|
||||
|
||||
|
||||
def test_reels_loop_repost_execution(mock_device):
|
||||
"""
|
||||
TDD Test: Verifies that the bot attempts to repost a Reel if resonance is high
|
||||
and repost_percentage allows it.
|
||||
"""
|
||||
# 1. Setup Cognitive Stack & Configs
|
||||
mock_cognitive_stack = {
|
||||
"dopamine": MagicMock(),
|
||||
"resonance": MagicMock(),
|
||||
"zero_engine": MagicMock(),
|
||||
"nav_graph": MagicMock(),
|
||||
"telepathic": MagicMock(),
|
||||
"growth_brain": MagicMock(),
|
||||
"darwin": MagicMock(),
|
||||
"active_inference": MagicMock(),
|
||||
"swarm": MagicMock(),
|
||||
"radome": MagicMock(),
|
||||
"crm": MagicMock(),
|
||||
}
|
||||
|
||||
# Configure growth_brain to allow repost, disallow comment/double-tap
|
||||
mock_cognitive_stack["growth_brain"].wants_to_double_tap.return_value = False
|
||||
mock_cognitive_stack["growth_brain"].wants_to_repost.return_value = True
|
||||
mock_cognitive_stack["growth_brain"].wants_to_comment.return_value = False
|
||||
mock_cognitive_stack["growth_brain"].evaluate_hesitation.return_value = False
|
||||
mock_cognitive_stack["radome"].sanitize_xml.side_effect = lambda x: x
|
||||
|
||||
# Simulate a single post interaction then exit
|
||||
mock_cognitive_stack["dopamine"].is_app_session_over.side_effect = [False, True]
|
||||
mock_cognitive_stack["dopamine"].wants_to_change_feed.return_value = False
|
||||
mock_cognitive_stack["dopamine"].wants_to_doomscroll.return_value = False
|
||||
|
||||
# High resonance to trigger repost
|
||||
mock_cognitive_stack["resonance"].calculate_resonance.return_value = 0.95
|
||||
|
||||
configs = MagicMock()
|
||||
configs.args.repost_percentage = 100
|
||||
configs.args.interact_percentage = 100
|
||||
configs.args.likes_percentage = 0
|
||||
configs.args.comment_percentage = 0
|
||||
configs.args.follow_percentage = 0
|
||||
configs.args.visit_profiles = 0
|
||||
|
||||
session_state = MagicMock()
|
||||
session_state.check_limit.return_value = (False, False, False, False)
|
||||
|
||||
# 2. Mock Reels UI
|
||||
# Reels usually have 'clips_viewer' or similar in the hierarchy
|
||||
reels_xml = """<?xml version='1.0' ?>
|
||||
<hierarchy>
|
||||
<node resource-id="com.instagram.android:id/row_feed_profile_header" bounds="[0,200][1080,300]" />
|
||||
<node resource-id="com.instagram.android:id/clips_author_username" text="test_user" />
|
||||
<node resource-id="com.instagram.android:id/clips_media_component" content-desc="Check out this cool reel #repost #viral" />
|
||||
<node resource-id="com.instagram.android:id/clips_video_container" />
|
||||
<node resource-id="com.instagram.android:id/direct_share_button" content-desc="Share" />
|
||||
</hierarchy>"""
|
||||
|
||||
mock_device.dump_hierarchy.return_value = reels_xml
|
||||
mock_device.dump_hierarchy.return_value = reels_xml
|
||||
|
||||
# Repost Sheet XML
|
||||
repost_sheet_xml = """<?xml version='1.0' ?>
|
||||
<hierarchy>
|
||||
<node resource-id="com.instagram.android:id/bottom_sheet_container">
|
||||
<node text="Repost" content-desc="Repost interaction button with two arrows" />
|
||||
</node>
|
||||
</hierarchy>"""
|
||||
|
||||
# 3. Setup Telepathic Engine Mocks
|
||||
mock_telepathic = mock_cognitive_stack["telepathic"]
|
||||
# First call: find interaction buttons
|
||||
mock_telepathic._extract_semantic_nodes.return_value = [{"x": 100, "y": 100, "semantic_string": "share button"}]
|
||||
|
||||
# Logic for finding nodes — return proper attributes for content extraction
|
||||
def mock_find_best_node(xml, intent, **kwargs):
|
||||
intent_lower = intent.lower() if isinstance(intent, str) else ""
|
||||
if "repost" in intent_lower:
|
||||
return {
|
||||
"x": 500,
|
||||
"y": 2000,
|
||||
"bounds": "[400,1950][600,2050]",
|
||||
"skip": False,
|
||||
"original_attribs": {"text": "Repost", "desc": ""},
|
||||
}
|
||||
if "author" in intent_lower or "username" in intent_lower:
|
||||
return {
|
||||
"x": 100,
|
||||
"y": 250,
|
||||
"text": "test_user",
|
||||
"content-desc": "",
|
||||
"bounds": "[0,200][1080,300]",
|
||||
"skip": False,
|
||||
"original_attribs": {"text": "test_user", "desc": ""},
|
||||
}
|
||||
if "media" in intent_lower or "image" in intent_lower or "video" in intent_lower:
|
||||
return {
|
||||
"x": 540,
|
||||
"y": 1200,
|
||||
"text": "",
|
||||
"content-desc": "Check out this cool reel #repost #viral",
|
||||
"bounds": "[0,300][1080,2100]",
|
||||
"skip": False,
|
||||
"original_attribs": {"text": "", "desc": "Check out this cool reel #repost #viral"},
|
||||
}
|
||||
return {"x": 100, "y": 100, "bounds": "[90,90][110,110]", "skip": False}
|
||||
|
||||
mock_telepathic.find_best_node.side_effect = mock_find_best_node
|
||||
|
||||
# Simulate share button transition success
|
||||
mock_cognitive_stack["nav_graph"]._execute_transition.return_value = True
|
||||
|
||||
# 4. Execute Feed Loop for Reels
|
||||
with (
|
||||
patch("GramAddict.core.bot_flow.TelepathicEngine") as MockEngine,
|
||||
patch("GramAddict.core.telepathic_engine.TelepathicEngine.get_instance", return_value=mock_telepathic),
|
||||
patch("GramAddict.core.bot_flow._humanized_click") as mock_click,
|
||||
patch("GramAddict.core.bot_flow.sleep"),
|
||||
patch("GramAddict.core.physics.timing.sleep"),
|
||||
patch("GramAddict.core.bot_flow.dump_ui_state"),
|
||||
):
|
||||
MockEngine.get_instance.return_value = mock_telepathic
|
||||
|
||||
# Resilient state-based mock for dump_hierarchy
|
||||
comment_sheet_xml = """<?xml version='1.0' ?><hierarchy><node resource-id="com.instagram.android:id/layout_comment_thread" /><node resource-id="com.instagram.android:id/bottom_sheet_container" /></hierarchy>"""
|
||||
|
||||
state = {"current": reels_xml}
|
||||
|
||||
def side_effect_func(*args, **kwargs):
|
||||
return state["current"]
|
||||
|
||||
mock_device.dump_hierarchy.side_effect = side_effect_func
|
||||
|
||||
# We need to change the state when transition is called
|
||||
mock_cognitive_stack["nav_graph"]._execute_transition
|
||||
|
||||
def mocked_execute(transition_name):
|
||||
if transition_name == "tap_comment_button":
|
||||
state["current"] = comment_sheet_xml
|
||||
elif transition_name == "tap_share_button":
|
||||
state["current"] = repost_sheet_xml
|
||||
return True
|
||||
|
||||
mock_cognitive_stack["nav_graph"]._execute_transition.side_effect = mocked_execute
|
||||
|
||||
from GramAddict.core.bot_flow import _run_zero_latency_feed_loop
|
||||
|
||||
_run_zero_latency_feed_loop(
|
||||
mock_device,
|
||||
mock_cognitive_stack["zero_engine"],
|
||||
mock_cognitive_stack["nav_graph"],
|
||||
configs,
|
||||
session_state,
|
||||
"ReelsFeed",
|
||||
mock_cognitive_stack,
|
||||
is_reels=True,
|
||||
)
|
||||
|
||||
# 5. Assertions
|
||||
# Should click the share button (via transition) and the repost button (direct click)
|
||||
assert mock_cognitive_stack["nav_graph"]._execute_transition.called_with("tap_share_button")
|
||||
|
||||
# Check if _humanized_click was called for the Repost button (x=500, y=2000)
|
||||
click_args = [call.args for call in mock_click.call_args_list]
|
||||
repost_clicked = any(args[1] == 500 and args[2] == 2000 for args in click_args)
|
||||
|
||||
assert repost_clicked, "Repost button was not clicked on Reels share sheet"
|
||||
assert mock_telepathic.confirm_click.called_with("Repost interaction button with two arrows")
|
||||
@@ -1,40 +0,0 @@
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from GramAddict.core.bot_flow import _run_zero_latency_feed_loop
|
||||
|
||||
|
||||
def test_run_zero_latency_feed_loop_name_error_regression():
|
||||
"""
|
||||
TDD RED PHASE: This test should FAIL with NameError: name '_detect_ad_structural' is not defined.
|
||||
"""
|
||||
mock_device = MagicMock()
|
||||
mock_device.dump_hierarchy.return_value = "<?xml version='1.0' ?><hierarchy><node text='test'/></hierarchy>"
|
||||
|
||||
# Mock DopamineEngine
|
||||
mock_dopamine = MagicMock()
|
||||
mock_dopamine.is_app_session_over.side_effect = [False, True] # Run once then exit
|
||||
mock_dopamine.wants_to_change_feed.return_value = False
|
||||
mock_dopamine.wants_to_doomscroll.return_value = False
|
||||
|
||||
mock_stack = {"dopamine": mock_dopamine, "radome": MagicMock(), "ai": MagicMock(), "growth": MagicMock()}
|
||||
|
||||
mock_session_state = MagicMock()
|
||||
mock_session_state.check_limit.return_value = (False,) # any() will be False
|
||||
|
||||
# We want it to reach line 896 where _detect_ad_structural is called
|
||||
mock_zero_engine = MagicMock()
|
||||
mock_nav_graph = MagicMock()
|
||||
mock_configs = MagicMock()
|
||||
|
||||
# Mocking globals
|
||||
with (
|
||||
patch("GramAddict.core.bot_flow._humanized_scroll"),
|
||||
patch("GramAddict.core.bot_flow.logger"),
|
||||
patch("GramAddict.core.bot_flow.random_sleep"),
|
||||
patch("GramAddict.core.bot_flow.ResonanceEngine"),
|
||||
patch("GramAddict.core.bot_flow.ZeroLatencyEngine"),
|
||||
):
|
||||
# Should now run without NameError
|
||||
_run_zero_latency_feed_loop(
|
||||
mock_device, mock_zero_engine, mock_nav_graph, mock_configs, mock_session_state, "ExploreFeed", mock_stack
|
||||
)
|
||||
@@ -1,53 +0,0 @@
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
from GramAddict.core.resonance_engine import ResonanceEngine
|
||||
|
||||
|
||||
def test_resonance_engine_bootstraps_persona_from_config(monkeypatch):
|
||||
"""
|
||||
TDD Test: When the ResonanceEngine is instantiated with persona_interests,
|
||||
it must successfully compute a persona vector and be able to calculate
|
||||
meaningful resonance scores (> 0.5 default) for matching content.
|
||||
"""
|
||||
# Mock the Qdrant DBs
|
||||
mock_content_db = MagicMock()
|
||||
mock_persona_db = MagicMock()
|
||||
|
||||
# Simulate a vector embedding
|
||||
def fake_get_embedding(text):
|
||||
if not text:
|
||||
return None
|
||||
# Return a dummy vector
|
||||
return [0.1] * 768
|
||||
|
||||
mock_content_db._get_embedding = MagicMock(side_effect=fake_get_embedding)
|
||||
|
||||
# We will simulate cosine similarity calculation.
|
||||
# Since both will be [0.1]*768, similarity would be 1.0.
|
||||
def fake_calculate_similarity(vec1, vec2):
|
||||
if not vec1 or not vec2:
|
||||
return 0.5
|
||||
return 0.95
|
||||
|
||||
monkeypatch.setattr("GramAddict.core.resonance_engine.ContentMemoryDB", lambda: mock_content_db)
|
||||
monkeypatch.setattr("GramAddict.core.resonance_engine.PersonaMemoryDB", lambda: mock_persona_db)
|
||||
monkeypatch.setattr("GramAddict.core.resonance_engine.cosine_similarity", fake_calculate_similarity, raising=False)
|
||||
|
||||
# 1. Create with NO persona interests
|
||||
engine_blind = ResonanceEngine("test_user", persona_interests=[])
|
||||
score_blind = engine_blind.calculate_resonance({"description": "Beautiful mountain sunset"})
|
||||
|
||||
assert score_blind == 0.5, "Blind engine should return exactly 0.5"
|
||||
assert engine_blind._persona_vector is None
|
||||
|
||||
# 2. Create WITH persona interests
|
||||
engine_smart = ResonanceEngine("test_user", persona_interests=["travel", "landscape"])
|
||||
|
||||
assert engine_smart._persona_vector is not None, "Persona vector must be bootstrapped!"
|
||||
|
||||
# Mocking semantic search behavior in ResonanceEngine:
|
||||
# Actually, calculate_resonance uses self.content_memory._get_embedding(text)
|
||||
# Let's mock the internal similarity function if it's there.
|
||||
|
||||
# We must ensure that target_audience is properly wired in bot_flow!
|
||||
# This test just verifies the engine side, we will also add a test to verify config parsing.
|
||||
@@ -1,84 +0,0 @@
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
from GramAddict.core.situational_awareness import SituationalAwarenessEngine, SituationType
|
||||
|
||||
|
||||
def test_sae_escalation_reset_on_situation_change():
|
||||
"""
|
||||
Test that the SAE resets its escalation counter if the situation type changes.
|
||||
This prevents the 'Nuclear Escalation' trap when transitioning from
|
||||
a system permission dialog to an in-app modal.
|
||||
"""
|
||||
device_mock = MagicMock()
|
||||
# Mocking a sequence of dumps:
|
||||
# 1-3: System Permission Dialog
|
||||
# 4-6: In-App Modal
|
||||
# 7: Clear
|
||||
|
||||
# Needs 8 calls because the first attempt gets initial_xml if provided,
|
||||
# but subsequent attempts call dump_hierarchy(). In execute_escape we also call dump_hierarchy.
|
||||
# Actually, let's just make dump_hierarchy yield from a generator, but also
|
||||
# the SAE perceive is what we care about.
|
||||
|
||||
# We will patch `perceive` to directly return our mock situations
|
||||
sae = SituationalAwarenessEngine.get_instance(device_mock)
|
||||
|
||||
situations = [
|
||||
# Attempt 0
|
||||
SituationType.OBSTACLE_SYSTEM, # perceive
|
||||
SituationType.OBSTACLE_SYSTEM, # post-perceive -> success=False (counter=1)
|
||||
# Attempt 1
|
||||
SituationType.OBSTACLE_SYSTEM, # perceive
|
||||
SituationType.OBSTACLE_SYSTEM, # post-perceive -> success=False (counter=2)
|
||||
# Attempt 2
|
||||
SituationType.OBSTACLE_SYSTEM, # perceive
|
||||
SituationType.OBSTACLE_MODAL, # post-perceive -> success=False (counter=3)
|
||||
# Attempt 3 (new situation perceived!) -> situation_attempts resets to 0
|
||||
SituationType.OBSTACLE_MODAL, # perceive
|
||||
SituationType.OBSTACLE_MODAL, # post-perceive -> success=False (counter=1)
|
||||
# Attempt 4
|
||||
SituationType.OBSTACLE_MODAL, # perceive
|
||||
SituationType.OBSTACLE_MODAL, # post-perceive -> success=False (counter=2)
|
||||
# Attempt 5
|
||||
SituationType.OBSTACLE_MODAL, # perceive
|
||||
SituationType.NORMAL, # post-perceive -> success=True!
|
||||
]
|
||||
|
||||
sae.perceive = MagicMock(side_effect=situations)
|
||||
sae._plan_escape_via_llm = MagicMock()
|
||||
sae._execute_escape = MagicMock()
|
||||
|
||||
from GramAddict.core.situational_awareness import EscapeAction
|
||||
|
||||
# Let the LLM "plan" something so it doesn't crash
|
||||
# Each plan needs a unique reason to not be caught by failed_this_session perfectly if it's the same coordinate?
|
||||
# Actually, the recall check does: failed_this_session.add(action_key)
|
||||
# If the LLM keeps returning the same coordinates, it might be an issue.
|
||||
# We can just return different EscapeActions on side_effect
|
||||
sae._plan_escape_via_llm.side_effect = [
|
||||
EscapeAction("tap_coordinates", 100, 100, "mock_1"),
|
||||
EscapeAction("tap_coordinates", 101, 100, "mock_2"),
|
||||
EscapeAction("tap_coordinates", 102, 100, "mock_3"),
|
||||
EscapeAction("tap_coordinates", 103, 100, "mock_4"),
|
||||
EscapeAction("tap_coordinates", 104, 100, "mock_5"),
|
||||
EscapeAction("tap_coordinates", 105, 100, "mock_6"),
|
||||
]
|
||||
|
||||
# Since execute_escape checks the device dump, we just mock device.dump_hierarchy to return garbage
|
||||
# The actual situation check relies on perceive
|
||||
device_mock.dump_hierarchy.return_value = "<mock/>"
|
||||
|
||||
success = sae.ensure_clear_screen(max_attempts=10)
|
||||
|
||||
assert success is True
|
||||
assert sae.perceive.call_count == 12
|
||||
|
||||
# 6 LLM calls total
|
||||
assert sae._plan_escape_via_llm.call_count == 6
|
||||
|
||||
# We should ensure that app_start (nuclear escalation) was NEVER called.
|
||||
# We can check the actions executed
|
||||
app_starts = [
|
||||
args[0][0].action_type for args in sae._execute_escape.call_args_list if args[0][0].action_type == "app_start"
|
||||
]
|
||||
assert len(app_starts) == 0
|
||||
@@ -1,93 +0,0 @@
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
from GramAddict.core.goap import GoalExecutor, ScreenType
|
||||
|
||||
|
||||
def test_semantic_poison_guard_rejects_hallucinations(monkeypatch):
|
||||
"""
|
||||
TDD Test: Verifiziert, dass ein Klick auf 'tap messages tab', der
|
||||
versehentlich im Reels-Feed landet (Halluzination), rigoros als Gift
|
||||
verworfen wird, anstatt die Konfidenz auf 1.0 zu setzen.
|
||||
"""
|
||||
device = MagicMock()
|
||||
executor = GoalExecutor(device, "test_user")
|
||||
|
||||
# 1. Wir behaupten, das Device klickt erfolgreich
|
||||
device.click = MagicMock()
|
||||
|
||||
# 2. Die UI ändert sich: Vor dem Klick waren wir auf Home, danach auf Reels
|
||||
# (Obwohl 'tap messages tab' zu DM_INBOX führen sollte)
|
||||
xml_home = "<hierarchy><node resource-id='home'/></hierarchy>"
|
||||
xml_reels = "<hierarchy><node resource-id='reels'/></hierarchy>"
|
||||
|
||||
device.dump_hierarchy = MagicMock(side_effect=[xml_home, xml_reels])
|
||||
|
||||
# Mock perceive() passend zur echten Engine, so dass es REELS erkennt
|
||||
def fake_perceive(xml=""):
|
||||
if "reels" in xml:
|
||||
return {"screen_type": ScreenType.REELS_FEED, "available_actions": [], "context": {}}
|
||||
return {"screen_type": ScreenType.HOME_FEED, "available_actions": [], "context": {}}
|
||||
|
||||
executor.perceive = MagicMock(side_effect=fake_perceive)
|
||||
|
||||
engine_mock = MagicMock()
|
||||
engine_mock.find_best_node.return_value = {"node": "fake_node"}
|
||||
executor.planner.knowledge.TAB_ACTIONS = {"direct_tab": "tap messages tab"}
|
||||
|
||||
# Speed up
|
||||
monkeypatch.setattr("GramAddict.core.goap.time.sleep", lambda x: None)
|
||||
|
||||
monkeypatch.setattr("GramAddict.core.telepathic_engine.TelepathicEngine.get_instance", lambda: engine_mock)
|
||||
|
||||
# Führe Action aus
|
||||
result = executor._execute_action("tap messages tab", goal="open messages")
|
||||
|
||||
# ASSERT: The Poison Guard SHOULD reject the navigation
|
||||
# because it landed on REELS_FEED instead of DM_INBOX.
|
||||
assert result is False, "Aktion 'tap messages tab' die nach REELS führt, MUSS False zurückgeben!"
|
||||
|
||||
# ASSERT: Die Engine MUSS angewiesen werden, den Klick zu verwerfen ("Poison Guard")
|
||||
engine_mock.reject_click.assert_called_with("tap messages tab")
|
||||
engine_mock.confirm_click.assert_not_called()
|
||||
|
||||
|
||||
def test_goap_misplaced_blame_path_execution(monkeypatch):
|
||||
"""
|
||||
TDD Test: Verifiziert, dass ein korrekter erster Zwischenschritt eines Pfades
|
||||
(z.B. 'tap profile tab' das zu 'own_profile' führt)
|
||||
erfolgreich gewertet wird, auch wenn das finale Ziel ('open following list')
|
||||
noch nicht direkt dadurch erreicht wurde.
|
||||
"""
|
||||
device = MagicMock()
|
||||
executor = GoalExecutor(device, "test_user")
|
||||
|
||||
# Fake UI Transition: Klick auf Profile Tab öffnet das Profil
|
||||
device.dump_hierarchy = MagicMock(side_effect=["<home/>", "<profile/>", "<profile/>"])
|
||||
device.click = MagicMock()
|
||||
|
||||
def fake_perceive(xml=""):
|
||||
if "profile" in xml:
|
||||
return {"screen_type": ScreenType.OWN_PROFILE, "available_actions": [], "context": {}}
|
||||
return {"screen_type": ScreenType.HOME_FEED, "available_actions": [], "context": {}}
|
||||
|
||||
executor.perceive = MagicMock(side_effect=fake_perceive)
|
||||
|
||||
engine_mock = MagicMock()
|
||||
engine_mock.find_best_node.return_value = {"node": "fake_node"}
|
||||
monkeypatch.setattr("GramAddict.core.telepathic_engine.TelepathicEngine.get_instance", lambda: engine_mock)
|
||||
|
||||
# Die Navigation führt zu ScreenType.OWN_PROFILE, Ziel ist "open following list".
|
||||
monkeypatch.setattr("GramAddict.core.goap.time.sleep", lambda x: None)
|
||||
|
||||
# _execute_recalled_path ruft _execute_action mehrfach auf.
|
||||
# Für den Test prüfen wir direkt was _execute_action beim ERSTEN Schritt macht:
|
||||
|
||||
# Simuliere 'tap profile tab' während unser Langzeitziel 'open following list' ist!
|
||||
result = executor._execute_action("tap profile tab", goal="open following list")
|
||||
|
||||
# ASSERT: Das MUß True sein, da die UI sich entscheidend zu einem gültigen Zustand bewegt hat!
|
||||
assert (
|
||||
result is True
|
||||
), "Misplaced Blame! legitimer Teilschritt wurde als Fehlschlag verworfen, weil das Endziel nicht direkt erreicht wurde."
|
||||
engine_mock.confirm_click.assert_called_with("tap profile tab")
|
||||
engine_mock.reject_click.assert_not_called()
|
||||
@@ -1,67 +0,0 @@
|
||||
"""
|
||||
TDD RED: Verify that wipe_all_ai_caches exists, is importable, and
|
||||
correctly wipes all global (non-user-specific) Qdrant collections.
|
||||
|
||||
This test catches the production error:
|
||||
ERROR | Failed to wipe global AI caches: cannot import name 'wipe_all_ai_caches'
|
||||
"""
|
||||
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def mock_qdrant_client():
|
||||
"""Ensure qdrant_client is mocked at the module level for import safety."""
|
||||
# This is already handled by session-scoped conftest, but we guard explicitly
|
||||
pass
|
||||
|
||||
|
||||
class TestWipeAllAiCachesExists:
|
||||
"""The function must be importable from qdrant_memory without errors."""
|
||||
|
||||
def test_wipe_all_ai_caches_is_importable(self):
|
||||
"""RED: This must not raise ImportError."""
|
||||
from GramAddict.core.qdrant_memory import wipe_all_ai_caches
|
||||
|
||||
assert callable(wipe_all_ai_caches)
|
||||
|
||||
def test_wipe_all_ai_caches_calls_wipe_on_all_global_collections(self):
|
||||
"""RED: The function must instantiate and wipe all global memory DBs."""
|
||||
from GramAddict.core.qdrant_memory import wipe_all_ai_caches
|
||||
|
||||
# Patch QdrantBase.__init__ to prevent real Qdrant connections
|
||||
with patch("GramAddict.core.qdrant_memory.QdrantBase.__init__", return_value=None):
|
||||
with patch("GramAddict.core.qdrant_memory.QdrantBase.wipe_collection") as mock_wipe:
|
||||
# Make is_connected return True so wipe_collection doesn't short-circuit
|
||||
with patch(
|
||||
"GramAddict.core.qdrant_memory.QdrantBase.is_connected",
|
||||
new_callable=lambda: property(lambda self: True),
|
||||
):
|
||||
wipe_all_ai_caches()
|
||||
|
||||
# Must wipe at least the known global collections
|
||||
assert mock_wipe.call_count >= 6, (
|
||||
f"Expected at least 6 global collection wipes, got {mock_wipe.call_count}. "
|
||||
f"Global collections: HeuristicMemoryDB, UIMemoryDB, CommentMemoryDB, "
|
||||
f"ContentMemoryDB, ScreenMemoryDB, NavigationMemoryDB"
|
||||
)
|
||||
|
||||
|
||||
class TestBlankStartCodePathIntegrity:
|
||||
"""
|
||||
The blank_start code path in bot_flow must NOT silently swallow ImportErrors.
|
||||
If a function doesn't exist, it's a code defect, not a runtime condition.
|
||||
"""
|
||||
|
||||
def test_blank_start_wipe_does_not_catch_import_error(self):
|
||||
"""
|
||||
The production code catches `Exception` broadly on Line 197 of bot_flow.py.
|
||||
ImportError IS an Exception subclass, so it gets swallowed silently.
|
||||
This test verifies that wipe_all_ai_caches actually exists (the root cause).
|
||||
"""
|
||||
# If this import works, the blank_start try/except will never hit ImportError
|
||||
from GramAddict.core.qdrant_memory import wipe_all_ai_caches
|
||||
|
||||
assert wipe_all_ai_caches is not None
|
||||
Reference in New Issue
Block a user