summaryrefslogtreecommitdiffstats
path: root/apt/cache.py
blob: 5893c0eddcfe4ab31836334a606f85648a4e8fc9 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
1001
1002
1003
1004
# cache.py - apt cache abstraction
#
#  Copyright (c) 2005-2009 Canonical
#
#  Author: Michael Vogt <michael.vogt@ubuntu.com>
#
#  This program is free software; you can redistribute it and/or
#  modify it under the terms of the GNU General Public License as
#  published by the Free Software Foundation; either version 2 of the
#  License, or (at your option) any later version.
#
#  This program is distributed in the hope that it will be useful,
#  but WITHOUT ANY WARRANTY; without even the implied warranty of
#  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
#  GNU General Public License for more details.
#
#  You should have received a copy of the GNU General Public License
#  along with this program; if not, write to the Free Software
#  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307
#  USA

from __future__ import annotations

import fnmatch
import os
import warnings
import weakref
from collections.abc import Callable, Iterator, KeysView
from typing import Any, cast

import apt_pkg

import apt.progress.text
from apt.package import Package, Version
from apt.progress.base import AcquireProgress, InstallProgress, OpProgress


class FetchCancelledException(IOError):
    """Exception that is thrown when the user cancels a fetch operation."""


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."""


class CacheClosedException(Exception):
    """Exception that is thrown when the cache is used after close()."""


class _WrappedLock:
    """Wraps an apt_pkg.FileLock to raise LockFailedException.

    Initialized using a directory path."""

    def __init__(self, path: str) -> None:
        self._path = path
        self._lock = apt_pkg.FileLock(os.path.join(path, "lock"))

    def __enter__(self) -> None:
        try:
            return self._lock.__enter__()
        except apt_pkg.Error as e:
            raise LockFailedException(
                ("Failed to lock directory %s: %s") % (self._path, e)
            )

    def __exit__(self, typ: object, value: object, traceback: object) -> None:
        return self._lock.__exit__(typ, value, traceback)


class Cache:
    """Dictionary-like package cache.

    The APT cache file contains a hash table mapping names of binary
    packages to their metadata. A Cache object is the in-core
    representation of the same. It provides access to APTs idea of the
    list of available packages.

    The cache can be used like a mapping from package names to Package
    objects (although only getting items is supported).

    Keyword arguments:
    progress -- a OpProgress object,
    rootdir  -- an alternative root directory. if that is given the system
    sources.list and system lists/files are not read, only file relative
    to the given rootdir,
    memonly  -- build the cache in memory only.


    .. versionchanged:: 1.0

        The cache now supports package names with special architecture
        qualifiers such as :all and :native. It does not export them
        in :meth:`keys()`, though, to keep :meth:`keys()` a unique set.
    """

    def __init__(
        self,
        progress: OpProgress | None = None,
        rootdir: str | None = None,
        memonly: bool = False,
    ) -> None:
        self._cache: apt_pkg.Cache = cast(apt_pkg.Cache, None)
        self._depcache: apt_pkg.DepCache = cast(apt_pkg.DepCache, None)
        self._records: apt_pkg.PackageRecords = cast(
            apt_pkg.PackageRecords, None
        )  # noqa
        self._list: apt_pkg.SourceList = cast(apt_pkg.SourceList, None)
        self._callbacks: dict[str, list[Callable[..., None] | str]] = {}  # noqa
        self._callbacks2: dict[
            str, list[tuple[Callable[..., Any], tuple[Any, ...], dict[Any, Any]]]
        ] = {}  # noqa
        self._weakref: weakref.WeakValueDictionary[str, apt.Package] = (
            weakref.WeakValueDictionary()
        )  # noqa
        self._weakversions: weakref.WeakSet[Version] = weakref.WeakSet()  # noqa
        self._changes_count = -1
        self._sorted_set: list[str] | None = None

        self.connect("cache_post_open", "_inc_changes_count")
        self.connect("cache_post_change", "_inc_changes_count")
        if memonly:
            # force apt to build its caches in memory
            apt_pkg.config.set("Dir::Cache::pkgcache", "")
        if rootdir:
            rootdir = os.path.abspath(rootdir)
            if os.path.exists(rootdir + "/etc/apt/apt.conf"):
                apt_pkg.read_config_file(apt_pkg.config, rootdir + "/etc/apt/apt.conf")
            if os.path.isdir(rootdir + "/etc/apt/apt.conf.d"):
                apt_pkg.read_config_dir(apt_pkg.config, rootdir + "/etc/apt/apt.conf.d")
            apt_pkg.config.set("Dir", rootdir)
            apt_pkg.config.set("Dir::State::status", rootdir + "/var/lib/dpkg/status")
            # also set dpkg to the rootdir path so that its called for the
            # --print-foreign-architectures call
            apt_pkg.config.set(
                "Dir::bin::dpkg", os.path.join(rootdir, "usr", "bin", "dpkg")
            )
            # create required dirs/files when run with special rootdir
            # automatically
            self._check_and_create_required_dirs(rootdir)
            # Call InitSystem so the change to Dir::State::Status is actually
            # recognized (LP: #320665)
            apt_pkg.init_system()

        # Prepare a lock object (context manager for archive lock)
        archive_dir = apt_pkg.config.find_dir("Dir::Cache::Archives")
        self._archive_lock = _WrappedLock(archive_dir)

        self.open(progress)

    def fix_broken(self) -> None:
        """Fix broken packages."""
        self._depcache.fix_broken()

    def _inc_changes_count(self) -> None:
        """Increase the number of changes"""
        self._changes_count += 1

    def _check_and_create_required_dirs(self, rootdir: str) -> None:
        """
        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",
        ]
        for d in dirs:
            if not os.path.exists(rootdir + d):
                # print "creating: ", rootdir + d
                os.makedirs(rootdir + d)
        for f in files:
            if not os.path.exists(rootdir + f):
                open(rootdir + f, "w").close()

    def _run_callbacks(self, name: str) -> None:
        """internal helper to run a callback"""
        if name in self._callbacks:
            for callback in self._callbacks[name]:
                if callback == "_inc_changes_count":
                    self._inc_changes_count()
                else:
                    callback()  # type: ignore

        if name in self._callbacks2:
            for callback, args, kwds in self._callbacks2[name]:
                callback(self, *args, **kwds)

    def open(self, progress: OpProgress | None = None) -> None:
        """Open the package cache, after that it can be used like
        a dictionary
        """
        if progress is None:
            progress = apt.progress.base.OpProgress()
        # close old cache on (re)open
        self.close()
        self.op_progress = progress
        self._run_callbacks("cache_pre_open")

        self._cache = apt_pkg.Cache(progress)
        self._depcache = apt_pkg.DepCache(self._cache)
        self._records = apt_pkg.PackageRecords(self._cache)
        self._list = apt_pkg.SourceList()
        self._list.read_main_list()
        self._sorted_set = None
        self.__remap()

        self._have_multi_arch = len(apt_pkg.get_architectures()) > 1

        progress.done()
        self._run_callbacks("cache_post_open")

    def __remap(self) -> None:
        """Called after cache reopen() to relocate to new cache.

        Relocate objects like packages and versions from the old
        underlying cache to the new one.
        """
        for key in list(self._weakref.keys()):
            try:
                pkg = self._weakref[key]
            except KeyError:
                continue

            try:
                pkg._pkg = self._cache[pkg._pkg.name, pkg._pkg.architecture]
            except LookupError:
                del self._weakref[key]

        for ver in list(self._weakversions):
            # Package has been reseated above, reseat version
            for v in ver.package._pkg.version_list:
                # Requirements as in debListParser::SameVersion
                if (
                    v.hash == ver._cand.hash
                    and (v.size == 0 or ver._cand.size == 0 or v.size == ver._cand.size)
                    and v.multi_arch == ver._cand.multi_arch
                    and v.ver_str == ver._cand.ver_str
                ):
                    ver._cand = v
                    break
            else:
                self._weakversions.remove(ver)

    def close(self) -> None:
        """Close the package cache"""
        # explicitely free the FDs that _records has open
        del self._records
        self._records = cast(apt_pkg.PackageRecords, None)

    def __enter__(self) -> Cache:
        """Enter the with statement"""
        return self

    def __exit__(self, exc_type: object, exc_value: object, traceback: object) -> None:
        """Exit the with statement"""
        self.close()

    def __getitem__(self, key: object) -> Package:
        """look like a dictionary (get key)"""
        try:
            key = str(key)
            rawpkg = self._cache[key]
        except KeyError:
            raise KeyError("The cache has no package named %r" % key)

        # It might be excluded due to not having a version or something
        if not self.__is_real_pkg(rawpkg):
            raise KeyError("The cache has no package named %r" % key)

        pkg = self._rawpkg_to_pkg(rawpkg)

        return pkg

    def get(self, key: object, default: object = None) -> Any:
        """Return *self*[*key*] or *default* if *key* not in *self*.

        .. versionadded:: 1.1
        """
        try:
            return self[key]
        except KeyError:
            return default

    def _rawpkg_to_pkg(self, rawpkg: apt_pkg.Package) -> Package:
        """Returns the apt.Package object for an apt_pkg.Package object.

        .. versionadded:: 1.0.0
        """
        fullname = rawpkg.get_fullname(pretty=True)

        return self._weakref.setdefault(fullname, Package(self, rawpkg))

    def __iter__(self) -> Iterator[Package]:
        # We iterate sorted over package names here. With this we read the
        # package lists linearly if we need to access the package records,
        # instead of having to do thousands of random seeks; the latter
        # is disastrous if we use compressed package indexes, and slower than
        # necessary for uncompressed indexes.
        for pkgname in self.keys():
            pkg = Package(self, self._cache[pkgname])
            yield self._weakref.setdefault(pkgname, pkg)

    def __is_real_pkg(self, rawpkg: apt_pkg.Package) -> bool:
        """Check if the apt_pkg.Package provided is a real package."""
        return rawpkg.has_versions

    def has_key(self, key: object) -> bool:
        return key in self

    def __contains__(self, key: object) -> bool:
        try:
            return self.__is_real_pkg(self._cache[str(key)])
        except KeyError:
            return False

    def __len__(self) -> int:
        return len(self.keys())

    def keys(self) -> list[str]:
        if self._sorted_set is None:
            self._sorted_set = sorted(
                p.get_fullname(pretty=True)
                for p in self._cache.packages
                if self.__is_real_pkg(p)
            )
        return list(self._sorted_set)  # We need a copy here, caller may modify

    def get_changes(self) -> list[Package]:
        """Get the marked changes"""
        changes = []
        marked_keep = self._depcache.marked_keep
        for rawpkg in self._cache.packages:
            if not marked_keep(rawpkg):
                changes.append(self._rawpkg_to_pkg(rawpkg))
        return changes

    def upgrade(self, dist_upgrade: bool = False) -> None:
        """Upgrade all packages.

        If the parameter *dist_upgrade* is True, new dependencies will be
        installed as well (and conflicting packages may be removed). The
        default value is False.
        """
        self.cache_pre_change()
        self._depcache.upgrade(dist_upgrade)
        self.cache_post_change()

    @property
    def required_download(self) -> int:
        """Get the size of the packages that are required to download."""
        if self._records is None:
            raise CacheClosedException("Cache object used after close() called")
        pm = apt_pkg.PackageManager(self._depcache)
        fetcher = apt_pkg.Acquire()
        pm.get_archives(fetcher, self._list, self._records)
        return fetcher.fetch_needed

    @property
    def required_space(self) -> int:
        """Get the size of the additional required space on the fs."""
        return self._depcache.usr_size

    @property
    def req_reinstall_pkgs(self) -> set[str]:
        """Return the packages not downloadable packages in reqreinst state."""
        reqreinst = set()
        get_candidate_ver = self._depcache.get_candidate_ver
        states = frozenset(
            (apt_pkg.INSTSTATE_REINSTREQ, apt_pkg.INSTSTATE_HOLD_REINSTREQ)
        )
        for pkg in self._cache.packages:
            cand = get_candidate_ver(pkg)
            if cand and not cand.downloadable and pkg.inst_state in states:
                reqreinst.add(pkg.get_fullname(pretty=True))
        return reqreinst

    def _run_fetcher(
        self, fetcher: apt_pkg.Acquire, allow_unauthenticated: bool | None
    ) -> 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()

        # now check the result (this is the code from apt-get.cc)
        failed = False
        err_msg = ""
        for item in fetcher.items:
            if item.status == item.STAT_DONE:
                continue
            if item.STAT_IDLE:
                continue
            err_msg += f"Failed to fetch {item.desc_uri} {item.error_text}\n"
            failed = True

        # we raise a exception if the download failed or it was cancelt
        if res == fetcher.RESULT_CANCELLED:
            raise FetchCancelledException(err_msg)
        elif failed:
            raise FetchFailedException(err_msg)
        return res

    def _fetch_archives(
        self,
        fetcher: apt_pkg.Acquire,
        pm: apt_pkg.PackageManager,
        allow_unauthenticated: bool | None = None,
    ) -> int:
        """fetch the needed archives"""
        if self._records is None:
            raise CacheClosedException("Cache object used after close() called")

        # 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, allow_unauthenticated)

    def fetch_archives(
        self,
        progress: AcquireProgress | None = None,
        fetcher: apt_pkg.Acquire | None = None,
        allow_unauthenticated: bool | None = None,
    ) -> int:
        """Fetch the archives for all packages marked for install/upgrade.

        You can specify either an :class:`apt.progress.base.AcquireProgress()`
        object for the parameter *progress*, or specify an already
        existing :class:`apt_pkg.Acquire` object for the parameter *fetcher*.

        The return value of the function is undefined. If an error occurred,
        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:
            raise ValueError("Takes a progress or a an Acquire object")
        if progress is None:
            progress = apt.progress.text.AcquireProgress()
        if fetcher is None:
            fetcher = apt_pkg.Acquire(progress)

        with self._archive_lock:
            return self._fetch_archives(
                fetcher, apt_pkg.PackageManager(self._depcache), allow_unauthenticated
            )

    def is_virtual_package(self, pkgname: str) -> bool:
        """Return whether the package is a virtual package."""
        try:
            pkg = self._cache[pkgname]
        except KeyError:
            return False
        else:
            return bool(pkg.has_provides and not pkg.has_versions)

    def get_providing_packages(
        self,
        pkgname: str,
        candidate_only: bool = True,
        include_nonvirtual: bool = False,
    ) -> list[Package]:
        """Return a list of all packages providing a package.

        Return a list of packages which provide the virtual package of the
        specified name.

        If 'candidate_only' is False, return all packages with at
        least one version providing the virtual package. Otherwise,
        return only those packages where the candidate version
        provides the virtual package.

        If 'include_nonvirtual' is True then it will search for all
        packages providing pkgname, even if pkgname is not itself
        a virtual pkg.
        """

        providers: set[Package] = set()
        get_candidate_ver = self._depcache.get_candidate_ver
        try:
            vp = self._cache[pkgname]
            if vp.has_versions and not include_nonvirtual:
                return list(providers)
        except KeyError:
            return list(providers)

        for provides, providesver, version in vp.provides_list:
            rawpkg = version.parent_pkg
            if not candidate_only or (version == get_candidate_ver(rawpkg)):
                providers.add(self._rawpkg_to_pkg(rawpkg))
        return list(providers)

    def update(
        self,
        fetch_progress: AcquireProgress | None = None,
        pulse_interval: int = 0,
        raise_on_error: bool = True,
        sources_list: str | None = None,
    ) -> int:
        """Run the equivalent of apt-get update.

        You probably want to call open() afterwards, in order to utilise the
        new cache. Otherwise, the old cache will be used which can lead to
        strange bugs.

        The first parameter *fetch_progress* may be set to an instance of
        apt.progress.FetchProgress, the default is apt.progress.FetchProgress()
        .
        sources_list -- Update a alternative sources.list than the default.
        Note that the sources.list.d directory is ignored in this case
        """
        with _WrappedLock(apt_pkg.config.find_dir("Dir::State::Lists")):
            if sources_list:
                old_sources_list = apt_pkg.config.find("Dir::Etc::sourcelist")
                old_sources_list_d = apt_pkg.config.find("Dir::Etc::sourceparts")
                old_cleanup = apt_pkg.config.find("APT::List-Cleanup")
                apt_pkg.config.set(
                    "Dir::Etc::sourcelist", os.path.abspath(sources_list)
                )
                apt_pkg.config.set("Dir::Etc::sourceparts", "xxx")
                apt_pkg.config.set("APT::List-Cleanup", "0")
                slist = apt_pkg.SourceList()
                slist.read_main_list()
            else:
                slist = self._list

            try:
                if fetch_progress is None:
                    fetch_progress = apt.progress.base.AcquireProgress()
                try:
                    res = self._cache.update(fetch_progress, slist, pulse_interval)
                except SystemError as e:
                    raise FetchFailedException(e)
                if not res and raise_on_error:
                    raise FetchFailedException()
                else:
                    return res
            finally:
                if sources_list:
                    apt_pkg.config.set("Dir::Etc::sourcelist", old_sources_list)
                    apt_pkg.config.set("Dir::Etc::sourceparts", old_sources_list_d)
                    apt_pkg.config.set("APT::List-Cleanup", old_cleanup)

    def install_archives(
        self, pm: apt_pkg.PackageManager, install_progress: InstallProgress
    ) -> int:
        """
        The first parameter *pm* refers to an object returned by
        apt_pkg.PackageManager().

        The second parameter *install_progress* refers to an InstallProgress()
        object of the module apt.progress.

        This releases a system lock in newer versions, if there is any,
        and reestablishes it afterwards.
        """
        # compat with older API
        try:
            install_progress.startUpdate()  # type: ignore
        except AttributeError:
            install_progress.start_update()

        did_unlock = apt_pkg.pkgsystem_is_locked()
        if did_unlock:
            apt_pkg.pkgsystem_unlock_inner()

        try:
            res = install_progress.run(pm)
        finally:
            if did_unlock:
                apt_pkg.pkgsystem_lock_inner()

        try:
            install_progress.finishUpdate()  # type: ignore
        except AttributeError:
            install_progress.finish_update()
        return res

    def commit(
        self,
        fetch_progress: AcquireProgress | None = None,
        install_progress: InstallProgress | None = None,
        allow_unauthenticated: bool | None = None,
    ) -> bool:
        """Apply the marked changes to the cache.

        The first parameter, *fetch_progress*, refers to a FetchProgress()
        object as found in apt.progress, the default being
        apt.progress.FetchProgress().

        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,
        # raise exceptions when a download or install fails
        # and send proper error strings to the application.
        # Current a failed download will just display "error"
        # which is less than optimal!

        if fetch_progress is None:
            fetch_progress = apt.progress.base.AcquireProgress()
        if install_progress is None:
            install_progress = apt.progress.base.InstallProgress()

        assert install_progress is not None

        with apt_pkg.SystemLock():
            pm = apt_pkg.PackageManager(self._depcache)
            fetcher = apt_pkg.Acquire(fetch_progress)
            with self._archive_lock:
                while True:
                    # fetch archives first
                    res = self._fetch_archives(fetcher, pm, allow_unauthenticated)

                    # then install
                    res = self.install_archives(pm, install_progress)
                    if res == pm.RESULT_COMPLETED:
                        break
                    elif res == pm.RESULT_FAILED:
                        raise SystemError("installArchives() failed")
                    elif res == pm.RESULT_INCOMPLETE:
                        pass
                    else:
                        raise SystemError(
                            "internal-error: unknown result "
                            "code from InstallArchives: %s" % res
                        )
                    # reload the fetcher for media swaping
                    fetcher.shutdown()
        return res == pm.RESULT_COMPLETED

    def clear(self) -> None:
        """Unmark all changes"""
        self._depcache.init()

    # cache changes

    def cache_post_change(self) -> None:
        "called internally if the cache has changed, emit a signal then"
        self._run_callbacks("cache_post_change")

    def cache_pre_change(self) -> None:
        """called internally if the cache is about to change, emit
        a signal then"""
        self._run_callbacks("cache_pre_change")

    def connect(self, name: str, callback: Callable[..., None] | str) -> None:
        """Connect to a signal.

        .. deprecated:: 1.0

            Please use connect2() instead, as this function is very
            likely to cause a memory leak.
        """
        if callback != "_inc_changes_count":
            warnings.warn(
                "connect() likely causes a reference" " cycle, use connect2() instead",
                RuntimeWarning,
                2,
            )
        if name not in self._callbacks:
            self._callbacks[name] = []
        self._callbacks[name].append(callback)

    def connect2(
        self, name: str, callback: Callable[..., Any], *args: object, **kwds: object
    ) -> None:
        """Connect to a signal.

        The callback will be passed the cache as an argument, and
        any arguments passed to this function. Make sure that, if you
        pass a method of a class as your callback, your class does not
        contain a reference to the cache.

        Cyclic references to the cache can cause issues if the Cache object
        is replaced by a new one, because the cache keeps a lot of objects and
        tens of open file descriptors.

        currently only used for cache_{post,pre}_{changed,open}.

        .. versionadded:: 1.0
        """
        if name not in self._callbacks2:
            self._callbacks2[name] = []
        self._callbacks2[name].append((callback, args, kwds))

    def actiongroup(self) -> apt_pkg.ActionGroup:
        """Return an `ActionGroup` object for the current cache.

        Action groups can be used to speedup actions. The action group is
        active as soon as it is created, and disabled when the object is
        deleted or when release() is called.

        You can use the action group as a context manager, this is the
        recommended way::

            with cache.actiongroup():
                for package in my_selected_packages:
                    package.mark_install()

        This way, the action group is automatically released as soon as the
        with statement block is left. It also has the benefit of making it
        clear which parts of the code run with a action group and which
        don't.
        """
        return apt_pkg.ActionGroup(self._depcache)

    @property
    def dpkg_journal_dirty(self) -> bool:
        """Return True if the dpkg was interrupted

        All dpkg operations will fail until this is fixed, the action to
        fix the system if dpkg got interrupted is to run
        'dpkg --configure -a' as root.
        """
        dpkg_status_dir = os.path.dirname(
            apt_pkg.config.find_file("Dir::State::status")
        )
        for f in os.listdir(os.path.join(dpkg_status_dir, "updates")):
            if fnmatch.fnmatch(f, "[0-9]*"):
                return True
        return False

    @property
    def broken_count(self) -> int:
        """Return the number of packages with broken dependencies."""
        return self._depcache.broken_count

    @property
    def delete_count(self) -> int:
        """Return the number of packages marked for deletion."""
        return self._depcache.del_count

    @property
    def install_count(self) -> int:
        """Return the number of packages marked for installation."""
        return self._depcache.inst_count

    @property
    def keep_count(self) -> int:
        """Return the number of packages marked as keep."""
        return self._depcache.keep_count


