Files
manual_slop/tests/test_hooks.py

103 lines
3.8 KiB
Python

import os
import sys
sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))
import pytest
from unittest.mock import patch
import gui
import api_hooks
import urllib.request
import json
import threading
import time
def test_hooks_enabled_via_cli():
with patch.object(sys, 'argv', ['gui.py', '--enable-test-hooks']):
app = gui.App()
assert app.test_hooks_enabled is True
def test_hooks_disabled_by_default():
with patch.object(sys, 'argv', ['gui.py']):
if 'SLOP_TEST_HOOKS' in os.environ:
del os.environ['SLOP_TEST_HOOKS']
app = gui.App()
assert getattr(app, 'test_hooks_enabled', False) is False
def test_hooks_enabled_via_env():
with patch.object(sys, 'argv', ['gui.py']):
with patch.dict(os.environ, {'SLOP_TEST_HOOKS': '1'}):
app = gui.App()
assert app.test_hooks_enabled is True
def test_ipc_server_starts_and_responds():
app_mock = gui.App()
app_mock.test_hooks_enabled = True
server = api_hooks.HookServer(app_mock, port=0)
server.start()
# Wait for server to start
time.sleep(0.5)
actual_port = server.server.server_address[1]
base_url = f"http://127.0.0.1:{actual_port}"
try:
req = urllib.request.Request(f"{base_url}/status")
with urllib.request.urlopen(req) as response:
assert response.status == 200
data = json.loads(response.read().decode())
assert data.get("status") == "ok"
# Test project GET
req = urllib.request.Request(f"{base_url}/api/project")
with urllib.request.urlopen(req) as response:
assert response.status == 200
data = json.loads(response.read().decode())
assert "project" in data
# Test session GET
req = urllib.request.Request(f"{base_url}/api/session")
with urllib.request.urlopen(req) as response:
assert response.status == 200
data = json.loads(response.read().decode())
assert "session" in data
# Test project POST
project_data = {"project": {"foo": "bar"}}
req = urllib.request.Request(
f"{base_url}/api/project",
method="POST",
data=json.dumps(project_data).encode("utf-8"),
headers={'Content-Type': 'application/json'})
with urllib.request.urlopen(req) as response:
assert response.status == 200
assert app_mock.project == {"foo": "bar"}
# Test session POST
session_data = {"session": {"entries": [{"role": "User", "content": "hi"}]}}
req = urllib.request.Request(
f"{base_url}/api/session",
method="POST",
data=json.dumps(session_data).encode("utf-8"),
headers={'Content-Type': 'application/json'})
with urllib.request.urlopen(req) as response:
assert response.status == 200
assert app_mock.disc_entries == [{"role": "User", "content": "hi"}]
# Test GUI queue hook
gui_data = {"action": "set_value", "item": "test_item", "value": "test_value"}
req = urllib.request.Request(
f"{base_url}/api/gui",
method="POST",
data=json.dumps(gui_data).encode("utf-8"),
headers={'Content-Type': 'application/json'})
with urllib.request.urlopen(req) as response:
assert response.status == 200
# Instead of checking DPG (since we aren't running the real main loop in tests),
# check if it got queued in app_mock
assert hasattr(app_mock, '_pending_gui_tasks')
assert len(app_mock._pending_gui_tasks) == 1
assert app_mock._pending_gui_tasks[0] == gui_data
finally:
server.stop()