test_core_review.py #2

#!/usr/bin/env python3
"""Offline regression coverage for shared logs, snapshots and binary downloads."""
import importlib.util
import json
import os
from pathlib import Path
import shutil
import subprocess
import sys
import tempfile
import unittest
from unittest.mock import patch

SDP = Path(__file__).resolve().parents[1]
BIN = SDP / 'Server/Unix/p4/common/bin'
BASH = os.environ.get('SDP_TEST_BASH', shutil.which('bash'))
spec = importlib.util.spec_from_file_location('review_sdplog', BIN / 'sdp_structured_log.py')
logger = importlib.util.module_from_spec(spec)
spec.loader.exec_module(logger)


class CoreReview(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory(prefix='sdp-core-review-')
        self.addCleanup(self.tmp.cleanup)
        self.root = Path(self.tmp.name)
        self.logs = self.root / 'logs'
        self.mock = self.root / 'mockbin'
        for p in [self.logs, self.mock, self.root / 'metrics', self.root / 'common/bin']:
            p.mkdir(parents=True)
        self.env = {'PATH': str(self.mock) + ':' + str(Path(BASH).parent) + ':/usr/bin:/bin',
                    'HOME': str(self.root), 'HOSTNAME': 'review-host', 'LOGS': str(self.logs),
                    'SDP_TEST_ROOT': str(self.root), 'SDP_INSTANCE': '1',
                    'SDP_LOG_FORMAT': 'jsonl', 'SDP_METRICS_DIR': str(self.root / 'metrics')}
        self.executable('flock', '#!' + sys.executable + '\nimport fcntl,sys\nfcntl.flock(int(sys.argv[-1]),fcntl.LOCK_EX)\n')

    def executable(self, name, content):
        p = self.mock / name
        p.write_text(content)
        p.chmod(0o755)
        return p

    def run_bash(self, script, args=(), extra=None):
        env = dict(self.env)
        env.update(extra or {})
        return subprocess.run([BASH, str(script), *args], env=env, cwd=self.root,
                              text=True, capture_output=True, timeout=20)

    def test_warning_then_nonzero_is_failure(self):
        with patch.dict(os.environ, self.env):
            ev = logger.SDPEvent('example.py')
            ev.set_status('warning')
            ev.emit(1)
        record = json.loads((self.logs / 'example.jsonl').read_text())
        self.assertEqual(record['status'], 'failure')
        self.assertEqual(record['exit_code'], 1)

    def test_config_check_stops_when_logging_library_fails(self):
        lib = self.root / 'common/lib'
        lib.mkdir()
        (lib / 'logging.lib').write_text('return 1\n')
        result = self.run_bash(BIN / 'ccheck.sh', extra={'SDP_ROOT': str(self.root)})
        self.assertEqual(result.returncode, 1)
        self.assertIn('Failed to load bash lib', result.stderr)
        self.assertNotIn('bail: command not found', result.stderr)

    def test_prom_boolean_redaction_and_fixed_metric_names(self):
        with patch.dict(os.environ, self.env):
            ev = logger.SDPEvent('example.py', mode='prom')
            ev.set_error('password=sensitive token=also-sensitive')
            ev.add(ready=True, invalid=float('nan'), success=77, last_run_timestamp_seconds=0)
            ev.add(**{'exit-code': 77, 'custom-field': 1, 'custom_field': 2})
            ev.emit(1)
        output = (self.root / 'metrics/p4_sdp_example.prom').read_text()
        self.assertNotIn('sensitive', output)
        self.assertIn('ready{instance="1",p4port=""} 1', output)
        self.assertNotIn('_invalid{', output)
        self.assertEqual(sum(line.startswith('p4_sdp_example_success{') for line in output.splitlines()), 1)
        self.assertEqual(sum(line.startswith('p4_sdp_example_last_run_timestamp_seconds{') for line in output.splitlines()), 1)
        self.assertEqual(sum(line.startswith('p4_sdp_example_exit_code{') for line in output.splitlines()), 1)
        self.assertEqual(sum(line.startswith('p4_sdp_example_custom_field{') for line in output.splitlines()), 1)

    def test_bash_logger_error_does_not_change_success(self):
        script = self.root / 'test.sh'
        script.write_text('set -e\nsource "' + str(BIN / 'sdp_structured_log.sh') + '"\n'
                          'sdp_log_init example.sh\nrm "$_SDP_STDERR_FILE"\nexit 0\n')
        result = self.run_bash(script)
        self.assertEqual(result.returncode, 0, result.stderr)
        self.assertEqual(json.loads((self.logs / 'example.jsonl').read_text())['status'], 'success')

    def test_bash_prom_does_not_duplicate_fixed_metrics(self):
        script = self.root / 'test.sh'
        script.write_text('source "' + str(BIN / 'sdp_structured_log.sh') + '"\n'
                          'sdp_log_init example.sh\nsdp_add_field success 77\nsdp_add_field last_run_timestamp_seconds 0\n'
                          'sdp_add_field exit-code 77\nsdp_add_field custom-field 1\nsdp_add_field custom_field 2\n')
        result = self.run_bash(script, extra={'SDP_LOG_FORMAT': 'prom'})
        self.assertEqual(result.returncode, 0, result.stderr)
        output = (self.root / 'metrics/p4_sdp_example.prom').read_text()
        self.assertEqual(sum(line.startswith('p4_sdp_example_success{') for line in output.splitlines()), 1)
        self.assertEqual(sum(line.startswith('p4_sdp_example_last_run_timestamp_seconds{') for line in output.splitlines()), 1)
        self.assertEqual(sum(line.startswith('p4_sdp_example_exit_code{') for line in output.splitlines()), 1)
        self.assertEqual(sum(line.startswith('p4_sdp_example_custom_field{') for line in output.splitlines()), 1)

    def snapshot_fixture(self):
        common = self.root / 'common/bin'
        (common / 'p4_vars').write_text('export LOGS="$SDP_TEST_ROOT/logs"\nexport P4ROOT="$SDP_TEST_ROOT/root"\nexport P4DBIN="$SDP_TEST_ROOT/mockbin/p4d"\n')
        shutil.copy2(BIN / 'sdp_structured_log.sh', common / 'sdp_structured_log.sh')
        login = common / 'p4login'
        login.write_text('#!' + BASH + '\nexit 0\n')
        login.chmod(0o755)
        self.executable('p4d', '#!' + BASH + '\necho "a@b@c@d@e@f@journal@42"\n')
        self.executable('aws', '#!' + sys.executable + r'''
import json, os, sys
from pathlib import Path
root = Path(os.environ['SDP_TEST_ROOT'])
a = sys.argv[1:]
with (root/'aws-calls').open('a') as f: f.write(json.dumps(a)+'\n')
cmd = a[1]
if cmd == os.environ.get('AWS_FAIL'):
    print('AWS mock failure: '+cmd,file=sys.stderr); sys.exit(7)
account = '123456789012'
if cmd == 'get-caller-identity': result = {'Account': account}
elif cmd == 'describe-volumes':
    name = json.loads(a[a.index('--filters')+1])[0]['Values'][0]
    volume = {'root':'vol-1','hxdepots':'vol-2','hxlogs':'vol-3'}[name.rsplit('-',1)[1]]
    result = {'Volumes': [{'VolumeId': volume, 'Tags': [{'Key':'Name','Value':name}], 'Attachments':[{'InstanceId':'i-abc','State':'attached'}]}]}
elif cmd == 'describe-instances':
    result = {'Reservations':[{'Instances':[{'InstanceId':'i-abc','RootDeviceName':'/dev/root','BlockDeviceMappings':[
        {'DeviceName':'/dev/root' if i==1 else '/dev/data'+str(i),'Ebs':{'VolumeId':'vol-'+str(i)}} for i in range(1,4)]}]}]}
elif cmd == 'create-snapshots':
    tags = json.loads(a[a.index('--tag-specifications')+1])[0]['Tags']
    snapshots = [{'SnapshotId':'snap-a'+str(i),'VolumeId':'vol-'+str(i), 'OwnerId':account, 'Tags':tags, 'State':'pending', 'StartTime':'2099-01-01T00:00:00Z'} for i in range(1,4)]
    (root/'created-snapshots').write_text(json.dumps(snapshots))
    result = {'Snapshots':snapshots}
elif cmd == 'wait':
    snapshots = json.loads((root/'created-snapshots').read_text())
    for snap in snapshots: snap['State'] = 'completed'
    (root/'created-snapshots').write_text(json.dumps(snapshots))
    result = None
elif cmd == 'describe-snapshots':
    snapshots = json.loads((root/'created-snapshots').read_text())
    if '--filters' in a:
        if os.environ.get('SNAPSHOT_MODE'):
            for i,snap in enumerate(snapshots):
                snap['SnapshotId'] = 'snap-b'+str(i)
                snap['StartTime'] = 'invalid' if os.environ['SNAPSHOT_MODE']=='invalid-date' else '2000-01-01T00:00:00Z'
                for tag in snap['Tags']:
                    if tag['Key']=='SDPRun': tag['Value']='20000101T000000Z-012345abcdef'
        else: snapshots=[]
    result = {'Snapshots':snapshots}
elif cmd == 'delete-snapshot': result = None
else: raise RuntimeError('Unexpected mock command '+cmd)
if result is not None: print(json.dumps(result))
''')
        cloud = SDP/'Server/Unix/p4/common/cloud'
        script = self.root/'cloud/aws/bin/snapshot.sh'
        script.parent.mkdir(parents=True)
        (self.root/'cloud/bin').mkdir(parents=True)
        shutil.copy2(cloud/'bin/sdp_snapshot.py', self.root/'cloud/bin/sdp_snapshot.py')
        script.write_text((cloud/'aws/bin/snapshot.sh').read_text().replace('/p4/common/bin', str(common)))
        return script

    def test_snapshot_records_and_tags_returned_ids(self):
        result = self.run_bash(self.snapshot_fixture(), ('1',))
        self.assertEqual(result.returncode, 0, result.stderr)
        calls = [json.loads(l) for l in (self.root/'aws-calls').read_text().splitlines()]
        create, = [c for c in calls if c[1]=='create-snapshots']
        tags = json.loads(create[create.index('--tag-specifications')+1])[0]['Tags']
        self.assertIn({'Key':'SDPJournal','Value':'42'}, tags)
        wait, = [c for c in calls if c[1]=='wait']
        self.assertEqual(wait[wait.index('--snapshot-ids')+1:wait.index('--output')], ['snap-a1','snap-a2','snap-a3'])
        self.assertFalse(any(c[1]=='delete-snapshot' for c in calls))
        self.assertIn('Snapshot set completed', (self.logs/'snapshot.log').read_text())

    def test_snapshot_failed_creation_with_tags_reports_failure(self):
        result = self.run_bash(self.snapshot_fixture(), ('1',), {'AWS_FAIL':'create-snapshots'})
        self.assertEqual(result.returncode, 1)
        self.assertIn('AWS mock failure', (self.logs/'snapshot.log').read_text())
        self.assertEqual(json.loads((self.logs/'snapshot.jsonl').read_text())['status'], 'failure')

    def test_snapshot_failed_counter_read_blocks_aws_even_with_valid_output(self):
        script = self.snapshot_fixture()
        self.executable('p4d', '#!' + BASH + '\necho "a@b@c@d@e@f@journal@42"\nexit 7\n')
        result = self.run_bash(script, ('1',))
        self.assertNotEqual(result.returncode, 0)
        self.assertFalse((self.root/'aws-calls').exists())
        self.assertIn('Could not read the journal counter', (self.logs/'snapshot.log').read_text())

    def test_snapshot_invalid_date_cannot_expire_snapshot(self):
        result = self.run_bash(self.snapshot_fixture(), ('1',), {'SNAPSHOT_MODE':'invalid-date'})
        self.assertEqual(result.returncode, 1)
        calls = [json.loads(l) for l in (self.root/'aws-calls').read_text().splitlines()]
        self.assertFalse(any(c[1]=='delete-snapshot' for c in calls))

    def test_snapshot_delete_failure_is_not_reported_as_success(self):
        result = self.run_bash(self.snapshot_fixture(), ('1',), {'SNAPSHOT_MODE':'old','AWS_FAIL':'delete-snapshot'})
        self.assertEqual(result.returncode, 1)
        self.assertIn('Could not delete expired snapshot', (self.logs/'snapshot.log').read_text())

    def downloader_fixture(self):
        self.executable('sleep', '#!' + BASH + '\nexit 0\n')
        self.executable('uname', '#!' + BASH + '\ncase "$1" in -m) echo x86_64;; -s) echo Linux;; -r) echo 5.15.0;; esac\n')
        self.executable('curl', '#!' + sys.executable + r'''
import os, sys
from pathlib import Path
root=Path(os.environ['SDP_TEST_ROOT']); args=sys.argv[1:]
name=args[-1].rsplit('/',1)[-1]
countfile=root/(name+'.calls')
count=int(countfile.read_text())+1 if countfile.exists() else 1
countfile.write_text(str(count))
if os.environ.get('DOWNLOAD_MODE')=='fail' or (os.environ.get('DOWNLOAD_MODE')=='retry_then_fail' and (name=='p4d' or count==1)):
    print('mock HTTP failure',file=sys.stderr); sys.exit(22)
dest=Path(args[args.index('-o')+1])
if name.startswith('bin.'):
    dest.write_text('<a href="p4api-test.tgz">API</a>\n')
elif name.endswith('.tgz'):
    if os.environ.get('DOWNLOAD_MODE')=='api_invalid':
        dest.write_text('invalid archive\n')
    else:
        import tarfile, io
        with tarfile.open(dest, 'w:gz') as archive:
            item=tarfile.TarInfo('p4api-test/README')
            item.size=4
            archive.addfile(item, io.BytesIO(b'test'))
elif os.environ.get('DOWNLOAD_MODE')=='invalid': dest.write_text('invalid downloaded content\n')
else: dest.write_text('#!/bin/sh\necho "Rev. mock/2026/100"\n')
''')
        return SDP/'p4_binaries/get_p4_binaries.sh'

    def old_binary(self, name='p4', mode=0o755):
        path=self.root/name
        path.write_text('#!/bin/sh\necho "Rev. original"\n')
        path.chmod(mode)
        return path

    def test_download_failure_preserves_existing_binary(self):
        old=self.old_binary(); contents=old.read_bytes()
        result=self.run_bash(self.downloader_fixture(), ('-sbd',str(self.root),'-b','p4'), {'DOWNLOAD_MODE':'fail'})
        self.assertNotEqual(result.returncode, 0)
        self.assertEqual(old.read_bytes(),contents)
        self.assertIn('mock HTTP failure',(self.root/'get_p4_binaries.log').read_text())

    def test_invalid_download_preserves_existing_binary(self):
        old=self.old_binary(); contents=old.read_bytes()
        result=self.run_bash(self.downloader_fixture(), ('-sbd',str(self.root),'-b','p4'), {'DOWNLOAD_MODE':'invalid'})
        self.assertNotEqual(result.returncode, 0)
        self.assertEqual(old.read_bytes(),contents)

    def test_retry_success_does_not_mask_next_binary_failure(self):
        self.old_binary(); old=self.old_binary('p4d'); contents=old.read_bytes()
        result=self.run_bash(self.downloader_fixture(), ('-sbd',str(self.root),'-b','p4,p4d'), {'DOWNLOAD_MODE':'retry_then_fail'})
        self.assertNotEqual(result.returncode, 0)
        self.assertIn('Rev. mock', (self.root/'p4').read_text())
        self.assertEqual((self.root/'p4').stat().st_mode & 0o777, 0o755)
        self.assertEqual(old.read_bytes(),contents)

    def test_dry_run_keeps_existing_permissions(self):
        old=self.old_binary(mode=0o600)
        result=self.run_bash(self.downloader_fixture(), ('-sbd',str(self.root),'-b','p4','-n'))
        self.assertEqual(result.returncode, 0,result.stderr)
        self.assertEqual(old.stat().st_mode & 0o777, 0o600)
        self.assertFalse((self.root/'p4.calls').exists())

    def test_invalid_api_download_keeps_existing_archive(self):
        archive=self.root/'p4api-test.tgz'
        archive.write_bytes(b'previous archive')
        result=self.run_bash(self.downloader_fixture(), ('-sbd',str(self.root),'-b','none','-api'), {'DOWNLOAD_MODE':'api_invalid'})
        self.assertNotEqual(result.returncode, 0)
        self.assertEqual(archive.read_bytes(), b'previous archive')

    def test_valid_api_download_is_installed(self):
        result=self.run_bash(self.downloader_fixture(), ('-sbd',str(self.root),'-b','none','-api'))
        self.assertEqual(result.returncode, 0, result.stdout+result.stderr)
        self.assertTrue((self.root/'p4api-test.tgz').exists())
        self.assertEqual((self.root/'p4api-test.tgz').stat().st_mode & 0o777, 0o644)


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