127 lines
5.2 KiB
Python
127 lines
5.2 KiB
Python
import threading
|
|
import unittest
|
|
from unittest.mock import Mock, patch
|
|
from pathlib import Path
|
|
|
|
import numpy as np
|
|
from PIL import Image
|
|
|
|
from automation.engine import Desktop, Matcher, Region, Runner, WorkflowError, load_flow
|
|
|
|
ROOT = Path(__file__).resolve().parent.parent
|
|
|
|
|
|
class FakeDesktop:
|
|
def __init__(self, images):
|
|
self.images = images
|
|
self.index = 0
|
|
self.clicks = []
|
|
|
|
def screenshot(self):
|
|
return self.images[min(self.index, len(self.images) - 1)]
|
|
|
|
def click(self, point, region, size):
|
|
self.clicks.append(point)
|
|
self.index += 1
|
|
|
|
|
|
class EngineTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.matcher = Matcher(ROOT / 'assets')
|
|
self.flow = load_flow(ROOT / 'workflows/pig_claim.json')
|
|
self.region = Region(0, 0, 562, 1002)
|
|
|
|
def state(self, specs):
|
|
image = Image.new('RGB', (562, 1002), (45, 62, 78))
|
|
for spec in specs:
|
|
template = Image.open(ROOT / 'assets' / spec['template']).convert('RGB')
|
|
x, y, _, _ = spec['search']
|
|
image.paste(template, (x + 5, y + 5))
|
|
return image
|
|
|
|
def test_retina_coordinate_mapping(self):
|
|
region = Region(400, 100, 562, 1002)
|
|
self.assertEqual(region.mouse_point((0.5, 0.5), (2000, 1400), (1000, 700)), (340, 300))
|
|
|
|
def test_invalid_region_and_click(self):
|
|
with self.assertRaises(WorkflowError):
|
|
Region(0, 0, 800, 400).validate((1000, 1000))
|
|
with self.assertRaises(WorkflowError):
|
|
self.region.mouse_point((1.5, 0.5), (562, 1002), (562, 1002))
|
|
|
|
def test_home_not_detected_behind_dark_popup(self):
|
|
spec = self.flow['finish']
|
|
frame = self.matcher.frame(self.state(spec), self.region)
|
|
self.assertTrue(self.matcher.matches(frame, spec))
|
|
self.assertFalse(self.matcher.matches((frame * 0.45).astype(np.uint8), spec))
|
|
|
|
def test_claim_workflow_and_no_sweep(self):
|
|
home = self.state(self.flow['finish'])
|
|
modal = self.state(self.flow['steps'][1]['expect'])
|
|
outcome = Image.new('RGB', (562, 1002), (80, 90, 120))
|
|
desktop = FakeDesktop([home, modal, outcome, home])
|
|
logs = []
|
|
Runner(desktop, self.matcher, self.region, logs.append).run(self.flow)
|
|
self.assertEqual(desktop.clicks, [[0.135, 0.607], [0.747, 0.738], [0.5, 0.875]])
|
|
self.assertIn('到账数量尚未验证', logs[-1])
|
|
|
|
def test_preview_does_not_click(self):
|
|
desktop = FakeDesktop([self.state(self.flow['finish'])])
|
|
Runner(desktop, self.matcher, self.region, lambda _: None).run(self.flow, True)
|
|
self.assertEqual(desktop.clicks, [])
|
|
|
|
def test_unchanged_claim_stops_without_retry(self):
|
|
home = self.state(self.flow['finish'])
|
|
modal = self.state(self.flow['steps'][1]['expect'])
|
|
desktop = FakeDesktop([home, modal, modal])
|
|
steps = [dict(step) for step in self.flow['steps']]
|
|
steps[1]['timeout'] = 0.8
|
|
flow = dict(self.flow, steps=steps)
|
|
with self.assertRaises(WorkflowError):
|
|
Runner(desktop, self.matcher, self.region, lambda _: None).run(flow)
|
|
self.assertEqual(desktop.clicks, [[0.135, 0.607], [0.747, 0.738]])
|
|
|
|
def test_unrecognized_screen_stops_without_click(self):
|
|
desktop = FakeDesktop([Image.new('RGB', (562, 1002))])
|
|
flow = dict(self.flow, steps=[dict(self.flow['steps'][0], timeout=0.05)])
|
|
runner = Runner(desktop, self.matcher, self.region, lambda _: None)
|
|
# Directly check the same initial guard with a short timeout.
|
|
with self.assertRaises(WorkflowError):
|
|
runner.wait_state(flow['steps'][0]['expect'], timeout=0.05)
|
|
self.assertEqual(desktop.clicks, [])
|
|
|
|
def test_stop_before_actions(self):
|
|
stop = threading.Event()
|
|
stop.set()
|
|
desktop = FakeDesktop([])
|
|
with self.assertRaises(WorkflowError):
|
|
Runner(desktop, self.matcher, self.region, stop=stop).run(self.flow)
|
|
self.assertEqual(desktop.clicks, [])
|
|
|
|
def test_permission_denied_blocks_mouse_event(self):
|
|
desktop = Desktop.__new__(Desktop)
|
|
desktop.api = Mock()
|
|
with patch('automation.permissions.mouse_permission_granted', return_value=False):
|
|
with self.assertRaisesRegex(WorkflowError, '辅助功能权限未授予'):
|
|
desktop.click((0.5, 0.5), self.region, (562, 1002))
|
|
desktop.api.click.assert_not_called()
|
|
desktop.api.mouseDown.assert_not_called()
|
|
|
|
def test_held_click_releases_after_interruption(self):
|
|
desktop = Desktop.__new__(Desktop)
|
|
desktop.api = Mock()
|
|
desktop.api.size.return_value = (562, 1002)
|
|
desktop.api.FAILSAFE = True
|
|
with patch('automation.permissions.mouse_permission_granted', return_value=True):
|
|
with patch('automation.engine.time.sleep', side_effect=RuntimeError('interrupted')):
|
|
with self.assertRaises(RuntimeError):
|
|
desktop.click((0.5, 0.5), self.region, (562, 1002))
|
|
desktop.api.moveTo.assert_called_once_with(281, 501, duration=0.2)
|
|
desktop.api.mouseDown.assert_called_once()
|
|
desktop.api.mouseUp.assert_called_once()
|
|
self.assertTrue(desktop.api.FAILSAFE)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|