"""Executed ONLY by the isolated backend test. Never uses the host system bus.""" import array import fcntl import json import os from pathlib import Path import select import socket import struct import subprocess import sys import tempfile import threading import time import unittest from urllib.parse import unquote from unittest.mock import patch, Mock assert os.environ.get('ARCHY_TEST_ISOLATED') == '1', 'isolated runner required' sys.path.insert(0, str(Path(__file__).resolve().parents[1])) import gi gi.require_version('Gio', '2.0') from gi.repository import Gio, GLib import c01_network_settings_nm as nm import c01_network_settings_owner as owner import c01_network_settings_watch as watch V = GLib.Variant D = nm.ROOT + '/Devices/1' P = nm.ROOT + '/Settings/1' A = nm.ROOT + '/ActiveConnection/1' IP4 = nm.ROOT + '/IP4Config/1' AP = nm.ROOT + '/AccessPoint/1' class FakeNM: """Actual disposable message bus service, not a mocked adapter method.""" def __init__(self): self.test_bus = Gio.TestDBus.new(Gio.TestDBusFlags.NONE) self.test_bus.up() flags = Gio.DBusConnectionFlags.AUTHENTICATION_CLIENT | Gio.DBusConnectionFlags.MESSAGE_BUS_CONNECTION address = self.test_bus.get_bus_address() self.server = Gio.DBusConnection.new_for_address_sync(address, flags, None, None) self.client = Gio.DBusConnection.new_for_address_sync(address, flags, None, None) self.server.call_sync('org.freedesktop.DBus', '/org/freedesktop/DBus', 'org.freedesktop.DBus', 'RequestName', V('(su)', (nm.SERVICE, 0)), None, Gio.DBusCallFlags.NONE, 1000, None) self.version, self.applied_version, self.device_type = 3, 7, 1 self.calls, self.fail_method, self.bad_readback = [], None, False self.pending_state_reads, self.state_reads = 0, 0 self.ap_ssid, self.ap_rsn = b'fixture-only', 0x100 self.settings = { 'connection': {'id': V('s', 'operator'), 'uuid': V('s', '12345678-1234-1234-1234-123456789abc'), 'type': V('s', '802-3-ethernet')}, 'ipv4': {'method': V('s', 'manual'), 'address-data': V('aa{sv}', [{'address': V('s', '192.0.2.8'), 'prefix': V('u', 24)}]), 'gateway': V('s', '192.0.2.1'), 'dns': V('au', [0x09090909]), 'ignore-auto-dns': V('b', True), 'route-metric': V('x', 77), 'dns-search': V('as', ['operator.example'])}, 'ipv6': {'method': V('s', 'auto')}, '802-3-ethernet': {'mtu': V('u', 1400)}} self.applied = dict(self.settings) self.registrations = [] self.register(nm.ROOT, nm.SERVICE, ''' ''') self.register(nm.ROOT + '/Settings', nm.SERVICE + '.Settings', ''' ''') self.register(D, nm.DEVICE, ''' ''') self.register(IP4, nm.SERVICE + '.IP4Config', ''' ''') self.register(D, nm.DEVICE + '.Wireless', ''' ''') self.register(AP, nm.SERVICE + '.AccessPoint', ''' ''') self.register(P, nm.CONNECTION, ''' ''') self.register(A, nm.SERVICE + '.Connection.Active', ''' ''') self.loop = GLib.MainLoop() self.thread = threading.Thread(target=self.loop.run) self.thread.start() def register(self, path, interface, xml): info = Gio.DBusNodeInfo.new_for_xml('' + xml + '') self.registrations.append(self.server.register_object(path, info.interfaces[0], self.method, self.prop, None)) def prop(self, connection, sender, path, interface, name): if name == 'Ssid': return V('ay', list(self.ap_ssid)) if name == 'RsnFlags': return V('u', self.ap_rsn) if name == 'State' and path == D: self.state_reads += 1 if self.pending_state_reads: self.pending_state_reads -= 1 return V('u', 70) return V('u', 100) if name == 'Ip4Config': return V('o', IP4) if name == 'AddressData': group = self.applied['ipv4'] if group['method'].unpack() == 'manual': return group['address-data'] return V('aa{sv}', [{'address': V('s', '198.51.100.9'), 'prefix': V('u', 24)}]) if name == 'Gateway': return self.applied['ipv4'].get('gateway', V('s', '')) return {'Interface': V('s', 'eth0'), 'DeviceType': V('u', self.device_type), 'Managed': V('b', True), 'Real': V('b', True), 'Software': V('b', False), 'ActiveConnection': V('o', A), 'Connection': V('o', P), 'State': V('u', 2), 'VersionId': V('t', self.version)}[name] def method(self, connection, sender, path, interface, method, args, invocation): self.calls.append(method) if method == self.fail_method: invocation.return_dbus_error('org.freedesktop.NetworkManager.Error.Failed', 'private NM error must not escape') return try: if method == 'GetDevices': result = V('(ao)', ([D],)) elif method == 'GetAllAccessPoints': result = V('(ao)', ([AP],)) elif method == 'ListConnections': result = V('(ao)', ([P],)) elif method == 'GetSettings': public = {group: dict(values) for group, values in self.settings.items()} public.get('802-11-wireless-security', {}).pop('psk', None) result = V('(a{sa{sv}})', (public,)) elif method == 'GetAppliedConnection': result = V('(a{sa{sv}}t)', (self.applied, self.applied_version)) elif method == 'Update2': assert args.unpack()[2]['version-id'] == self.version self.settings = nm.variant_settings(args.get_child_value(0)) self.version += 1 result = V('(a{sv})', ({},)) elif method == 'Reapply': assert args.unpack()[1] == self.applied_version self.applied = nm.variant_settings(args.get_child_value(0)) if self.bad_readback: self.applied['ipv4']['method'] = V('s', 'disabled') self.applied_version += 1 result = V('()', ()) elif method == 'AddConnection2': self.settings = nm.variant_settings(args.get_child_value(0)) self.version += 1 result = V('(oa{sv})', (P, {})) elif method == 'ActivateConnection': assert args.unpack() == (P, D, '/') self.applied = self.settings result = V('(o)', (A,)) else: raise AssertionError(method) invocation.return_value(result) except BaseException: invocation.return_dbus_error('org.freedesktop.NetworkManager.Error.Failed', 'fixture assertion') def close(self): self.client.close_sync(None) self.server.close_sync(None) self.loop.quit() self.thread.join(2) self.test_bus.down() class ExecutorTests(unittest.TestCase): def setUp(self): self.bus = FakeNM() self.directory = tempfile.TemporaryDirectory() self.fd = os.open(self.directory.name, os.O_RDONLY | os.O_DIRECTORY) self.journal = owner.Journal(self.fd) self.owner = owner.Owner(self.journal, lambda: nm.NetworkManager(self.bus.client)) def tearDown(self): if self.owner.thread: self.owner.thread.join(3) self.bus.close() os.close(self.fd) self.directory.cleanup() def execute(self, mutation): first = self.owner.submit(mutation) self.owner.thread.join(5) self.assertFalse(self.owner.thread.is_alive()) return self.owner.status(first['ticket'])['phase'] def test_dhcp_actual_bus_preserves_operator_dns_routes_and_mtu(self): self.assertEqual(self.execute({'Ethernet': {'interface': 'eth0', 'ipv4': 'Dhcp'}}), 'Succeeded') settings = self.bus.settings self.assertEqual(settings['ipv4']['method'].unpack(), 'auto') self.assertEqual(settings['ipv4']['address-data'].unpack(), []) self.assertEqual(settings['ipv4']['gateway'].unpack(), '') self.assertEqual(settings['ipv4']['dns'].unpack(), [0x09090909]) self.assertEqual(settings['ipv4']['route-metric'].unpack(), 77) self.assertEqual(settings['802-3-ethernet']['mtu'].unpack(), 1400) self.assertEqual(self.bus.calls.count('Update2'), 1) self.assertFalse(self.journal.blocked()) def test_static_and_mixed_dns_actual_bus(self): mutation = {'Ethernet': {'interface': 'eth0', 'ipv4': {'Static': {'address': '192.0.2.2', 'prefix': 24, 'gateway': '192.0.2.1', 'dns': {'Servers': ['1.1.1.1', '2606:4700:4700::1111']}}}}} self.assertEqual(self.execute(mutation), 'Succeeded') self.assertEqual(self.bus.settings['ipv4']['address-data'].unpack()[0]['address'], '192.0.2.2') self.assertEqual(self.bus.settings['ipv6']['dns'].unpack()[0][-2:], [17, 17]) self.assertTrue(self.bus.settings['ipv6']['ignore-auto-dns'].unpack()) def test_ethernet_waits_for_device_activation_and_address(self): self.bus.pending_state_reads = 2 self.assertEqual(self.execute({'Ethernet': {'interface': 'eth0', 'ipv4': 'Dhcp'}}), 'Succeeded') self.assertGreaterEqual(self.bus.state_reads, 3) def test_system_dns_clears_both_families_preserves_search(self): self.assertEqual(self.execute({'Dns': {'policy': 'Automatic'}}), 'Succeeded') for group in ('ipv4', 'ipv6'): self.assertEqual(self.bus.settings[group]['dns'].unpack(), []) self.assertFalse(self.bus.settings[group]['ignore-auto-dns'].unpack()) self.assertEqual(self.bus.settings['ipv4']['dns-search'].unpack(), ['operator.example']) def test_wifi_secret_dbus_profile_and_activation(self): self.bus.device_type = 2 mutation = {'Wifi': {'ssid': 'fixture-only', 'password': 'fixture-secret'}} self.assertEqual(self.execute(mutation), 'Succeeded') self.assertIn('AddConnection2', self.bus.calls) self.assertIn('ActivateConnection', self.bus.calls) self.assertEqual(self.bus.settings['802-11-wireless-security']['psk'].unpack(), 'fixture-secret') self.assertEqual(mutation['Wifi']['password'], '') record = Path(self.directory.name, 'operation.json').read_text() self.assertNotIn('fixture-secret', record) self.assertNotIn('fixture-only', record) def test_operator_wifi_profile_not_deleted_or_overwritten(self): self.bus.device_type = 2 self.bus.settings['802-11-wireless'] = {'ssid': V('ay', list(b'operator-wifi'))} self.assertEqual(self.execute({'Wifi': {'ssid': 'operator-wifi', 'password': 'fixture-secret'}}), 'Rejected') self.assertNotIn('Update2', self.bus.calls) self.assertNotIn('AddConnection2', self.bus.calls) self.assertFalse(self.journal.blocked()) def test_wpa3_long_password_selects_sae_and_preserves_full_secret(self): self.bus.device_type, self.bus.ap_rsn = 2, 0x400 secret = 'owner-chosen long passphrase ' * 5 mutation = {'Wifi': {'ssid': 'fixture-only', 'password': secret}} self.assertEqual(self.execute(mutation), 'Succeeded') security = self.bus.settings['802-11-wireless-security'] self.assertEqual(security['key-mgmt'].unpack(), 'sae') self.assertEqual(security['psk'].unpack(), secret) self.assertEqual(security['pmf'].unpack(), 3) self.assertEqual(mutation['Wifi']['password'], '') def test_wpa2_long_nonhex_password_rejected_before_profile_mutation(self): self.bus.device_type = 2 self.assertEqual(self.execute({'Wifi': {'ssid': 'fixture-only', 'password': 'z' * 80}}), 'Rejected') self.assertNotIn('AddConnection2', self.bus.calls) self.assertNotIn('ActivateConnection', self.bus.calls) self.assertFalse(self.journal.blocked()) def test_sae_advertised_by_another_ssid_does_not_authorize_long_wpa2_password(self): self.bus.device_type, self.bus.ap_rsn, self.bus.ap_ssid = 2, 0x400, b'another-network' self.assertEqual(self.execute({'Wifi': {'ssid': 'fixture-only', 'password': 'z' * 80}}), 'Rejected') self.assertNotIn('AddConnection2', self.bus.calls) def test_wpa2_raw_hex_key_is_preserved(self): self.bus.device_type = 2 self.assertEqual(self.execute({'Wifi': {'ssid': 'fixture-only', 'password': 'a' * 64}}), 'Succeeded') self.assertEqual(self.bus.settings['802-11-wireless-security']['key-mgmt'].unpack(), 'wpa-psk') self.assertEqual(self.bus.settings['802-11-wireless-security']['psk'].unpack(), 'a' * 64) def test_saved_sae_profile_never_downgrades_for_short_password(self): self.bus.device_type = 2 self.bus.settings['802-11-wireless'] = {'ssid': V('ay', list(b'owned-wifi'))} self.bus.settings['user'] = {'data': V('a{ss}', {'archipelago.network-settings': '1'})} self.bus.settings['802-11-wireless-security'] = {'key-mgmt': V('s', 'sae')} self.assertEqual(self.execute({'Wifi': {'ssid': 'owned-wifi', 'password': 'short'}}), 'Succeeded') self.assertEqual(self.bus.settings['802-11-wireless-security']['key-mgmt'].unpack(), 'sae') def test_owned_wifi_preserves_explicit_security_and_refuses_downgrade(self): self.bus.device_type = 2 self.bus.settings['802-11-wireless'] = {'ssid': V('ay', list(b'owned-wifi'))} self.bus.settings['user'] = {'data': V('a{ss}', {'archipelago.network-settings': '1'})} self.bus.settings['802-11-wireless-security'] = {'key-mgmt': V('s', 'wpa-psk'), 'pmf': V('u', 3)} self.assertEqual(self.execute({'Wifi': {'ssid': 'owned-wifi', 'password': ''}}), 'Rejected') self.assertNotIn('Update2', self.bus.calls) self.assertEqual(self.execute({'Wifi': {'ssid': 'owned-wifi', 'password': 'fixture-secret'}}), 'Succeeded') self.assertEqual(self.bus.settings['802-11-wireless-security']['pmf'].unpack(), 3) def test_partial_failure_fences_restart_and_never_retries(self): self.bus.fail_method = 'Reapply' self.assertEqual(self.execute({'Ethernet': {'interface': 'eth0', 'ipv4': 'Dhcp'}}), 'Unknown') restored = owner.Owner(owner.Journal(self.fd), lambda: nm.NetworkManager(self.bus.client)) self.assertEqual(restored.submit({'Dns': {'policy': 'Automatic'}})['phase'], 'Rejected') self.assertEqual(self.bus.calls.count('Update2'), 1) self.assertNotIn('private NM error', Path(self.directory.name, 'operation.json').read_text()) def test_readback_mismatch_is_not_success(self): self.bus.bad_readback = True self.assertEqual(self.execute({'Ethernet': {'interface': 'eth0', 'ipv4': 'Dhcp'}}), 'Unknown') self.assertTrue(self.journal.blocked()) def test_expired_before_effect_no_write(self): adapter = nm.NetworkManager(self.bus.client) with self.assertRaises(ValueError): adapter.execute({'Dns': {'policy': 'Automatic'}}, 0, lambda paths: self.fail('late mutation'), lambda: None) self.assertNotIn('Update2', self.bus.calls) def test_real_nonroot_peer_owner_loop_and_supervisor_watch(self): # Real nonroot connect(), not a root-created application socketpair. self.bus.pending_state_reads = 2 address = '\0network-settings-test-' + os.urandom(16).hex() listener = socket.socket(socket.AF_UNIX, socket.SOCK_SEQPACKET) listener.setsockopt(socket.SOL_SOCKET, socket.SO_PASSCRED, 1) listener.bind(address) listener.listen(1) listener.settimeout(3) ready_read, ready_write = os.pipe() app_code = '''import os,socket,json,time,sys os.setgroups([]); os.setgid(65534); os.setuid(65534) s=socket.socket(socket.AF_UNIX,socket.SOCK_SEQPACKET) s.connect(bytes.fromhex(sys.argv[1]).decode()) if len(sys.argv)>2: owner_pid=os.read(int(sys.argv[3]),64).decode().strip() os.dup2(s.fileno(),11,inheritable=True) os.execve(int(sys.argv[2]),[sys.argv[4],'--exact','network::dns::settings::transport::tests::network_settings_transport_application_child','--nocapture'],{'ARCHY_TEST_ISOLATED':'1','ARCHY_NETWORK_OWNER_PID':owner_pid,'RUST_MIN_STACK':'16777216'}) s.settimeout(5) s.send(json.dumps({'version':1,'sequence':1,'operation':{'Submit':{'Ethernet':{'interface':'eth0','ipv4':'Dhcp'}}}}).encode()) r=json.loads(s.recv(4096)); seq=1 while r['receipt']['phase'] not in ('Succeeded','Rejected','Unknown'): time.sleep(.05); seq+=1 s.send(json.dumps({'version':1,'sequence':seq,'operation':{'Status':r['receipt']['ticket']}}).encode()) r=json.loads(s.recv(4096)) assert r['receipt']['phase']=='Succeeded',r print('verified',flush=True) time.sleep(1) ''' executable = os.environ.get('ARCHY_NETWORK_TEST_EXE') # The shared target has private ancestors. Retain the already-built ELF # descriptor rather than weakening report-directory permissions for nobody. executable_fd = os.open(executable, os.O_RDONLY | os.O_CLOEXEC) if executable else None if executable_fd is not None: original = executable_fd executable_fd = fcntl.fcntl(original, fcntl.F_DUPFD_CLOEXEC, 32) os.close(original) arguments = [str(executable_fd), str(ready_read), executable] if executable else [] app = subprocess.Popen(['/usr/bin/python3', '-I', '-c', app_code, address.encode().hex(), *arguments], pass_fds=(ready_read, *([executable_fd] if executable else [])), stdout=subprocess.PIPE, stderr=subprocess.PIPE) if executable_fd is not None: os.close(executable_fd) os.close(ready_read) control, child_control = socket.socketpair(socket.AF_UNIX, socket.SOCK_SEQPACKET) for channel in (control, child_control): channel.setsockopt(socket.SOL_SOCKET, socket.SO_PASSCRED, 1) worker = None accepted = None bus_link = None try: accepted, _ = listener.accept() accepted.setsockopt(socket.SOL_SOCKET, socket.SO_PASSCRED, 1) self.assertEqual(struct.unpack('3i', accepted.getsockopt(socket.SOL_SOCKET, socket.SO_PEERCRED, 12)), (app.pid, 65534, 65534)) # Inject only the disposable bus in this fixture interpreter. Production # native embedding clears the environment and uses Gio's system bus. code = '''import os,sys,fcntl fds=[fcntl.fcntl(int(n),fcntl.F_DUPFD_CLOEXEC,32) for n in sys.argv[1:4]] for target,fd in zip((3,4,5),fds): os.dup2(fd,target); os.close(fd) for original in map(int,sys.argv[1:4]): if original not in (3,4,5): os.close(original) if len(sys.argv)>6: os.execve(sys.argv[6],[sys.argv[6],'network-settings-owner-v1'],{}) sys.path.insert(0,sys.argv[4]) import gi gi.require_version('Gio','2.0') from gi.repository import Gio import c01_network_settings_owner as owner import c01_network_settings_nm as nm original=nm.NetworkManager address=sys.argv[5] def factory(): bus=Gio.DBusConnection.new_for_address_sync(address,Gio.DBusConnectionFlags.AUTHENTICATION_CLIENT|Gio.DBusConnectionFlags.MESSAGE_BUS_CONNECTION,None,None) return original(bus) owner.Owner.__init__.__defaults__=(factory,) owner.main() ''' native = os.environ.get('ARCHY_NETWORK_NATIVE_EXE') if native: # /run is the isolated runner's private tmpfs, never the host bus. os.makedirs('/run/dbus', mode=0o755, exist_ok=True) bus_link = Path('/run/dbus/system_bus_socket') self.assertFalse(bus_link.exists()) address = self.bus.test_bus.get_bus_address() self.assertTrue(address.startswith('unix:path=')) bus_link.symlink_to(unquote(address[len('unix:path='):].split(',')[0])) worker = subprocess.Popen(['/usr/bin/python3', '-I', '-c', code, str(child_control.fileno()), str(accepted.fileno()), str(self.fd), str(Path(__file__).resolve().parents[1]), self.bus.test_bus.get_bus_address(), *([native] if native else [])], pass_fds=(child_control.fileno(), accepted.fileno(), self.fd), stdout=subprocess.DEVNULL, stderr=subprocess.PIPE) child_control.close() observer = watch.Watch(control, accepted, worker.pid) meta = os.fstat(self.fd) owner.send(control, {'version': 1, 'process_id': worker.pid, 'application': {'pid': app.pid, 'uid': 65534, 'gid': 65534}, 'journal_device': meta.st_dev, 'journal_inode': meta.st_ino}) if executable: os.write(ready_write, str(worker.pid).encode()) os.close(ready_write) ready_write = None end = time.monotonic() + 8 while time.monotonic() < end and app.poll() is None: if not observer.receive(): # App exit revokes this dedicated role; that is expected only # after the real typed client has completed successfully. app.wait(timeout=1) break time.sleep(.02) stdout, stderr = app.communicate(timeout=3) self.assertEqual(app.returncode, 0, stderr.decode()) self.assertIn(b'verified', stdout) self.assertEqual(self.bus.calls.count('Update2'), 1) self.assertGreaterEqual(self.bus.state_reads, 3) self.assertEqual(owner.Journal(self.fd).record['phase'], 'Succeeded') if native: print('NATIVE network-settings-owner-v1 -> disposable NM bus: verified', flush=True) finally: if ready_write is not None: os.close(ready_write) for process in (app, worker): if process is not None: if process.poll() is None: process.kill() process.communicate(timeout=3) control.close() child_control.close() if accepted: accepted.close() listener.close() if bus_link is not None and bus_link.is_symlink(): bus_link.unlink() class ProtocolTests(unittest.TestCase): def test_supervisor_role_budget_retirement_and_admission_refusal(self): import c01_supervisor as supervisor self.assertEqual(supervisor.lifecycle.ROLES['network-settings'], (128 * 1024**2, 16)) with self.assertRaises(supervisor.AdmissionUnavailable): supervisor.admit({}, None, None) class Children: stopping = False results = {} def request_stop(self): self.stopping = True def poll(self): return [] instance = object.__new__(supervisor.Supervisor) instance.children = Children() instance.failure = None instance.watch = None instance.tor_watch = None instance.power_lease = None instance.network_settings_watch = Mock() instance.network_settings_watch.receive.return_value = False instance.access_lease = instance.runtime_lease = instance.network_lease = None instance.completed_one_shots = {'administrator'} self.assertFalse(instance.tick()) instance.network_settings_watch.revoke.assert_called() instance.network_settings_watch.release_reaped.assert_not_called() instance.children.results = {'network-settings': 0} self.assertFalse(instance.tick()) instance.network_settings_watch.release_reaped.assert_called_once() def test_owner_validates_decoded_types_and_unknown_fields(self): invalid = [ {'Ethernet': {'interface': 'eth0', 'ipv4': {'Static': {'address': 1, 'prefix': 24, 'gateway': None, 'dns': 'Preserve'}}}}, {'Wifi': {'ssid': 'a', 'password': '', 'argv': []}}, {'Dns': {'policy': {'Servers': ['::']}}}, {'Dns': {'policy': 'Preserve'}}, ] for mutation in invalid: with self.assertRaises(ValueError): nm.validate(mutation) with self.assertRaises(ValueError): owner.validate_ticket({'epoch': [True] * 16, 'sequence': 1}) def pair(self): a, b = socket.socketpair(socket.AF_UNIX, socket.SOCK_SEQPACKET) a.setsockopt(socket.SOL_SOCKET, socket.SO_PASSCRED, 1) self.addCleanup(a.close) self.addCleanup(b.close) return a, b def test_authenticated_packets_and_wrong_peer(self): a, b = self.pair() owner.send(b, {'ok': True}) self.assertEqual(owner.receive(a, (os.getpid(), os.getuid(), os.getgid())), {'ok': True}) owner.send(b, {'ok': True}) with self.assertRaises(ValueError): owner.receive(a, (os.getpid() + 1, os.getuid(), os.getgid())) def test_rights_truncation_duplicate_keys_refused(self): a, b = self.pair() b.sendmsg([b'{}'], [(socket.SOL_SOCKET, socket.SCM_RIGHTS, array.array('i', [b.fileno()]))]) with self.assertRaises(ValueError): owner.receive(a, (os.getpid(), os.getuid(), os.getgid())) b.send(b'{"a":1,"a":2}') with self.assertRaises(ValueError): owner.receive(a, (os.getpid(), os.getuid(), os.getgid())) b.send(b' ' * 4097) with self.assertRaises(ValueError): owner.receive(a, (os.getpid(), os.getuid(), os.getgid())) def test_watch_heartbeat_never_extends_operation_deadline(self): a, b = self.pair() app, remote = self.pair() with patch.object(watch, 'now', return_value=100): observer = watch.Watch(a, app, os.getpid()) b.send(owner.PROGRESS.pack(1, 1, 0, 0, b'x' * 16)) self.assertTrue(observer.receive()) b.send(owner.PROGRESS.pack(1, 2, 1, 1000, b'x' * 16)) self.assertTrue(observer.receive()) b.send(owner.PROGRESS.pack(1, 3, 1, 1000, b'x' * 16)) self.assertTrue(observer.receive()) with patch.object(watch, 'now', return_value=1000): self.assertFalse(observer.receive()) self.assertEqual(remote.recv(1), b'') def test_watch_replay_and_peer_death(self): a, b = self.pair() app, remote = self.pair() observer = watch.Watch(a, app, os.getpid()) raw = owner.PROGRESS.pack(1, 1, 0, 0, b'x' * 16) b.send(raw) self.assertTrue(observer.receive()) b.send(raw) self.assertFalse(observer.receive()) self.assertEqual(remote.recv(1), b'') def test_journal_symlink_refused(self): with tempfile.TemporaryDirectory() as directory: Path(directory, 'operation.json').symlink_to('/dev/null') fd = os.open(directory, os.O_RDONLY | os.O_DIRECTORY) try: with self.assertRaises(OSError): owner.Journal(fd) finally: os.close(fd) def test_owner_deadline_single_slot_and_late_success_fence(self): entered, release = threading.Event(), threading.Event() class Delayed: def execute(self, mutation, end, before, verifying): before(['fixture-profile']) entered.set() if not release.wait(3): raise ValueError('fixture release timeout') verifying() with tempfile.TemporaryDirectory() as directory: fd = os.open(directory, os.O_RDONLY | os.O_DIRECTORY) instance = owner.Owner(owner.Journal(fd), Delayed) try: first = instance.submit({'Dns': {'policy': 'Automatic'}}) self.assertTrue(entered.wait(2)) with self.assertRaises(ValueError): instance.submit({'Dns': {'policy': 'Automatic'}}) with patch.object(owner, 'now', return_value=instance.end): self.assertEqual(instance.status(first['ticket'])['phase'], 'Unknown') release.set() instance.thread.join(3) self.assertEqual(instance.status(first['ticket'])['phase'], 'Unknown') self.assertTrue(owner.Journal(fd).blocked()) finally: release.set() instance.thread.join(3) os.close(fd) def test_idle_watch_stall_and_control_peer_death(self): a, b = self.pair() app, remote = self.pair() with patch.object(watch, 'now', return_value=100): observer = watch.Watch(a, app, os.getpid()) b.send(owner.PROGRESS.pack(1, 1, 0, 0, b'x' * 16)) self.assertTrue(observer.receive()) with patch.object(watch, 'now', return_value=2_000_000_100): self.assertFalse(observer.receive()) self.assertEqual(remote.recv(1), b'') a, b = self.pair() app, remote = self.pair() observer = watch.Watch(a, app, os.getpid()) b.close() self.assertFalse(observer.receive()) self.assertEqual(remote.recv(1), b'') if __name__ == '__main__': unittest.main()