class ProblemResolver:
    """Resolve problems due to dependencies and conflicts.

    The first argument 'cache' is an instance of apt.Cache.
    """

    def __init__(self, cache: Cache) -> None:
        self._resolver = apt_pkg.ProblemResolver(cache._depcache)
        self._cache = cache

    def clear(self, package: Package) -> None:
        """Reset the package to the default state."""
        self._resolver.clear(package._pkg)

    def protect(self, package: Package) -> None:
        """Protect a package so it won't be removed."""
        self._resolver.protect(package._pkg)

    def remove(self, package: Package) -> None:
        """Mark a package for removal."""
        self._resolver.remove(package._pkg)

    def resolve(self) -> None:
        """Resolve dependencies, try to remove packages where needed."""
        self._cache.cache_pre_change()
        self._resolver.resolve()
        self._cache.cache_post_change()

    def resolve_by_keep(self) -> None:
        """Resolve dependencies, do not try to remove packages."""
        self._cache.cache_pre_change()
        self._resolver.resolve_by_keep()
        self._cache.cache_post_change()

    def keep_phased_updates(self) -> None:
        """Keep back phased updates."""
        self._cache.cache_pre_change()
        self._resolver.keep_phased_updates()
        self._cache.cache_post_change()


# ----------------------------- experimental interface


