X-Git-Url: http://git.scottworley.com/pinch/blobdiff_plain/1b7ac5ff572bbdd96a5edc1768ab9192e4458db1..8ec8e7607393df1f7be14ab20e5dc1b3a8e1832d:/pinch.py?ds=inline diff --git a/pinch.py b/pinch.py index 1a3a94b..c101dc3 100644 --- a/pinch.py +++ b/pinch.py @@ -293,14 +293,25 @@ def parse_channel(v: Verification, channel_html: str) \ v.status(git_revision) v.ok() v.status('Verifying git commit label') + assert git_commit_node.previousSibling is not None v.result(git_commit_node.previousSibling.nodeValue == 'Git commit ') v.status('Parsing table') table: Dict[str, ChannelTableEntry] = {} for row in d.getElementsByTagName('tr')[1:]: + assert isinstance( + row.childNodes[0].firstChild, xml.dom.minidom.Element) + assert isinstance( + row.childNodes[0].firstChild.firstChild, xml.dom.minidom.Text) name = row.childNodes[0].firstChild.firstChild.nodeValue + assert name is not None url = row.childNodes[0].firstChild.getAttribute('href') + assert row.childNodes[1].firstChild is not None + assert row.childNodes[1].firstChild.nodeValue is not None size = int(row.childNodes[1].firstChild.nodeValue) + assert row.childNodes[2].firstChild is not None + assert row.childNodes[2].firstChild.firstChild is not None + assert row.childNodes[2].firstChild.firstChild.nodeValue is not None digest = Digest16(row.childNodes[2].firstChild.firstChild.nodeValue) table[name] = ChannelTableEntry(url=url, digest=digest, size=size) v.ok() @@ -320,43 +331,24 @@ def digest_file(filename: str) -> Digest16: return Digest16(hasher.hexdigest()) -@functools.lru_cache -def _experimental_flag_needed(v: Verification) -> bool: - v.status('Checking Nix version') - process = subprocess.run(['nix', '--help'], stdout=subprocess.PIPE) - v.result(process.returncode == 0) - return b'--experimental-features' in process.stdout - - -def _nix_command(v: Verification) -> List[str]: - return ['nix', '--experimental-features', - 'nix-command'] if _experimental_flag_needed(v) else ['nix'] +_NIX_COMMAND = ['nix', '--experimental-features', 'nix-command'] def to_Digest16(v: Verification, digest32: Digest32) -> Digest16: v.status('Converting digest to base16') - process = subprocess.run(_nix_command(v) + [ - 'to-base16', - '--type', + process = subprocess.run(_NIX_COMMAND + [ + 'hash', + 'convert', + '--hash-algo', 'sha256', + '--to', + 'base16', digest32], stdout=subprocess.PIPE) v.result(process.returncode == 0) return Digest16(process.stdout.decode().strip()) -def to_Digest32(v: Verification, digest16: Digest16) -> Digest32: - v.status('Converting digest to base32') - process = subprocess.run(_nix_command(v) + [ - 'to-base32', - '--type', - 'sha256', - digest16], - stdout=subprocess.PIPE) - v.result(process.returncode == 0) - return Digest32(process.stdout.decode().strip()) - - def fetch_with_nix_prefetch_url( v: Verification, url: str, @@ -685,23 +677,35 @@ def updateCommand(args: argparse.Namespace) -> None: assert isinstance(sp, AliasSearchPath) # For mypy exprs[section] = exprs[sp.alias_of] - command = [ - 'nix-env', - '--profile', - args.profile, - '--show-trace', - '--file', - '', - '--install', - '--remove-all', - ] + search_paths + ['--from-expression'] + [ - exprs[name] % name for name in sorted(exprs.keys())] - if args.dry_run: - print(' '.join(map(shlex.quote, command))) - else: - v.status('Installing channels with nix-env') - process = subprocess.run(command) - v.result(process.returncode == 0) + with tempfile.NamedTemporaryFile() as unpack_channel_nix: + unpack_channel_nix.write(b''' + { name, channelName, src, }: + derivation { + inherit name channelName src; + builder = "builtin:unpack-channel"; + system = "builtin"; + preferLocalBuild = true; + } + ''') + unpack_channel_nix.flush() + + command = [ + 'nix-env', + '--profile', + args.profile, + '--show-trace', + '--file', + unpack_channel_nix.name, + '--install', + '--remove-all', + ] + search_paths + ['--from-expression'] + [ + exprs[name] % name for name in sorted(exprs.keys())] + if args.dry_run: + print(' '.join(map(shlex.quote, command))) + else: + v.status('Installing channels with nix-env') + process = subprocess.run(command) + v.result(process.returncode == 0) def main() -> None: