upgrades: Return reason when checking for dist upgrade

With default logging settings, only print the result.

Tests: Checked output for following cases:
- Dist upgrade flag exists.
- System is Debian testing.
- Latest stable release without --test.
- Automatic updates not enabled.
- Not enough free space.
- Dist upgrade started.

Signed-off-by: James Valleroy <jvalleroy@mailbox.org>
Reviewed-by: Veiko Aasa <veiko17@disroot.org>
This commit is contained in:
James Valleroy 2021-01-27 20:56:28 -05:00 committed by Veiko Aasa
parent 0b9c4c92e9
commit 906816f7cb
No known key found for this signature in database
GPG Key ID: 478539CAE680674E

View File

@ -5,6 +5,7 @@ Configures or runs unattended-upgrades
""" """
import argparse import argparse
import logging
import os import os
import pathlib import pathlib
import re import re
@ -309,20 +310,21 @@ def _add_apt_preferences():
def _check_dist_upgrade(test_upgrade=False): def _check_dist_upgrade(test_upgrade=False):
"""Check for new stable release. If there is one, and updates are """Check for new stable release, if updates are enabled, and if there is
enabled, return True. enough free space for the dist upgrade.
If test_upgrade is True, also check for upgrade to testing. If test_upgrade is True, also check for upgrade to testing.
Returns (boolean, string) indicating if the upgrade is ready, and a reason
if not.
""" """
if dist_upgrade_flag.exists(): if dist_upgrade_flag.exists():
print('Found previously interrupted dist-upgrade.') return (True, 'Found previously interrupted dist-upgrade.')
return True
release, dist = get_current_release() release, dist = get_current_release()
if release in ['unstable', 'testing']: if release in ['unstable', 'testing']:
print(f'System release is {release}. Skip checking for new stable ' return (False, f'System release is {release}. Skip checking for new '
'release.') 'stable release.')
return False
check_dists = ['stable'] check_dists = ['stable']
if test_upgrade: if test_upgrade:
@ -335,42 +337,38 @@ def _check_dist_upgrade(test_upgrade=False):
protocol = _get_protocol() protocol = _get_protocol()
if protocol == 'tor+http': if protocol == 'tor+http':
command.insert(0, 'torsocks') command.insert(0, 'torsocks')
print('Package download over Tor is enabled.') logging.info('Package download over Tor is enabled.')
try: try:
output = subprocess.check_output(command).decode() output = subprocess.check_output(command).decode()
except (subprocess.CalledProcessError, FileNotFoundError): except (subprocess.CalledProcessError, FileNotFoundError):
print(f'Error while checking for new {check_dist} release') logging.warning('Error while checking for new %s release',
check_dist)
else: else:
for line in output.split('\n'): for line in output.split('\n'):
if line.startswith('Codename:'): if line.startswith('Codename:'):
codename = line.split()[1] codename = line.split()[1]
if not codename: if not codename:
print('"Codename:" not found in release file.') return (False, '"Codename:" not found in release file.')
return False
if codename == dist: if codename == dist:
print(f'{dist} is already the latest release.') return (False, f'{dist} is already the latest release.')
return False
if not _check_auto(): if not _check_auto():
print('Automatic updates are not enabled.') return (False, 'Automatic updates are not enabled.')
return False
if check_dist == 'testing' and not test_upgrade: if check_dist == 'testing' and not test_upgrade:
print(f'Skipping dist-upgrade to {check_dist} since --test is not ' return (False, f'Skipping dist-upgrade to {check_dist} since --test is'
'set.') ' not set.')
return False
output = subprocess.check_output(['df', '--output=avail,pcent', '/']) output = subprocess.check_output(['df', '--output=avail,pcent', '/'])
output = output.decode().split('\n')[1].split() output = output.decode().split('\n')[1].split()
free_space, free_percent = int(output[0]), int(output[1][:-1]) free_space, free_percent = int(output[0]), int(output[1][:-1])
if free_space < 5000000 or free_percent < 10: if free_space < 5000000 or free_percent < 10:
print('Not enough free space in /.') return (False, 'Not enough free space in /.')
return False
print(f'Upgrading from {dist} to {codename}...') logging.info('Upgrading from %s to %s...', dist, codename)
with open(SOURCES_LIST, 'r') as sources_list: with open(SOURCES_LIST, 'r') as sources_list:
lines = sources_list.readlines() lines = sources_list.readlines()
@ -388,9 +386,9 @@ def _check_dist_upgrade(test_upgrade=False):
sources_list.write(new_line) sources_list.write(new_line)
print('Dist upgrade in progress. Setting flag.') logging.info('Dist upgrade in progress. Setting flag.')
dist_upgrade_flag.touch(mode=0o660) dist_upgrade_flag.touch(mode=0o660)
return True return (True, 'Started dist upgrade.')
def _perform_dist_upgrade(): def _perform_dist_upgrade():
@ -479,7 +477,9 @@ def subcommand_start_dist_upgrade(arguments):
Check if a new stable release is available, and start dist-upgrade process Check if a new stable release is available, and start dist-upgrade process
if updates are enabled. if updates are enabled.
""" """
if _check_dist_upgrade(arguments.test): upgrade_ready, reason = _check_dist_upgrade(arguments.test)
print(reason)
if upgrade_ready:
with open(DIST_UPGRADE_SERVICE_PATH, 'w') as service_file: with open(DIST_UPGRADE_SERVICE_PATH, 'w') as service_file:
service_file.write(DIST_UPGRADE_SERVICE) service_file.write(DIST_UPGRADE_SERVICE)