class Filter:
    """Filter base class"""

    def apply(self, pkg: Package) -> bool:
        """Filter function, return True if the package matchs a
        filter criteria and False otherwise
        """
        return True


class MarkedChangesFilter(Filter):
    """Filter that returns all marked changes"""

    def apply(self, pkg: Package) -> bool:
        if pkg.marked_install or pkg.marked_delete or pkg.marked_upgrade:
            return True
        else:
            return False


class InstalledFilter(Filter):
    """Filter that returns all installed packages.

    .. versionadded:: 1.0.0
    """

    def apply(self, pkg: Package) -> bool:
        return pkg.is_installed


class _FilteredCacheHelper:
    """Helper class for FilteredCache to break a reference cycle."""

    def __init__(self, cache: Cache) -> None:
        # Do not keep a reference to the cache, or you have a cycle!

        self._filtered: dict[str, bool] = {}
        self._filters: list[Filter] = []
        cache.connect2("cache_post_change", self.filter_cache_post_change)
        cache.connect2("cache_post_open", self.filter_cache_post_change)

    def _reapply_filter(self, cache: Cache) -> None:
        "internal helper to refilter"
        # Do not keep a reference to the cache, or you have a cycle!
        self._filtered = {}
        for pkg in cache:
            for f in self._filters:
                if f.apply(pkg):
                    self._filtered[pkg.name] = True
                    break

    def set_filter(self, filter: Filter) -> None:
        """Set the current active filter."""
        self._filters = []
        self._filters.append(filter)

    def filter_cache_post_change(self, cache: Cache) -> None:
        """Called internally if the cache changes, emit a signal then."""
        # Do not keep a reference to the cache, or you have a cycle!
        self._reapply_filter(cache)


