Enhance log sharing capability (#4526)

This commit is contained in:
Daniel Girtler 2026-05-15 21:38:17 +10:00 committed by GitHub
parent e48ca45b0b
commit 516a61d8af
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
6 changed files with 158 additions and 91 deletions

View File

@ -41,6 +41,7 @@ repos:
additional_dependencies:
- pydantic
- pytest
- hypothesis
- cryptography
- textual
- repo: local

View File

@ -6,7 +6,7 @@ import urllib.error
import urllib.parse
from argparse import ArgumentParser, Namespace
from dataclasses import dataclass, field
from enum import StrEnum
from enum import Enum, StrEnum
from pathlib import Path
from typing import Any, Self
from urllib.request import Request, urlopen
@ -35,6 +35,10 @@ from archinstall.lib.version import get_version
from archinstall.tui.components import tui
class SubCommand(Enum):
SHARE_LOG = 'share-log'
@p_dataclass
class Arguments:
config: Path | None = None
@ -58,6 +62,8 @@ class Arguments:
advanced: bool = False
verbose: bool = False
command: SubCommand | None = None
class ArchConfigType(StrEnum):
VERSION = 'version'
@ -365,13 +371,13 @@ class ArchConfig:
class ArchConfigHandler:
def __init__(self) -> None:
self._parser: ArgumentParser = self._define_arguments()
args: Arguments = self._parse_args()
self._args = args
self._add_sub_parsers()
self._args: Arguments = self._parse_args()
config = self._parse_config()
try:
self._config = ArchConfig.from_config(config, args)
self._config = ArchConfig.from_config(config, self._args)
self._config.version = get_version()
except ValueError as err:
warn(str(err))
@ -397,8 +403,13 @@ class ArchConfigHandler:
def print_help(self) -> None:
self._parser.print_help()
def _add_sub_parsers(self) -> None:
subparsers = self._parser.add_subparsers(dest='command', help='Available subcommands')
_ = subparsers.add_parser(SubCommand.SHARE_LOG.value, help='Upload log file to public server')
def _define_arguments(self) -> ArgumentParser:
parser = ArgumentParser(formatter_class=argparse.ArgumentDefaultsHelpFormatter)
parser.add_argument(
'-v',
'--version',

View File

@ -185,6 +185,10 @@ class Logger:
def path(self) -> Path:
return self._path / 'install.log'
@path.setter
def path(self, value: Path) -> None:
self._path = value
@property
def directory(self) -> Path:
return self._path
@ -212,6 +216,17 @@ class Logger:
level_name = logging.getLevelName(level)
f.write(f'[{ts}] - {level_name} - {content}\n')
def get_content(self, max_bytes: int | None = None) -> bytes:
content = self.path.read_bytes()
if max_bytes is not None:
size = self.path.stat().st_size
if size > max_bytes:
content = content[-max_bytes:]
return content
logger = Logger()
@ -295,6 +310,11 @@ def _stylize_output(
return f'\033[{ansi}m{text}\033[0m'
def _timestamp() -> str:
now = datetime.now(tz=UTC)
return now.strftime('%Y-%m-%d %H:%M:%S')
def info(
*msgs: str,
level: int = logging.INFO,
@ -306,11 +326,6 @@ def info(
log(*msgs, level=level, fg=fg, bg=bg, reset=reset, font=font)
def _timestamp() -> str:
now = datetime.now(tz=UTC)
return now.strftime('%Y-%m-%d %H:%M:%S')
def debug(
*msgs: str,
level: int = logging.DEBUG,
@ -368,35 +383,20 @@ def log(
def share_install_log(
paste_url: str = 'https://paste.rs',
max_size: int = 10 * 1024 * 1024,
confirm: Callable[[str], bool] = lambda _: True,
) -> int:
paste_url: str,
max_bytes: int | None = None,
) -> str | None:
log_path = logger.path
if not log_path.exists():
info(f'Log file not found: {log_path}')
return 1
return None
size = log_path.stat().st_size
if size == 0:
content = logger.get_content(max_bytes=max_bytes)
if len(content) == 0:
info(f'Log file is empty: {log_path}')
return 1
if size > max_size:
info(f'Log file exceeds {max_size} bytes, uploading last {max_size} bytes')
content = log_path.read_bytes()[-max_size:]
else:
content = log_path.read_bytes()
header = f'About to upload {log_path} ({len(content)} bytes) to {paste_url}\n\n'
header += 'The log may contain hostname, mirror URLs, package list and partition layout.\n'
header += 'The uploaded paste is public.\n\n'
header += 'Continue?'
if not confirm(header):
info('Cancelled.')
return 1
return None
try:
req = urllib.request.Request(paste_url, data=content)
@ -404,12 +404,10 @@ def share_install_log(
url = response.read().decode().strip()
except urllib.error.URLError as e:
info(f'Upload failed: {e}')
return 1
return None
if not url.startswith('http'):
info(f'Unexpected response from {paste_url}: {url[:200]!r}')
return 1
return None
# raw print so the URL is pipe-friendly (no ANSI colors, no log prefix)
print(url)
return 0
return url

View File

@ -8,13 +8,13 @@ import time
import traceback
from pathlib import Path
from archinstall.lib.args import ArchConfigHandler
from archinstall.lib.args import ArchConfigHandler, SubCommand
from archinstall.lib.disk.utils import disk_layouts
from archinstall.lib.hardware import SysInfo
from archinstall.lib.menu.helpers import Confirmation
from archinstall.lib.network.wifi_handler import WifiHandler
from archinstall.lib.networking import ping
from archinstall.lib.output import debug, error, info, share_install_log, warn
from archinstall.lib.output import debug, error, info, logger, share_install_log, warn
from archinstall.lib.packages.util import check_version_upgrade
from archinstall.lib.pacman.pacman import Pacman
from archinstall.lib.translationhandler import tr, translation_handler
@ -75,17 +75,36 @@ def _list_scripts() -> str:
return '\n'.join(lines)
def _tui_confirm(header: str) -> bool:
async def _ask() -> bool:
def _share_log_command() -> None:
paste_url: str = 'https://paste.rs'
log_path = logger.path
max_size = 10 * 1024 * 1024 # max supported size by paste.rs
content = logger.get_content(max_bytes=max_size).decode()
header = tr('About to upload "{}" to the publicly accessible {}').format(log_path, paste_url) + '\n\n'
header += tr('Do you want to continue?')
group = MenuItemGroup.yes_no()
group.set_preview_for_all(lambda _: content)
async def _confirm() -> bool:
result = await Confirmation(
group=MenuItemGroup.yes_no(),
header=header,
allow_skip=False,
preset=False,
group=group,
preview_header='Log content',
preview_location='bottom',
).show()
return result.get_value()
return tui.run(_ask)
result = tui.run(_confirm)
if result is True:
res = share_install_log(paste_url=paste_url, max_bytes=max_size)
if res is not None:
info(tr('Log uploaded successfully. URL: {}').format(res))
else:
error(tr('Failed to upload log.'))
def run() -> int:
@ -94,15 +113,19 @@ def run() -> int:
OR straight as a module: python -m archinstall
In any case we will be attempting to load the provided script to be run from the scripts/ folder
"""
if 'share-log' in sys.argv:
return share_install_log(confirm=_tui_confirm)
arch_config_handler = ArchConfigHandler()
if '--help' in sys.argv or '-h' in sys.argv:
arch_config_handler.print_help()
return 0
match arch_config_handler.args.command:
case SubCommand.SHARE_LOG:
_share_log_command()
exit(0)
case None:
pass
script = arch_config_handler.get_script()
if script == 'list':

View File

@ -40,6 +40,7 @@ dev = [
"ruff==0.15.13",
"pylint==4.0.5",
"pytest==9.0.3",
"hypothesis>=6.152.4",
]
doc = ["sphinx"]

View File

@ -1,94 +1,127 @@
# pylint: disable=redefined-outer-name
import string
import urllib.error
from io import BytesIO
from pathlib import Path
from unittest.mock import MagicMock, patch
from unittest.mock import patch
import pytest
from hypothesis import HealthCheck, given, settings
from hypothesis import strategies as st
from archinstall.lib.output import share_install_log
urls = st.builds(
'{}://{}.{}/{}'.format,
st.sampled_from(['http', 'https']),
st.text(alphabet=string.ascii_lowercase, min_size=3, max_size=10),
st.sampled_from(['com', 'net', 'org', 'rs']),
st.text(alphabet=string.ascii_lowercase + string.digits, min_size=0, max_size=8),
)
@pytest.fixture()
max_bytes = st.one_of(st.none(), st.integers(min_value=1, max_value=130))
random_paths = st.lists(
st.text(
alphabet=string.ascii_lowercase + string.digits,
min_size=1,
max_size=10,
),
min_size=1,
max_size=5,
).map(lambda parts: Path(*parts))
@pytest.fixture
def log_file(tmp_path: Path) -> Path:
log_dir = tmp_path / 'archinstall'
log_dir.mkdir()
return log_dir / 'install.log'
def _fake_logger(log_file: Path) -> MagicMock:
mock = MagicMock()
mock.path = log_file
return mock
# def _fake_logger(log_file: Path) -> MagicMock:
# mock = MagicMock()
# mock.path = log_file
# return mock
def test_file_not_found(tmp_path: Path) -> None:
missing = tmp_path / 'no-such' / 'install.log'
with patch('archinstall.lib.output.logger', _fake_logger(missing)):
assert share_install_log() == 1
@given(paste_url=urls, max_byte=max_bytes, sub_path=random_paths)
@settings(max_examples=3, suppress_health_check=[HealthCheck.function_scoped_fixture])
def test_file_not_found(
tmp_path: Path,
sub_path: Path,
paste_url: str,
max_byte: int | None,
) -> None:
missing_log = tmp_path / sub_path / 'install.log'
with patch('archinstall.lib.output.logger._path', new=missing_log):
assert share_install_log(paste_url, max_byte) is None
def test_empty_file(log_file: Path) -> None:
@given(paste_url=urls, max_byte=max_bytes)
@settings(max_examples=3, suppress_health_check=[HealthCheck.function_scoped_fixture])
def test_empty_file(log_file: Path, paste_url: str, max_byte: int | None) -> None:
log_file.write_bytes(b'')
with patch('archinstall.lib.output.logger', _fake_logger(log_file)):
assert share_install_log() == 1
with patch('archinstall.lib.output.logger._path', new=log_file.parent):
# with patch('archinstall.lib.output.logger', _fake_logger(log_file)):
assert share_install_log(paste_url, max_byte) is None
def test_user_cancels(log_file: Path) -> None:
@given(paste_url=urls, resp_url=urls, max_byte=max_bytes)
@settings(max_examples=3, suppress_health_check=[HealthCheck.function_scoped_fixture])
def test_successful_upload(log_file: Path, resp_url: str, paste_url: str, max_byte: int | None) -> None:
log_file.write_text('some log content')
with patch('archinstall.lib.output.logger', _fake_logger(log_file)):
assert share_install_log(confirm=lambda _: False) == 1
def test_successful_upload(log_file: Path) -> None:
log_file.write_text('some log content')
fake_response = BytesIO(b'https://paste.rs/abc.def')
fake_response = BytesIO(resp_url.encode())
with (
patch('archinstall.lib.output.logger', _fake_logger(log_file)),
patch('urllib.request.urlopen', return_value=fake_response) as mock_open,
patch('archinstall.lib.output.logger._path', new=log_file.parent),
patch('urllib.request.urlopen', return_value=fake_response),
):
result = share_install_log()
assert result == 0
req = mock_open.call_args[0][0]
assert req.data == b'some log content'
result = share_install_log(paste_url, max_byte)
assert result == resp_url
def test_truncation(log_file: Path) -> None:
max_size = 100
@given(paste_url=urls, resp_url=urls, max_byte=max_bytes)
@settings(max_examples=3, suppress_health_check=[HealthCheck.function_scoped_fixture])
def test_truncation(log_file: Path, resp_url: str, paste_url: str, max_byte: int | None) -> None:
content = b'A' * 50 + b'B' * 80
log_file.write_bytes(content)
fake_response = BytesIO(b'https://paste.rs/abc.def')
fake_response = BytesIO(resp_url.encode())
exptected_byte = len(content) if max_byte is None else max_byte
with (
patch('archinstall.lib.output.logger', _fake_logger(log_file)),
patch('archinstall.lib.output.logger._path', new=log_file.parent),
patch('urllib.request.urlopen', return_value=fake_response) as mock_open,
):
result = share_install_log(max_size=max_size)
assert result == 0
_ = share_install_log(paste_url, max_byte)
req = mock_open.call_args[0][0]
assert len(req.data) == max_size
assert req.data == content[-max_size:]
assert len(req.data) == exptected_byte
assert req.data == content[-exptected_byte:]
def test_network_error(log_file: Path) -> None:
@given(paste_url=urls, max_byte=max_bytes)
@settings(max_examples=3, suppress_health_check=[HealthCheck.function_scoped_fixture])
def test_network_error(log_file: Path, paste_url: str, max_byte: int | None) -> None:
log_file.write_text('some log content')
with (
patch('archinstall.lib.output.logger', _fake_logger(log_file)),
patch('archinstall.lib.output.logger._path', new=log_file.parent),
patch('urllib.request.urlopen', side_effect=urllib.error.URLError('no network')),
):
assert share_install_log() == 1
assert share_install_log(paste_url, max_byte) is None
def test_unexpected_response(log_file: Path) -> None:
@given(paste_url=urls, max_byte=max_bytes)
@settings(max_examples=3, suppress_health_check=[HealthCheck.function_scoped_fixture])
def test_unexpected_response(log_file: Path, paste_url: str, max_byte: int | None) -> None:
log_file.write_text('some log content')
fake_response = BytesIO(b'ERROR: something went wrong')
with (
patch('archinstall.lib.output.logger', _fake_logger(log_file)),
patch('archinstall.lib.output.logger._path', new=log_file.parent),
patch('urllib.request.urlopen', return_value=fake_response),
):
assert share_install_log() == 1
assert share_install_log(paste_url, max_byte) is None