diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 91fd6f336a..de2bcef4b9 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -41,6 +41,7 @@ repos: additional_dependencies: - pydantic - pytest + - hypothesis - cryptography - textual - repo: local diff --git a/archinstall/lib/args.py b/archinstall/lib/args.py index efe9baa2f3..14aff6c69c 100644 --- a/archinstall/lib/args.py +++ b/archinstall/lib/args.py @@ -6,7 +6,7 @@ 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.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 @@ def from_config(cls, args_config: dict[str, Any], args: Arguments) -> Self: 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 @@ def get_script(self) -> str: 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', diff --git a/archinstall/lib/output.py b/archinstall/lib/output.py index 86a47ed1db..bc9bc7a3c1 100644 --- a/archinstall/lib/output.py +++ b/archinstall/lib/output.py @@ -185,6 +185,10 @@ def __init__(self, path: Path = Path('/var/log/archinstall')) -> None: 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 @@ def log(self, level: int, content: str) -> None: 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: - 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() + content = logger.get_content(max_bytes=max_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 + if len(content) == 0: + info(f'Log file is empty: {log_path}') + 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 diff --git a/archinstall/main.py b/archinstall/main.py index 9b91681b00..f30b01d35b 100644 --- a/archinstall/main.py +++ b/archinstall/main.py @@ -8,13 +8,13 @@ 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': diff --git a/pyproject.toml b/pyproject.toml index 636c80179f..133bda54af 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -40,6 +40,7 @@ dev = [ "ruff==0.15.13", "pylint==4.0.5", "pytest==9.0.3", + "hypothesis>=6.152.4", ] doc = ["sphinx"] diff --git a/tests/test_share_log.py b/tests/test_share_log.py index c1c3ca84d7..a6706da850 100644 --- a/tests/test_share_log.py +++ b/tests/test_share_log.py @@ -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: - log_file.write_bytes(b'') - with patch('archinstall.lib.output.logger', _fake_logger(log_file)): - assert share_install_log() == 1 +@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'') -def test_user_cancels(log_file: Path) -> 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 + 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_successful_upload(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') - 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 - req = mock_open.call_args[0][0] - assert len(req.data) == max_size - assert req.data == content[-max_size:] + _ = share_install_log(paste_url, max_byte) + req = mock_open.call_args[0][0] + 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