188 lines
11 KiB
Python
188 lines
11 KiB
Python
"""Courses déterministes aux frontières externes, sans processus ni IPC réel."""
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
from common import *
|
|
from test_prewarm import Sway
|
|
from login_focus import checkpoint, UUID
|
|
from session_v2.executor import Executor
|
|
from session_v2.anchors import Anchors
|
|
from session_v2.recovery import Recovery, RecordedSway, focus_proof
|
|
from session_v2.ipc import workspace
|
|
from session_v2.errors import Failure
|
|
from session_v2.storage import read
|
|
from session_v2.startup import acknowledge, status
|
|
from session_v2.observability import Attempt
|
|
|
|
|
|
class BoundaryTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.tmp.cleanup)
|
|
self.runtime = Path(self.tmp.name)
|
|
self.sway = Sway()
|
|
self.sway.add('kitty', 11, focus=False)
|
|
self.sway.add('firefox', 12)
|
|
workspace(self.sway.value, '1')['layout'] = 'splith'
|
|
workspace(self.sway.value, '10')['layout'] = 'splith'
|
|
self.data = checkpoint()
|
|
self.engine = Executor(self.sway, Catalog(), object(), self.runtime, '/forbidden', process_identity=lambda pid: 123)
|
|
|
|
def test_guard_replacement_and_extra_refuse_before_any_mutation(self):
|
|
for mode in ('replacement', 'extra', 'pid', 'starttime', 'placement'):
|
|
with self.subTest(mode=mode), tempfile.TemporaryDirectory() as directory:
|
|
self.setUp()
|
|
self.engine.runtime = Path(directory)
|
|
prepared = self.engine._observe_prepared()
|
|
proof = focus_proof(self.sway)
|
|
def guard(*args):
|
|
node = next(n for n in workspace(self.sway.value, '1')['nodes'] if n['id'] == 11)
|
|
if mode == 'replacement': node['id'] = 21
|
|
if mode == 'extra': self.sway.add('kitty', 21, focus=False)
|
|
if mode == 'pid': node['pid'] += 1
|
|
if mode == 'starttime': self.engine.process_identity = lambda pid: 456
|
|
if mode == 'placement':
|
|
workspace(self.sway.value, '1')['nodes'].remove(node)
|
|
workspace(self.sway.value, '10')['nodes'].append(node)
|
|
from unittest.mock import Mock
|
|
return Mock(ready={'helpers_suspended': 0}, finish=lambda: {'helpers_resumed': 0, 'anchors_cleaned': 0})
|
|
from unittest.mock import Mock
|
|
anchors = Mock(owned=lambda n: False, close=lambda: None)
|
|
with patch('session_v2.executor.Anchors', return_value=anchors), patch('session_v2.executor.Guard', side_effect=guard):
|
|
with self.assertRaises(Failure):
|
|
self.engine._apply(self.data, 1, -1, proof, prepared)
|
|
self.assertEqual(self.sway.commands, [])
|
|
attempt = read(self.engine.runtime / 'session-v2-restore-attempt.json')
|
|
self.assertNotEqual(attempt['status'], 'success')
|
|
|
|
def test_extra_after_first_move_refuses_before_second_staging(self):
|
|
from unittest.mock import Mock
|
|
from session_v2.recovery import command_focus
|
|
from session_v2.ipc import walk
|
|
def command(text):
|
|
import json
|
|
self.sway.commands.append(text)
|
|
target = int(text.split('=', 1)[1].split(']', 1)[0])
|
|
operation = text.split('] ', 1)[1]
|
|
focus = command_focus(self.sway.value, target, operation)
|
|
node = next(n for n in walk(self.sway.value) if n['id'] == target)
|
|
for parent in walk(self.sway.value):
|
|
if node in parent.get('nodes', []):
|
|
parent['nodes'].remove(node)
|
|
break
|
|
stage = json.loads(operation.removeprefix('move container to workspace '))
|
|
destination = dict(id=90, type='workspace', name=stage, layout='splith', nodes=[node])
|
|
self.sway.value['nodes'][0]['nodes'].append(destination)
|
|
for n in walk(self.sway.value):
|
|
n['focused'] = n['id'] == focus['con_id'] if focus['con_id'] is not None else n.get('name') == focus['workspace']
|
|
self.sway.add('kitty', 21, focus=False)
|
|
self.sway.command = command
|
|
anchors = Mock(owned=lambda node: False, close=lambda: None)
|
|
guard = Mock(ready={'helpers_suspended': 0}, finish=lambda: {'helpers_resumed': 0, 'anchors_cleaned': 0})
|
|
prepared, proof = self.engine._observe_prepared(), focus_proof(self.sway)
|
|
# La reprise est hors de cette frontière : aucun cleanup ne doit
|
|
# masquer combien de commandes de staging ont effectivement été émises.
|
|
with patch('session_v2.executor.Anchors', return_value=anchors), \
|
|
patch('session_v2.executor.Guard', return_value=guard), patch.object(Recovery, 'recover'):
|
|
with self.assertRaisesRegex(Failure, 'APPLICATION_MEMBERSHIP_CHANGED'):
|
|
self.engine._apply(self.data, 1, -1, proof, prepared)
|
|
self.assertEqual(len(self.sway.commands), 1)
|
|
self.assertFalse(any('[con_id=21]' in text for text in self.sway.commands))
|
|
anchors.build.assert_not_called()
|
|
|
|
def final_fixture(self):
|
|
recovery = Recovery(self.sway, self.runtime, 'b'*32, 'labfy-v2-stage-'+'b'*32, start=lambda pid: 123)
|
|
recovery.baseline({'kitty', 'firefox', 'limusic-app'})
|
|
for node in self.sway.windows(): recovery.admit(node)
|
|
self.engine.sway = RecordedSway(self.sway, recovery)
|
|
anchors = Anchors.__new__(Anchors)
|
|
anchors.sway, anchors.data = self.engine.sway, self.data
|
|
anchors.owned = lambda n: False
|
|
mapping = {'F': 12, 'K': 11, 'M': 5}
|
|
# Le checkpoint impose F,K ; le préwarm initial créait K,F.
|
|
workspace(self.sway.value, '1')['nodes'].reverse()
|
|
for entry in recovery.data['windows']:
|
|
entry['expected'] = recovery.snapshot(entry['origin']['con_id'])
|
|
return recovery, anchors, mapping
|
|
|
|
def test_last_provider_capture_cannot_hide_structure_or_identity_change(self):
|
|
for mode in ('layout', 'swap', 'pid', 'starttime', 'extra', 'floating', 'output', 'focus', 'uuid'):
|
|
with self.subTest(mode=mode):
|
|
self.setUp()
|
|
recovery, anchors, mapping = self.final_fixture()
|
|
def capture(members):
|
|
ws = workspace(self.sway.value, '1')
|
|
if mode == 'layout': ws['layout'] = 'splitv'
|
|
if mode == 'swap': ws['nodes'].reverse()
|
|
if mode == 'pid': ws['nodes'][0]['pid'] += 1
|
|
if mode == 'starttime': recovery.start = lambda pid: 456
|
|
if mode == 'extra': self.sway.add('kitty', 21, focus=False)
|
|
if mode == 'floating': ws['nodes'][0]['floating'] = 'user_on'
|
|
if mode == 'output': self.sway.value['nodes'][0]['name'] = 'OTHER'
|
|
if mode == 'focus':
|
|
from session_v2.ipc import walk
|
|
for node in walk(self.sway.value): node['focused'] = node['id'] == 11
|
|
return {UUID: 11 if mode == 'uuid' else 12}
|
|
status(self.runtime, self.sway.session, self.data, True)
|
|
attempt = Attempt(self.runtime / 'session-v2-restore-attempt.json', 'b'*32, self.data)
|
|
attempt.update(status='failed', reason='TEST_ONLY')
|
|
from unittest.mock import Mock
|
|
self.engine.provider = Mock(capture=Mock(side_effect=capture))
|
|
with self.assertRaises(Failure):
|
|
self.engine._verify(self.data, mapping, {a['application_id']: Catalog().validate(a) for a in self.data['applications']}, anchors)
|
|
self.engine.provider.capture.assert_called_once()
|
|
with self.assertRaises(Failure): acknowledge(self.runtime, self.sway.session, 'restored', 'b'*32)
|
|
|
|
def test_unchanged_final_observation_completes(self):
|
|
recovery, anchors, mapping = self.final_fixture()
|
|
from unittest.mock import Mock
|
|
self.engine.provider = Mock(capture=lambda nodes: {UUID: 12})
|
|
specs = {a['application_id']: Catalog().validate(a) for a in self.data['applications']}
|
|
self.engine._verify(self.data, mapping, specs, anchors)
|
|
recovery.complete(lambda tree: self.engine._verify_structure(self.data, mapping, specs, anchors, tree))
|
|
self.assertEqual(recovery.data['status'], 'complete')
|
|
|
|
def test_unary_parent_normalization_preserves_canonical_layout(self):
|
|
recovery, anchors, mapping = self.final_fixture()
|
|
ws = workspace(self.sway.value, '1')
|
|
ws['nodes'] = [dict(id=90, type='con', layout='splith', nodes=ws['nodes'])]
|
|
specs = {a['application_id']: Catalog().validate(a) for a in self.data['applications']}
|
|
recovery.complete(lambda tree: self.engine._verify_structure(self.data, mapping, specs, anchors, tree))
|
|
self.assertEqual(recovery.data['status'], 'complete')
|
|
|
|
def test_layout_change_in_second_split_focus_observation_refuses(self):
|
|
recovery, anchors, mapping = self.final_fixture()
|
|
from session_v2.ipc import walk
|
|
ws = workspace(self.sway.value, '1')
|
|
for node in walk(self.sway.value): node['focused'] = False
|
|
parent = dict(id=90, type='con', layout='splith', nodes=ws['nodes'], focused=True)
|
|
ws['nodes'] = [parent]
|
|
recovery.data['focus_expected'] = dict(con_id=90, workspace='1')
|
|
raw_tree, calls = self.sway.tree, []
|
|
def tree():
|
|
calls.append(True)
|
|
if len(calls) == 2: parent['layout'] = 'splitv'
|
|
return raw_tree()
|
|
self.sway.tree = tree
|
|
specs = {a['application_id']: Catalog().validate(a) for a in self.data['applications']}
|
|
with self.assertRaisesRegex(Failure, 'FINAL_TREE_MISMATCH'):
|
|
self.engine._verify_structure(self.data, mapping, specs, anchors, self.sway.tree())
|
|
self.assertEqual(len(calls), 2)
|
|
|
|
def test_complete_refuses_full_identity_placement_and_pending(self):
|
|
for mode in ('pid', 'starttime', 'placement', 'pending', 'focus_pending', 'swap'):
|
|
with self.subTest(mode=mode):
|
|
self.setUp()
|
|
recovery, _, _ = self.final_fixture()
|
|
node = workspace(self.sway.value, '1')['nodes'][0]
|
|
if mode == 'pid': node['pid'] += 1
|
|
if mode == 'starttime': recovery.start = lambda pid: 456
|
|
if mode == 'placement': node['floating'] = 'user_on'
|
|
if mode == 'pending': recovery.data['windows'][0]['pending'] = True
|
|
if mode == 'focus_pending': recovery.data['focus_pending'] = True
|
|
if mode == 'swap': workspace(self.sway.value, '1')['nodes'].reverse()
|
|
with self.assertRaises(Failure): recovery.complete()
|
|
self.assertNotEqual(recovery.data['status'], 'complete')
|
|
|
|
|
|
if __name__ == '__main__': unittest.main()
|