test_baseline_maintenance.py #1

#!/usr/bin/env python3
"""Paired, offline deployed-archive/candidate behavioral contracts.

SDP_BASELINE_SOURCE must name a separately exported SDP tree. Tests never fetch
Perforce sources or connect to a server. SDP_CANDIDATE_SOURCE is optional.
"""
import json
import os
from pathlib import Path
import shlex
import shutil
import subprocess
import sys
import tempfile
import unittest

CANDIDATE = Path(os.environ.get('SDP_CANDIDATE_SOURCE', Path(__file__).resolve().parents[1]))
HARNESS = r'''
import contextlib, importlib.util, io, json, os, subprocess, sys, time, types
from pathlib import Path
root, scenario = Path(sys.argv[1]), sys.argv[2]
sys.path.insert(0, str(root / 'Maintenance'))
# No native P4 module or installed /p4 logger can be used by these fixtures.
class P4Exception(Exception): pass
sys.modules['P4'] = types.SimpleNamespace(P4=lambda: None, P4Exception=P4Exception)
sys.modules['sdp_structured_log'] = None
import sdputils
class Utils:
    server='fixture:1666'; p4user='fixture'; p4='/mock/p4'
    def __init__(self,*args): pass
    def get(self,key,*args,**kwargs):
        return {'weeks':'4','userweeks':'8','deleteweeks':'12','domain':'example.test'}.get(key,'')
    def login(self,*args): pass  # Baseline import-time login is intercepted; no process/network.
sdputils.SDPUtils=Utils
calls=[]; mutations=[]; status=0; evidence={}
Path('unrelated.txt').write_text('preserve this report\n')
def load(name,args=()):
    sys.argv=[name+'.py',*args]
    spec=importlib.util.spec_from_file_location('paired_'+name,root/'Maintenance'/(name+'.py'))
    mod=importlib.util.module_from_spec(spec); spec.loader.exec_module(mod)
    return mod
console=io.StringIO()
with contextlib.redirect_stdout(console), contextlib.redirect_stderr(console):
    try:
        if scenario.startswith('clients_'):
            now=2_000_000_000; week=604800
            time.time=lambda:now
            if scenario=='clients_cap':
                records=[{'client':str(n),'Access':'1'} for n in range(250)]
            else:
                records=[{'client':'old-z','Access':str(now-5*week)},
                         {'Client':'at-cutoff','access':str(now-4*week)},
                         {'client':'recent','Access':str(now-4*week+1)},
                         {'client':'SWARM-internal','Access':'1'},
                         {'client':'bad-date','Access':'invalid'},
                         {'client':'missing-date'},
                         {'client':'old-a','Access':'1'}]
            before=[dict(row) for row in records]
            if scenario in ('clients_activity','clients_recheck_failure'):
                records[0]['Access']=str(now)
            mod=load('unload_clients')
            class API:
                def run(self,*args):
                    calls.append(args)
                    if args[0]=='clients':
                        if len(args)>1:
                            if scenario=='clients_recheck_failure': raise P4Exception('access query denied')
                            return [row for row in records if (row.get('client') or row.get('Client'))==args[-1]]
                        return records
                    if args[0]=='unload':
                        mutations.append(list(args)); return []
                    raise AssertionError(args)
                def disconnect(self): evidence['disconnected']=True
            api=API(); mod.connect_p4=lambda instance:api
            argv=['1','--limit','200']
            if scenario in ('clients_activity','clients_recheck_failure') and hasattr(sdputils,'write_client_snapshot'):
                sdputils.write_client_snapshot('snapshot.json','1','fixture:1666',4,before,now)
                argv+=['--clients-file','snapshot.json']
            status=mod.main(argv)
        elif scenario.startswith('cleanup_'):
            mod=load('remove_empty_pending_changes')
            def command(args,**kwargs):
                calls.append(args)
                if 'changes' in args:
                    return subprocess.CompletedProcess(args,0,'10,alice\n11,Swarm-bot\n12,bob\n13,alice\n')
                if 'describe' in args:
                    if args[-1]=='12': return subprocess.CompletedProcess(args,0,'#DO-NOT-DELETE retain\n')
                    if args[-1]=='13' and scenario=='cleanup_describe_failure':
                        raise subprocess.CalledProcessError(7,args,stderr='describe denied')
                    return subprocess.CompletedProcess(args,0,'disposable empty change\n')
                if 'change' in args:
                    mutations.append(args[1:])
                    failed=args[-1]=='13' and scenario=='cleanup_delete_failure'
                    return subprocess.CompletedProcess(args,7 if failed else 0,'','delete denied' if failed else '')
                raise AssertionError(args)
            mod.subprocess.run=command
            status=mod.main()
        elif scenario.startswith('group_'):
            mod=load('removeusersfromgroup',['1','alice','team'])
            group='Group: team\nMaxResults: 1000\nOwners:\n\towner\nSubgroups:\n\tchild\nUsers:\n\talice\n\tbob\n'
            if scenario=='group_scratch': Path('team.group').write_text('unrelated group report\n')
            def command(args,**kwargs):
                calls.append(args)
                if '-o' in args: return subprocess.CompletedProcess(args,0,group)
                mutations.append({'argv':args[1:],'input':kwargs['input']})
                if scenario=='group_write_failure':
                    if kwargs.get('check'): raise subprocess.CalledProcessError(7,args,stderr='write denied')
                    return subprocess.CompletedProcess(args,7,'','write denied')
                return subprocess.CompletedProcess(args,0,'')
            mod.subprocess.run=command
            def checked(args, **kwargs):
                kwargs.setdefault('check', True)
                return command(args, **kwargs)
            sdputils.run_checked=checked
            mod.main()
        elif scenario=='user_membership':
            mod=load('maintain_user_from_groups')
            def command(args,**kwargs):
                calls.append(args)
                verb=args[1]
                text=''
                if verb=='users': text='alice <[email protected]> (Alice) accessed 2026/01/01\nbob <[email protected]> (Bob) accessed 2026/01/01\nperforce <[email protected]> (Admin) accessed 2026/01/01\n'
                elif verb=='groups': text='team\n'
                elif verb=='group': text='Group: team\nUsers:\n\talice\n\tnew-user\n\tperforce\n'
                elif verb=='-Ztag': text=''
                elif '-o' in args: text='User: new-user\nEmail: [email protected]\n'
                elif 'user' in args: mutations.append({'argv':args[1:],'input':kwargs.get('input')})
                else: raise AssertionError(args)
                if 'stdout' in kwargs: kwargs['stdout'].write(text)
                return subprocess.CompletedProcess(args,0,text)
            mod.subprocess.run=command
            sdputils.run_checked=command
            mod.main()
        elif scenario.startswith('protect_'):
            mod=load('clean_protect',['groups.txt','protect.txt'])
            Path('groups.txt').write_text('target\n')
            if scenario=='protect_hosts':
                text='Protections:\n\tread group target * //...\n\twrite group target 10.0.0.0/8 //depot/...\n\tread group keeper 10.0.0.1 //...\n\tread user target * //...\n'
            elif scenario=='protect_format':
                text='# retained comment  \r\nProtections:\r\n\tread group keeper * //...  \r\n'
            else:
                text='Protections:\n\tread group target * //...\n\tread group keeper * //...\n\tread user target * //...\n'
            Path('protect.txt').write_bytes(text.encode())
            output=io.StringIO(newline='')
            with contextlib.redirect_stdout(output): mod.main()
            evidence['output']=output.getvalue()
        else: raise AssertionError(scenario)
    except subprocess.CalledProcessError as error:
        status=error.returncode
    except SystemExit as error:
        status=error.code or 0
files={p.name:p.read_text() for p in Path('.').iterdir() if p.name in ['unrelated.txt','team.group']}
if scenario=='group_write_failure':
    evidence['recovery_specs']=[p.read_text() for p in Path('.').rglob('*') if p.is_file() and p.name in ['team.group','group.spec']]
print(json.dumps({'status':status or 0,'mutations':mutations,'files':files,'evidence':evidence}))
'''


