Marc Mintel
42eabb7bda
- Implement wipe_all_ai_caches() in qdrant_memory.py (was a phantom function referenced but never created, causing ERROR on every blank_start) - Move imports OUT of try/except in bot_flow.py blank_start block so that ImportError/NameError crash loudly instead of being silently swallowed - Add Production Integrity Guard (check_production_integrity) to detect MagicMock poisoning at startup - Add missing TelepathicEngine import in bot_flow.py - Fix conftest.py: move sys.modules monkeypatching into session fixture to prevent global environment poisoning on test import - Add TDD test test_wipe_all_ai_caches.py proving importability and correct wipe behavior across all 8 global Qdrant collections - Delete root garbage: patch_sae_tests.py, test_debug.py, test_mock.py, tmp_bot_flow.py, test_e2e_output*.txt
414 lines
16 KiB
Python
414 lines
16 KiB
Python
import os
|
|
import sys
|
|
import time
|
|
from unittest.mock import MagicMock
|
|
|
|
import pytest
|
|
|
|
from GramAddict.core import utils
|
|
|
|
|
|
@pytest.fixture(scope="session", autouse=True)
|
|
def global_qdrant_mock():
|
|
"""
|
|
Force Qdrant mocking globally across ALL E2E tests so we never
|
|
block on connection refused trying to hit localhost:6344.
|
|
Moved to a fixture to avoid poisoning the global sys.modules on import.
|
|
"""
|
|
mock_qdrant = MagicMock()
|
|
|
|
# Setup correct return types for dimension check warnings in qdrant_memory
|
|
mock_collection = MagicMock()
|
|
mock_collection.config.params.vectors.size = 768
|
|
mock_qdrant.get_collection.return_value = mock_collection
|
|
|
|
# We use a wrapper to ensure the mock is only active when we want it
|
|
sys.modules["qdrant_client"].QdrantClient = MagicMock(return_value=mock_qdrant)
|
|
|
|
yield mock_qdrant
|
|
|
|
# Optional: cleanup if needed, but for E2E it's usually fine to keep it for the session
|
|
|
|
|
|
@pytest.fixture
|
|
def e2e_device_dump_injector(request):
|
|
"""
|
|
Provides a factory to mock device.dump_hierarchy using real XML files.
|
|
Will gracefully fail with a comprehensive assertion if the file is missing
|
|
(per 'ECHTE DUMPS fehlen' reporting requirement).
|
|
"""
|
|
if request.config.getoption("--live"):
|
|
return lambda *args, **kwargs: None
|
|
|
|
def _inject_dump(device_mock, xml_filename):
|
|
fix_dir = os.path.join(os.path.dirname(os.path.dirname(__file__)), "fixtures")
|
|
xml_path = os.path.join(fix_dir, xml_filename)
|
|
|
|
if not os.path.exists(xml_path):
|
|
pytest.fail(
|
|
f"MISSING REAL DUMP: required XML fixture '{xml_filename}' for full E2E workflow testing could not be found at {xml_path}. FAKE_NOTHING policy implies dropping this test execution until it is captured.",
|
|
pytrace=False,
|
|
)
|
|
|
|
with open(xml_path, "r") as f:
|
|
real_xml = f.read()
|
|
|
|
device_mock.dump_hierarchy.return_value = real_xml
|
|
return real_xml
|
|
|
|
return _inject_dump
|
|
|
|
|
|
class VirtualClock:
|
|
def __init__(self):
|
|
self.time = 0.0
|
|
self.animation_target_time = 0.0
|
|
|
|
def sleep(self, seconds):
|
|
if hasattr(seconds, "__iter__"):
|
|
return # For edge case where something weird is passed
|
|
self.time += float(seconds)
|
|
|
|
|
|
clock = VirtualClock()
|
|
|
|
|
|
@pytest.fixture
|
|
def dynamic_e2e_dump_injector(monkeypatch, request):
|
|
"""
|
|
State-Machine Injector: Replaces dump_hierarchy dynamically when transitions occur.
|
|
Validates that the Telepathic Engine's pathfinding truly worked.
|
|
It now inherently simulates UI animation delays. If a dump is requested
|
|
LESS than 1.5 virtual seconds after a transition, it returns a garbage animating UI.
|
|
"""
|
|
if request.config.getoption("--live"):
|
|
return lambda *args, **kwargs: None
|
|
|
|
def _inject(device_mock, state_map, initial_xml):
|
|
from GramAddict.core.q_nav_graph import QNavGraph
|
|
|
|
fix_dir = os.path.join(os.path.dirname(os.path.dirname(__file__)), "fixtures")
|
|
|
|
def load_xml(filename):
|
|
path = os.path.join(fix_dir, filename)
|
|
if not os.path.exists(path):
|
|
pytest.fail(f"MISSING REAL DUMP: {filename} not found.")
|
|
with open(path, "r") as f:
|
|
return f.read()
|
|
|
|
# History stack to allow "back" navigation
|
|
device_mock._xml_history = [load_xml(initial_xml)]
|
|
device_mock._current_active_xml = device_mock._xml_history[-1]
|
|
|
|
import uuid
|
|
|
|
def _dump_hierarchy_hook():
|
|
if clock.time < clock.animation_target_time:
|
|
pytest.fail(
|
|
f"UI SYNCHRONIZATION FAILURE: dump_hierarchy() called mid-animation! "
|
|
f"Virtual Clock is at {clock.time:.1f}s but UI needs until {clock.animation_target_time:.1f}s to settle. "
|
|
f"Add a time.sleep() guard before interacting with the UI after a click.",
|
|
pytrace=False,
|
|
)
|
|
xml = device_mock._current_active_xml
|
|
if xml and "</hierarchy>" in xml:
|
|
xml = xml.replace("</hierarchy>", f'<node sid="{uuid.uuid4()}" /></hierarchy>')
|
|
return xml
|
|
|
|
device_mock.dump_hierarchy.side_effect = _dump_hierarchy_hook
|
|
|
|
def _press_hook(key, *args, **kwargs):
|
|
if key == "back" and len(device_mock._xml_history) > 1:
|
|
device_mock._xml_history.pop()
|
|
device_mock._current_active_xml = device_mock._xml_history[-1]
|
|
clock.animation_target_time = clock.time + 1.5
|
|
|
|
device_mock.press.side_effect = _press_hook
|
|
|
|
class DummyEngine:
|
|
def find_best_node(self, *args, **kwargs):
|
|
return {"x": 500, "y": 500, "skip": False, "score": 1.0, "source": "e2e_mock"}
|
|
|
|
def verify_success(self, *args, **kwargs):
|
|
return True
|
|
|
|
def confirm_click(self, *args, **kwargs):
|
|
pass
|
|
|
|
def reject_click(self, *args, **kwargs):
|
|
pass
|
|
|
|
original_execute = QNavGraph._execute_transition
|
|
from GramAddict.core.goap import GoalExecutor
|
|
|
|
original_goap_execute = GoalExecutor._execute_action
|
|
|
|
def _mock_execute_transition(nav_self, action, zero_engine=None, max_retries=2):
|
|
if action == "tap_post_username":
|
|
return True
|
|
|
|
original_click = nav_self.device.click
|
|
|
|
def _click_hook(obj=None, *args, **kwargs):
|
|
original_click(obj, *args, **kwargs)
|
|
if action in state_map:
|
|
new_xml = load_xml(state_map[action])
|
|
device_mock._xml_history.append(new_xml)
|
|
device_mock._current_active_xml = new_xml
|
|
clock.animation_target_time = clock.time + 1.5
|
|
|
|
nav_self.device.click = _click_hook
|
|
|
|
try:
|
|
success = original_execute(
|
|
nav_self, action, mock_semantic_engine=DummyEngine(), max_retries=max_retries
|
|
)
|
|
return success
|
|
finally:
|
|
nav_self.device.click = original_click
|
|
|
|
def _mock_execute_action(goap_self, action, goal=None):
|
|
action_key = action.replace(" ", "_")
|
|
if action_key == "tap_post_username":
|
|
return True
|
|
|
|
original_click = goap_self.device.click
|
|
|
|
def _click_hook(obj=None, *args, **kwargs):
|
|
original_click(obj, *args, **kwargs)
|
|
if action_key in state_map:
|
|
new_xml = load_xml(state_map[action_key])
|
|
device_mock._xml_history.append(new_xml)
|
|
device_mock._current_active_xml = new_xml
|
|
clock.animation_target_time = clock.time + 1.5
|
|
elif action in state_map:
|
|
new_xml = load_xml(state_map[action])
|
|
device_mock._xml_history.append(new_xml)
|
|
device_mock._current_active_xml = new_xml
|
|
clock.animation_target_time = clock.time + 1.5
|
|
|
|
goap_self.device.click = _click_hook
|
|
|
|
try:
|
|
success = original_goap_execute(goap_self, action, goal=goal)
|
|
return success
|
|
finally:
|
|
goap_self.device.click = original_click
|
|
|
|
monkeypatch.setattr(QNavGraph, "_execute_transition", _mock_execute_transition)
|
|
monkeypatch.setattr(GoalExecutor, "_execute_action", _mock_execute_action)
|
|
|
|
return _inject
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def mock_all_delays(monkeypatch, request):
|
|
"""
|
|
Replaces all humanized hardware delays specifically for the E2E test suite
|
|
with a Virtual Clock. Ensures loops evaluate instantly but preserves chronological
|
|
dependency for our Animation Simulator.
|
|
"""
|
|
if request.config.getoption("--live"):
|
|
return
|
|
|
|
global clock
|
|
clock.time = 0.0 # reset for test
|
|
clock.animation_target_time = 0.0
|
|
|
|
def simulate_sleep(seconds):
|
|
clock.sleep(seconds)
|
|
|
|
def money_sleep(x):
|
|
return simulate_sleep(x)
|
|
|
|
def random_sleep(a=1.0, b=2.0, *args, **kwargs):
|
|
return simulate_sleep(max(1.5, float(a)))
|
|
|
|
monkeypatch.setattr(time, "sleep", money_sleep)
|
|
monkeypatch.setattr(utils, "random_sleep", random_sleep)
|
|
monkeypatch.setattr(utils, "sleep", money_sleep)
|
|
|
|
# Needs to capture specific module sleeps depending on how they imported it
|
|
try:
|
|
from GramAddict.core import bot_flow
|
|
|
|
monkeypatch.setattr(bot_flow, "sleep", money_sleep)
|
|
monkeypatch.setattr(bot_flow.random, "uniform", lambda a, b: float(a)) # deterministic lower bound
|
|
if hasattr(bot_flow, "random_sleep"):
|
|
monkeypatch.setattr(bot_flow, "random_sleep", random_sleep)
|
|
|
|
from GramAddict.core import q_nav_graph
|
|
|
|
monkeypatch.setattr(q_nav_graph.random, "uniform", lambda a, b: float(a))
|
|
if hasattr(q_nav_graph, "random_sleep"):
|
|
monkeypatch.setattr(q_nav_graph, "random_sleep", random_sleep)
|
|
|
|
from GramAddict.core import goap
|
|
|
|
if hasattr(goap, "random"):
|
|
monkeypatch.setattr(goap.random, "uniform", lambda a, b: float(a))
|
|
if hasattr(goap, "random_sleep"):
|
|
monkeypatch.setattr(goap, "random_sleep", random_sleep)
|
|
|
|
monkeypatch.setattr(utils.random, "uniform", lambda a, b: float(a))
|
|
|
|
from GramAddict.core import device_facade
|
|
|
|
monkeypatch.setattr(device_facade, "sleep", money_sleep)
|
|
monkeypatch.setattr(device_facade.random, "uniform", lambda a, b: float(a))
|
|
if hasattr(device_facade, "random_sleep"):
|
|
monkeypatch.setattr(device_facade, "random_sleep", random_sleep)
|
|
except Exception as e:
|
|
print(f"Mocking delays exception: {e}")
|
|
|
|
# Standardize DarwinEngine across tests to prevent mockup math errors on session end
|
|
try:
|
|
from GramAddict.core.darwin_engine import DarwinEngine
|
|
|
|
monkeypatch.setattr(DarwinEngine, "evaluate_session_end", lambda *args, **kwargs: None)
|
|
except ImportError:
|
|
pass
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def mock_identity_guard(monkeypatch):
|
|
import GramAddict.core.bot_flow
|
|
|
|
monkeypatch.setattr(GramAddict.core.bot_flow, "verify_and_switch_account", lambda *args, **kwargs: True)
|
|
|
|
|
|
@pytest.fixture
|
|
def e2e_configs():
|
|
import argparse
|
|
from unittest.mock import MagicMock
|
|
|
|
args = argparse.Namespace(
|
|
username="testuser",
|
|
device="emulator-5554",
|
|
app_id="com.instagram.android",
|
|
debug=True,
|
|
feed=None,
|
|
carousel_percentage=0,
|
|
carousel_count="1",
|
|
explore=None,
|
|
reels=None,
|
|
stories=None,
|
|
interact_percentage=100,
|
|
likes_count="2-3",
|
|
likes_percentage=100,
|
|
follow_percentage=100,
|
|
comment_percentage=100,
|
|
stories_count="1-2",
|
|
stories_percentage=100,
|
|
working_hours=[0.0, 24.0],
|
|
time_delta_session=0,
|
|
speed_multiplier=1.0,
|
|
disable_filters=False,
|
|
interaction_users_amount="1",
|
|
scrape_profiles=False,
|
|
disable_ai_messaging=True,
|
|
total_unfollows_limit=0,
|
|
ai_telepathic_url="http://localhost",
|
|
ai_telepathic_model="llama3",
|
|
ai_condenser_url="http://localhost",
|
|
dry_run_comments=False,
|
|
visual_vibe_check_percentage=0,
|
|
)
|
|
|
|
configs = MagicMock()
|
|
configs.args = args
|
|
configs.username = "testuser"
|
|
|
|
# Realistically mock get_plugin_config
|
|
def get_plugin_config_mock(plugin_name):
|
|
# Return a dict that simulates what's in the args for that plugin
|
|
mapping = {
|
|
"likes": {"count": args.likes_count, "percentage": args.likes_percentage},
|
|
"comment": {
|
|
"percentage": args.comment_percentage,
|
|
"dry_run": args.dry_run_comments,
|
|
},
|
|
"follow": {"percentage": args.follow_percentage},
|
|
"stories": {"count": args.stories_count, "percentage": args.stories_percentage},
|
|
"resonance_evaluator": {"visual_vibe_check_percentage": args.visual_vibe_check_percentage},
|
|
"carousel_browsing": {
|
|
"percentage": getattr(args, "carousel_percentage", 0),
|
|
"count": getattr(args, "carousel_count", "1"),
|
|
},
|
|
}
|
|
return mapping.get(plugin_name, {})
|
|
|
|
configs.get_plugin_config.side_effect = get_plugin_config_mock
|
|
return configs
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def mock_sae_perceive(request, monkeypatch):
|
|
"""
|
|
Mock SAE.perceive for all E2E tests EXCEPT the ones actually testing SAE.
|
|
This prevents the tests from hitting the local Qdrant/Ollama instances
|
|
and failing due to non-deterministic LLM output or missing caches.
|
|
"""
|
|
if "test_e2e_sae.py" in str(request.node.fspath):
|
|
return
|
|
if "test_e2e_real_llm_learning.py" in str(request.node.fspath):
|
|
return
|
|
if request.config.getoption("--live"):
|
|
return
|
|
|
|
import GramAddict.core.situational_awareness
|
|
|
|
monkeypatch.setattr(
|
|
GramAddict.core.situational_awareness.SituationalAwarenessEngine,
|
|
"perceive",
|
|
lambda self, xml: GramAddict.core.situational_awareness.SituationType.NORMAL,
|
|
)
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def setup_e2e_plugin_registry():
|
|
"""Ensures that all standard plugins are registered for E2E tests."""
|
|
from GramAddict.core.behaviors import PluginRegistry
|
|
from GramAddict.core.behaviors.ad_guard import AdGuardPlugin
|
|
from GramAddict.core.behaviors.anomaly_handler import AnomalyHandlerPlugin
|
|
from GramAddict.core.behaviors.carousel_browsing import CarouselBrowsingPlugin
|
|
from GramAddict.core.behaviors.close_friends_guard import CloseFriendsGuardPlugin
|
|
from GramAddict.core.behaviors.comment import CommentPlugin
|
|
from GramAddict.core.behaviors.darwin_dwell import DarwinDwellPlugin
|
|
from GramAddict.core.behaviors.follow import FollowPlugin
|
|
from GramAddict.core.behaviors.grid_like import GridLikePlugin
|
|
from GramAddict.core.behaviors.like import LikePlugin
|
|
from GramAddict.core.behaviors.obstacle_guard import ObstacleGuardPlugin
|
|
from GramAddict.core.behaviors.perfect_snapping import PerfectSnappingPlugin
|
|
from GramAddict.core.behaviors.post_data_extraction import PostDataExtractionPlugin
|
|
from GramAddict.core.behaviors.post_interaction import PostInteractionPlugin
|
|
from GramAddict.core.behaviors.profile_guard import ProfileGuardPlugin
|
|
from GramAddict.core.behaviors.profile_visit import ProfileVisitPlugin
|
|
from GramAddict.core.behaviors.rabbit_hole import RabbitHolePlugin
|
|
from GramAddict.core.behaviors.repost import RepostPlugin
|
|
from GramAddict.core.behaviors.resonance_evaluator import ResonanceEvaluatorPlugin
|
|
from GramAddict.core.behaviors.story_view import StoryViewPlugin
|
|
|
|
PluginRegistry.reset()
|
|
plugin_registry = PluginRegistry.get_instance()
|
|
plugin_registry.register(ProfileGuardPlugin())
|
|
plugin_registry.register(StoryViewPlugin())
|
|
plugin_registry.register(FollowPlugin())
|
|
plugin_registry.register(GridLikePlugin())
|
|
plugin_registry.register(CarouselBrowsingPlugin())
|
|
plugin_registry.register(AdGuardPlugin())
|
|
plugin_registry.register(CloseFriendsGuardPlugin())
|
|
plugin_registry.register(AnomalyHandlerPlugin())
|
|
plugin_registry.register(ObstacleGuardPlugin())
|
|
plugin_registry.register(PerfectSnappingPlugin())
|
|
plugin_registry.register(PostDataExtractionPlugin())
|
|
plugin_registry.register(ResonanceEvaluatorPlugin())
|
|
plugin_registry.register(RabbitHolePlugin())
|
|
plugin_registry.register(DarwinDwellPlugin())
|
|
plugin_registry.register(ProfileVisitPlugin())
|
|
plugin_registry.register(LikePlugin())
|
|
plugin_registry.register(CommentPlugin())
|
|
plugin_registry.register(RepostPlugin())
|
|
plugin_registry.register(PostInteractionPlugin())
|
|
yield plugin_registry
|