Support old mypy version 0.701

[pinch] / pinch.py

diff --git a/pinch.py b/pinch.py
index 9b845e3414df7a5c66a0cd940890a8562ebf7da8..b5e3b81567894101a9ee0d9cc9de34f253c15dd1 100644 (file)
--- a/pinch.py
+++ b/pinch.py
@@ -11,6 +11,7 @@ import shlex
 import shutil
 import subprocess
 import sys
+import tarfile
 import tempfile
 import types
 import urllib.parse
@@ -97,6 +98,12 @@ class AliasPin(NamedTuple):
     pass
 
 
+class SymlinkPin(NamedTuple):
+    @property
+    def release_name(self) -> str:
+        return 'link'
+
+
 class GitPin(NamedTuple):
     git_revision: str
     release_name: str
@@ -109,7 +116,15 @@ class ChannelPin(NamedTuple):
     tarball_sha256: str
 
 
-Pin = Union[AliasPin, GitPin, ChannelPin]
+Pin = Union[AliasPin, SymlinkPin, GitPin, ChannelPin]
+
+
+def copy_to_nix_store(v: Verification, filename: str) -> str:
+    v.status('Putting tarball in Nix store')
+    process = subprocess.run(
+        ['nix-store', '--add', filename], stdout=subprocess.PIPE)
+    v.result(process.returncode == 0)
+    return process.stdout.decode().strip()  # type: ignore  # (for old mypy)
 
 
 class AliasSearchPath(NamedTuple):
@@ -120,6 +135,22 @@ class AliasSearchPath(NamedTuple):
         return AliasPin()
 
 
+class SymlinkSearchPath(NamedTuple):
+    path: str
+
+    # pylint: disable=no-self-use
+    def pin(self, _: Verification, __: Optional[Pin]) -> SymlinkPin:
+        return SymlinkPin()
+
+    def fetch(self, v: Verification, _: Pin) -> str:
+        with tempfile.TemporaryDirectory() as td:
+            archive_filename = os.path.join(td, 'link.tar.gz')
+            os.symlink(self.path, os.path.join(td, 'link'))
+            with tarfile.open(archive_filename, mode='x:gz') as t:
+                t.add(os.path.join(td, 'link'), arcname='link')
+            return copy_to_nix_store(v, archive_filename)
+
+
 class GitSearchPath(NamedTuple):
     git_ref: str
     git_repo: str
@@ -168,7 +199,10 @@ class ChannelSearchPath(NamedTuple):
             v, pin.tarball_url, Digest16(pin.tarball_sha256))
 
 
-SearchPath = Union[AliasSearchPath, GitSearchPath, ChannelSearchPath]
+SearchPath = Union[AliasSearchPath,
+                   SymlinkSearchPath,
+                   GitSearchPath,
+                   ChannelSearchPath]
 TarrableSearchPath = Union[GitSearchPath, ChannelSearchPath]
 
 
@@ -204,7 +238,7 @@ def fetch_channel(
         v: Verification, channel: ChannelSearchPath) -> Tuple[str, str]:
     v.status('Fetching channel')
     request = urllib.request.urlopen(channel.channel_url, timeout=10)
-    channel_html = request.read()
+    channel_html = request.read().decode()
     forwarded_url = request.geturl()
     v.result(request.status == 200)  # type: ignore  # (for old mypy)
     v.check('Got forwarded', channel.channel_url != forwarded_url)
@@ -513,11 +547,7 @@ def git_get_tarball(
             git.wait()
             v.result(git.returncode == 0 and xz.returncode == 0)
 
-        v.status('Putting tarball in Nix store')
-        process = subprocess.run(
-            ['nix-store', '--add', output_filename], stdout=subprocess.PIPE)
-        v.result(process.returncode == 0)
-        store_tarball = process.stdout.decode().strip()
+        store_tarball = copy_to_nix_store(v, output_filename)
 
         os.makedirs(os.path.dirname(cache_file), exist_ok=True)
         open(cache_file, 'w').write(store_tarball)
@@ -614,13 +644,14 @@ def read_config_section(
         'alias': (AliasSearchPath, AliasPin),
         'channel': (ChannelSearchPath, ChannelPin),
         'git': (GitSearchPath, GitPin),
+        'symlink': (SymlinkSearchPath, SymlinkPin),
     }
     SP, P = mapping[conf['type']]
     _, all_fields = filter_dict(dict(conf.items()), set(['type']))
     pin_fields, remaining_fields = filter_dict(all_fields, set(P._fields))
     # Error suppression works around https://github.com/python/mypy/issues/9007
     pin_present = pin_fields != {} or P._fields == ()
-    pin = P(**pin_fields) if pin_present else None  # type:ignore[call-arg]
+    pin = P(**pin_fields) if pin_present else None  # type: ignore
     return SP(**remaining_fields), pin