class BaselineMaintenance(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        value = os.environ.get('SDP_BASELINE_SOURCE')
        if not value:
            raise RuntimeError('SDP_BASELINE_SOURCE must name the exported deployed-archive SDP tree')
        cls.baseline = Path(value)
        for root in [cls.baseline, CANDIDATE]:
            if not (root / 'Maintenance/sdputils.py').is_file():
                raise RuntimeError('Missing Maintenance sources under ' + str(root))

    def run_case(self, root, scenario):
        with tempfile.TemporaryDirectory(prefix='sdp-paired-maint-') as tmp:
            env = dict(os.environ, LOGS=tmp, SDP_LOG_FORMAT='off')
            result = subprocess.run([sys.executable, '-c', HARNESS, str(root), scenario],
                                    cwd=tmp, env=env, capture_output=True, text=True, timeout=20)
            self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
            return json.loads(result.stdout)

    def pair(self, scenario):
        return self.run_case(self.baseline, scenario), self.run_case(CANDIDATE, scenario)

    def inventory(self, root, scenario):
        with tempfile.TemporaryDirectory(prefix='sdp-paired-inventory-') as tmp:
            work = Path(tmp)
            common = work / 'common'
            common.mkdir()
            fake = common / 'p4'
            fake.write_text('#!' + sys.executable + '\n' + r'''
import json, marshal, os, sys
from pathlib import Path
a=sys.argv[1:]; structured='-G' in a; scenario=os.environ['FIXTURE_CASE']
commands=['depots','dirs','files']; command=next(x for x in commands if x in a)
target=a[-1]
rows=[]; rc=0
if command=='depots': rows=[{'code':'stat','name':'depot'}]
elif command=='dirs':
    rows=[{'code':'stat','dir':'//depot/sub'}]
    if scenario=='space': rows.append({'code':'stat','dir':'//depot/space dir'})
    if scenario=='failure': rows.append({'code':'stat','dir':'//depot/denied'})
elif target=='//depot/*':
    if scenario=='root': rows=[{'code':'stat','depotFile':'//depot/root.txt'}]
elif target=='//depot/sub/...': rows=[{'code':'stat','depotFile':'//depot/sub/file.txt'}]
elif target=='//depot/space dir/...': rows=[{'code':'stat','depotFile':'//depot/space dir/a file.txt'}]
elif target=='//depot/denied/...':
    rows=[{'code':'error','generic':6,'severity':3,'data':'permission denied'}]; rc=7
else:
    rows=[{'code':'error','generic':17,'severity':2,'data':'no such file(s)'}]; rc=1
if command=='files' and not rows:
    rows=[{'code':'error','generic':17,'severity':2,'data':'no such file(s)'}]; rc=1
for row in rows:
    if structured:
        marshal.dump({k.encode():v.encode() if isinstance(v,str) else v for k,v in row.items()},sys.stdout.buffer)
    elif row['code']=='error': print(row['data'],file=sys.stderr)
    elif command=='dirs': print('... dir '+row['dir'])
    else: print(row['name'] if command=='depots' else row['depotFile'])
sys.exit(rc)
''')
            fake.chmod(0o755)
            (common / 'p4_vars').write_text('export P4BIN=' + shlex.quote(str(fake)) + '\n')
            (common / 'sdp_structured_log.sh').write_text('sdp_run_log_init() { :; }\n_sdp_on_exit() { :; }\n')
            wrapper = work / 'create_p4_filelist.sh'
            wrapper.write_text((root / 'Maintenance/create_p4_filelist.sh').read_text().replace('/p4/common/bin/', str(common) + '/'))
            if (root / 'Maintenance/file_inventory.py').exists():
                shutil.copyfile(root / 'Maintenance/file_inventory.py', work / 'file_inventory.py')
            (work / 'p4_file_list.txt').write_text('previous complete inventory\n')
            (work / 'unrelated.txt').write_text('preserve\n')
            result = subprocess.run([os.environ.get('SDP_TEST_BASH', shutil.which('bash')), str(wrapper), '1'],
                                    cwd=work, env=dict(os.environ, FIXTURE_CASE=scenario, PATH=str(common) + os.pathsep + os.environ.get('PATH', '')),
                                    capture_output=True, text=True, timeout=20)
            return {'status':result.returncode, 'inventory':(work / 'p4_file_list.txt').read_text(),
                    'unrelated':(work / 'unrelated.txt').read_text()}

    def test_client_age_cutoff_order_swarm_and_invalid_access_match_baseline(self):
        baseline, candidate = self.pair('clients_policy')
        self.assertEqual(candidate, baseline)
        self.assertEqual([row[-1] for row in candidate['mutations']], ['old-z','at-cutoff','old-a'])

    def test_client_200_cap_mutations_status_and_disconnect_match_baseline(self):
        baseline, candidate = self.pair('clients_cap')
        self.assertEqual(candidate, baseline)
        self.assertEqual(len(candidate['mutations']), 200)
        self.assertEqual(candidate['mutations'][-1][-1], '199')

    def test_snapshot_recheck_preserves_baseline_fresh_activity_decision(self):
        baseline, candidate = self.pair('clients_activity')
        self.assertEqual(candidate, baseline)
        self.assertNotIn('old-z', [row[-1] for row in candidate['mutations']])

    def test_snapshot_recheck_failure_explicitly_stops_candidate_before_mutation(self):
        baseline, candidate = self.pair('clients_recheck_failure')
        # Baseline has no snapshot/recheck path. A failed added safety query must
        # fail closed, not continue using the earlier successful report.
        self.assertEqual(baseline['status'], 0)
        self.assertTrue(baseline['mutations'])
        self.assertEqual(candidate['status'], 1)
        self.assertEqual(candidate['mutations'], [])
        self.assertEqual(candidate['files'], baseline['files'])

    def test_pending_cleanup_retains_skip_and_failure_policies(self):
        for scenario, status, deleted in [('cleanup_success',0,['10','13']),
                                          ('cleanup_describe_failure',1,['10']),
                                          ('cleanup_delete_failure',1,['10','13'])]:
            with self.subTest(scenario=scenario):
                baseline, candidate = self.pair(scenario)
                self.assertEqual(candidate['mutations'], baseline['mutations'])
                self.assertEqual(candidate['files'], baseline['files'])
                self.assertEqual(candidate['status'], status)
                # Approved bug fix: baseline swallowed describe failures.
                self.assertEqual(baseline['status'], 0 if scenario=='cleanup_describe_failure' else status)
                self.assertEqual([row[-1] for row in candidate['mutations']], deleted)

    def test_group_mutation_preserves_members_owners_subgroups_and_limits(self):
        baseline, candidate = self.pair('group_success')
        self.assertEqual(candidate, baseline)
        spec=candidate['mutations'][0]['input']
        self.assertNotIn('\talice\n', spec)
        for expected in ('\tbob\n','\towner\n','\tchild\n','MaxResults: 1000'):
            self.assertIn(expected, spec)

    def test_approved_private_scratch_change_preserves_unrelated_group_file(self):
        baseline, candidate = self.pair('group_scratch')
        self.assertEqual(candidate['status'], baseline['status'])
        self.assertEqual(candidate['mutations'], baseline['mutations'])
        self.assertNotIn('team.group', baseline['files'])
        self.assertEqual(candidate['files']['team.group'], 'unrelated group report\n')
        self.assertEqual(candidate['files']['unrelated.txt'], baseline['files']['unrelated.txt'])

    def test_failed_group_write_preserves_same_recovery_spec_and_failure_status(self):
        baseline, candidate = self.pair('group_write_failure')
        # Approved bug fix: unchecked baseline writes reported success and
        # deleted recovery data; candidate retains the exact attempted spec.
        self.assertEqual(baseline['status'], 0)
        self.assertEqual(candidate['status'], 7)
        self.assertEqual(candidate['mutations'], baseline['mutations'])
        self.assertEqual(baseline['evidence']['recovery_specs'], [])
        self.assertEqual(candidate['evidence']['recovery_specs'], [candidate['mutations'][0]['input']])
        self.assertNotIn('team.group', candidate['files'])
        self.assertNotIn('team.group', baseline['files'])
        self.assertEqual(candidate['files']['unrelated.txt'], baseline['files']['unrelated.txt'])

    def test_membership_user_creation_deletion_policy_matches_baseline(self):
        baseline, candidate = self.pair('user_membership')
        self.assertEqual(candidate, baseline)
        self.assertEqual(candidate['mutations'][0]['argv'], ['user','-FDy','bob'])
        self.assertIn('User: new-user', candidate['mutations'][1]['input'])

    def test_classic_protection_selection_matches_baseline(self):
        baseline, candidate = self.pair('protect_classic')
        self.assertEqual(candidate, baseline)
        self.assertNotIn('group target', candidate['evidence']['output'])
        self.assertIn('user target', candidate['evidence']['output'])

    def test_approved_host_restricted_protection_removal_changes_only_selected_group(self):
        baseline, candidate = self.pair('protect_hosts')
        selected='\twrite group target 10.0.0.0/8 //depot/...\n'
        self.assertIn(selected, baseline['evidence']['output'])
        self.assertEqual(candidate['evidence']['output'], baseline['evidence']['output'].replace(selected,''))
        self.assertEqual(candidate['files'], baseline['files'])
        self.assertEqual(candidate['status'], baseline['status'])

    def test_approved_protection_format_preservation_retains_crlf_and_spaces(self):
        baseline, candidate = self.pair('protect_format')
        expected='# retained comment  \r\nProtections:\r\n\tread group keeper * //...  \r\n'
        self.assertEqual(candidate['evidence']['output'], expected)
        self.assertEqual(baseline['evidence']['output'], '# retained comment\nProtections:\n\tread group keeper * //...\n')
        self.assertEqual(candidate['files'], baseline['files'])

    def test_approved_inventory_completeness_adds_root_file_without_losing_nested_file(self):
        baseline, candidate = self.inventory(self.baseline, 'root'), self.inventory(CANDIDATE, 'root')
        self.assertEqual(baseline, {'status':0,'inventory':'//depot/sub/file.txt\n','unrelated':'preserve\n'})
        self.assertEqual(candidate, {'status':0,'inventory':'//depot/root.txt\n//depot/sub/file.txt\n','unrelated':'preserve\n'})

    def test_approved_inventory_spaces_are_complete_instead_of_partial_failure(self):
        baseline, candidate = self.inventory(self.baseline, 'space'), self.inventory(CANDIDATE, 'space')
        self.assertEqual(baseline, {'status':0,'inventory':'//depot/sub/file.txt\n','unrelated':'preserve\n'})
        self.assertEqual(candidate, {'status':0,'inventory':'//depot/sub/file.txt\n//depot/space dir/a file.txt\n','unrelated':'preserve\n'})

    def test_approved_atomic_inventory_preserves_previous_output_on_late_query_failure(self):
        baseline, candidate = self.inventory(self.baseline, 'failure'), self.inventory(CANDIDATE, 'failure')
        self.assertEqual(baseline, {'status':0,'inventory':'//depot/sub/file.txt\n','unrelated':'preserve\n'})
        self.assertEqual(candidate, {'status':7,'inventory':'previous complete inventory\n','unrelated':'preserve\n'})


if __name__ == '__main__':
    unittest.main()