class FilteredCache:
    """A package cache that is filtered.

    Can work on a existing cache or create a new one
    """

    def __init__(
        self, cache: Cache | None = None, progress: OpProgress | None = None
    ) -> None:
        if cache is None:
            self.cache = Cache(progress)
        else:
            self.cache = cache
        self._helper = _FilteredCacheHelper(self.cache)

    def __len__(self) -> int:
        return len(self._helper._filtered)

    def __getitem__(self, key: str) -> Package:
        return self.cache[key]

    def __iter__(self) -> Iterator[Package]:
        for pkgname in self._helper._filtered:
            yield self.cache[pkgname]

    def keys(self) -> KeysView[str]:
        return self._helper._filtered.keys()

    def has_key(self, key: object) -> bool:
        return key in self

    def __contains__(self, key: object) -> bool:
        try:
            # Normalize package name for multi arch
            return self.cache[key].name in self._helper._filtered
        except KeyError:
            return False

    def set_filter(self, filter: Filter) -> None:
        """Set the current active filter."""
        self._helper.set_filter(filter)
        self.cache.cache_post_change()

    def filter_cache_post_change(self) -> None:
        """Called internally if the cache changes, emit a signal then."""
        self._helper.filter_cache_post_change(self.cache)

    def __getattr__(self, key: str) -> Any:
        """we try to look exactly like a real cache."""
        return getattr(self.cache, key)


