"""Fake-owner/source contracts only; these do not compile WitcherScript.""" import copy import os from pathlib import Path import tempfile import threading import unittest from unittest.mock import patch from witcher_devkit import cli_app, debugger, native from witcher_devkit.debug_journal import Journal from witcher_devkit.debug_runtime import Runtime, validate, OperationUncertain class RequestWorld: def __init__(self): self.calls, self.rows = [], [] self.ticks = 0 self.fail = self.bad_page = False self.dropped = 0 self.originals = {} self.tags = {} self.status = dict(window_id=1, generation=17, actor_ids=[22, 23], active=False, armed=False, records=0, dropped=0, mode='observe', scores=[0, 0], instance_count_fields_modified=False, ticket_sources_modified=False) def call(self, command, args='', timeout=5): self.calls.append((command, args)) data = {} if command == 'snapshot': data = dict(ready=True, ticks=self.ticks, input_ticks=0, remaining_ticks=0, input_button='None', world=dict(devkit_paused=True, saves_locked=True, player=dict(alive=True))) elif command == 'request_algorithm_arm': parts = args.split('|') self.status.update(actor_ids=list(map(int,parts[1].split(','))), mode=parts[4], scores=list(map(float, parts[5].split(','))), active=True, armed=True, records=0) data = copy.deepcopy(self.status) elif command == 'advance': self.ticks += int(args) self.rows = [] for i, actor in enumerate(self.status['actor_ids']): original = self.originals.get(actor, 100+i) returned = self.status['scores'][i] if self.status['mode']=='score' and original > 0 else original self.rows.append(dict(sequence=i+1, window_id=1, generation=17, actor_id=actor, algorithm_instance=i+1, original_score=original, returned_score=returned, binding_tag=self.tags.get(actor,''), binding_tag_status='declared' if self.tags.get(actor) else 'untagged')) self.status['records'] = len(self.rows) elif command == 'request_algorithm_stop': if self.fail: raise TimeoutError('stop acknowledgement lost') self.status.update(active=False, armed=False, dropped=self.dropped) data = copy.deepcopy(self.status) elif command == 'request_algorithm_status': data = copy.deepcopy(self.status) elif command == 'request_algorithm_page': start = int(args.split('|')[-1]) data = dict(window_id=1, generation=17, start=start, total=len(self.rows), dropped=self.dropped, records=copy.deepcopy(self.rows[start:start+4])) if self.bad_page: data['window_id'] = 2 return dict(ok=True, generation=17, data=data) class RequestTests(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory() self.addCleanup(self.temp.cleanup) self.runtime = Runtime(Journal(Path(self.temp.name)), threading.Event()) self.world = RequestWorld() self.enterContext(patch.object(native, 'call', side_effect=self.world.call)) self.enterContext(patch.object(native, 'state', return_value={})) def run_window(self, **kw): args = dict(actors=['17|22', '17|23'], ticks=3) args.update(kw) return self.runtime.execute('request-algorithm', args, 5) def test_finite_owner_window_and_disarm(self): result = self.run_window()['request_algorithm'] self.assertTrue(result['runtime_restoration_verified']) self.assertTrue(result['collection']['complete']) self.assertFalse(self.world.status['armed']) self.assertEqual(self.world.calls.count(('advance', '3')), 1) self.assertEqual(self.world.calls.count(('request_algorithm_stop', '17|1')), 1) payload = next(a for c, a in self.world.calls if c == 'request_algorithm_arm').split('|') self.assertEqual(payload[:3], ['17', '22,23', '3']) self.assertTrue(1 <= int(payload[3]) <= 60000) self.assertEqual(payload[4:], ['observe', '0.000000,0.000000']) def test_scores_are_ordered_and_only_one_step(self): self.run_window(mode='score', scores=[121, 122]) payload = next(a for c, a in self.world.calls if c == 'request_algorithm_arm') self.assertTrue(payload.endswith('|score|121.000000,122.000000')) def test_all_cardinalities_observe_and_ordered_score(self): for count in range(1,9): # Reverse IDs to catch accidental sorting or positional assumptions. ids = list(reversed(range(22,22+count))) refs = [f'17|{actor}' for actor in ids] for mode in ('observe','score'): with self.subTest(count=count,mode=mode): scores = [121+i for i in range(count)] if mode=='score' else [] result = self.run_window(actors=refs,mode=mode,scores=scores)['request_algorithm'] self.assertEqual(result['arm']['data']['actor_ids'],ids) self.assertEqual(result['stop']['data']['actor_ids'],ids) collection = result['collection'] self.assertTrue(collection['complete']) self.assertEqual([row['actor_id'] for row in collection['records']],ids) self.assertEqual(len(collection['pages']), (count+3)//4) expected = scores if mode=='score' else list(range(100,100+count)) self.assertEqual([row['returned_score'] for row in collection['records']],expected) self.assertFalse(self.world.status['armed']) def test_nonpositive_originals_unchanged_among_eight_actors(self): refs = [f'17|{actor}' for actor in range(22,30)] self.world.originals = {22:0,29:-5} result = self.run_window(actors=refs,mode='score',scores=[125]*8)['request_algorithm'] rows = result['collection']['records'] self.assertEqual([row['returned_score'] for row in rows],[0]+[125]*6+[-5]) self.assertTrue(result['collection']['complete']) def test_tags_follow_actual_actor_not_selection_position(self): self.world.tags={22:'A',23:'B'} for refs in (['17|22','17|23'],['17|23','17|22']): result=self.run_window(actors=refs)['request_algorithm'] rows=result['collection']['records'] self.assertEqual({row['actor_id']:row['binding_tag'] for row in rows},{22:'A',23:'B'}) self.assertTrue(all(row['original_score']==row['returned_score'] for row in rows)) self.world.tags={} rows=self.run_window()['request_algorithm']['collection']['records'] self.assertTrue(all(row['binding_tag_status']=='untagged' and row['binding_tag']=='' for row in rows)) def test_oversized_tag_is_not_accepted_as_attestation(self): self.world.tags={22:'x'*65} with self.assertRaises(OperationUncertain): self.run_window() def test_raw_protocol_cardinality_bounds_and_mode_constraints(self): for count in range(1,9): ids = ','.join(map(str,range(count))) for mode, score in [('observe','0'),('score','1000000.000000')]: payload=f"17|{ids}|600|60000|{mode}|{','.join([score]*count)}" self.assertLessEqual(len(payload),512) native.validate('request_algorithm_arm',payload) for payload in ( '17||3|1000|observe|0', '17|22,022|3|1000|observe|0,0', '17|512|3|1000|observe|0', '17|0,1,2,3,4,5,6,7,8|3|1000|observe|0,0,0,0,0,0,0,0,0', '17|22,23|3|1000|observe|0', '17|22|3|1000|observe|0,0', '17|22|3|1000|observe|1', '17|22|3|1000|score|0', '17|22|3|1000|score|0.0000001', '17|22|3|1000|score|1000001', '17|22|3|1000|score|nan', '17|22|3|1000|score|inf', '17|22|3|1000|score|-1', '17|22|3|1000|score|1,', '17|22|23|3|1000|observe|0|0'): with self.subTest(payload=payload), self.assertRaises(ValueError): native.validate('request_algorithm_arm',payload) def test_cancel_before_dispatch_still_disarms(self): self.runtime.cancelled.set() self.assertEqual(self.run_window()['stop_reason'], 'cancelled') self.assertFalse(self.world.status['armed']) self.assertFalse(any(c == 'advance' for c, _ in self.world.calls)) def test_drop_marks_collection_incomplete(self): self.world.dropped = 1 self.assertFalse(self.run_window()['request_algorithm']['collection']['complete']) def test_readback_is_nonmutating_and_retains_records(self): self.run_window() self.world.calls.clear() result = self.runtime.execute('inspect', dict(view='request-algorithm'), 5) self.assertEqual(len(result['data']['collection']['records']), 2) self.assertEqual([c for c, _ in self.world.calls], ['request_algorithm_status', 'request_algorithm_page']) def test_stale_reference_does_not_arm(self): with self.assertRaises(ValueError): self.runtime.execute('request-algorithm', dict(actors=['18|22','18|23'],ticks=3), 5) self.assertFalse(any(c == 'request_algorithm_arm' for c, _ in self.world.calls)) def test_wall_stop_and_cleanup_without_tick_progress(self): with patch.object(self.runtime, 'block', return_value=dict(ok=True,stop_reason='wall_budget')): result = self.run_window() self.assertEqual(result['stop_reason'], 'wall_budget') self.assertFalse(self.world.status['armed']) def test_cleanup_and_page_identity_failure_are_uncertain(self): for field in ('fail', 'bad_page'): with self.subTest(field=field): setattr(self.world, field, True) with self.assertRaises(OperationUncertain): self.run_window() setattr(self.world, field, False) def test_validation(self): base = dict(actors=['17|22', '17|23'], ticks=3) validate('request-algorithm', base) for changes in (dict(actors=[]), dict(actors=[f'17|{i}' for i in range(9)]), dict(actors=['17|22', '17|022']), dict(actors=['17|22', '18|23']), dict(actors=[[], {}]), dict(actors=['17|22','17|23','17|23']), dict(actors=['17|22','17|23','18|24']), dict(actors=['17|512']), dict(mode='count'), dict(algorithm='attack'), dict(mode='score',scores=[100]), dict(mode='score',scores=[100,101,102]), dict(mode='score', scores=[0, 100]), dict(mode='score', scores=[float('nan'), 100]), dict(mode='score', scores=[1e-9, 100]), dict(scores=[100, 100]), dict(ticks=601), dict(input='0|0|0|0|None|3')): with self.subTest(changes=changes), self.assertRaises(ValueError): validate('request-algorithm', dict(base, **changes)) with self.assertRaises(ValueError): validate('request-algorithm',dict(base,binding_tag='host-cannot-supply-runtime-tag')) def test_cli_contract(self): a = cli_app.parser().parse_args(['debug', 'request-algorithm', '--actor', '17|22', '--actor', '17|23', '--ticks', '3']) from witcher_devkit.debug_cli import operation self.assertEqual(operation(a)[0], 'request-algorithm') a.wall_seconds = 61 with self.assertRaises(ValueError): operation(a) a = cli_app.parser().parse_args(['debug','request-algorithm','--actor','17|22','--ticks','3']) self.assertEqual(operation(a)[1]['actors'],['17|22']) def test_wrapper_source_preserves_original_and_never_writes_count(self): source = (Path(__file__).resolve().parents[1] / 'scripts/request_algorithm.ws.in').read_text() wrapper = source[source.index('@wrapMethod(CTicketAlgorithmApproach)'):] self.assertEqual(wrapper.count('wrappedMethod()'), 1) self.assertNotIn('overrideTicketsCount =', source) self.assertNotIn('TicketSourceOverrideRequest', source) self.assertNotIn('ShouldAskForTicket(', source) self.assertIn('pdRAMode == "score" && original > 0', source) self.assertIn('if(!PD_RequestLeaseAlive())', source) self.assertIn('pdRARecords.Size() >= 128', source) self.assertIn('ids.Size() < 1 || ids.Size() > 8 || scores.Size() != ids.Size()',source) self.assertIn('for(j = 0; j < i; j += 1)',source) self.assertIn('for(i = 0; i < pdRAActors.Size(); i += 1)',source) self.assertNotIn('pdRAIds.Size() != 2',source) self.assertNotIn('pdRAScores.Size() != 2',source) self.assertIn('GetTickCount64()', (Path(__file__).resolve().parents[1] / 'native/PathInput.cpp').read_text()) self.assertEqual(source.count('algorithm.PDRequestBindingTag()'),1) self.assertLess(source.index('returned = original'),source.index('algorithm.PDRequestBindingTag()')) self.assertIn('quotedTag = PDQuote(rawTag)',source) self.assertNotIn('PDQuote(tagStatus)',source) class BrokerTemporaryTests(unittest.TestCase): def test_lifetime_temp_exists_and_environment_is_restored(self): with tempfile.TemporaryDirectory() as directory: seen = [] def owned(mode, seconds): current = Path(os.environ['TMPDIR']) seen.append(current) self.assertTrue(current.is_dir()) self.assertEqual(os.environ['TMP'], str(current)) self.assertEqual(os.environ['TEMP'], str(current)) self.assertEqual(current.parent, Path(directory)/'tmp') raise RuntimeError('startup failure') with patch.object(native, 'STATE_ROOT', Path(directory)), \ patch.object(debugger, 'serve_owned', side_effect=owned), \ patch.dict(os.environ, {'TMPDIR': '/missing/terminal/tmp', 'TMP': 'old', 'TEMP': 'old'}): with self.assertRaises(RuntimeError): debugger.serve('open', 10) self.assertEqual(os.environ['TMPDIR'], '/missing/terminal/tmp') self.assertFalse(seen[0].exists())