diff --git a/python/online/fxreader/pr34/commands_typed/archlinux/cli/diff.py b/python/online/fxreader/pr34/commands_typed/archlinux/cli/diff.py new file mode 100644 index 0000000..9f8c675 --- /dev/null +++ b/python/online/fxreader/pr34/commands_typed/archlinux/cli/diff.py @@ -0,0 +1,159 @@ +"""Diff CLI: compare two compiled requirement lists.""" + +import argparse +import dataclasses +import logging +import pathlib + +logger = logging.getLogger(__name__) + + +@dataclasses.dataclass +class diff_entry_t: + name: str + kind: str # 'added', 'removed', 'version', 'url', 'sha256' + old: str = '' + new: str = '' + + +def parse_compiled(txt: str) -> dict[str, dict[str, str]]: + """Parse compiled requirements text into {name: {version, url, sha256}}.""" + result: dict[str, dict[str, str]] = {} + pending_url = '' + + for line in txt.splitlines(): + line = line.strip() + if line == '': + continue + if line.startswith('#'): + url = line[1:].strip() + # strip trailing annotation like "URL # pinned" + if ' #' in url: + url = url.split(' #', 1)[0].strip() + pending_url = url + continue + + # strip trailing inline comment like "pkg==1.0 --hash=... # pinned" + if ' #' in line: + line = line.split(' #', 1)[0].strip() + + parts = line.split() + pkg_spec = parts[0] + sha256 = '' + for p in parts[1:]: + if p.startswith('--hash=sha256:'): + sha256 = p[len('--hash=sha256:'):] + + if '==' in pkg_spec: + name, version = pkg_spec.split('==', 1) + else: + name = pkg_spec + version = '' + + result[name] = { + 'version': version, + 'url': pending_url, + 'sha256': sha256, + } + pending_url = '' + + return result + + +def compute_diff( + old: dict[str, dict[str, str]], + new: dict[str, dict[str, str]], + diff_url: bool = False, + diff_checksum: bool = False, +) -> list[diff_entry_t]: + entries: list[diff_entry_t] = [] + + all_names = sorted(set(old.keys()) | set(new.keys())) + + for name in all_names: + if name not in old: + entries.append(diff_entry_t(name=name, kind='added', new=new[name]['version'])) + continue + + if name not in new: + entries.append(diff_entry_t(name=name, kind='removed', old=old[name]['version'])) + continue + + o = old[name] + n = new[name] + + if o['version'] != n['version']: + entries.append(diff_entry_t(name=name, kind='version', old=o['version'], new=n['version'])) + + if diff_url and o['url'] != n['url']: + entries.append(diff_entry_t(name=name, kind='url', old=o['url'], new=n['url'])) + + if diff_checksum and o['sha256'] != n['sha256']: + entries.append(diff_entry_t(name=name, kind='sha256', old=o['sha256'], new=n['sha256'])) + + return entries + + +def format_diff(entries: list[diff_entry_t]) -> str: + lines: list[str] = [] + for e in entries: + if e.kind == 'added': + lines.append('+ %s==%s' % (e.name, e.new)) + elif e.kind == 'removed': + lines.append('- %s==%s' % (e.name, e.old)) + elif e.kind == 'version': + lines.append('~ %s: %s -> %s' % (e.name, e.old, e.new)) + elif e.kind == 'url': + lines.append('U %s: %s -> %s' % (e.name, e.old, e.new)) + elif e.kind == 'sha256': + lines.append('H %s: %s -> %s' % (e.name, e.old, e.new)) + return '\n'.join(lines) + + +def main(args: list[str]) -> int: + parser = argparse.ArgumentParser( + prog='online-fxreader-pr34-archlinux diff', + description='Compare two compiled requirement lists.', + ) + parser.add_argument( + 'old', + help='path to old compiled requirements file', + ) + parser.add_argument( + 'new', + help='path to new compiled requirements file', + ) + parser.add_argument( + '--diff-url', + default=False, + action=argparse.BooleanOptionalAction, + help='include url differences (default: off)', + ) + parser.add_argument( + '--diff-checksum', + default=False, + action=argparse.BooleanOptionalAction, + help='include sha256 checksum differences (default: off)', + ) + + options = parser.parse_args(args) + + old_txt = pathlib.Path(options.old).read_text() + new_txt = pathlib.Path(options.new).read_text() + + old_parsed = parse_compiled(old_txt) + new_parsed = parse_compiled(new_txt) + + entries = compute_diff( + old_parsed, + new_parsed, + diff_url=options.diff_url, + diff_checksum=options.diff_checksum, + ) + + if len(entries) == 0: + print('no differences') + return 0 + + print(format_diff(entries)) + return 0 diff --git a/python/online/fxreader/pr34/commands_typed/archlinux/tests/test_diff.py b/python/online/fxreader/pr34/commands_typed/archlinux/tests/test_diff.py new file mode 100644 index 0000000..4043180 --- /dev/null +++ b/python/online/fxreader/pr34/commands_typed/archlinux/tests/test_diff.py @@ -0,0 +1,188 @@ +import unittest + +from ..cli.diff import ( + compute_diff, + diff_entry_t, + format_diff, + parse_compiled, +) + + +class TestParseCompiled(unittest.TestCase): + def test_simple(self) -> None: + txt = ( + '# https://example.com/core/bash-5.2-1-x86_64.pkg.tar.zst\n' + 'bash==5.2-1\n' + ) + parsed = parse_compiled(txt) + self.assertEqual(parsed['bash']['version'], '5.2-1') + self.assertEqual(parsed['bash']['url'], 'https://example.com/core/bash-5.2-1-x86_64.pkg.tar.zst') + + def test_with_hash(self) -> None: + txt = ( + '# https://example.com/bash.pkg\n' + 'bash==5.2-1 --hash=sha256:abc123\n' + ) + parsed = parse_compiled(txt) + self.assertEqual(parsed['bash']['sha256'], 'abc123') + + def test_pinned_annotation_stripped(self) -> None: + txt = ( + '# https://example.com/bash.pkg # pinned\n' + 'bash==5.2-1\n' + ) + parsed = parse_compiled(txt) + self.assertEqual(parsed['bash']['url'], 'https://example.com/bash.pkg') + + def test_multiple(self) -> None: + txt = ( + '# https://example.com/bash.pkg\n' + 'bash==5.2-1\n' + '# https://example.com/glibc.pkg\n' + 'glibc==2.38-1\n' + ) + parsed = parse_compiled(txt) + self.assertEqual(len(parsed), 2) + self.assertIn('bash', parsed) + self.assertIn('glibc', parsed) + + def test_empty(self) -> None: + self.assertEqual(parse_compiled(''), {}) + + def test_no_url(self) -> None: + txt = 'bash==5.2-1\n' + parsed = parse_compiled(txt) + self.assertEqual(parsed['bash']['version'], '5.2-1') + self.assertEqual(parsed['bash']['url'], '') + + +class TestComputeDiff(unittest.TestCase): + def test_no_changes(self) -> None: + old = {'bash': {'version': '5.2-1', 'url': 'u', 'sha256': ''}} + new = {'bash': {'version': '5.2-1', 'url': 'u', 'sha256': ''}} + self.assertEqual(compute_diff(old, new), []) + + def test_added(self) -> None: + old: dict[str, dict[str, str]] = {} + new = {'bash': {'version': '5.2-1', 'url': '', 'sha256': ''}} + entries = compute_diff(old, new) + self.assertEqual(len(entries), 1) + self.assertEqual(entries[0].kind, 'added') + self.assertEqual(entries[0].name, 'bash') + self.assertEqual(entries[0].new, '5.2-1') + + def test_removed(self) -> None: + old = {'bash': {'version': '5.2-1', 'url': '', 'sha256': ''}} + new: dict[str, dict[str, str]] = {} + entries = compute_diff(old, new) + self.assertEqual(len(entries), 1) + self.assertEqual(entries[0].kind, 'removed') + self.assertEqual(entries[0].old, '5.2-1') + + def test_version_change(self) -> None: + old = {'bash': {'version': '5.2-1', 'url': '', 'sha256': ''}} + new = {'bash': {'version': '5.3-1', 'url': '', 'sha256': ''}} + entries = compute_diff(old, new) + self.assertEqual(len(entries), 1) + self.assertEqual(entries[0].kind, 'version') + self.assertEqual(entries[0].old, '5.2-1') + self.assertEqual(entries[0].new, '5.3-1') + + def test_url_change_requires_flag(self) -> None: + old = {'bash': {'version': '5.2-1', 'url': 'http://a/bash.pkg', 'sha256': ''}} + new = {'bash': {'version': '5.2-1', 'url': 'http://b/bash.pkg', 'sha256': ''}} + # off by default + self.assertEqual(compute_diff(old, new), []) + # enabled + entries = compute_diff(old, new, diff_url=True) + self.assertEqual(len(entries), 1) + self.assertEqual(entries[0].kind, 'url') + + def test_sha256_change_requires_flag(self) -> None: + old = {'bash': {'version': '5.2-1', 'url': '', 'sha256': 'aaa'}} + new = {'bash': {'version': '5.2-1', 'url': '', 'sha256': 'bbb'}} + self.assertEqual(compute_diff(old, new), []) + entries = compute_diff(old, new, diff_checksum=True) + self.assertEqual(len(entries), 1) + self.assertEqual(entries[0].kind, 'sha256') + + def test_multiple_changes(self) -> None: + old = { + 'bash': {'version': '5.2-1', 'url': 'u1', 'sha256': 'aaa'}, + 'glibc': {'version': '2.38-1', 'url': '', 'sha256': ''}, + 'removed': {'version': '1.0', 'url': '', 'sha256': ''}, + } + new = { + 'bash': {'version': '5.3-1', 'url': 'u2', 'sha256': 'bbb'}, + 'glibc': {'version': '2.38-1', 'url': '', 'sha256': ''}, + 'added': {'version': '2.0', 'url': '', 'sha256': ''}, + } + entries = compute_diff(old, new, diff_url=True, diff_checksum=True) + + kinds = {(e.name, e.kind) for e in entries} + self.assertIn(('bash', 'version'), kinds) + self.assertIn(('bash', 'url'), kinds) + self.assertIn(('bash', 'sha256'), kinds) + self.assertIn(('removed', 'removed'), kinds) + self.assertIn(('added', 'added'), kinds) + + def test_empty_side_shows_when_flag_enabled(self) -> None: + """When old has empty url but new has one, flag makes it visible.""" + old = {'bash': {'version': '5.2-1', 'url': '', 'sha256': ''}} + new = {'bash': {'version': '5.2-1', 'url': 'http://b/bash', 'sha256': 'x'}} + self.assertEqual(compute_diff(old, new), []) + entries = compute_diff(old, new, diff_url=True, diff_checksum=True) + kinds = {e.kind for e in entries} + self.assertIn('url', kinds) + self.assertIn('sha256', kinds) + + +class TestFormatDiff(unittest.TestCase): + def test_added(self) -> None: + entries = [diff_entry_t(name='bash', kind='added', new='5.2-1')] + self.assertEqual(format_diff(entries), '+ bash==5.2-1') + + def test_removed(self) -> None: + entries = [diff_entry_t(name='bash', kind='removed', old='5.2-1')] + self.assertEqual(format_diff(entries), '- bash==5.2-1') + + def test_version(self) -> None: + entries = [diff_entry_t(name='bash', kind='version', old='5.2-1', new='5.3-1')] + self.assertEqual(format_diff(entries), '~ bash: 5.2-1 -> 5.3-1') + + def test_url(self) -> None: + entries = [diff_entry_t(name='bash', kind='url', old='u1', new='u2')] + self.assertIn('U bash:', format_diff(entries)) + + def test_sha256(self) -> None: + entries = [diff_entry_t(name='bash', kind='sha256', old='aaa', new='bbb')] + self.assertIn('H bash:', format_diff(entries)) + + +class TestDiffEndToEnd(unittest.TestCase): + def test_full_flow(self) -> None: + old_txt = ( + '# https://example.com/bash-5.2-1.pkg\n' + 'bash==5.2-1 --hash=sha256:aaa\n' + '# https://example.com/glibc-2.38-1.pkg\n' + 'glibc==2.38-1 --hash=sha256:bbb\n' + ) + new_txt = ( + '# https://example.com/bash-5.3-1.pkg\n' + 'bash==5.3-1 --hash=sha256:ccc\n' + '# https://example.com/glibc-2.38-1.pkg\n' + 'glibc==2.38-1 --hash=sha256:bbb\n' + '# https://example.com/python-3.12.pkg\n' + 'python==3.12-1\n' + ) + + old_parsed = parse_compiled(old_txt) + new_parsed = parse_compiled(new_txt) + entries = compute_diff(old_parsed, new_parsed, diff_url=True, diff_checksum=True) + + kinds = {(e.name, e.kind) for e in entries} + self.assertIn(('bash', 'version'), kinds) + self.assertIn(('bash', 'sha256'), kinds) + self.assertIn(('bash', 'url'), kinds) + self.assertIn(('python', 'added'), kinds) + self.assertNotIn(('glibc', 'version'), kinds) diff --git a/python/online/fxreader/pr34/commands_typed/logging.py b/python/online/fxreader/pr34/commands_typed/logging.py index abc4bf8..0ee02fd 100644 --- a/python/online/fxreader/pr34/commands_typed/logging.py +++ b/python/online/fxreader/pr34/commands_typed/logging.py @@ -1,16 +1,77 @@ +import enum +import json import logging +import logging.handlers +import pathlib + from typing import ( + Any, Optional, ) -def setup(level: Optional[int] = None) -> None: +PLAIN_FORMAT: str = '%(levelname)s:%(name)s:%(message)s:%(process)d:%(asctime)s:%(pathname)s:%(funcName)s:%(lineno)s' + + +class format_t(enum.Enum): + plain = 'plain' + json = 'json' + + +class json_formatter_t(logging.Formatter): + def format(self, record: logging.LogRecord) -> str: + payload: dict[str, Any] = { + 'level': record.levelname, + 'name': record.name, + 'message': record.getMessage(), + 'process': record.process, + 'time': self.formatTime(record), + 'pathname': record.pathname, + 'funcName': record.funcName, + 'lineno': record.lineno, + } + if record.exc_info: + payload['exc_info'] = self.formatException(record.exc_info) + return json.dumps(payload) + + +def setup( + level: Optional[int] = None, + format: format_t = format_t.plain, + log_dir: Optional[pathlib.Path] = None, + log_name: str = 'app', + max_bytes: int = 16 * 1024 * 1024, + backup_count: int = 5, + use_console: bool = True, +) -> None: if level is None: level = logging.INFO - logging.basicConfig( - level=level, - format=( - '%(levelname)s:%(name)s:%(message)s:%(process)d:%(asctime)s:%(pathname)s:%(funcName)s:%(lineno)s' - ), - ) + if format is format_t.json: + formatter: logging.Formatter = json_formatter_t() + else: + formatter = logging.Formatter(PLAIN_FORMAT) + + root = logging.getLogger() + root.setLevel(level) + + # clear existing handlers so repeated setup() calls don't stack + for h in list(root.handlers): + root.removeHandler(h) + + if use_console: + console = logging.StreamHandler() + console.setLevel(level) + console.setFormatter(formatter) + root.addHandler(console) + + if log_dir is not None: + log_dir.mkdir(parents=True, exist_ok=True) + file_handler = logging.handlers.RotatingFileHandler( + filename=str(log_dir / ('%s.log' % log_name)), + maxBytes=max_bytes, + backupCount=backup_count, + ) + file_handler.setLevel(level) + file_handler.setFormatter(formatter) + root.addHandler(file_handler)