"""Synthetic-only: no Docker/socket, real /proc/cgroup reads, process enumeration or main().""" import copy import importlib.util from pathlib import Path from types import SimpleNamespace import unittest from unittest.mock import patch AREA = Path(__file__).parent spec = importlib.util.spec_from_file_location('memory_measurement_extended', AREA / 'measure-memory.py') m = importlib.util.module_from_spec(spec) spec.loader.exec_module(m) MEMINFO = 'MemTotal: 2048 kB\nMemAvailable: 1024 kB\nSwapTotal: 4096 kB\nSwapFree: 3072 kB\n' VMSTAT = 'pswpin 20\npswpout 30\noom_kill 0\n' PSI = 'some avg10=0.00 avg60=1.25 avg300=100.00 total=123456\nfull avg10=0.00 avg60=0.01 avg300=1.00 total=0\n' ID = 'a' * 64 CG = '/sys/fs/cgroup/system.slice/docker-' + ID + '.scope' class Synthetic(unittest.TestCase): def setUp(self): self.reads = [] self.gets = [] self.before = {'Id': ID, 'Image': 'sha256:' + 'b' * 64, 'RestartCount': 0, 'State': {'StartedAt': '2026-11-02T00:00:00Z', 'Pid': 123, 'Running': True, 'OOMKilled': False}, 'HostConfig': {'Memory': 512, 'MemorySwap': 1024}, 'Config': {'Env': ['SECRET=never-retain']}} self.stats = {'id': ID, 'memory_stats': {'usage': 100, 'limit': 2048, 'stats': {'inactive_file': 30}}} self.files = {'/proc/meminfo': MEMINFO, '/proc/vmstat': VMSTAT, '/proc/pressure/memory': PSI, '/proc/self/mountinfo': '1 0 0:1 / /sys/fs/cgroup rw - cgroup2 cgroup rw\n', '/proc/123/cgroup': '0::/system.slice/docker-' + ID + '.scope\n', '/proc/123/stat': '123 (owned fixture) S ' + ' '.join(['0'] * 18 + ['987'] + ['0'] * 8), CG + '/memory.current': '100\n', CG + '/memory.peak': '200\n', CG + '/memory.events': 'low 0\nhigh 0\nmax 0\noom 0\noom_kill 0\n', CG + '/memory.swap.current': '0\n', CG + '/memory.swap.peak': '0\n', CG + '/memory.swap.events': 'high 0\nmax 0\nfail 0\n', CG + '/memory.max': '512\n', CG + '/memory.swap.max': '512\n', CG + '/cpu.stat': 'usage_usec 50\nuser_usec 30\nsystem_usec 20\nnr_periods 10\nnr_throttled 2\nthrottled_usec 1234\nnr_bursts 0\n'} def deny(*_args, **_kwargs): raise AssertionError('Real external/OS I/O is forbidden by synthetic test') # Module-local OS proxy avoids changing unittest/Python's filesystem operations. fake_os = SimpleNamespace(**vars(m.os)) fake_os.path = SimpleNamespace(**vars(m.os.path)) fake_os.sep = '/' # synthetic Linux path contract, not this Windows host fake_os.open = deny fake_os.stat = deny fake_os.path.realpath = deny self.enterContext(patch.object(m, 'os', fake_os)) self.enterContext(patch.object(m, 'read_bounded', side_effect=deny)) self.enterContext(patch.object(m, 'docker_get', side_effect=deny)) self.enterContext(patch.object(m.socket, 'socket', side_effect=deny)) def read(self, name, expected_device=None, deadline=None): self.reads.append(name) if name not in self.files: raise FileNotFoundError('synthetic missing') value = self.files[name] if isinstance(value, BaseException): raise value return value def observe(self, after=None, realpath=None, statter=None): after = copy.deepcopy(self.before) if after is None else after def get(name, deadline=None): self.gets.append(name) if self.gets == ['/containers/' + ID + '/json']: return copy.deepcopy(self.before) if self.gets == ['/containers/' + ID + '/json', '/containers/' + ID + '/stats?stream=false&one-shot=true']: return copy.deepcopy(self.stats) if self.gets == ['/containers/' + ID + '/json', '/containers/' + ID + '/stats?stream=false&one-shot=true', '/containers/' + ID + '/json']: return copy.deepcopy(after) raise AssertionError('Unexpected Docker path or extra request') def statter_default(name): allowed = {'/sys/fs/cgroup': (8, 1), CG: (8, 2), '/proc/123/root': (8, 3)} if name not in allowed: raise AssertionError('Unexpected stat path') dev, ino = allowed[name] return SimpleNamespace(st_dev=dev, st_ino=ino) with patch.object(m, 'read_bounded', side_effect=self.read), patch.object(m, 'docker_get', side_effect=get), \ patch.object(m.os.path, 'realpath', side_effect=realpath or (lambda p: p)), \ patch.object(m.os, 'stat', side_effect=statter or statter_default): return m.collect_container(ID) def test_limits_max_zero_finite_are_not_missing_or_host_ram(self): for value, expected in [('max\n', 'max'), ('max', 'max'), ('0\n', 0), ('512\n', 512), (str(2**64-1), 2**64-1)]: self.assertEqual(m.parse_limit(value), expected) for bad in ('MAX', '-1', ' 0', '00', '1.0', str(2**64), 'max\n\n'): with self.assertRaises(ValueError): m.parse_limit(bad) def test_cpu_throttling_units_required_fields_and_output_whitelist(self): parsed = m.parse_cpu_throttling(self.files[CG + '/cpu.stat']) self.assertEqual(parsed, {'nr_periods':10,'nr_throttled':2,'throttled_usec':1234}) for bad in ('nr_periods 1\nnr_throttled 0\n', 'nr_periods 1\nnr_periods 2\nnr_throttled 0\nthrottled_usec 0\n', 'nr_periods -1\nnr_throttled 0\nthrottled_usec 0\n'): with self.assertRaises(ValueError): m.parse_cpu_throttling(bad) def test_pressure_fixed_groups_percent_and_microseconds(self): parsed = m.parse_memory_pressure(PSI) self.assertEqual(parsed['some'], {'avg10Percent':'0.00','avg60Percent':'1.25','avg300Percent':'100.00','totalMicros':123456}) self.assertEqual(parsed['full']['totalMicros'], 0) for bad in (PSI.replace('100.00','100.01'), PSI.replace('1.25','nan'), PSI.replace('total=0','total=-1'), PSI.splitlines()[0], PSI.replace('full','some'), PSI+'extra\n', PSI.replace('total=0','total='+str(2**64))): with self.assertRaises(ValueError): m.parse_memory_pressure(bad) def test_docker_memoryswap_raw_minus1_zero_finite_and_missing_differ(self): for raw in (-1,0,512,2**63-1): item = copy.deepcopy(self.before); item['HostConfig']['MemorySwap'] = raw self.assertEqual(m.configured_memory_swap(item), raw) for bad in (None,False,True,'0',-2,2**63): item = copy.deepcopy(self.before); item['HostConfig']['MemorySwap'] = bad self.assertIsNone(m.configured_memory_swap(item)) for host in ({},None,[]): self.assertIsNone(m.configured_memory_swap({'HostConfig':host})) def test_real_collection_paths_are_only_fixed_host_and_retained_cgroup(self): row = self.observe() self.assertTrue(row['identityStableDuringSample']) self.assertTrue(row['resourceLimitsStableDuringSample']) self.assertEqual(row['configuredMemorySwapBytes'], 1024) self.assertEqual(row['cgroupV2']['status'], 'available') self.assertEqual(row['cgroupV2']['metrics']['memory.max'], 512) self.assertEqual(row['cgroupV2']['metrics']['cpu.stat']['throttled_usec'], 1234) self.assertTrue(set(self.reads).issubset(set(self.files))) self.assertNotIn('SECRET', str(row)); self.assertNotIn('Env', str(row)) self.assertEqual(len(self.gets), 3) def test_unlimited_limits_do_not_become_numeric_zero(self): self.files[CG+'/memory.max']='max\n'; self.files[CG+'/memory.swap.max']='max\n' row=self.observe();self.assertEqual(row['cgroupV2']['metrics']['memory.max'],'max');self.assertEqual(row['cgroupV2']['metrics']['memory.swap.max'],'max') def test_each_new_cgroup_source_missing_or_malformed_is_unknown(self): for name in ('memory.max','memory.swap.max','cpu.stat'): for value in (None,'not-a-counter\n'): with self.subTest(name=name,value=value): self.setUp_fixture_only() if value is None: self.files.pop(CG+'/'+name) else: self.files[CG+'/'+name]=value row=self.observe();self.assertEqual(row['cgroupV2']['status'],'partial');self.assertIsNone(row['cgroupV2']['metrics'][name]);self.assertIn(name,row['cgroupV2']['unavailable']) def setUp_fixture_only(self): # Recreate dictionary, not guards/patches; no real I/O. self.files.update({CG+'/memory.max':'512\n',CG+'/memory.swap.max':'512\n',CG+'/cpu.stat':'nr_periods 10\nnr_throttled 2\nthrottled_usec 1234\n'});self.gets=[] def test_missing_or_changed_swap_configuration_cannot_authenticate_limits(self): after=copy.deepcopy(self.before);after['HostConfig']['MemorySwap']=2048 row=self.observe(after);self.assertFalse(row['resourceLimitsStableDuringSample']);self.assertEqual(row['cgroupV2']['status'],'partial');self.assertIsNone(row['cgroupV2']['metrics']['memory.swap.max']) self.gets=[];self.before['HostConfig'].pop('MemorySwap');after=copy.deepcopy(self.before) row=self.observe(after);self.assertIsNone(row['configuredMemorySwapBytes']);self.assertEqual(row['configuredMemorySwapStatus'],'unavailable');self.assertIsNone(row['resourceLimitsStableDuringSample']);self.assertEqual(row['cgroupV2']['status'],'partial') def test_memory_limit_change_cannot_reuse_stable_container_identity_as_limit_proof(self): after=copy.deepcopy(self.before);after['HostConfig']['Memory']=1024 row=self.observe(after) self.assertTrue(row['identityStableDuringSample']) self.assertFalse(row['resourceLimitsStableDuringSample']) for key in ('memory.max','memory.swap.max'): self.assertIsNone(row['cgroupV2']['metrics'][key]) self.assertEqual(row['cgroupV2']['unavailable'][key],'container-resource-limits-changed-or-unavailable') def test_legacy_projection_fields_keep_cache_limit_and_rss_distinctions(self): item=copy.deepcopy(self.before);item['HostConfig'].pop('MemorySwap');item['HostConfig']['Memory']=0 row=m.project_container(item,self.stats,ID) self.assertEqual(row['usageIncludingCacheBytes'],100) self.assertEqual(row['usageMinusInactiveFileBytes'],70) self.assertEqual(row['configuredLimitBytes'],0) self.assertEqual(row['reportedLimitBytes'],2048) self.assertEqual(row['configuredMemorySwapStatus'],'unavailable') self.assertIsNone(row['configuredMemorySwapBytes']) def test_identity_drift_symlink_and_pid_generation_still_refuse(self): after=copy.deepcopy(self.before);after['State']['Pid']=456 row=self.observe(after);self.assertFalse(row['identityStableDuringSample']);self.assertIsNone(row['resourceLimitsStableDuringSample']);self.assertEqual(row['cgroupV2']['status'],'unavailable') self.gets=[];row=self.observe(realpath=lambda p:'/other' if p.endswith('memory.swap.max') else p);self.assertEqual(row['cgroupV2']['status'],'partial');self.assertIsNone(row['cgroupV2']['metrics']['memory.swap.max']) self.gets=[] with patch.object(m,'process_generation',side_effect=[(1,8,3),(2,8,3)]): self.assertEqual(self.observe()['cgroupV2']['status'],'unavailable') def test_host_collection_whitelists_one_psi_file_and_missing_is_explicit(self): with patch.object(m,'read_bounded',side_effect=self.read): host=m.collect_host();self.assertEqual(host['memoryPressure']['status'],'available');self.assertEqual(host['MemTotalBytes'],2048*1024) self.assertEqual(self.reads,['/proc/meminfo','/proc/vmstat','/proc/pressure/memory']) self.files.pop('/proc/pressure/memory');host=m.collect_host();self.assertEqual(host['memoryPressure']['status'],'unavailable');self.assertIsNone(host['memoryPressure']['metrics']) def test_deadline_never_becomes_partial_or_success(self): self.files[CG+'/cpu.stat']=m.ObservationDeadlineExceeded('fixture-deadline') with self.assertRaises(m.ObservationDeadlineExceeded): self.observe() with patch.object(m,'read_bounded',side_effect=[MEMINFO,VMSTAT,m.ObservationDeadlineExceeded('fixture-deadline')]): with self.assertRaises(m.ObservationDeadlineExceeded): m.collect_host() def test_summary_exposes_unknown_and_never_authorizes_runtime(self): row=self.observe() with patch.object(m,'read_bounded',side_effect=self.read):host=m.collect_host() sample={'elapsedMs':0,'host':host,'containers':[row]} out=m.summarize([sample]);self.assertTrue(out['allResourceLimitsStableDuringSample']);self.assertTrue(out['allHostMemoryPressureAvailable']);self.assertFalse(out['acceptancePassed']) row['resourceLimitsStableDuringSample']=None;host['memoryPressure']['status']='unavailable' out=m.summarize([sample]);self.assertFalse(out['allResourceLimitsStableDuringSample']);self.assertFalse(out['allHostMemoryPressureAvailable']);self.assertFalse(out['acceptancePassed']) if __name__=='__main__': unittest.main()