add # 新增偷菜功能

tests/test_crop_identity.py

@@ -116,6 +116,8 @@ class MigrationTests(unittest.IsolatedAsyncioTestCase):
     def create_v1(self, path, unknown=False, version=1):
         with closing(sqlite3.connect(path)) as db, db:
             for statement in SCHEMA:
+                if statement.startswith('CREATE TABLE thefts('):
+                    continue
                 if version == 1 and statement.startswith("CREATE TABLE inventory("):
                     statement = """CREATE TABLE inventory(
                         identity TEXT, name TEXT, seeds INTEGER,

tests/test_stealing.py

@@ -0,0 +1,156 @@
+import asyncio
+from contextlib import closing
+from dataclasses import replace
+from pathlib import Path
+import sqlite3
+import tempfile
+import unittest
+from unittest.mock import patch
+
+from farm_core import Catalogue, FarmService, Identity, SQLiteStore
+from farm_core.domain import FarmError, Plant, Plot
+from farm_core.repository import SQLiteRepository
+from farm_core.yield_rules import remaining_yield
+
+
+class StealingTests(unittest.IsolatedAsyncioTestCase):
+    async def asyncSetUp(self):
+        self.temp = tempfile.TemporaryDirectory()
+        self.store = SQLiteStore(Path(self.temp.name) / 'farm.db')
+        await self.store.initialize()
+        self.now = 100000
+        self.plant = Plant(1, '测试菜', 0, 1, False, 0, 2, 10,
+                           1, 100, 2, 100)
+        self.service = FarmService(self.store, Catalogue([self.plant]),
+                                   lambda: self.now)
+        self.owner = Identity('test', 'local', 'qq', '1')
+        self.thief = replace(self.owner, user_id='2')
+        for who in (self.owner, self.thief):
+            await self.service.register(who, who.user_id, 'register')
+        self.fixture = 0
+
+    async def asyncTearDown(self):
+        await self.store.close()
+        self.temp.cleanup()
+
+    async def prepare(self, plots, owner=None):
+        self.fixture += 1
+        await self.store.transact(
+            owner or self.owner, f'fixture-{self.fixture}', 'fixture', {},
+            self.now, lambda repo: [repo.save_plot(p) for p in plots],
+        )
+
+    async def test_each_target_plot_random_bounds_replay_and_repeat(self):
+        await self.prepare([Plot(i, self.plant, self.now) for i in range(1, 7)])
+        with patch('farm_core.application.secrets.randbelow',
+                   side_effect=[0, 2, 0, 2, 0, 2]):
+            result = await self.service.steal(self.thief, self.owner, 'first')
+        self.assertEqual([i['quantity'] for i in result.data['items']],
+                         [1, 3, 1, 3, 1, 3])
+        self.assertEqual(result.data['remaining_attempts'], 4)
+        self.assertEqual(await self.service.steal(
+            self.thief, self.owner, 'first'), result)
+        repeated = await self.service.steal(self.thief, self.owner, 'again')
+        self.assertEqual(repeated.data['quantity'], 0)
+        self.assertEqual(repeated.data['skipped']['already_stolen'], 6)
+        self.assertEqual(repeated.data['remaining_attempts'], 4)
+        self.assertEqual((await self.service.snapshot(self.thief))
+                         .inventory[0].crops, 12)
+
+    async def test_protection_odd_yield_pests_and_bonus(self):
+        odd = replace(self.plant, harvest=9)
+        await self.prepare([
+            Plot(1, odd, self.now, stolen_count=3),
+            Plot(2, self.plant, self.now, stolen_count=5),
+            Plot(3, self.plant, self.now, yield_bonus=20,
+                 maturity_pests=3, stolen_count=4),
+        ])
+        with patch('farm_core.application.secrets.randbelow', return_value=2):
+            result = await self.service.steal(self.thief, self.owner, 'steal')
+        self.assertEqual([i['quantity'] for i in result.data['items']], [1, 1])
+        self.assertEqual(result.data['skipped']['protected'], 1)
+        plots = (await self.service.snapshot(self.owner)).plots
+        self.assertEqual([remaining_yield(p) for p in plots[:3]], [5, 5, 5])
+
+    async def test_harvest_preserves_remaining_and_resets_next_season(self):
+        await self.prepare([Plot(1, self.plant, self.now)])
+        with patch('farm_core.application.secrets.randbelow', return_value=2):
+            await self.service.steal(self.thief, self.owner, 'first')
+        receipt = await self.service.harvest(self.owner, 'harvest')
+        self.assertEqual(receipt.data['quantity'], 7)
+        self.assertEqual((await self.service.snapshot(self.owner))
+                         .plots[0].stolen_count, 0)
+        self.now += 1000
+        again = await self.service.steal(self.thief, self.owner, 'next')
+        self.assertGreater(again.data['quantity'], 0)
+
+    async def test_concurrent_thieves_never_cross_protected_half(self):
+        await self.prepare([Plot(1, self.plant, self.now)])
+        thieves = [replace(self.thief, user_id=str(i)) for i in range(3, 13)]
+        for who in thieves:
+            await self.service.register(who, who.user_id, 'register')
+        with patch('farm_core.application.secrets.randbelow', return_value=2):
+            results = await asyncio.gather(*[
+                self.service.steal(who, self.owner, 'steal') for who in thieves
+            ])
+        self.assertEqual(sum(r.data['quantity'] for r in results), 5)
+        self.assertEqual(remaining_yield(
+            (await self.service.snapshot(self.owner)).plots[0]), 5)
+
+    async def test_daily_quota_failed_visits_free_persistent_and_midnight(self):
+        self.now = 86400-28800-1
+        empty = await self.service.steal(self.thief, self.owner, 'empty')
+        self.assertEqual(empty.data['remaining_attempts'], 5)
+        for i in range(5):
+            owner = replace(self.owner, user_id=str(i+20))
+            await self.service.register(owner, str(i), 'register')
+            await self.prepare([Plot(1, self.plant, self.now)], owner)
+            await self.service.steal(self.thief, owner, f'steal-{i}')
+        await self.store.close()
+        await self.store.initialize()
+        with self.assertRaisesRegex(FarmError, 'steal_daily_limit'):
+            await self.service.steal(self.thief, self.owner, 'limit')
+        self.now += 1
+        await self.prepare([Plot(1, self.plant, self.now)])
+        receipt = await self.service.steal(self.thief, self.owner, 'new-day')
+        self.assertEqual(receipt.data['remaining_attempts'], 4)
+
+    async def test_denied_targets_and_atomic_rollback(self):
+        for target, code in (
+            (self.thief, 'cannot_steal_self'),
+            (replace(self.owner, realm='other'), 'different_farm_scope'),
+            (replace(self.owner, user_id='missing'), 'target_not_registered'),
+        ):
+            with self.assertRaisesRegex(FarmError, code):
+                await self.service.steal(self.thief, target, code)
+        await self.prepare([Plot(1, self.plant, self.now)])
+        before = [await self.service.snapshot(w)
+                  for w in (self.owner, self.thief)]
+        with patch.object(SQLiteRepository, 'record_theft',
+                          side_effect=sqlite3.OperationalError('fixture')):
+            with self.assertRaises(Exception):
+                await self.service.steal(self.thief, self.owner, 'rollback')
+        self.assertEqual([await self.service.snapshot(w)
+                          for w in (self.owner, self.thief)], before)
+        receipt = await self.service.steal(self.thief, self.owner, 'rollback')
+        self.assertGreater(receipt.data['quantity'], 0)
+
+    async def test_v3_migration_preserves_plots_and_initializes_theft_state(self):
+        await self.prepare([Plot(1, self.plant, self.now)])
+        before = await self.service.snapshot(self.owner)
+        await self.store.close()
+        with closing(sqlite3.connect(self.store.path)) as db, db:
+            db.execute('DROP TABLE thefts')
+            db.execute('ALTER TABLE plots DROP COLUMN stolen_count')
+            db.execute('PRAGMA user_version=3')
+        await self.store.initialize()
+        self.assertEqual(await self.service.snapshot(self.owner), before)
+        await self.service.steal(self.thief, self.owner, 'after-migration')
+
+    async def test_concurrent_harvest_and_steal_conserve_yield(self):
+        await self.prepare([Plot(1, self.plant, self.now)])
+        results = await asyncio.gather(
+            self.service.harvest(self.owner, 'harvest'),
+            self.service.steal(self.thief, self.owner, 'steal'),
+        )
+        self.assertEqual(sum(r.data['quantity'] for r in results), 10)

tests/test_warehouse.py

@@ -72,5 +72,10 @@ class WarehouseTests(unittest.TestCase):
                                        kind, 1, 1)
             with Image.open(BytesIO(content)) as image:
                 self.assertEqual(image.format, 'JPEG')
-                self.assertEqual(image.size, (1024, 1536))
+                self.assertEqual(image.size, (768, 1152))
             self.assertLess(len(content), 500000)
+            high = render_warehouse(renderer, rows, self.snapshot,
+                                    kind, 1, 1, high=True)
+            with Image.open(BytesIO(high)) as image:
+                self.assertEqual(image.size, (1024, 1536))
+            self.assertGreater(len(high), len(content))

tests/test_zhenxun_runtime.py

@@ -98,6 +98,13 @@ class ZhenxunRuntimeTests(unittest.IsolatedAsyncioTestCase):
             query.assert_awaited_once()
         with Image.open(BytesIO(content)) as image:
             self.assertEqual(image.format, 'JPEG')
+            self.assertEqual(image.size, (768, 1152))
+        high = await self.call('seeds', '1 高清', message=None)
+        with Image.open(BytesIO(high)) as image:
+            self.assertEqual(image.size, (1024, 1536))
+        self.assertIn('胡萝卜（#1） 2粒', await self.call(
+            'seeds', '文字', message=None,
+        ))
         self.assertEqual(before, await self.runtime.service.snapshot(
             self.runtime_identity(),
         ))
@@ -115,6 +122,37 @@ class ZhenxunRuntimeTests(unittest.IsolatedAsyncioTestCase):
         finally:
             self.runtime.rendering = rendering
 
+    async def test_steal_target_arguments_and_text_receipts(self):
+        from farm_core.domain import Plot
+        await self.call('register')
+        await self.call('register', user_id='2')
+        owner = replace(self.runtime_identity(), user_id='2')
+        plant = self.runtime.service.catalogue.get('胡萝卜')
+        now = self.runtime.service.clock()
+        await self.runtime.service.store.transact(
+            owner, 'fixture', 'fixture', {}, now,
+            lambda repo: repo.save_plot(Plot(1, plant, now)),
+        )
+        self.assertIn('偷菜', await self.call('steal'))
+        before = await self.runtime.service.snapshot(owner)
+        for account in ('1', '2', '9999'):
+            self.assertIn('不支持直接输入QQ号', await self.call(
+                'steal', account, message='numeric-'+account,
+            ))
+        self.assertEqual(await self.runtime.service.snapshot(owner), before)
+        self.assertIn('自己', await self.call('steal', target_user_id='1'))
+        self.assertIn('开通', await self.call('steal', target_user_id='9999'))
+        result = await self.call('steal', message='steal', target_user_id='2')
+        self.assertIsInstance(result, str)
+        self.assertIn('胡萝卜', result)
+        self.assertIn('4/5', result)
+        repeated = await self.call('steal', message='again', target_user_id='2')
+        self.assertIn('本季已偷', repeated)
+        self.assertIn('4/5', repeated)
+        self.assertIn('目标', await self.call(
+            'steal', '2', target_user_id='2', message='invalid',
+        ))
+
     async def test_original_commands_transactions_and_png(self):
         self.assertNotIn("测试农场", self.module.COMMANDS)
         self.assertIn("开通农场", self.module.COMMANDS)
@@ -442,6 +480,27 @@ class ZhenxunRuntimeTests(unittest.IsolatedAsyncioTestCase):
                     SimpleNamespace(extract_plain_text=lambda:""),
                 )
                 self.assertTrue(sent)
+                from unittest.mock import AsyncMock
+
+                class Arguments(list):
+                    def extract_plain_text(self):
+                        return ''
+
+                for segment in (
+                    SimpleNamespace(type='at', data={'qq': '123'}),
+                    SimpleNamespace(type='mention_user',
+                                    data={'user_id': '123'}),
+                ):
+                    with patch.object(imported.runtime, 'execute',
+                                      new=AsyncMock(return_value='结果')) as run:
+                        await hooks['commands']['偷菜'](
+                            bot, SimpleNamespace(message_id='steal'),
+                            SimpleNamespace(user=SimpleNamespace(
+                                id='u', name='用户',
+                            )), Arguments([segment]),
+                        )
+                        self.assertEqual(run.await_args.kwargs[
+                            'target_user_id'], '123')
             finally:
                 await hooks["close"][0]()
             self.assertFalse(imported.runtime.started)