X-Git-Url: http://git.scottworley.com/pinch/blobdiff_plain/faff8642d1b68a819a41186ec720a0f00f43f712..b896966bd140bc9c3c965c70597517ede83d5b86:/pinch.py diff --git a/pinch.py b/pinch.py index a8af6a6..bfc6ac6 100644 --- a/pinch.py +++ b/pinch.py @@ -1,3 +1,4 @@ +from abc import ABC, abstractmethod import argparse import configparser import filecmp @@ -85,12 +86,22 @@ class ChannelTableEntry(types.SimpleNamespace): url: str -class SearchPath(types.SimpleNamespace): +class SearchPath(types.SimpleNamespace, ABC): release_name: str + @abstractmethod + def pin(self, v: Verification, conf: configparser.SectionProxy) -> None: + pass -class Channel(SearchPath): + +class AliasSearchPath(SearchPath): alias_of: str + + def pin(self, v: Verification, conf: configparser.SectionProxy) -> None: + assert not hasattr(self, 'git_repo') + + +class Channel(SearchPath): channel_html: bytes channel_url: str forwarded_url: str @@ -100,6 +111,36 @@ class Channel(SearchPath): old_git_revision: str table: Dict[str, ChannelTableEntry] + def pin(self, v: Verification, conf: configparser.SectionProxy) -> None: + if hasattr(self, 'git_revision'): + self.old_git_revision = self.git_revision + del self.git_revision + + if 'channel_url' in conf: + pin_channel(v, self) + conf['release_name'] = self.release_name + conf['tarball_url'] = self.table['nixexprs.tar.xz'].absolute_url + conf['tarball_sha256'] = self.table['nixexprs.tar.xz'].digest + else: + git_fetch(v, self) + conf['release_name'] = git_revision_name(v, self) + conf['git_revision'] = self.git_revision + + def fetch(self, v: Verification, section: str, + conf: configparser.SectionProxy) -> str: + if 'git_repo' not in conf or 'release_name' not in conf: + raise Exception( + 'Cannot update unpinned channel "%s" (Run "pin" before "update")' % + section) + + if 'channel_url' in conf: + return fetch_with_nix_prefetch_url( + v, conf['tarball_url'], Digest16( + conf['tarball_sha256'])) + + ensure_git_rev_available(v, self) + return git_get_tarball(v, self) + def compare(a: str, b: str) -> Tuple[List[str], List[str], List[str]]: @@ -493,7 +534,9 @@ def git_revision_name(v: Verification, channel: Channel) -> str: process.stdout.decode().strip()) -def read_search_path(conf: configparser.SectionProxy) -> Channel: +def read_search_path(conf: configparser.SectionProxy) -> SearchPath: + if 'alias_of' in conf: + return AliasSearchPath(**dict(conf.items())) return Channel(**dict(conf.items())) @@ -503,6 +546,18 @@ def read_config(filename: str) -> configparser.ConfigParser: return config +def read_config_files( + filenames: Iterable[str]) -> Dict[str, configparser.SectionProxy]: + merged_config: Dict[str, configparser.SectionProxy] = {} + for file in filenames: + config = read_config(file) + for section in config.sections(): + if section in merged_config: + raise Exception('Duplicate channel "%s"' % section) + merged_config[section] = config[section] + return merged_config + + def pin(args: argparse.Namespace) -> None: v = Verification() config = read_config(args.channels_file) @@ -510,72 +565,31 @@ def pin(args: argparse.Namespace) -> None: if args.channels and section not in args.channels: continue - channel = read_search_path(config[section]) - - if hasattr(channel, 'alias_of'): - assert not hasattr(channel, 'git_repo') - continue - - if hasattr(channel, 'git_revision'): - channel.old_git_revision = channel.git_revision - del channel.git_revision + sp = read_search_path(config[section]) - if 'channel_url' in config[section]: - pin_channel(v, channel) - config[section]['release_name'] = channel.release_name - config[section]['tarball_url'] = channel.table['nixexprs.tar.xz'].absolute_url - config[section]['tarball_sha256'] = channel.table['nixexprs.tar.xz'].digest - else: - git_fetch(v, channel) - config[section]['release_name'] = git_revision_name(v, channel) - config[section]['git_revision'] = channel.git_revision + sp.pin(v, config[section]) with open(args.channels_file, 'w') as configfile: config.write(configfile) -def fetch_channel( - v: Verification, - section: str, - conf: configparser.SectionProxy) -> str: - if 'git_repo' not in conf or 'release_name' not in conf: - raise Exception( - 'Cannot update unpinned channel "%s" (Run "pin" before "update")' % - section) - - if 'channel_url' in conf: - return fetch_with_nix_prefetch_url( - v, conf['tarball_url'], Digest16( - conf['tarball_sha256'])) - - channel = read_search_path(conf) - ensure_git_rev_available(v, channel) - return git_get_tarball(v, channel) - - def update(args: argparse.Namespace) -> None: v = Verification() - config = configparser.ConfigParser() exprs: Dict[str, str] = {} - configs = [read_config(filename) for filename in args.channels_file] - for config in configs: - for section in config.sections(): - if 'alias_of' in config[section]: - assert 'git_repo' not in config[section] - continue - tarball = fetch_channel(v, section, config[section]) - if section in exprs: - raise Exception('Duplicate channel "%s"' % section) - exprs[section] = ( - 'f: f { name = "%s"; channelName = "%%s"; src = builtins.storePath "%s"; }' % - (config[section]['release_name'], tarball)) - - for config in configs: - for section in config.sections(): - if 'alias_of' in config[section]: - if section in exprs: - raise Exception('Duplicate channel "%s"' % section) - exprs[section] = exprs[str(config[section]['alias_of'])] + config = read_config_files(args.channels_file) + for section in config: + sp = read_search_path(config[section]) + if isinstance(sp, AliasSearchPath): + assert 'git_repo' not in config[section] + continue + tarball = sp.fetch(v, section, config[section]) + exprs[section] = ( + 'f: f { name = "%s"; channelName = "%%s"; src = builtins.storePath "%s"; }' % + (config[section]['release_name'], tarball)) + + for section in config: + if 'alias_of' in config[section]: + exprs[section] = exprs[str(config[section]['alias_of'])] command = [ 'nix-env',