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 from %s' % channel.channel_url)
 251     with urllib.request.urlopen(channel.channel_url, timeout=10) as request:
 252         channel_html = request.read().decode()
 253         forwarded_url = request.geturl()
 254         v.result(request.status == 200)
 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 @functools.lru_cache
 306 def _experimental_flag_needed(v: Verification) -> bool:
 307     v.status('Checking Nix version')
 308     process = subprocess.run(['nix', '--help'], stdout=subprocess.PIPE)
 309     v.result(process.returncode == 0)
 310     return b'--experimental-features' in process.stdout
 311
 312
 313 def _nix_command(v: Verification) -> List[str]:
 314     return ['nix', '--experimental-features',
 315             'nix-command'] if _experimental_flag_needed(v) else ['nix']
 316
 317
 318 def to_Digest16(v: Verification, digest32: Digest32) -> Digest16:
 319     v.status('Converting digest to base16')
 320     process = subprocess.run(_nix_command(v) + [
 321         'to-base16',
 322         '--type',
 323         'sha256',
 324         digest32],
 325         stdout=subprocess.PIPE)
 326     v.result(process.returncode == 0)
 327     return Digest16(process.stdout.decode().strip())
 328
 329
 330 def to_Digest32(v: Verification, digest16: Digest16) -> Digest32:
 331     v.status('Converting digest to base32')
 332     process = subprocess.run(_nix_command(v) + [
 333         'to-base32',
 334         '--type',
 335         'sha256',
 336         digest16],
 337         stdout=subprocess.PIPE)
 338     v.result(process.returncode == 0)
 339     return Digest32(process.stdout.decode().strip())
 340
 341
 342 def fetch_with_nix_prefetch_url(
 343         v: Verification,
 344         url: str,
 345         digest: Digest16) -> str:
 346     v.status('Fetching %s' % url)
 347     process = subprocess.run(
 348         ['nix-prefetch-url', '--print-path', url, digest], stdout=subprocess.PIPE)
 349     v.result(process.returncode == 0)
 350     prefetch_digest, path, empty = process.stdout.decode().split('\n')
 351     assert empty == ''
 352     v.check("Verifying nix-prefetch-url's digest",
 353             to_Digest16(v, Digest32(prefetch_digest)) == digest)
 354     v.status("Verifying digest of %s" % path)
 355     file_digest = digest_file(path)
 356     v.result(file_digest == digest)
 357     return path  # type: ignore  # (for old mypy)
 358
 359
 360 def fetch_resources(
 361         v: Verification,
 362         pin: GitPin,
 363         forwarded_url: str,
 364         table: Dict[str, ChannelTableEntry]) -> None:
 365     for resource in ['git-revision', 'nixexprs.tar.xz']:
 366         fields = table[resource]
 367         fields.absolute_url = urllib.parse.urljoin(forwarded_url, fields.url)
 368         fields.file = fetch_with_nix_prefetch_url(
 369             v, fields.absolute_url, fields.digest)
 370     v.status('Verifying git commit on main page matches git commit in table')
 371     with open(table['git-revision'].file, encoding='utf-8') as rev_file:
 372         v.result(rev_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     with subprocess.Popen(
 451             ['git', '-C', git_cache.git_cachedir(channel.git_repo), 'archive', pin.git_revision],
 452             stdout=subprocess.PIPE) as git:
 453         with subprocess.Popen(['tar', 'x', '-C', dest, '-f', '-'], stdin=git.stdout) as tar:
 454             if git.stdout:
 455                 git.stdout.close()
 456             tar.wait()
 457             git.wait()
 458             v.result(git.returncode == 0 and tar.returncode == 0)
 459
 460
 461 def git_get_tarball(
 462         v: Verification,
 463         channel: TarrableSearchPath,
 464         pin: GitPin) -> str:
 465     cache_file = tarball_cache_file(channel, pin)
 466     if os.path.exists(cache_file):
 467         with open(cache_file, encoding='utf-8') as f:
 468             cached_tarball = f.read(9999)
 469             if os.path.exists(cached_tarball):
 470                 return cached_tarball
 471
 472     with tempfile.TemporaryDirectory() as output_dir:
 473         output_filename = os.path.join(
 474             output_dir, pin.release_name + '.tar.xz')
 475         with open(output_filename, 'w', encoding='utf-8') as output_file:
 476             v.status(
 477                 'Generating tarball for git revision %s' %
 478                 pin.git_revision)
 479             with subprocess.Popen(
 480                     ['git', '-C', git_cache.git_cachedir(channel.git_repo),
 481                         'archive', '--prefix=%s/' % pin.release_name, pin.git_revision],
 482                     stdout=subprocess.PIPE) as git:
 483                 with subprocess.Popen(['xz'], stdin=git.stdout, stdout=output_file) as xz:
 484                     xz.wait()
 485                     git.wait()
 486                     v.result(git.returncode == 0 and xz.returncode == 0)
 487
 488         store_tarball = copy_to_nix_store(v, output_filename)
 489
 490         os.makedirs(os.path.dirname(cache_file), exist_ok=True)
 491         with open(cache_file, 'w', encoding='utf-8') as f:
 492             f.write(store_tarball)
 493         return store_tarball  # type: ignore  # (for old mypy)
 494
 495
 496 def check_channel_metadata(
 497         v: Verification,
 498         pin: GitPin,
 499         channel_contents: str) -> None:
 500     v.status('Verifying git commit in channel tarball')
 501     with open(os.path.join(channel_contents, pin.release_name, '.git-revision'),
 502               encoding='utf-8') as f:
 503         v.result(f.read(999) == pin.git_revision)
 504
 505     v.status(
 506         'Verifying version-suffix is a suffix of release name %s:' %
 507         pin.release_name)
 508     with open(os.path.join(channel_contents, pin.release_name, '.version-suffix'),
 509               encoding='utf-8') as f:
 510         version_suffix = f.read(999)
 511     v.status(version_suffix)
 512     v.result(pin.release_name.endswith(version_suffix))
 513
 514
 515 def check_channel_contents(
 516         v: Verification,
 517         channel: TarrableSearchPath,
 518         table: Dict[str, ChannelTableEntry],
 519         pin: GitPin) -> None:
 520     with tempfile.TemporaryDirectory() as channel_contents, \
 521             tempfile.TemporaryDirectory() as git_contents:
 522
 523         extract_tarball(v, table, channel_contents)
 524         check_channel_metadata(v, pin, channel_contents)
 525
 526         git_checkout(v, channel, pin, git_contents)
 527
 528         compare_tarball_and_git(v, pin, channel_contents, git_contents)
 529
 530         v.status('Removing temporary directories')
 531     v.ok()
 532
 533
 534 def git_revision_name(
 535         v: Verification,
 536         channel: TarrableSearchPath,
 537         git_revision: str) -> str:
 538     v.status('Getting commit date')
 539     process = subprocess.run(['git',
 540                               '-C',
 541                               git_cache.git_cachedir(channel.git_repo),
 542                               'log',
 543                               '-n1',
 544                               '--format=%ct-%h',
 545                               '--abbrev=11',
 546                               '--no-show-signature',
 547                               git_revision],
 548                              stdout=subprocess.PIPE)
 549     v.result(process.returncode == 0 and process.stdout != b'')
 550     return '%s-%s' % (os.path.basename(channel.git_repo),
 551                       process.stdout.decode().strip())
 552
 553
 554 K = TypeVar('K')
 555 V = TypeVar('V')
 556
 557
 558 def partition_dict(pred: Callable[[K, V], bool],
 559                    d: Dict[K, V]) -> Tuple[Dict[K, V], Dict[K, V]]:
 560     selected: Dict[K, V] = {}
 561     remaining: Dict[K, V] = {}
 562     for k, v in d.items():
 563         if pred(k, v):
 564             selected[k] = v
 565         else:
 566             remaining[k] = v
 567     return selected, remaining
 568
 569
 570 def filter_dict(d: Dict[K, V], fields: Set[K]
 571                 ) -> Tuple[Dict[K, V], Dict[K, V]]:
 572     return partition_dict(lambda k, v: k in fields, d)
 573
 574
 575 def read_config_section(
 576         conf: configparser.SectionProxy) -> Tuple[SearchPath, Optional[Pin]]:
 577     mapping: Mapping[str, Tuple[Type[SearchPath], Type[Pin]]] = {
 578         'alias': (AliasSearchPath, AliasPin),
 579         'channel': (ChannelSearchPath, ChannelPin),
 580         'git': (GitSearchPath, GitPin),
 581         'symlink': (SymlinkSearchPath, SymlinkPin),
 582     }
 583     SP, P = mapping[conf['type']]
 584     _, all_fields = filter_dict(dict(conf.items()), set(['type']))
 585     pin_fields, remaining_fields = filter_dict(all_fields, set(P._fields))
 586     # Error suppression works around https://github.com/python/mypy/issues/9007
 587     pin_present = pin_fields != {} or P._fields == ()
 588     pin = P(**pin_fields) if pin_present else None  # type: ignore
 589     return SP(**remaining_fields), pin
 590
 591
 592 def read_pinned_config_section(
 593         section: str, conf: configparser.SectionProxy) -> Tuple[SearchPath, Pin]:
 594     sp, pin = read_config_section(conf)
 595     if pin is None:
 596         raise Exception(
 597             'Cannot update unpinned channel "%s" (Run "pin" before "update")' %
 598             section)
 599     return sp, pin
 600
 601
 602 def read_config(filename: str) -> configparser.ConfigParser:
 603     config = configparser.ConfigParser()
 604     with open(filename, encoding='utf-8') as f:
 605         config.read_file(f, filename)
 606     return config
 607
 608
 609 def read_config_files(
 610         filenames: Iterable[str]) -> Dict[str, configparser.SectionProxy]:
 611     merged_config: Dict[str, configparser.SectionProxy] = {}
 612     for file in filenames:
 613         config = read_config(file)
 614         for section in config.sections():
 615             if section in merged_config:
 616                 raise Exception('Duplicate channel "%s"' % section)
 617             merged_config[section] = config[section]
 618     return merged_config
 619
 620
 621 def pinCommand(args: argparse.Namespace) -> None:
 622     v = Verification()
 623     config = read_config(args.channels_file)
 624     for section in config.sections():
 625         if args.channels and section not in args.channels:
 626             continue
 627
 628         sp, old_pin = read_config_section(config[section])
 629
 630         config[section].update(sp.pin(v, old_pin)._asdict())
 631
 632     with open(args.channels_file, 'w', encoding='utf-8') as configfile:
 633         config.write(configfile)
 634
 635
 636 def updateCommand(args: argparse.Namespace) -> None:
 637     v = Verification()
 638     exprs: Dict[str, str] = {}
 639     profile_manifest = os.path.join(args.profile, "manifest.nix")
 640     search_paths: List[str] = [
 641         "-I", "pinch_profile=" + args.profile,
 642         "-I", "pinch_profile_manifest=" + os.readlink(profile_manifest)
 643     ] if os.path.exists(profile_manifest) else []
 644     config = {
 645         section: read_pinned_config_section(section, conf) for section,
 646         conf in read_config_files(
 647             args.channels_file).items()}
 648     alias, nonalias = partition_dict(
 649         lambda k, v: isinstance(v[0], AliasSearchPath), config)
 650
 651     for section, (sp, pin) in sorted(nonalias.items()):
 652         assert not isinstance(sp, AliasSearchPath)  # mypy can't see through
 653         assert not isinstance(pin, AliasPin)        # partition_dict()
 654         tarball = sp.fetch(v, pin)
 655         search_paths.extend(["-I", "pinch_tarball_for_%s=%s" %
 656                             (pin.release_name, tarball)])
 657         exprs[section] = (
 658             'f: f { name = "%s"; channelName = "%%s"; src = builtins.storePath "%s"; }' %
 659             (pin.release_name, tarball))
 660
 661     for section, (sp, pin) in alias.items():
 662         assert isinstance(sp, AliasSearchPath)  # For mypy
 663         exprs[section] = exprs[sp.alias_of]
 664
 665     command = [
 666         'nix-env',
 667         '--profile',
 668         args.profile,
 669         '--show-trace',
 670         '--file',
 671         '<nix/unpack-channel.nix>',
 672         '--install',
 673         '--remove-all',
 674     ] + search_paths + ['--from-expression'] + [
 675         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()