Commit 4f56b10d authored by Matthias Klumpp's avatar Matthias Klumpp
Browse files

Merge branch 'master' of salsa.debian.org/apt-team/python-apt

parents f547b3ef c97d4159
Pipeline #57046 failed with stages
in 3 minutes and 47 seconds
image: debian:testing
image: ubuntu:focal
variables:
DEBIAN_FRONTEND: noninteractive
test:
typing:
script:
- apt update
- apt install -q -y python3-pip
- apt build-dep -q -y ./
- python3 -m pip install -U mypy
- env python3 tests/testmanual_pep8.py
- env MYPYPATH=$PWD/typehinting/ mypy ./apt
- env MYPYPATH=$PWD/typehinting/ mypy --strict ./apt
pep8:
script:
- apt update
- apt install -q -y pycodestyle
- env python3 tests/testmanual_pycodestyle.py
test:
script:
- apt update
- apt install -q -y software-properties-common
- add-apt-repository ppa:deity/sid
- apt update
- apt build-dep -q -y ./
- dpkg-buildpackage
- make -C doc html
artifacts:
paths:
- doc/build/html
- build/sphinx/html/
pages:
stage: deploy
script:
- mv doc/build/html/ public
- mv build/sphinx/html/ public
artifacts:
paths:
- public
......
......@@ -4,13 +4,11 @@ sudo: required
services:
- docker
env:
- DISTRO=debian:buster
- DISTRO=ubuntu:cosmic
- DISTRO=ubuntu:disco
- DISTRO=ubuntu:eoan
install:
- sed -i -e "s#1000#$(id -u)#g" -e "s#debian:testing#$DISTRO#g" Dockerfile
- docker build --tag=apt-ci .
script:
- docker run --rm -w $PWD -v $HOME/.ccache:$HOME/.ccache -v $PWD:$PWD --user=travis apt-ci env ./debian/rules build-arch
- docker run --rm -w $PWD -v $PWD:$PWD --user=travis apt-ci env python3 tests/testmanual_pep8.py
- docker run --rm -w $PWD -v $PWD:$PWD --user=travis apt-ci env python3 tests/testmanual_pycodestyle.py
- docker run --rm -w $PWD -v $PWD:$PWD --user=travis apt-ci env MYPYPATH=$PWD/typehinting/ mypy ./apt
......@@ -23,11 +23,12 @@ from __future__ import print_function
import apt_pkg
# import some fancy classes
from apt.package import Package
from apt.cache import Cache, ProblemResolver
from apt.package import Package as Package, Version as Version
from apt.cache import Cache as Cache, ProblemResolver as ProblemResolver
Cache # pyflakes
ProblemResolver # pyflakes
from apt.cdrom import Cdrom
Version # pyflakes
from apt.cdrom import Cdrom as Cdrom
# init the package system, but do not re-initialize config
if "APT" not in apt_pkg.config:
......
#!/usr/bin/env python
#!/usr/bin/python3
# -*- coding: utf-8 -*-
# auth - authentication key management
#
......@@ -37,16 +37,7 @@ import tempfile
import apt_pkg
from apt_pkg import gettext as _
if sys.version_info.major > 2:
unicode = str
try:
from typing import List, Tuple, Union
List # pyflakes
Tuple # pyflakes
Union # pyflakes
except ImportError:
pass
from typing import List, Optional, Tuple
class AptKeyError(Exception):
......@@ -75,7 +66,7 @@ class TrustedKey(object):
def _call_apt_key_script(*args, **kwargs):
# type: (str, Union[str, bytes]) -> str
# type: (str, Optional[str]) -> str
"""Run the apt-key script with the given arguments."""
conf = None
cmd = [apt_pkg.config.find_file("Dir::Bin::Apt-Key", "/usr/bin/apt-key")]
......@@ -100,9 +91,6 @@ def _call_apt_key_script(*args, **kwargs):
stderr=subprocess.PIPE)
stdin = kwargs.get("stdin", None)
# py2 needs this encoded, py3.3 will crash if it is
if sys.version_info.major < 3 and isinstance(stdin, unicode):
stdin = stdin.encode("utf-8")
output, stderr = proc.communicate(stdin) # type: str, str
......
......@@ -26,32 +26,14 @@ import os
import warnings
import weakref
try:
from typing import (Any, Callable, Dict, Iterator, List, Optional,
Set, Tuple, Union, cast, KeysView)
Any # pyflakes
Callable # pyflakes
Dict # pyflakes
Iterator # pyflakes
KeysView # pyflakes
List # pyflakes
Optional # pyflakes
Set # pyflakes
Tuple # pyflakes
Union # pyflakes
except ImportError:
def cast(typ, obj): # type: ignore
return obj
pass
from typing import (Any, Callable, Dict, Iterator, List, Optional,
Set, Tuple, Union, cast, KeysView)
import apt_pkg
from apt.package import Package, Version
import apt.progress.text
from apt.progress.base import AcquireProgress, InstallProgress, OpProgress
OpProgress # pyflakes
InstallProgress # pyflakes
AcquireProgress # pyflakes
Version # pyflakes
class FetchCancelledException(IOError):
......@@ -62,6 +44,10 @@ class FetchFailedException(IOError):
"""Exception that is thrown when fetching fails."""
class UntrustedException(FetchFailedException):
"""Exception that is thrown when fetching fails for trust reasons"""
class LockFailedException(IOError):
"""Exception that is thrown when locking fails."""
......@@ -120,15 +106,15 @@ class Cache(object):
"""
def __init__(self, progress=None, rootdir=None, memonly=False):
# type: (OpProgress, str, bool) -> None
# type: (Optional[OpProgress], Optional[str], bool) -> None
self._cache = cast(apt_pkg.Cache, None) # type: apt_pkg.Cache
self._depcache = cast(apt_pkg.DepCache, None) # type: apt_pkg.DepCache
self._records = cast(apt_pkg.PackageRecords, None) # type: apt_pkg.PackageRecords # nopep8
self._records = cast(apt_pkg.PackageRecords, None) # type: apt_pkg.PackageRecords # noqa
self._list = cast(apt_pkg.SourceList, None) # type: apt_pkg.SourceList
self._callbacks = {} # type: Dict[str, List[Union[Callable[..., None],str]]] # nopep8
self._callbacks2 = {} # type: Dict[str, List[Tuple[Callable[..., Any], Tuple[Any, ...], Dict[Any,Any]]]] # nopep8
self._weakref = weakref.WeakValueDictionary() # type: weakref.WeakValueDictionary[str, apt.Package] # nopep8
self._weakversions = weakref.WeakSet() # type: weakref.WeakSet[Version] # nopep8
self._callbacks = {} # type: Dict[str, List[Union[Callable[..., None],str]]] # noqa
self._callbacks2 = {} # type: Dict[str, List[Tuple[Callable[..., Any], Tuple[Any, ...], Dict[Any,Any]]]] # noqa
self._weakref = weakref.WeakValueDictionary() # type: weakref.WeakValueDictionary[str, apt.Package] # noqa
self._weakversions = weakref.WeakSet() # type: weakref.WeakSet[Version] # noqa
self._changes_count = -1
self._sorted_set = None # type: Optional[List[str]]
......@@ -181,14 +167,16 @@ class Cache(object):
check if the required apt directories/files are there and if
not create them
"""
files = ["/var/lib/dpkg/status",
"/etc/apt/sources.list",
]
dirs = ["/var/lib/dpkg",
"/etc/apt/",
"/var/cache/apt/archives/partial",
"/var/lib/apt/lists/partial",
]
files = [
"/var/lib/dpkg/status",
"/etc/apt/sources.list",
]
dirs = [
"/var/lib/dpkg",
"/etc/apt/",
"/var/cache/apt/archives/partial",
"/var/lib/apt/lists/partial",
]
for d in dirs:
if not os.path.exists(rootdir + d):
#print "creating: ", rootdir + d
......@@ -209,10 +197,10 @@ class Cache(object):
if name in self._callbacks2:
for callback, args, kwds in self._callbacks2[name]:
callback(self, *args, **kwds)
callback(self, *args, **kwds)
def open(self, progress=None):
# type: (OpProgress) -> None
# type: (Optional[OpProgress]) -> None
""" Open the package cache, after that it can be used like
a dictionary
"""
......@@ -416,8 +404,17 @@ class Cache(object):
reqreinst.add(pkg.get_fullname(pretty=True))
return reqreinst
def _run_fetcher(self, fetcher):
# type: (apt_pkg.Acquire) -> int
def _run_fetcher(self, fetcher, allow_unauthenticated):
# type: (apt_pkg.Acquire, Optional[bool]) -> int
if allow_unauthenticated is None:
allow_unauthenticated = apt_pkg.config.find_b("APT::Get::"
"AllowUnauthenticated", False)
untrusted = [item for item in fetcher.items if not item.is_trusted]
if untrusted and not allow_unauthenticated:
raise UntrustedException("Untrusted packages:\n%s" %
"\n".join(i.desc_uri for i in untrusted))
# do the actual fetching
res = fetcher.run()
......@@ -440,8 +437,12 @@ class Cache(object):
raise FetchFailedException(err_msg)
return res
def _fetch_archives(self, fetcher, pm):
# type: (apt_pkg.Acquire, apt_pkg.PackageManager) -> int
def _fetch_archives(self,
fetcher, # type: apt_pkg.Acquire
pm, # type: apt_pkg.PackageManager
allow_unauthenticated=None, # type: Optional[bool]
):
# type: (...) -> int
""" fetch the needed archives """
if self._records is None:
raise CacheClosedException(
......@@ -450,12 +451,17 @@ class Cache(object):
# this may as well throw a SystemError exception
if not pm.get_archives(fetcher, self._list, self._records):
return False
# now run the fetcher, throw exception if something fails to be
# fetched
return self._run_fetcher(fetcher)
def fetch_archives(self, progress=None, fetcher=None):
# type: (AcquireProgress, apt_pkg.Acquire) -> int
return self._run_fetcher(fetcher, allow_unauthenticated)
def fetch_archives(self,
progress=None, # type: Optional[AcquireProgress]
fetcher=None, # type: Optional[apt_pkg.Acquire]
allow_unauthenticated=None, # type: Optional[bool]
):
# type: (...) -> int
"""Fetch the archives for all packages marked for install/upgrade.
You can specify either an :class:`apt.progress.base.AcquireProgress()`
......@@ -466,6 +472,10 @@ class Cache(object):
an exception of type :class:`FetchFailedException` or
:class:`FetchCancelledException` is raised.
The keyword-only parameter *allow_unauthenticated* specifies whether
to allow unauthenticated downloads. If not specified, it defaults to
the configuration option `APT::Get::AllowUnauthenticated`.
.. versionadded:: 0.8.0
"""
if progress is not None and fetcher is not None:
......@@ -477,7 +487,8 @@ class Cache(object):
with self._archive_lock:
return self._fetch_archives(fetcher,
apt_pkg.PackageManager(self._depcache))
apt_pkg.PackageManager(self._depcache),
allow_unauthenticated)
def is_virtual_package(self, pkgname):
# type: (str) -> bool
......@@ -524,7 +535,7 @@ class Cache(object):
def update(self, fetch_progress=None, pulse_interval=0,
raise_on_error=True, sources_list=None):
# type: (AcquireProgress, int, bool, str) -> int
# type: (Optional[AcquireProgress], int, bool, Optional[str]) -> int
"""Run the equivalent of apt-get update.
You probably want to call open() afterwards, in order to utilise the
......@@ -607,8 +618,12 @@ class Cache(object):
install_progress.finish_update()
return res
def commit(self, fetch_progress=None, install_progress=None):
# type: (AcquireProgress, InstallProgress) -> bool
def commit(self,
fetch_progress=None, # type: Optional[AcquireProgress]
install_progress=None, # type: Optional[InstallProgress]
allow_unauthenticated=None, # type: Optional[bool]
):
# type: (...) -> bool
"""Apply the marked changes to the cache.
The first parameter, *fetch_progress*, refers to a FetchProgress()
......@@ -617,6 +632,10 @@ class Cache(object):
The second parameter, *install_progress*, is a
apt.progress.InstallProgress() object.
The keyword-only parameter *allow_unauthenticated* specifies whether
to allow unauthenticated downloads. If not specified, it defaults to
the configuration option `APT::Get::AllowUnauthenticated`.
"""
# FIXME:
# use the new acquire/pkgmanager interface here,
......@@ -638,7 +657,8 @@ class Cache(object):
with self._archive_lock:
while True:
# fetch archives first
res = self._fetch_archives(fetcher, pm)
res = self._fetch_archives(fetcher, pm,
allow_unauthenticated)
# then install
res = self.install_archives(pm, install_progress)
......@@ -790,11 +810,6 @@ class ProblemResolver(object):
"""Reset the package to the default state."""
self._resolver.clear(package._pkg)
def install_protect(self):
# type: () -> None
"""mark protected packages for install or removal."""
self._resolver.install_protect()
def protect(self, package):
# type: (Package) -> None
"""Protect a package so it won't be removed."""
......@@ -899,7 +914,7 @@ class FilteredCache(object):
"""
def __init__(self, cache=None, progress=None):
# type: (Cache, OpProgress) -> None
# type: (Optional[Cache], Optional[OpProgress]) -> None
if cache is None:
self.cache = Cache(progress)
else:
......@@ -991,7 +1006,7 @@ def _test():
apt_pkg.config.set("Dir::Cache::Archives", "/tmp/pytest")
pm = apt_pkg.PackageManager(cache._depcache)
fetcher = apt_pkg.Acquire(apt.progress.text.AcquireProgress())
cache._fetch_archives(fetcher, pm)
cache._fetch_archives(fetcher, pm, None)
#sys.exit(1)
print("Testing filtered cache (argument is old cache)")
......
......@@ -22,6 +22,7 @@
"""Classes related to cdrom handling."""
from __future__ import print_function
from typing import Optional
import glob
import apt_pkg
......@@ -46,7 +47,7 @@ class Cdrom(apt_pkg.Cdrom):
"""
def __init__(self, progress=None, mountpoint=None, nomount=True):
# type: (CdromProgress, str, bool) -> None
# type: (Optional[CdromProgress], Optional[str], bool) -> None
apt_pkg.Cdrom.__init__(self)
if progress is None:
self._progress = CdromProgress()
......@@ -62,12 +63,12 @@ class Cdrom(apt_pkg.Cdrom):
apt_pkg.config.set("APT::CDROM::NoMount", "false")
def add(self, progress=None):
# type: (CdromProgress) -> bool
# type: (Optional[CdromProgress]) -> bool
"""Add cdrom to the sources.list."""
return apt_pkg.Cdrom.add(self, progress or self._progress)
def ident(self, progress=None):
# type: (CdromProgress) -> str
# type: (Optional[CdromProgress]) -> str
"""Identify the cdrom."""
return apt_pkg.Cdrom.ident(self, progress or self._progress)
......
......@@ -26,19 +26,7 @@ import gzip
import os
import sys
try:
from typing import Dict, Iterable, List, Optional, Set, Tuple, Union, cast
Dict # pyflakes
Iterable # pyflakes
List # pyflakes
Optional # pyflakes
Set # pyflakes
Tuple # pyflakes
Union # pyflakes
except ImportError:
def cast(typ, obj): # type: ignore
return obj
pass
from typing import Dict, Iterable, List, Optional, Set, Tuple, Union, cast
from apt_pkg import gettext as _
from io import BytesIO
......@@ -61,14 +49,14 @@ class DebPackage(object):
debug = 0
def __init__(self, filename=None, cache=None):
# type: (str, apt.Cache) -> None
# type: (Optional[str], Optional[apt.Cache]) -> None
if cache is None:
cache = apt.Cache()
self._cache = cache
self._debfile = cast(apt_inst.DebFile, None)
self.pkgname = ""
self.filename = None # type: Optional[str]
self._sections = {} # type: Union[Dict[str, str], apt_pkg.TagSection]
self._sections = {} # type: Union[Dict[str, str], apt_pkg.TagSection[str]] # noqa
self._need_pkgs = [] # type: List[str]
self._check_was_run = False
self._failure_string = ""
......@@ -80,7 +68,7 @@ class DebPackage(object):
# type: (str) -> None
""" open given debfile """
self._dbg(3, "open '%s'" % filename)
self._need_pkgs = [] # type: List[str]
self._need_pkgs = []
self._installed_conflicts = set() # type: Set[str]
self._failure_string = ""
self.filename = filename
......@@ -729,7 +717,7 @@ class DebPackage(object):
print(msg, file=sys.stderr)
def install(self, install_progress=None):
# type: (apt.progress.base.InstallProgress) -> int
# type: (Optional[apt.progress.base.InstallProgress]) -> int
"""Install the package."""
if self.filename is None:
raise apt_pkg.Error("No filename specified")
......@@ -752,7 +740,7 @@ class DscSrcPackage(DebPackage):
"""A locally available source package."""
def __init__(self, filename=None, cache=None):
# type: (str, apt.Cache) -> None
# type: (Optional[str], Optional[apt.Cache]) -> None
DebPackage.__init__(self, None, cache)
self.filename = filename # type: Optional[str]
self._depends = [] # type: List[List[Tuple[str, str, str]]]
......@@ -806,7 +794,7 @@ class DscSrcPackage(DebPackage):
if 'Binary' in sec:
self.binaries = [b.strip() for b in
sec['Binary'].split(',')]
for tag in sec.keys(): # type: ignore
for tag in sec.keys():
if tag in sec:
self._sections[tag] = sec[tag]
finally:
......
......@@ -28,52 +28,14 @@ import re
import socket
import subprocess
import threading
threading # pyflakes
try:
from http.client import BadStatusLine
from urllib.error import HTTPError
from urllib.request import urlopen
except ImportError:
from httplib import BadStatusLine # type: ignore
from urllib2 import HTTPError, urlopen # type: ignore
try:
from typing import (Any, Iterable, Iterator, List, Optional, Set,
Tuple, Union, no_type_check, overload, Mapping,
Sequence)
Any # pyflakes
Iterable # pyflakes
Iterator # pyflakes
List # pyflakes
Optional # pyflakes
Set # pyflakes
Tuple # pyflakes
Union # pyflakes
overload # pyflakes
Sequence # pyflakes
except ImportError:
import collections
class GenericWrapper(object):
"""Takes a non-generic type and adds __getitem__"""
def __init__(self, value):
# type: (type) -> None
self.value = value
def __getitem__(self, key):
# type: (type) -> type
return self.value
List = GenericWrapper(list) # type: ignore
Mapping = GenericWrapper(collections.Mapping) # type: ignore
Sequence = GenericWrapper(collections.Sequence) # type: ignore
Any = None
def no_type_check(arg):
# type: (Any) -> Any
return arg
pass
from http.client import BadStatusLine
from urllib.error import HTTPError
from urllib.request import urlopen
from typing import (Any, Iterable, Iterator, List, Optional, Set,
Tuple, Union, no_type_check, Mapping,
Sequence)
import apt_pkg
import apt.progress.text
......@@ -82,24 +44,19 @@ from apt.progress.base import (
AcquireProgress,
InstallProgress,
)
AcquireProgress # pyflakes
InstallProgress # pyflakes
from apt_pkg import gettext as _
__all__ = ('BaseDependency', 'Dependency', 'Origin', 'Package', 'Record',
'Version', 'VersionList')
if sys.version_info.major >= 3:
unicode = str
def _file_is_same(path, size, md5):
# type: (str, int, str) -> bool
def _file_is_same(path, size, hashes):
# type: (str, int, apt_pkg.HashStringList) -> bool
"""Return ``True`` if the file is the same."""
if os.path.exists(path) and os.path.getsize(path) == size:
with open(path) as fobj:
return apt_pkg.md5sum(fobj) == md5
return apt_pkg.Hashes(fobj).hashes == hashes
return False
......@@ -107,6 +64,10 @@ class FetchError(Exception):
"""Raised when a file could not be fetched."""
class UntrustedError(FetchError):
"""Raised when a file did not have a trusted hash."""
class BaseDependency(object):
"""A single dependency."""
......@@ -418,7 +379,7 @@ class Record(Mapping[Any, Any]):
def __iter__(self):
# type: () -> Iterator[str]
return iter(self._rec.keys()) # type: ignore
return iter(self._rec.keys())
def iteritems(self):
# type: () -> Iterable[Tuple[object, str]]
......@@ -427,7 +388,7 @@ class Record(Mapping[Any, Any]):
yield key, self._rec[key]
def get(self, key, default=None):
# type: (object, object) -> object
# type: (str, object) -> object
"""Return record[key] if key in record, else *default*.
The parameter *default* must be either a string or None.
......@@ -533,6 +494,7 @@ class Version(object):
def _records(self):
# type: () -> apt_pkg.PackageRecords
"""Internal helper that moves the Records to the right position."""
# If changing lookup, change fetch_binary() as well
if not self.package._pcache._records.lookup(self._cand.file_list[0]):
raise LookupError("Could not lookup record")
......@@ -631,7 +593,7 @@ class Version(object):
"Please report.") % (self.package.name)
try:
if not isinstance(dsc, unicode):
if not isinstance(dsc, str):
# Only convert where needed (i.e. Python 2.X)
dsc = dsc.decode("utf-8")
except UnicodeDecodeError as err:
......@@ -849,8 +811,9 @@ class Version(object):
except StopIteration:
return None
def fetch_binary(self, destdir='', progress=None):
# type: (str, AcquireProgress) -> str
def fetch_binary(self, destdir='', progress=None,
allow_unauthenticated=None):
# type: (str, Optional[AcquireProgress], Optional[bool]) -> str
"""Fetch the binary version of the package.
The parameter *destdir* specifies the directory where the package will
......@@ -860,15 +823,39 @@ class Version(object):
object. If not specified or None, apt.progress.text.AcquireProgress()
is used.
The keyword-only parameter *allow_unauthenticated* specifies whether
to allow unauthenticated downloads. If not specified, it defaults to
the configuration option `APT::Get::AllowUnauthenticated`.
.. versionadded:: 0.7.10
"""
if allow_unauthenticated is None:
allow_unauthenticated = apt_pkg.config.find_b("APT::Get::"
"AllowUnauthenticated", False)
base = os.path.basename(self._records.filename)
destfile = os.path.join(destdir, base)