]> git.scottworley.com Git - pinch/blame - pinch.py
Fix tests after restricted mode support update
[pinch] / pinch.py
CommitLineData
0e5e611d 1import argparse
f15e458d 2import configparser
2f96f32a
SW
3import filecmp
4import functools
736c25eb 5import getpass
2f96f32a
SW
6import hashlib
7import operator
8import os
9import os.path
9a78329e 10import shlex
2f96f32a 11import shutil
73bec7e8 12import subprocess
9d7844bb 13import sys
0afcdb2a 14import tarfile
2f96f32a 15import tempfile
89e79125 16import types
2f96f32a
SW
17import urllib.parse
18import urllib.request
19import xml.dom.minidom
20
21from typing import (
9d2c406b 22 Callable,
2f96f32a
SW
23 Dict,
24 Iterable,
25 List,
7f4c3ace 26 Mapping,
0a4ff8dd 27 NamedTuple,
73bec7e8 28 NewType,
d7cfdb22 29 Optional,
567a6783 30 Set,
2f96f32a 31 Tuple,
7f4c3ace 32 Type,
567a6783 33 TypeVar,
0a4ff8dd 34 Union,
2f96f32a
SW
35)
36
d06918bc
SW
37import git_cache
38
3603dde2
SW
39# Use xdg module when it's less painful to have as a dependency
40
41
9f936f16 42class XDG(NamedTuple):
3603dde2
SW
43 XDG_CACHE_HOME: str
44
45
46xdg = XDG(
47 XDG_CACHE_HOME=os.getenv(
48 'XDG_CACHE_HOME',
49 os.path.expanduser('~/.cache')))
26125a28
SW
50
51
2f96f32a
SW
52class VerificationError(Exception):
53 pass
54
55
56class Verification:
57
58 def __init__(self) -> None:
59 self.line_length = 0
60
61 def status(self, s: str) -> None:
9d7844bb 62 print(s, end=' ', file=sys.stderr, flush=True)
2f96f32a
SW
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)
dd1026fe 72 cols = shutil.get_terminal_size().columns or 80
2f96f32a 73 pad = (cols - (self.line_length + length)) % cols
9d7844bb 74 print(' ' * pad + self._color(message, color), file=sys.stderr)
2f96f32a
SW
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
faff8642
SW
87Digest16 = NewType('Digest16', str)
88Digest32 = NewType('Digest32', str)
89
90
91class ChannelTableEntry(types.SimpleNamespace):
92 absolute_url: str
93 digest: Digest16
94 file: str
95 size: int
96 url: str
97
98
0a4ff8dd
SW
99class AliasPin(NamedTuple):
100 pass
101
102
0afcdb2a
SW
103class SymlinkPin(NamedTuple):
104 @property
105 def release_name(self) -> str:
106 return 'link'
107
108
0a4ff8dd
SW
109class GitPin(NamedTuple):
110 git_revision: str
faff8642
SW
111 release_name: str
112
0a4ff8dd
SW
113
114class ChannelPin(NamedTuple):
115 git_revision: str
116 release_name: str
117 tarball_url: str
118 tarball_sha256: str
119
120
0afcdb2a
SW
121Pin = Union[AliasPin, SymlinkPin, GitPin, ChannelPin]
122
123
124def 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)
530104d7 129 return process.stdout.decode().strip() # type: ignore # (for old mypy)
0a4ff8dd
SW
130
131
96063a51
SW
132def 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
567a6783 141class AliasSearchPath(NamedTuple):
faff8642 142 alias_of: str
2fa9cbea 143
567a6783
SW
144 # pylint: disable=no-self-use
145 def pin(self, _: Verification, __: Optional[Pin]) -> AliasPin:
0a4ff8dd 146 return AliasPin()
2fa9cbea
SW
147
148
0afcdb2a
SW
149class 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:
96063a51 157 return symlink_archive(v, self.path)
0afcdb2a
SW
158
159
567a6783 160class GitSearchPath(NamedTuple):
faff8642
SW
161 git_ref: str
162 git_repo: str
faff8642 163
567a6783 164 def pin(self, v: Verification, old_pin: Optional[Pin]) -> GitPin:
d06918bc 165 _, new_revision = git_cache.fetch(self.git_repo, self.git_ref)
567a6783
SW
166 if old_pin is not None:
167 assert isinstance(old_pin, GitPin)
d06918bc 168 verify_git_ancestry(v, self, old_pin.git_revision, new_revision)
d7cfdb22
SW
169 return GitPin(release_name=git_revision_name(v, self, new_revision),
170 git_revision=new_revision)
4a82be40 171
567a6783
SW
172 def fetch(self, v: Verification, pin: Pin) -> str:
173 assert isinstance(pin, GitPin)
d06918bc
SW
174 git_cache.ensure_rev_available(
175 self.git_repo, self.git_ref, pin.git_revision)
567a6783 176 return git_get_tarball(v, self, pin)
b3ea39b8 177
b3ea39b8 178
567a6783
SW
179class ChannelSearchPath(NamedTuple):
180 channel_url: str
181 git_ref: str
182 git_repo: str
4a82be40 183
567a6783
SW
184 def pin(self, v: Verification, old_pin: Optional[Pin]) -> ChannelPin:
185 if old_pin is not None:
186 assert isinstance(old_pin, ChannelPin)
a67cfec9 187
567a6783
SW
188 channel_html, forwarded_url = fetch_channel(v, self)
189 table, new_gitpin = parse_channel(v, channel_html)
53a27350
SW
190 if old_pin is not None and old_pin.git_revision == new_gitpin.git_revision:
191 return old_pin
567a6783 192 fetch_resources(v, new_gitpin, forwarded_url, table)
d06918bc
SW
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)
567a6783 198 check_channel_contents(v, self, table, new_gitpin)
0a4ff8dd 199 return ChannelPin(
b17278e3 200 release_name=new_gitpin.release_name,
567a6783
SW
201 tarball_url=table['nixexprs.tar.xz'].absolute_url,
202 tarball_sha256=table['nixexprs.tar.xz'].digest,
3258ff2c 203 git_revision=new_gitpin.git_revision)
4a82be40 204
b3ea39b8 205 # pylint: disable=no-self-use
567a6783
SW
206 def fetch(self, v: Verification, pin: Pin) -> str:
207 assert isinstance(pin, ChannelPin)
b3ea39b8
SW
208
209 return fetch_with_nix_prefetch_url(
567a6783
SW
210 v, pin.tarball_url, Digest16(pin.tarball_sha256))
211
212
0afcdb2a
SW
213SearchPath = Union[AliasSearchPath,
214 SymlinkSearchPath,
215 GitSearchPath,
216 ChannelSearchPath]
567a6783 217TarrableSearchPath = Union[GitSearchPath, ChannelSearchPath]
b3ea39b8 218
4a82be40 219
dc038df0 220def compare(a: str, b: str) -> Tuple[List[str], List[str], List[str]]:
2f96f32a
SW
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
567a6783
SW
248def fetch_channel(
249 v: Verification, channel: ChannelSearchPath) -> Tuple[str, str]:
b2bb38e6 250 v.status('Fetching channel from %s' % channel.channel_url)
b17def3f 251 request = urllib.request.urlopen(channel.channel_url, timeout=10)
530104d7 252 channel_html = request.read().decode()
567a6783 253 forwarded_url = request.geturl()
7c4de64c 254 v.result(request.status == 200) # type: ignore # (for old mypy)
567a6783
SW
255 v.check('Got forwarded', channel.channel_url != forwarded_url)
256 return channel_html, forwarded_url
2f96f32a
SW
257
258
567a6783
SW
259def parse_channel(v: Verification, channel_html: str) \
260 -> Tuple[Dict[str, ChannelTableEntry], GitPin]:
2f96f32a 261 v.status('Parsing channel description as XML')
567a6783 262 d = xml.dom.minidom.parseString(channel_html)
2f96f32a
SW
263 v.ok()
264
3e6421c4
SW
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)
3e6421c4
SW
271
272 v.status('Extracting git commit:')
2f96f32a 273 git_commit_node = d.getElementsByTagName('tt')[0]
3258ff2c
SW
274 git_revision = git_commit_node.firstChild.nodeValue
275 v.status(git_revision)
2f96f32a
SW
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')
567a6783 281 table: Dict[str, ChannelTableEntry] = {}
2f96f32a
SW
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)
73bec7e8 286 digest = Digest16(row.childNodes[2].firstChild.firstChild.nodeValue)
567a6783 287 table[name] = ChannelTableEntry(url=url, digest=digest, size=size)
2f96f32a 288 v.ok()
567a6783 289 return table, GitPin(release_name=title_name, git_revision=git_revision)
2f96f32a
SW
290
291
dc038df0
SW
292def digest_string(s: bytes) -> Digest16:
293 return Digest16(hashlib.sha256(s).hexdigest())
294
295
73bec7e8
SW
296def 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
b1e8c1b0
SW
305@functools.lru_cache
306def _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
1d48f551
SW
311
312
313def _nix_command(v: Verification) -> List[str]:
1d48f551 314 return ['nix', '--experimental-features',
b1e8c1b0 315 'nix-command'] if _experimental_flag_needed(v) else ['nix']
1d48f551
SW
316
317
73bec7e8
SW
318def to_Digest16(v: Verification, digest32: Digest32) -> Digest16:
319 v.status('Converting digest to base16')
1d48f551
SW
320 process = subprocess.run(_nix_command(v) + [
321 'to-base16',
322 '--type',
323 'sha256',
324 digest32],
325 stdout=subprocess.PIPE)
73bec7e8
SW
326 v.result(process.returncode == 0)
327 return Digest16(process.stdout.decode().strip())
328
329
330def to_Digest32(v: Verification, digest16: Digest16) -> Digest32:
331 v.status('Converting digest to base32')
1d48f551
SW
332 process = subprocess.run(_nix_command(v) + [
333 'to-base32',
334 '--type',
335 'sha256',
336 digest16],
337 stdout=subprocess.PIPE)
73bec7e8
SW
338 v.result(process.returncode == 0)
339 return Digest32(process.stdout.decode().strip())
340
341
342def 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(
ba596fc0 348 ['nix-prefetch-url', '--print-path', url, digest], stdout=subprocess.PIPE)
73bec7e8
SW
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)
d66db88a 354 v.status("Verifying digest of %s" % path)
73bec7e8
SW
355 file_digest = digest_file(path)
356 v.result(file_digest == digest)
7c4de64c 357 return path # type: ignore # (for old mypy)
2f96f32a 358
73bec7e8 359
9343cf48
SW
360def fetch_resources(
361 v: Verification,
567a6783
SW
362 pin: GitPin,
363 forwarded_url: str,
364 table: Dict[str, ChannelTableEntry]) -> None:
2f96f32a 365 for resource in ['git-revision', 'nixexprs.tar.xz']:
567a6783
SW
366 fields = table[resource]
367 fields.absolute_url = urllib.parse.urljoin(forwarded_url, fields.url)
e434d96d
SW
368 fields.file = fetch_with_nix_prefetch_url(
369 v, fields.absolute_url, fields.digest)
73bec7e8 370 v.status('Verifying git commit on main page matches git commit in table')
567a6783 371 v.result(open(table['git-revision'].file).read(999) == pin.git_revision)
2f96f32a 372
971d3659 373
b17278e3 374def tarball_cache_file(channel: TarrableSearchPath, pin: GitPin) -> str:
eb0c6f1b
SW
375 return os.path.join(
376 xdg.XDG_CACHE_HOME,
377 'pinch/git-tarball',
378 '%s-%s-%s' %
379 (digest_string(channel.git_repo.encode()),
b17278e3
SW
380 pin.git_revision,
381 pin.release_name))
971d3659
SW
382
383
d7cfdb22
SW
384def verify_git_ancestry(
385 v: Verification,
386 channel: TarrableSearchPath,
d06918bc
SW
387 old_revision: str,
388 new_revision: str) -> None:
389 cachedir = git_cache.git_cachedir(channel.git_repo)
390 v.status('Verifying rev is an ancestor of previous rev %s' % old_revision)
971d3659
SW
391 process = subprocess.run(['git',
392 '-C',
393 cachedir,
394 'merge-base',
395 '--is-ancestor',
d06918bc
SW
396 old_revision,
397 new_revision])
971d3659
SW
398 v.result(process.returncode == 0)
399
dc038df0 400
925c801b
SW
401def compare_tarball_and_git(
402 v: Verification,
a72fdca9 403 pin: GitPin,
925c801b
SW
404 channel_contents: str,
405 git_contents: str) -> None:
406 v.status('Comparing channel tarball with git checkout')
407 match, mismatch, errors = compare(os.path.join(
a72fdca9 408 channel_contents, pin.release_name), git_contents)
925c801b
SW
409 v.ok()
410 v.check('%d files match' % len(match), len(match) > 0)
411 v.check('%d files differ' % len(mismatch), len(mismatch) == 0)
412 expected_errors = [
413 '.git-revision',
414 '.version-suffix',
415 'nixpkgs',
416 'programs.sqlite',
417 'svn-revision']
418 benign_errors = []
419 for ee in expected_errors:
420 if ee in errors:
421 errors.remove(ee)
422 benign_errors.append(ee)
423 v.check(
424 '%d unexpected incomparable files' %
425 len(errors),
426 len(errors) == 0)
427 v.check(
428 '(%d of %d expected incomparable files)' %
429 (len(benign_errors),
430 len(expected_errors)),
431 len(benign_errors) == len(expected_errors))
432
433
7d2c278f
SW
434def extract_tarball(
435 v: Verification,
567a6783 436 table: Dict[str, ChannelTableEntry],
7d2c278f 437 dest: str) -> None:
567a6783
SW
438 v.status('Extracting tarball %s' % table['nixexprs.tar.xz'].file)
439 shutil.unpack_archive(table['nixexprs.tar.xz'].file, dest)
925c801b
SW
440 v.ok()
441
442
7d2c278f
SW
443def git_checkout(
444 v: Verification,
445 channel: TarrableSearchPath,
3258ff2c 446 pin: GitPin,
7d2c278f 447 dest: str) -> None:
925c801b
SW
448 v.status('Checking out corresponding git revision')
449 git = subprocess.Popen(['git',
450 '-C',
d06918bc 451 git_cache.git_cachedir(channel.git_repo),
925c801b 452 'archive',
3258ff2c 453 pin.git_revision],
925c801b
SW
454 stdout=subprocess.PIPE)
455 tar = subprocess.Popen(
456 ['tar', 'x', '-C', dest, '-f', '-'], stdin=git.stdout)
de68382a
SW
457 if git.stdout:
458 git.stdout.close()
925c801b
SW
459 tar.wait()
460 git.wait()
461 v.result(git.returncode == 0 and tar.returncode == 0)
462
463
9343cf48
SW
464def git_get_tarball(
465 v: Verification,
466 channel: TarrableSearchPath,
467 pin: GitPin) -> str:
b17278e3 468 cache_file = tarball_cache_file(channel, pin)
eb0c6f1b
SW
469 if os.path.exists(cache_file):
470 cached_tarball = open(cache_file).read(9999)
471 if os.path.exists(cached_tarball):
472 return cached_tarball
473
736c25eb
SW
474 with tempfile.TemporaryDirectory() as output_dir:
475 output_filename = os.path.join(
9343cf48 476 output_dir, pin.release_name + '.tar.xz')
736c25eb
SW
477 with open(output_filename, 'w') as output_file:
478 v.status(
479 'Generating tarball for git revision %s' %
9343cf48 480 pin.git_revision)
736c25eb
SW
481 git = subprocess.Popen(['git',
482 '-C',
d06918bc 483 git_cache.git_cachedir(channel.git_repo),
736c25eb 484 'archive',
9343cf48
SW
485 '--prefix=%s/' % pin.release_name,
486 pin.git_revision],
736c25eb
SW
487 stdout=subprocess.PIPE)
488 xz = subprocess.Popen(['xz'], stdin=git.stdout, stdout=output_file)
489 xz.wait()
490 git.wait()
491 v.result(git.returncode == 0 and xz.returncode == 0)
492
0afcdb2a 493 store_tarball = copy_to_nix_store(v, output_filename)
eb0c6f1b
SW
494
495 os.makedirs(os.path.dirname(cache_file), exist_ok=True)
496 open(cache_file, 'w').write(store_tarball)
7c4de64c 497 return store_tarball # type: ignore # (for old mypy)
736c25eb
SW
498
499
f9cd7bdc
SW
500def check_channel_metadata(
501 v: Verification,
a72fdca9 502 pin: GitPin,
f9cd7bdc
SW
503 channel_contents: str) -> None:
504 v.status('Verifying git commit in channel tarball')
505 v.result(
506 open(
507 os.path.join(
508 channel_contents,
a72fdca9
SW
509 pin.release_name,
510 '.git-revision')).read(999) == pin.git_revision)
f9cd7bdc
SW
511
512 v.status(
513 'Verifying version-suffix is a suffix of release name %s:' %
a72fdca9 514 pin.release_name)
f9cd7bdc
SW
515 version_suffix = open(
516 os.path.join(
517 channel_contents,
a72fdca9 518 pin.release_name,
f9cd7bdc
SW
519 '.version-suffix')).read(999)
520 v.status(version_suffix)
a72fdca9 521 v.result(pin.release_name.endswith(version_suffix))
f9cd7bdc
SW
522
523
7d2c278f
SW
524def check_channel_contents(
525 v: Verification,
a72fdca9 526 channel: TarrableSearchPath,
567a6783 527 table: Dict[str, ChannelTableEntry],
a72fdca9 528 pin: GitPin) -> None:
dc038df0
SW
529 with tempfile.TemporaryDirectory() as channel_contents, \
530 tempfile.TemporaryDirectory() as git_contents:
925c801b 531
567a6783 532 extract_tarball(v, table, channel_contents)
a72fdca9 533 check_channel_metadata(v, pin, channel_contents)
f9cd7bdc 534
3258ff2c 535 git_checkout(v, channel, pin, git_contents)
925c801b 536
a72fdca9 537 compare_tarball_and_git(v, pin, channel_contents, git_contents)
925c801b 538
dc038df0 539 v.status('Removing temporary directories')
2f96f32a
SW
540 v.ok()
541
542
d7cfdb22
SW
543def git_revision_name(
544 v: Verification,
545 channel: TarrableSearchPath,
546 git_revision: str) -> str:
e3cae769
SW
547 v.status('Getting commit date')
548 process = subprocess.run(['git',
549 '-C',
d06918bc 550 git_cache.git_cachedir(channel.git_repo),
bed32182 551 'log',
e3cae769
SW
552 '-n1',
553 '--format=%ct-%h',
554 '--abbrev=11',
88af5903 555 '--no-show-signature',
d7cfdb22 556 git_revision],
ba596fc0 557 stdout=subprocess.PIPE)
de68382a 558 v.result(process.returncode == 0 and process.stdout != b'')
e3cae769
SW
559 return '%s-%s' % (os.path.basename(channel.git_repo),
560 process.stdout.decode().strip())
561
562
567a6783
SW
563K = TypeVar('K')
564V = TypeVar('V')
565
566
9d2c406b
SW
567def partition_dict(pred: Callable[[K, V], bool],
568 d: Dict[K, V]) -> Tuple[Dict[K, V], Dict[K, V]]:
567a6783
SW
569 selected: Dict[K, V] = {}
570 remaining: Dict[K, V] = {}
571 for k, v in d.items():
9d2c406b 572 if pred(k, v):
567a6783
SW
573 selected[k] = v
574 else:
575 remaining[k] = v
576 return selected, remaining
577
578
9d2c406b
SW
579def filter_dict(d: Dict[K, V], fields: Set[K]
580 ) -> Tuple[Dict[K, V], Dict[K, V]]:
581 return partition_dict(lambda k, v: k in fields, d)
582
583
d815b199 584def read_config_section(
567a6783
SW
585 conf: configparser.SectionProxy) -> Tuple[SearchPath, Optional[Pin]]:
586 mapping: Mapping[str, Tuple[Type[SearchPath], Type[Pin]]] = {
587 'alias': (AliasSearchPath, AliasPin),
588 'channel': (ChannelSearchPath, ChannelPin),
589 'git': (GitSearchPath, GitPin),
0afcdb2a 590 'symlink': (SymlinkSearchPath, SymlinkPin),
7f4c3ace 591 }
567a6783
SW
592 SP, P = mapping[conf['type']]
593 _, all_fields = filter_dict(dict(conf.items()), set(['type']))
594 pin_fields, remaining_fields = filter_dict(all_fields, set(P._fields))
595 # Error suppression works around https://github.com/python/mypy/issues/9007
596 pin_present = pin_fields != {} or P._fields == ()
530104d7 597 pin = P(**pin_fields) if pin_present else None # type: ignore
567a6783 598 return SP(**remaining_fields), pin
f8f5b125
SW
599
600
e8bd4979
SW
601def read_pinned_config_section(
602 section: str, conf: configparser.SectionProxy) -> Tuple[SearchPath, Pin]:
603 sp, pin = read_config_section(conf)
604 if pin is None:
605 raise Exception(
606 'Cannot update unpinned channel "%s" (Run "pin" before "update")' %
607 section)
608 return sp, pin
609
610
01ba0eb2
SW
611def read_config(filename: str) -> configparser.ConfigParser:
612 config = configparser.ConfigParser()
613 config.read_file(open(filename), filename)
614 return config
615
616
4603b1a7
SW
617def read_config_files(
618 filenames: Iterable[str]) -> Dict[str, configparser.SectionProxy]:
619 merged_config: Dict[str, configparser.SectionProxy] = {}
620 for file in filenames:
621 config = read_config(file)
622 for section in config.sections():
623 if section in merged_config:
624 raise Exception('Duplicate channel "%s"' % section)
625 merged_config[section] = config[section]
626 return merged_config
627
628
41b87c9c 629def pinCommand(args: argparse.Namespace) -> None:
2f96f32a 630 v = Verification()
01ba0eb2 631 config = read_config(args.channels_file)
5cfa8e11 632 for section in config.sections():
98853153
SW
633 if args.channels and section not in args.channels:
634 continue
736c25eb 635
d815b199 636 sp, old_pin = read_config_section(config[section])
17906b27 637
567a6783 638 config[section].update(sp.pin(v, old_pin)._asdict())
8fca6c28 639
0e5e611d 640 with open(args.channels_file, 'w') as configfile:
e434d96d 641 config.write(configfile)
2f96f32a
SW
642
643
41b87c9c 644def updateCommand(args: argparse.Namespace) -> None:
736c25eb 645 v = Verification()
da135b07 646 exprs: Dict[str, str] = {}
3b2117a3
SW
647 profile_manifest = os.path.join(args.profile, "manifest.nix")
648 search_paths: List[str] = [
649 "-I", "pinch_profile=" + args.profile,
650 "-I", "pinch_profile_manifest=" + os.readlink(profile_manifest)
651 ] if os.path.exists(profile_manifest) else []
9d2c406b
SW
652 config = {
653 section: read_pinned_config_section(section, conf) for section,
654 conf in read_config_files(
655 args.channels_file).items()}
656 alias, nonalias = partition_dict(
657 lambda k, v: isinstance(v[0], AliasSearchPath), config)
658
436195f0 659 for section, (sp, pin) in sorted(nonalias.items()):
9d2c406b
SW
660 assert not isinstance(sp, AliasSearchPath) # mypy can't see through
661 assert not isinstance(pin, AliasPin) # partition_dict()
567a6783 662 tarball = sp.fetch(v, pin)
436195f0
SW
663 search_paths.extend(["-I", "pinch_tarball_for_%s=%s" %
664 (pin.release_name, tarball)])
4603b1a7
SW
665 exprs[section] = (
666 'f: f { name = "%s"; channelName = "%%s"; src = builtins.storePath "%s"; }' %
9d2c406b 667 (pin.release_name, tarball))
4603b1a7 668
9d2c406b
SW
669 for section, (sp, pin) in alias.items():
670 assert isinstance(sp, AliasSearchPath) # For mypy
671 exprs[section] = exprs[sp.alias_of]
17906b27 672
9a78329e
SW
673 command = [
674 'nix-env',
675 '--profile',
9e8ed0ed 676 args.profile,
9a78329e
SW
677 '--show-trace',
678 '--file',
679 '<nix/unpack-channel.nix>',
680 '--install',
fc168c34 681 '--remove-all',
436195f0
SW
682 ] + search_paths + ['--from-expression'] + [
683 exprs[name] % name for name in sorted(exprs.keys())]
9a78329e
SW
684 if args.dry_run:
685 print(' '.join(map(shlex.quote, command)))
686 else:
687 v.status('Installing channels with nix-env')
688 process = subprocess.run(command)
689 v.result(process.returncode == 0)
736c25eb
SW
690
691
0e5e611d
SW
692def main() -> None:
693 parser = argparse.ArgumentParser(prog='pinch')
694 subparsers = parser.add_subparsers(dest='mode', required=True)
695 parser_pin = subparsers.add_parser('pin')
696 parser_pin.add_argument('channels_file', type=str)
98853153 697 parser_pin.add_argument('channels', type=str, nargs='*')
41b87c9c 698 parser_pin.set_defaults(func=pinCommand)
736c25eb 699 parser_update = subparsers.add_parser('update')
9a78329e 700 parser_update.add_argument('--dry-run', action='store_true')
9e8ed0ed
SW
701 parser_update.add_argument('--profile', default=(
702 '/nix/var/nix/profiles/per-user/%s/channels' % getpass.getuser()))
01ba0eb2 703 parser_update.add_argument('channels_file', type=str, nargs='+')
41b87c9c 704 parser_update.set_defaults(func=updateCommand)
0e5e611d
SW
705 args = parser.parse_args()
706 args.func(args)
707
708
b5964ec3
SW
709if __name__ == '__main__':
710 main()