Cron 系统任务表与用户任务表有不同契约

from pathlib import Path
from tempfile import TemporaryDirectory
import json
import os
import pwd
import shutil
import subprocess
import time

assert os.geteuid() == 0 and Path('/.dockerenv').exists(), 'use a disposable root Linux container'
assert subprocess.run(['pgrep', '-x', 'cron'], capture_output=True).returncode == 1
assert subprocess.run(['crontab', '-l'], capture_output=True).returncode != 0
assert shutil.which('root') is None
username = 'xbtime'
try:
    pwd.getpwnam(username)
except KeyError:
    pass
else:
    raise AssertionError('test account already exists')
system_table = Path('/etc/cron.d/xbcatm_time_test')
assert not system_table.exists()
subprocess.run(['useradd', '--create-home', '--shell', '/bin/sh', username], check=True)
account = pwd.getpwnam(username)
daemon = None
try:
    with TemporaryDirectory(prefix='cron-tables-') as directory:
        root = Path(directory)
        os.chown(root, account.pw_uid, account.pw_gid)
        worker = root / 'worker.py'
        worker.write_text('''from pathlib import Path
import json,os,sys
root=Path(__file__).parent
mode=sys.argv[1]
data={'uid':os.getuid(),'table':os.environ.get('LAB_TABLE'),
      'path':os.environ.get('PATH'),'shell':os.environ.get('SHELL'),
      'parent_only':os.environ.get('PARENT_ONLY')}
if mode=='percent': data['stdin']=sys.stdin.read()
(root/(mode+'.json')).write_text(json.dumps(data))
''')
        user_table = ('SHELL=/bin/sh\nPATH=/usr/bin:/bin\nLAB_TABLE=user\n'
                      f'* * * * * /usr/bin/python3 {worker} user\n'
                      f'* * * * * /usr/bin/python3 {worker} percent%first%second\n'
                      f'* * * * * root /usr/bin/id >{root}/wrong-column.err 2>&1\n')
        subprocess.run(['crontab', '-'], input=user_table, text=True, check=True)
        invalid = subprocess.run(['crontab', '-'], input='* * * * * /bin/true',
                                 text=True, capture_output=True)
        assert invalid.returncode != 0
        assert subprocess.check_output(['crontab', '-l'], text=True) == user_table
        system_table.write_text('SHELL=/bin/sh\nPATH=/usr/bin:/bin\nLAB_TABLE=system\n'
                                f'* * * * * {username} /usr/bin/python3 {worker} system\n')
        system_table.chmod(0o644)
        daemon = subprocess.Popen(['cron', '-f'], env={**os.environ, 'PARENT_ONLY': 'parent-demo'},
                                  stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
        try:
            deadline = time.monotonic() + 75
            names = ['user.json', 'system.json', 'percent.json', 'wrong-column.err']
            while time.monotonic() < deadline:
                assert daemon.poll() is None
                if all((root/name).exists() and (root/name).stat().st_size for name in names):
                    break
                time.sleep(0.2)
            else:
                raise AssertionError('scheduled table outputs were not produced')
            user = json.loads((root/'user.json').read_text())
            system = json.loads((root/'system.json').read_text())
            percent = json.loads((root/'percent.json').read_text())
            wrong = (root/'wrong-column.err').read_text()
            assert user['uid'] == 0 and user['table'] == 'user'
            assert system['uid'] == account.pw_uid and system['table'] == 'system'
            assert all(value['parent_only'] is None for value in (user, system))
            assert all(value['path'] == '/usr/bin:/bin' for value in (user, system))
            assert percent['stdin'] == 'first\nsecond\n'
            assert 'root: not found' in wrong
            print(json.dumps({'user_table': user, 'system_table': system,
                              'missing_final_newline_rejected': invalid.returncode,
                              'previous_table_preserved': True, 'percent_stdin': percent['stdin'],
                              'username_in_user_table_fails': wrong.strip()}), flush=True)
        finally:
            daemon.terminate()
            daemon.communicate(timeout=5)
            daemon = None
finally:
    if daemon is not None:
        daemon.terminate()
        daemon.communicate(timeout=5)
    system_table.unlink(missing_ok=True)
    subprocess.run(['crontab', '-r'], capture_output=True)
    subprocess.run(['userdel', '--remove', username], capture_output=True, check=True)