def cache_pre_changed(cache: Cache) -> None:
    print("cache pre changed")


def cache_post_changed(cache: Cache) -> None:
    print("cache post changed")


def _test() -> None:
    """Internal test code."""
    print("Cache self test")
    apt_pkg.init()
    cache = Cache(apt.progress.text.OpProgress())
    cache.connect2("cache_pre_change", cache_pre_changed)
    cache.connect2("cache_post_change", cache_post_changed)
    print("aptitude" in cache)
    pkg = cache["aptitude"]
    print(pkg.name)
    print(len(cache))

    for pkgname in cache.keys():
        assert cache[pkgname].name == pkgname

    cache.upgrade()
    changes = cache.get_changes()
    print(len(changes))
    for pkg in changes:
        assert pkg.name

    # see if fetching works
    for dirname in ["/tmp/pytest", "/tmp/pytest/partial"]:
        if not os.path.exists(dirname):
            os.mkdir(dirname)
    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, None)
    # sys.exit(1)

    print("Testing filtered cache (argument is old cache)")
    filtered = FilteredCache(cache)
    filtered.cache.connect2("cache_pre_change", cache_pre_changed)
    filtered.cache.connect2("cache_post_change", cache_post_changed)
    filtered.cache.upgrade()
    filtered.set_filter(MarkedChangesFilter())
    print(len(filtered))
    for pkgname in filtered.keys():
        assert pkgname == filtered[pkgname].name

    print(len(filtered))

    print("Testing filtered cache (no argument)")
    filtered = FilteredCache(progress=apt.progress.base.OpProgress())
    filtered.cache.connect2("cache_pre_change", cache_pre_changed)
    filtered.cache.connect2("cache_post_change", cache_post_changed)
    filtered.cache.upgrade()
    filtered.set_filter(MarkedChangesFilter())
    print(len(filtered))
    for pkgname in filtered.keys():
        assert pkgname == filtered[pkgname].name

    print(len(filtered))


if __name__ == "__main__":
    _test()