]> git.scottworley.com Git - pinch/blobdiff - pinch.py
Use a proper User-Agent when fetching channel info
[pinch] / pinch.py
index bdd7a6ebbf39e8af5581980c161c3bf73a0d2d83..bbafe6bd811a8860c2c59392a33ec4fe2fe31f6b 100644 (file)
--- a/pinch.py
+++ b/pinch.py
@@ -43,6 +43,8 @@ from typing import (
 
 import git_cache
 
+from version import pinch_version
+
 # Use xdg module when it's less painful to have as a dependency
 
 
@@ -252,7 +254,11 @@ def compare(a: str, b: str) -> Tuple[List[str], List[str], List[str]]:
 def fetch_channel(
         v: Verification, channel: ChannelSearchPath) -> Tuple[str, str]:
     v.status(f'Fetching channel from {channel.channel_url}')
-    with urllib.request.urlopen(channel.channel_url, timeout=10) as request:
+    with urllib.request.urlopen(
+            urllib.request.Request(
+                url=channel.channel_url,
+                headers={'User-Agent': f'pinch-{pinch_version}'}),
+            timeout=10) as request:
         channel_html = request.read().decode()
         forwarded_url = request.geturl()
         v.result(request.status == 200)
@@ -293,14 +299,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 +337,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,