feat: add offer reward claim workflow
This commit is contained in:
admin committed
1 parent
870ca82e8a
commit
b6bd78dc66
25 files changed
+1028
-5
No files matched your search
@@ -0,0 +1,126 @@
|
||||
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()
|
||||
@@ -0,0 +1,82 @@
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
from PIL import Image
|
||||
|
||||
from automation.engine import Matcher, Region, WorkflowError, load_flow
|
||||
from automation.offers import OfferRunner
|
||||
|
||||
ROOT = Path(__file__).resolve().parent.parent
|
||||
|
||||
|
||||
class OfferDesktop:
|
||||
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 OfferTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.flow = load_flow(ROOT / 'workflows/offer_claim.json')
|
||||
self.matcher = Matcher(ROOT / 'assets')
|
||||
self.region = Region(0, 0, 562, 1002)
|
||||
|
||||
def image(self, name):
|
||||
image = Image.new('RGB', (562, 1002), (45, 62, 78))
|
||||
for spec in self.flow['states'][name]:
|
||||
template = Image.open(ROOT / 'assets' / spec['template']).convert('RGB')
|
||||
x, y, _, _ = spec['search']
|
||||
image.paste(template, (x + 5, y + 5))
|
||||
return image
|
||||
|
||||
def runner(self, names):
|
||||
desktop = OfferDesktop([self.image(n) for n in names])
|
||||
runner = OfferRunner(desktop, self.matcher, self.flow, lambda _: self.region,
|
||||
log=lambda _: None, timeout=0.8)
|
||||
return desktop, runner
|
||||
|
||||
def test_full_claim_and_return(self):
|
||||
desktop, runner = self.runner(['home', 'available', 'reward', 'claimed', 'home'])
|
||||
self.assertEqual(runner.run(), 'claimed')
|
||||
self.assertEqual(desktop.clicks, [self.flow['clicks'][n] for n in ('open', 'claim', 'dismiss', 'back')])
|
||||
|
||||
def test_already_claimed_does_not_claim_again(self):
|
||||
desktop, runner = self.runner(['home', 'claimed', 'home'])
|
||||
self.assertEqual(runner.run(), 'already_claimed')
|
||||
self.assertEqual(desktop.clicks, [self.flow['clicks'][n] for n in ('open', 'back')])
|
||||
|
||||
def test_monthcard_entry_selects_battlepass(self):
|
||||
desktop, runner = self.runner(['home', 'menu', 'claimed', 'home'])
|
||||
self.assertEqual(runner.run(), 'already_claimed')
|
||||
self.assertEqual(desktop.clicks, [self.flow['clicks'][n] for n in ('open', 'battlepass', 'back')])
|
||||
|
||||
def test_missing_reward_stops_without_retry(self):
|
||||
desktop, runner = self.runner(['home', 'available', 'available'])
|
||||
with self.assertRaises(WorkflowError):
|
||||
runner.run()
|
||||
self.assertEqual(desktop.clicks, [self.flow['clicks'][n] for n in ('open', 'claim')])
|
||||
|
||||
def test_preview_never_clicks(self):
|
||||
desktop, runner = self.runner(['home'])
|
||||
self.assertEqual(runner.run(True), 'preview')
|
||||
self.assertEqual(desktop.clicks, [])
|
||||
|
||||
def test_window_movement_blocks_stale_click(self):
|
||||
desktop, runner = self.runner(['home'])
|
||||
observation = ('home', desktop.screenshot(), self.region)
|
||||
runner.region_provider = lambda _: Region(2, 0, 562, 1002)
|
||||
with self.assertRaises(WorkflowError):
|
||||
runner.click('open', observation)
|
||||
self.assertEqual(desktop.clicks, [])
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
Reference in new issue
Block a user