git.scottworley.com Git - pinch/blob - pinch.py

   1 import argparse
   2 import configparser
   3 import filecmp
   4 import functools
   5 import getpass
   6 import hashlib
   7 import operator
   8 import os
   9 import os.path
  10 import shlex
  11 import shutil
  12 import subprocess
  13 import sys
  14 import tarfile
  15 import tempfile
  16 import types
  17 import urllib.parse
  18 import urllib.request
  19 import xml.dom.minidom
  20
  21 from typing import (
  22     Callable,
  23     Dict,
  24     Iterable,
  25     List,
  26     Mapping,
  27     NamedTuple,
  28     NewType,
  29     Optional,
  30     Set,
  31     Tuple,
  32     Type,
  33     TypeVar,
  34     Union,
  35 )
  36
  37 import git_cache
  38
  39 # Use xdg module when it's less painful to have as a dependency
  40
  41
  42 class XDG(NamedTuple):
  43     XDG_CACHE_HOME: str
  44
  45
  46 xdg = XDG(
  47     XDG_CACHE_HOME=os.getenv(
  48         'XDG_CACHE_HOME',
  49         os.path.expanduser('~/.cache')))
  50
  51
  52 class VerificationError(Exception):
  53     pass
  54
  55
  56 class Verification:
  57
  58     def __init__(self) -> None:
  59         self.line_length = 0
  60
  61     def status(self, s: str) -> None:
  62         print(s, end=' ', file=sys.stderr, flush=True)
  63         self.line_length += 1 + len(s)  # Unicode??
  64
  65     @staticmethod
  66     def _color(s: str, c: int) -> str:
  67         return '\033[%2dm%s\033[00m' % (c, s)
  68
  69     def result(self, r: bool) -> None:
  70         message, color = {True: ('OK ', 92), False: ('FAIL', 91)}[r]
  71         length = len(message)
  72         cols = shutil.get_terminal_size().columns or 80
  73         pad = (cols - (self.line_length + length)) % cols
  74         print(' ' * pad + self._color(message, color), file=sys.stderr)
  75         self.line_length = 0
  76         if not r:
  77             raise VerificationError()
  78
  79     def check(self, s: str, r: bool) -> None:
  80         self.status(s)
  81         self.result(r)
  82
  83     def ok(self) -> None:
  84         self.result(True)
  85
  86
  87 Digest16 = NewType('Digest16', str)
  88 Digest32 = NewType('Digest32', str)
  89
  90
  91 class ChannelTableEntry(types.SimpleNamespace):
  92     absolute_url: str
  93     digest: Digest16
  94     file: str
  95     size: int
  96     url: str
  97
  98
  99 class AliasPin(NamedTuple):
 100     pass
 101
 102
 103 class SymlinkPin(NamedTuple):
 104     @property
 105     def release_name(self) -> str:
 106         return 'link'
 107
 108
 109 class GitPin(NamedTuple):
 110     git_revision: str
 111     release_name: str
 112
 113
 114 class ChannelPin(NamedTuple):
 115     git_revision: str
 116     release_name: str
 117     tarball_url: str
 118     tarball_sha256: str
 119
 120
 121 Pin = Union[AliasPin, SymlinkPin, GitPin, ChannelPin]
 122
 123
 124 def copy_to_nix_store(v: Verification, filename: str) -> str:
 125     v.status('Putting tarball in Nix store')
 126     process = subprocess.run(
 127         ['nix-store', '--add', filename], stdout=subprocess.PIPE)
 128     v.result(process.returncode == 0)
 129     return process.stdout.decode().strip()  # type: ignore  # (for old mypy)
 130
 131
 132 def symlink_archive(v: Verification, path: str) -> str:
 133     with tempfile.TemporaryDirectory() as td:
 134         archive_filename = os.path.join(td, 'link.tar.gz')
 135         os.symlink(path, os.path.join(td, 'link'))
 136         with tarfile.open(archive_filename, mode='x:gz') as t:
 137             t.add(os.path.join(td, 'link'), arcname='link')
 138         return copy_to_nix_store(v, archive_filename)
 139
 140
 141 class AliasSearchPath(NamedTuple):
 142     alias_of: str
 143
 144     # pylint: disable=no-self-use
 145     def pin(self, _: Verification, __: Optional[Pin]) -> AliasPin:
 146         return AliasPin()
 147
 148
 149 class SymlinkSearchPath(NamedTuple):
 150     path: str
 151
 152     # pylint: disable=no-self-use
 153     def pin(self, _: Verification, __: Optional[Pin]) -> SymlinkPin:
 154         return SymlinkPin()
 155
 156     def fetch(self, v: Verification, _: Pin) -> str:
 157         return symlink_archive(v, self.path)
 158
 159
 160 class GitSearchPath(NamedTuple):
 161     git_ref: str
 162     git_repo: str
 163
 164     def pin(self, v: Verification, old_pin: Optional[Pin]) -> GitPin:
 165         _, new_revision = git_cache.fetch(self.git_repo, self.git_ref)
 166         if old_pin is not None:
 167             assert isinstance(old_pin, GitPin)
 168             verify_git_ancestry(v, self, old_pin.git_revision, new_revision)
 169         return GitPin(release_name=git_revision_name(v, self, new_revision),
 170                       git_revision=new_revision)
 171
 172     def fetch(self, v: Verification, pin: Pin) -> str:
 173         assert isinstance(pin, GitPin)
 174         git_cache.ensure_rev_available(
 175             self.git_repo, self.git_ref, pin.git_revision)
 176         return git_get_tarball(v, self, pin)
 177
 178
 179 class ChannelSearchPath(NamedTuple):
 180     channel_url: str
 181     git_ref: str
 182     git_repo: str
 183
 184     def pin(self, v: Verification, old_pin: Optional[Pin]) -> ChannelPin:
 185         if old_pin is not None:
 186             assert isinstance(old_pin, ChannelPin)
 187
 188         channel_html, forwarded_url = fetch_channel(v, self)
 189         table, new_gitpin = parse_channel(v, channel_html)
 190         if old_pin is not None and old_pin.git_revision == new_gitpin.git_revision:
 191             return old_pin
 192         fetch_resources(v, new_gitpin, forwarded_url, table)
 193         git_cache.ensure_rev_available(
 194             self.git_repo, self.git_ref, new_gitpin.git_revision)
 195         if old_pin is not None:
 196             verify_git_ancestry(
 197                 v, self, old_pin.git_revision, new_gitpin.git_revision)
 198         check_channel_contents(v, self, table, new_gitpin)
 199         return ChannelPin(
 200             release_name=new_gitpin.release_name,
 201             tarball_url=table['nixexprs.tar.xz'].absolute_url,
 202             tarball_sha256=table['nixexprs.tar.xz'].digest,
 203             git_revision=new_gitpin.git_revision)
 204
 205     # pylint: disable=no-self-use
 206     def fetch(self, v: Verification, pin: Pin) -> str:
 207         assert isinstance(pin, ChannelPin)
 208
 209         return fetch_with_nix_prefetch_url(
 210             v, pin.tarball_url, Digest16(pin.tarball_sha256))
 211
 212
 213 SearchPath = Union[AliasSearchPath,
 214                    SymlinkSearchPath,
 215                    GitSearchPath,
 216                    ChannelSearchPath]
 217 TarrableSearchPath = Union[GitSearchPath, ChannelSearchPath]
 218
 219
 220 def compare(a: str, b: str) -> Tuple[List[str], List[str], List[str]]:
 221
 222     def throw(error: OSError) -> None:
 223         raise error
 224
 225     def join(x: str, y: str) -> str:
 226         return y if x == '.' else os.path.join(x, y)
 227
 228     def recursive_files(d: str) -> Iterable[str]:
 229         all_files: List[str] = []
 230         for path, dirs, files in os.walk(d, onerror=throw):
 231             rel = os.path.relpath(path, start=d)
 232             all_files.extend(join(rel, f) for f in files)
 233             for dir_or_link in dirs:
 234                 if os.path.islink(join(path, dir_or_link)):
 235                     all_files.append(join(rel, dir_or_link))
 236         return all_files
 237
 238     def exclude_dot_git(files: Iterable[str]) -> Iterable[str]:
 239         return (f for f in files if not f.startswith('.git/'))
 240
 241     files = functools.reduce(
 242         operator.or_, (set(
 243             exclude_dot_git(
 244                 recursive_files(x))) for x in [a, b]))
 245     return filecmp.cmpfiles(a, b, files, shallow=False)
 246
 247
 248 def fetch_channel(
 249         v: Verification, channel: ChannelSearchPath) -> Tuple[str, str]:
 250     v.status('Fetching channel')
 251     request = urllib.request.urlopen(channel.channel_url, timeout=10)
 252     channel_html = request.read().decode()
 253     forwarded_url = request.geturl()
 254     v.result(request.status == 200)  # type: ignore  # (for old mypy)
 255     v.check('Got forwarded', channel.channel_url != forwarded_url)
 256     return channel_html, forwarded_url
 257
 258
 259 def parse_channel(v: Verification, channel_html: str) \
 260         -> Tuple[Dict[str, ChannelTableEntry], GitPin]:
 261     v.status('Parsing channel description as XML')
 262     d = xml.dom.minidom.parseString(channel_html)
 263     v.ok()
 264
 265     v.status('Extracting release name:')
 266     title_name = d.getElementsByTagName(
 267         'title')[0].firstChild.nodeValue.split()[2]
 268     h1_name = d.getElementsByTagName('h1')[0].firstChild.nodeValue.split()[2]
 269     v.status(title_name)
 270     v.result(title_name == h1_name)
 271
 272     v.status('Extracting git commit:')
 273     git_commit_node = d.getElementsByTagName('tt')[0]
 274     git_revision = git_commit_node.firstChild.nodeValue
 275     v.status(git_revision)
 276     v.ok()
 277     v.status('Verifying git commit label')
 278     v.result(git_commit_node.previousSibling.nodeValue == 'Git commit ')
 279
 280     v.status('Parsing table')
 281     table: Dict[str, ChannelTableEntry] = {}
 282     for row in d.getElementsByTagName('tr')[1:]:
 283         name = row.childNodes[0].firstChild.firstChild.nodeValue
 284         url = row.childNodes[0].firstChild.getAttribute('href')
 285         size = int(row.childNodes[1].firstChild.nodeValue)
 286         digest = Digest16(row.childNodes[2].firstChild.firstChild.nodeValue)
 287         table[name] = ChannelTableEntry(url=url, digest=digest, size=size)
 288     v.ok()
 289     return table, GitPin(release_name=title_name, git_revision=git_revision)
 290
 291
 292 def digest_string(s: bytes) -> Digest16:
 293     return Digest16(hashlib.sha256(s).hexdigest())
 294
 295
 296 def digest_file(filename: str) -> Digest16:
 297     hasher = hashlib.sha256()
 298     with open(filename, 'rb') as f:
 299         # pylint: disable=cell-var-from-loop
 300         for block in iter(lambda: f.read(4096), b''):
 301             hasher.update(block)
 302     return Digest16(hasher.hexdigest())
 303
 304
 305 _experimental_flag_needed = None
 306
 307
 308 def _nix_command(v: Verification) -> List[str]:
 309     global _experimental_flag_needed
 310     if _experimental_flag_needed is None:
 311         v.status('Checking Nix version')
 312         process = subprocess.run(['nix', '--help'], stdout=subprocess.PIPE)
 313         v.result(process.returncode == 0)
 314         _experimental_flag_needed = b'--experimental-features' in process.stdout
 315     return ['nix', '--experimental-features',
 316             'nix-command'] if _experimental_flag_needed else ['nix']
 317
 318
 319 def to_Digest16(v: Verification, digest32: Digest32) -> Digest16:
 320     v.status('Converting digest to base16')
 321     process = subprocess.run(_nix_command(v) + [
 322         'to-base16',
 323         '--type',
 324         'sha256',
 325         digest32],
 326         stdout=subprocess.PIPE)
 327     v.result(process.returncode == 0)
 328     return Digest16(process.stdout.decode().strip())
 329
 330
 331 def to_Digest32(v: Verification, digest16: Digest16) -> Digest32:
 332     v.status('Converting digest to base32')
 333     process = subprocess.run(_nix_command(v) + [
 334         'to-base32',
 335         '--type',
 336         'sha256',
 337         digest16],
 338         stdout=subprocess.PIPE)
 339     v.result(process.returncode == 0)
 340     return Digest32(process.stdout.decode().strip())
 341
 342
 343 def fetch_with_nix_prefetch_url(
 344         v: Verification,
 345         url: str,
 346         digest: Digest16) -> str:
 347     v.status('Fetching %s' % url)
 348     process = subprocess.run(
 349         ['nix-prefetch-url', '--print-path', url, digest], stdout=subprocess.PIPE)
 350     v.result(process.returncode == 0)
 351     prefetch_digest, path, empty = process.stdout.decode().split('\n')
 352     assert empty == ''
 353     v.check("Verifying nix-prefetch-url's digest",
 354             to_Digest16(v, Digest32(prefetch_digest)) == digest)
 355     v.status("Verifying file digest")
 356     file_digest = digest_file(path)
 357     v.result(file_digest == digest)
 358     return path  # type: ignore  # (for old mypy)
 359
 360
 361 def fetch_resources(
 362         v: Verification,
 363         pin: GitPin,
 364         forwarded_url: str,
 365         table: Dict[str, ChannelTableEntry]) -> None:
 366     for resource in ['git-revision', 'nixexprs.tar.xz']:
 367         fields = table[resource]
 368         fields.absolute_url = urllib.parse.urljoin(forwarded_url, fields.url)
 369         fields.file = fetch_with_nix_prefetch_url(
 370             v, fields.absolute_url, fields.digest)
 371     v.status('Verifying git commit on main page matches git commit in table')
 372     v.result(open(table['git-revision'].file).read(999) == pin.git_revision)
 373
 374
 375 def tarball_cache_file(channel: TarrableSearchPath, pin: GitPin) -> str:
 376     return os.path.join(
 377         xdg.XDG_CACHE_HOME,
 378         'pinch/git-tarball',
 379         '%s-%s-%s' %
 380         (digest_string(channel.git_repo.encode()),
 381          pin.git_revision,
 382          pin.release_name))
 383
 384
 385 def verify_git_ancestry(
 386         v: Verification,
 387         channel: TarrableSearchPath,
 388         old_revision: str,
 389         new_revision: str) -> None:
 390     cachedir = git_cache.git_cachedir(channel.git_repo)
 391     v.status('Verifying rev is an ancestor of previous rev %s' % old_revision)
 392     process = subprocess.run(['git',
 393                               '-C',
 394                               cachedir,
 395                               'merge-base',
 396                               '--is-ancestor',
 397                               old_revision,
 398                               new_revision])
 399     v.result(process.returncode == 0)
 400
 401
 402 def compare_tarball_and_git(
 403         v: Verification,
 404         pin: GitPin,
 405         channel_contents: str,
 406         git_contents: str) -> None:
 407     v.status('Comparing channel tarball with git checkout')
 408     match, mismatch, errors = compare(os.path.join(
 409         channel_contents, pin.release_name), git_contents)
 410     v.ok()
 411     v.check('%d files match' % len(match), len(match) > 0)
 412     v.check('%d files differ' % len(mismatch), len(mismatch) == 0)
 413     expected_errors = [
 414         '.git-revision',
 415         '.version-suffix',
 416         'nixpkgs',
 417         'programs.sqlite',
 418         'svn-revision']
 419     benign_errors = []
 420     for ee in expected_errors:
 421         if ee in errors:
 422             errors.remove(ee)
 423             benign_errors.append(ee)
 424     v.check(
 425         '%d unexpected incomparable files' %
 426         len(errors),
 427         len(errors) == 0)
 428     v.check(
 429         '(%d of %d expected incomparable files)' %
 430         (len(benign_errors),
 431          len(expected_errors)),
 432         len(benign_errors) == len(expected_errors))
 433
 434
 435 def extract_tarball(
 436         v: Verification,
 437         table: Dict[str, ChannelTableEntry],
 438         dest: str) -> None:
 439     v.status('Extracting tarball %s' % table['nixexprs.tar.xz'].file)
 440     shutil.unpack_archive(table['nixexprs.tar.xz'].file, dest)
 441     v.ok()
 442
 443
 444 def git_checkout(
 445         v: Verification,
 446         channel: TarrableSearchPath,
 447         pin: GitPin,
 448         dest: str) -> None:
 449     v.status('Checking out corresponding git revision')
 450     git = subprocess.Popen(['git',
 451                             '-C',
 452                             git_cache.git_cachedir(channel.git_repo),
 453                             'archive',
 454                             pin.git_revision],
 455                            stdout=subprocess.PIPE)
 456     tar = subprocess.Popen(
 457         ['tar', 'x', '-C', dest, '-f', '-'], stdin=git.stdout)
 458     if git.stdout:
 459         git.stdout.close()
 460     tar.wait()
 461     git.wait()
 462     v.result(git.returncode == 0 and tar.returncode == 0)
 463
 464
 465 def git_get_tarball(
 466         v: Verification,
 467         channel: TarrableSearchPath,
 468         pin: GitPin) -> str:
 469     cache_file = tarball_cache_file(channel, pin)
 470     if os.path.exists(cache_file):
 471         cached_tarball = open(cache_file).read(9999)
 472         if os.path.exists(cached_tarball):
 473             return cached_tarball
 474
 475     with tempfile.TemporaryDirectory() as output_dir:
 476         output_filename = os.path.join(
 477             output_dir, pin.release_name + '.tar.xz')
 478         with open(output_filename, 'w') as output_file:
 479             v.status(
 480                 'Generating tarball for git revision %s' %
 481                 pin.git_revision)
 482             git = subprocess.Popen(['git',
 483                                     '-C',
 484                                     git_cache.git_cachedir(channel.git_repo),
 485                                     'archive',
 486                                     '--prefix=%s/' % pin.release_name,
 487                                     pin.git_revision],
 488                                    stdout=subprocess.PIPE)
 489             xz = subprocess.Popen(['xz'], stdin=git.stdout, stdout=output_file)
 490             xz.wait()
 491             git.wait()
 492             v.result(git.returncode == 0 and xz.returncode == 0)
 493
 494         store_tarball = copy_to_nix_store(v, output_filename)
 495
 496         os.makedirs(os.path.dirname(cache_file), exist_ok=True)
 497         open(cache_file, 'w').write(store_tarball)
 498         return store_tarball  # type: ignore  # (for old mypy)
 499
 500
 501 def check_channel_metadata(
 502         v: Verification,
 503         pin: GitPin,
 504         channel_contents: str) -> None:
 505     v.status('Verifying git commit in channel tarball')
 506     v.result(
 507         open(
 508             os.path.join(
 509                 channel_contents,
 510                 pin.release_name,
 511                 '.git-revision')).read(999) == pin.git_revision)
 512
 513     v.status(
 514         'Verifying version-suffix is a suffix of release name %s:' %
 515         pin.release_name)
 516     version_suffix = open(
 517         os.path.join(
 518             channel_contents,
 519             pin.release_name,
 520             '.version-suffix')).read(999)
 521     v.status(version_suffix)
 522     v.result(pin.release_name.endswith(version_suffix))
 523
 524
 525 def check_channel_contents(
 526         v: Verification,
 527         channel: TarrableSearchPath,
 528         table: Dict[str, ChannelTableEntry],
 529         pin: GitPin) -> None:
 530     with tempfile.TemporaryDirectory() as channel_contents, \
 531             tempfile.TemporaryDirectory() as git_contents:
 532
 533         extract_tarball(v, table, channel_contents)
 534         check_channel_metadata(v, pin, channel_contents)
 535
 536         git_checkout(v, channel, pin, git_contents)
 537
 538         compare_tarball_and_git(v, pin, channel_contents, git_contents)
 539
 540         v.status('Removing temporary directories')
 541     v.ok()
 542
 543
 544 def git_revision_name(
 545         v: Verification,
 546         channel: TarrableSearchPath,
 547         git_revision: str) -> str:
 548     v.status('Getting commit date')
 549     process = subprocess.run(['git',
 550                               '-C',
 551                               git_cache.git_cachedir(channel.git_repo),
 552                               'log',
 553                               '-n1',
 554                               '--format=%ct-%h',
 555                               '--abbrev=11',
 556                               '--no-show-signature',
 557                               git_revision],
 558                              stdout=subprocess.PIPE)
 559     v.result(process.returncode == 0 and process.stdout != b'')
 560     return '%s-%s' % (os.path.basename(channel.git_repo),
 561                       process.stdout.decode().strip())
 562
 563
 564 K = TypeVar('K')
 565 V = TypeVar('V')
 566
 567
 568 def partition_dict(pred: Callable[[K, V], bool],
 569                    d: Dict[K, V]) -> Tuple[Dict[K, V], Dict[K, V]]:
 570     selected: Dict[K, V] = {}
 571     remaining: Dict[K, V] = {}
 572     for k, v in d.items():
 573         if pred(k, v):
 574             selected[k] = v
 575         else:
 576             remaining[k] = v
 577     return selected, remaining
 578
 579
 580 def filter_dict(d: Dict[K, V], fields: Set[K]
 581                 ) -> Tuple[Dict[K, V], Dict[K, V]]:
 582     return partition_dict(lambda k, v: k in fields, d)
 583
 584
 585 def read_config_section(
 586         conf: configparser.SectionProxy) -> Tuple[SearchPath, Optional[Pin]]:
 587     mapping: Mapping[str, Tuple[Type[SearchPath], Type[Pin]]] = {
 588         'alias': (AliasSearchPath, AliasPin),
 589         'channel': (ChannelSearchPath, ChannelPin),
 590         'git': (GitSearchPath, GitPin),
 591         'symlink': (SymlinkSearchPath, SymlinkPin),
 592     }
 593     SP, P = mapping[conf['type']]
 594     _, all_fields = filter_dict(dict(conf.items()), set(['type']))
 595     pin_fields, remaining_fields = filter_dict(all_fields, set(P._fields))
 596     # Error suppression works around https://github.com/python/mypy/issues/9007
 597     pin_present = pin_fields != {} or P._fields == ()
 598     pin = P(**pin_fields) if pin_present else None  # type: ignore
 599     return SP(**remaining_fields), pin
 600
 601
 602 def read_pinned_config_section(
 603         section: str, conf: configparser.SectionProxy) -> Tuple[SearchPath, Pin]:
 604     sp, pin = read_config_section(conf)
 605     if pin is None:
 606         raise Exception(
 607             'Cannot update unpinned channel "%s" (Run "pin" before "update")' %
 608             section)
 609     return sp, pin
 610
 611
 612 def read_config(filename: str) -> configparser.ConfigParser:
 613     config = configparser.ConfigParser()
 614     config.read_file(open(filename), filename)
 615     return config
 616
 617
 618 def read_config_files(
 619         filenames: Iterable[str]) -> Dict[str, configparser.SectionProxy]:
 620     merged_config: Dict[str, configparser.SectionProxy] = {}
 621     for file in filenames:
 622         config = read_config(file)
 623         for section in config.sections():
 624             if section in merged_config:
 625                 raise Exception('Duplicate channel "%s"' % section)
 626             merged_config[section] = config[section]
 627     return merged_config
 628
 629
 630 def pinCommand(args: argparse.Namespace) -> None:
 631     v = Verification()
 632     config = read_config(args.channels_file)
 633     for section in config.sections():
 634         if args.channels and section not in args.channels:
 635             continue
 636
 637         sp, old_pin = read_config_section(config[section])
 638
 639         config[section].update(sp.pin(v, old_pin)._asdict())
 640
 641     with open(args.channels_file, 'w') as configfile:
 642         config.write(configfile)
 643
 644
 645 def updateCommand(args: argparse.Namespace) -> None:
 646     v = Verification()
 647     exprs: Dict[str, str] = {}
 648     config = {
 649         section: read_pinned_config_section(section, conf) for section,
 650         conf in read_config_files(
 651             args.channels_file).items()}
 652     alias, nonalias = partition_dict(
 653         lambda k, v: isinstance(v[0], AliasSearchPath), config)
 654
 655     for section, (sp, pin) in nonalias.items():
 656         assert not isinstance(sp, AliasSearchPath)  # mypy can't see through
 657         assert not isinstance(pin, AliasPin)        # partition_dict()
 658         tarball = sp.fetch(v, pin)
 659         exprs[section] = (
 660             'f: f { name = "%s"; channelName = "%%s"; src = builtins.storePath "%s"; }' %
 661             (pin.release_name, tarball))
 662
 663     for section, (sp, pin) in alias.items():
 664         assert isinstance(sp, AliasSearchPath)  # For mypy
 665         exprs[section] = exprs[sp.alias_of]
 666
 667     command = [
 668         'nix-env',
 669         '--profile',
 670         args.profile,
 671         '--show-trace',
 672         '--file',
 673         '<nix/unpack-channel.nix>',
 674         '--install',
 675         '--from-expression'] + [exprs[name] % name for name in sorted(exprs.keys())]
 676     if args.dry_run:
 677         print(' '.join(map(shlex.quote, command)))
 678     else:
 679         v.status('Installing channels with nix-env')
 680         process = subprocess.run(command)
 681         v.result(process.returncode == 0)
 682
 683
 684 def main() -> None:
 685     parser = argparse.ArgumentParser(prog='pinch')
 686     subparsers = parser.add_subparsers(dest='mode', required=True)
 687     parser_pin = subparsers.add_parser('pin')
 688     parser_pin.add_argument('channels_file', type=str)
 689     parser_pin.add_argument('channels', type=str, nargs='*')
 690     parser_pin.set_defaults(func=pinCommand)
 691     parser_update = subparsers.add_parser('update')
 692     parser_update.add_argument('--dry-run', action='store_true')
 693     parser_update.add_argument('--profile', default=(
 694         '/nix/var/nix/profiles/per-user/%s/channels' % getpass.getuser()))
 695     parser_update.add_argument('channels_file', type=str, nargs='+')
 696     parser_update.set_defaults(func=updateCommand)
 697     args = parser.parse_args()
 698     args.func(args)
 699
 700
 701 if __name__ == '__main__':
 702     main()