aboutsummaryrefslogtreecommitdiff
path: root/python/scripts/mkvenv.py
blob: d0b9c215ca2bf5a52b0b367087fa2e60524557c7 (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
"""
mkvenv - QEMU pyvenv bootstrapping utility

usage: mkvenv [-h] command ...

QEMU pyvenv bootstrapping utility

options:
  -h, --help  show this help message and exit

Commands:
  command     Description
    create    create a venv
    post_init
              post-venv initialization
    ensure    Ensure that the specified package is installed.
    ensuregroup
              Ensure that the specified package group is installed.

--------------------------------------------------

usage: mkvenv create [-h] target

positional arguments:
  target      Target directory to install virtual environment into.

options:
  -h, --help  show this help message and exit

--------------------------------------------------

usage: mkvenv post_init [-h]

options:
  -h, --help         show this help message and exit

--------------------------------------------------

usage: mkvenv ensure [-h] [--online] [--dir DIR] dep_spec...

positional arguments:
  dep_spec    PEP 508 Dependency specification, e.g. 'meson>=0.61.5'

options:
  -h, --help  show this help message and exit
  --online    Install packages from PyPI, if necessary.
  --dir DIR   Path to vendored packages where we may install from.

--------------------------------------------------

usage: mkvenv ensuregroup [-h] [--online] [--dir DIR] file group...

positional arguments:
  file        pointer to a TOML file
  group       section name in the TOML file

options:
  -h, --help  show this help message and exit
  --online    Install packages from PyPI, if necessary.
  --dir DIR   Path to vendored packages where we may install from.

"""

# Copyright (C) 2022-2023 Red Hat, Inc.
#
# Authors:
#  John Snow <jsnow@redhat.com>
#  Paolo Bonzini <pbonzini@redhat.com>
#
# This work is licensed under the terms of the GNU GPL, version 2 or
# later. See the COPYING file in the top-level directory.

import argparse
from importlib.metadata import (
    Distribution,
    EntryPoint,
    PackageNotFoundError,
    distribution,
    version,
)
from importlib.util import find_spec
import logging
import os
from pathlib import Path
import re
import shutil
import site
import subprocess
import sys
import sysconfig
from types import SimpleNamespace
from typing import (
    Any,
    Dict,
    Iterator,
    Optional,
    Sequence,
    Tuple,
    Union,
)
import venv


# Try to load distlib, with a fallback to pip's vendored version.
# HAVE_DISTLIB is checked below, just-in-time, so that mkvenv does not fail
# outside the venv or before a potential call to ensurepip in checkpip().
HAVE_DISTLIB = True
try:
    import distlib.scripts
    import distlib.version
except ImportError:
    try:
        # Reach into pip's cookie jar.  pylint and flake8 don't understand
        # that these imports will be used via distlib.xxx.
        from pip._vendor import distlib
        import pip._vendor.distlib.scripts  # noqa, pylint: disable=unused-import
        import pip._vendor.distlib.version  # noqa, pylint: disable=unused-import
    except ImportError:
        HAVE_DISTLIB = False

# Try to load tomllib, with a fallback to tomli.
# HAVE_TOMLLIB is checked below, just-in-time, so that mkvenv does not fail
# outside the venv or before a potential call to ensurepip in checkpip().
HAVE_TOMLLIB = True
try:
    import tomllib
except ImportError:
    try:
        import tomli as tomllib
    except ImportError:
        HAVE_TOMLLIB = False

# Do not add any mandatory dependencies from outside the stdlib:
# This script *must* be usable standalone!

DirType = Union[str, bytes, "os.PathLike[str]", "os.PathLike[bytes]"]
logger = logging.getLogger("mkvenv")


def inside_a_venv() -> bool:
    """Returns True if it is executed inside of a virtual environment."""
    return sys.prefix != sys.base_prefix


class Ouch(RuntimeError):
    """An Exception class we can't confuse with a builtin."""


class QemuEnvBuilder(venv.EnvBuilder):
    """
    An extension of venv.EnvBuilder for building QEMU's configure-time venv.

    The primary difference is that it emulates a "nested" virtual
    environment when invoked from inside of an existing virtual
    environment by including packages from the parent.  Also,
    "ensurepip" is replaced if possible with just recreating pip's
    console_scripts inside the virtual environment.

    Parameters for base class init:
      - system_site_packages: bool = False
      - clear: bool = False
      - symlinks: bool = False
      - upgrade: bool = False
      - with_pip: bool = False
      - prompt: Optional[str] = None
      - upgrade_deps: bool = False             (Since 3.9)
    """

    def __init__(self, *args: Any, **kwargs: Any) -> None:
        logger.debug("QemuEnvBuilder.__init__(...)")

        # For nested venv emulation:
        self.use_parent_packages = False
        if inside_a_venv():
            # Include parent packages only if we're in a venv and
            # system_site_packages was True.
            self.use_parent_packages = kwargs.pop(
                "system_site_packages", False
            )
            # Include system_site_packages only when the parent,
            # The venv we are currently in, also does so.
            kwargs["system_site_packages"] = sys.base_prefix in site.PREFIXES

        # ensurepip is slow: venv creation can be very fast for cases where
        # we allow the use of system_site_packages. Therefore, ensurepip is
        # replaced with our own script generation once the virtual environment
        # is setup.
        self.want_pip = kwargs.get("with_pip", False)
        if self.want_pip:
            if (
                kwargs.get("system_site_packages", False)
                and not need_ensurepip()
            ):
                kwargs["with_pip"] = False
            else:
                check_ensurepip()

        super().__init__(*args, **kwargs)

        # Make the context available post-creation:
        self._context: Optional[SimpleNamespace] = None

    def get_parent_libpath(self) -> Optional[str]:
        """Return the libpath of the parent venv, if applicable."""
        if self.use_parent_packages:
            return sysconfig.get_path("purelib")
        return None

    @staticmethod
    def compute_venv_libpath(context: SimpleNamespace) -> str:
        """
        Compatibility wrapper for context.lib_path for Python < 3.12
        """
        # Python 3.12+, not strictly necessary because it's documented
        # to be the same as 3.10 code below:
        if sys.version_info >= (3, 12):
            return context.lib_path

        # Python 3.10+
        if "venv" in sysconfig.get_scheme_names():
            lib_path = sysconfig.get_path(
                "purelib", scheme="venv", vars={"base": context.env_dir}
            )
            assert lib_path is not None
            return lib_path

        # For Python <= 3.9 we need to hardcode this. Fortunately the
        # code below was the same in Python 3.6-3.10, so there is only
        # one case.
        if sys.platform == "win32":
            return os.path.join(context.env_dir, "Lib", "site-packages")
        return os.path.join(
            context.env_dir,
            "lib",
            "python%d.%d" % sys.version_info[:2],
            "site-packages",
        )

    def ensure_directories(self, env_dir: DirType) -> SimpleNamespace:
        logger.debug("ensure_directories(env_dir=%s)", env_dir)
        self._context = super().ensure_directories(env_dir)
        return self._context

    def create(self, env_dir: DirType) -> None:
        logger.debug("create(env_dir=%s)", env_dir)
        super().create(env_dir)
        assert self._context is not None
        self.post_post_setup(self._context)

    def post_post_setup(self, context: SimpleNamespace) -> None:
        """
        The final, final hook. Enter the venv and run commands inside of it.
        """
        if self.use_parent_packages:
            # We're inside of a venv and we want to include the parent
            # venv's packages.
            parent_libpath = self.get_parent_libpath()
            assert parent_libpath is not None
            logger.debug("parent_libpath: %s", parent_libpath)

            our_libpath = self.compute_venv_libpath(context)
            logger.debug("our_libpath: %s", our_libpath)

            pth_file = os.path.join(our_libpath, "nested.pth")
            with open(pth_file, "w", encoding="UTF-8") as file:
                file.write(parent_libpath + os.linesep)

        if self.want_pip:
            args = [
                context.env_exe,
                __file__,
                "post_init",
            ]
            subprocess.run(args, check=True)

    def get_value(self, field: str) -> str:
        """
        Get a string value from the context namespace after a call to build.

        For valid field names, see:
        https://docs.python.org/3/library/venv.html#venv.EnvBuilder.ensure_directories
        """
        ret = getattr(self._context, field)
        assert isinstance(ret, str)
        return ret


def need_ensurepip() -> bool:
    """
    Tests for the presence of setuptools and pip.

    :return: `True` if we do not detect both packages.
    """
    # Don't try to actually import them, it's fraught with danger:
    # https://github.com/pypa/setuptools/issues/2993
    if find_spec("setuptools") and find_spec("pip"):
        return False
    return True


def check_ensurepip() -> None:
    """
    Check that we have ensurepip.

    Raise a fatal exception with a helpful hint if it isn't available.
    """
    if not find_spec("ensurepip"):
        msg = (
            "Python's ensurepip module is not found.\n"
            "It's normally part of the Python standard library, "
            "maybe your distribution packages it separately?\n"
            "Either install ensurepip, or alleviate the need for it in the "
            "first place by installing pip and setuptools for "
            f"'{sys.executable}'.\n"
            "(Hint: Debian puts ensurepip in its python3-venv package.)"
        )
        raise Ouch(msg)

    # ensurepip uses pyexpat, which can also go missing on us:
    if not find_spec("pyexpat"):
        msg = (
            "Python's pyexpat module is not found.\n"
            "It's normally part of the Python standard library, "
            "maybe your distribution packages it separately?\n"
            "Either install pyexpat, or alleviate the need for it in the "
            "first place by installing pip and setuptools for "
            f"'{sys.executable}'.\n\n"
            "(Hint: NetBSD's pkgsrc debundles this to e.g. 'py310-expat'.)"
        )
        raise Ouch(msg)


def make_venv(  # pylint: disable=too-many-arguments
    env_dir: Union[str, Path],
    system_site_packages: bool = False,
    clear: bool = True,
    symlinks: Optional[bool] = None,
    with_pip: bool = True,
) -> None:
    """
    Create a venv using `QemuEnvBuilder`.

    This is analogous to the `venv.create` module-level convenience
    function that is part of the Python stdblib, except it uses
    `QemuEnvBuilder` instead.

    :param env_dir: The directory to create/install to.
    :param system_site_packages:
        Allow inheriting packages from the system installation.
    :param clear: When True, fully remove any prior venv and files.
    :param symlinks:
        Whether to use symlinks to the target interpreter or not. If
        left unspecified, it will use symlinks except on Windows to
        match behavior with the "venv" CLI tool.
    :param with_pip:
        Whether to install "pip" binaries or not.
    """
    logger.debug(
        "%s: make_venv(env_dir=%s, system_site_packages=%s, "
        "clear=%s, symlinks=%s, with_pip=%s)",
        __file__,
        str(env_dir),
        system_site_packages,
        clear,
        symlinks,
        with_pip,
    )

    if symlinks is None:
        # Default behavior of standard venv CLI
        symlinks = os.name != "nt"

    builder = QemuEnvBuilder(
        system_site_packages=system_site_packages,
        clear=clear,
        symlinks=symlinks,
        with_pip=with_pip,
    )

    style = "non-isolated" if builder.system_site_packages else "isolated"
    nested = ""
    if builder.use_parent_packages:
        nested = f"(with packages from '{builder.get_parent_libpath()}') "
    print(
        f"mkvenv: Creating {style} virtual environment"
        f" {nested}at '{str(env_dir)}'",
        file=sys.stderr,
    )

    try:
        logger.debug("Invoking builder.create()")
        try:
            builder.create(str(env_dir))
        except SystemExit as exc:
            # Some versions of the venv module raise SystemExit; *nasty*!
            # We want the exception that prompted it. It might be a subprocess
            # error that has output we *really* want to see.
            logger.debug("Intercepted SystemExit from EnvBuilder.create()")
            raise exc.__cause__ or exc.__context__ or exc
        logger.debug("builder.create() finished")
    except subprocess.CalledProcessError as exc:
        logger.error("mkvenv subprocess failed:")
        logger.error("cmd: %s", exc.cmd)
        logger.error("returncode: %d", exc.returncode)

        def _stringify(data: Union[str, bytes]) -> str:
            if isinstance(data, bytes):
                return data.decode()
            return data

        lines = []
        if exc.stdout:
            lines.append("========== stdout ==========")
            lines.append(_stringify(exc.stdout))
            lines.append("============================")
        if exc.stderr:
            lines.append("========== stderr ==========")
            lines.append(_stringify(exc.stderr))
            lines.append("============================")
        if lines:
            logger.error(os.linesep.join(lines))

        raise Ouch("VENV creation subprocess failed.") from exc

    # print the python executable to stdout for configure.
    print(builder.get_value("env_exe"))


def _get_entry_points(packages: Sequence[str]) -> Iterator[str]:

    def _generator() -> Iterator[str]:
        for package in packages:
            try:
                entry_points: Iterator[EntryPoint] = \
                    iter(distribution(package).entry_points)
            except PackageNotFoundError:
                continue

            # The EntryPoints type is only available in 3.10+,
            # treat this as a vanilla list and filter it ourselves.
            entry_points = filter(
                lambda ep: ep.group == "console_scripts", entry_points
            )

            for entry_point in entry_points:
                yield f"{entry_point.name} = {entry_point.value}"

    return _generator()


def generate_console_scripts(
    packages: Sequence[str],
    python_path: Optional[str] = None,
    bin_path: Optional[str] = None,
) -> None:
    """
    Generate script shims for console_script entry points in @packages.
    """
    if python_path is None:
        python_path = sys.executable
    if bin_path is None:
        bin_path = sysconfig.get_path("scripts")
        assert bin_path is not None

    logger.debug(
        "generate_console_scripts(packages=%s, python_path=%s, bin_path=%s)",
        packages,
        python_path,
        bin_path,
    )

    if not packages:
        return

    maker = distlib.scripts.ScriptMaker(None, bin_path)
    maker.variants = {""}
    maker.clobber = False

    for entry_point in _get_entry_points(packages):
        for filename in maker.make(entry_point):
            logger.debug("wrote console_script '%s'", filename)


def pkgname_from_depspec(dep_spec: str) -> str:
    """
    Parse package name out of a PEP-508 depspec.

    See https://peps.python.org/pep-0508/#names
    """
    match = re.match(
        r"^([A-Z0-9]([A-Z0-9._-]*[A-Z0-9])?)", dep_spec, re.IGNORECASE
    )
    if not match:
        raise ValueError(
            f"dep_spec '{dep_spec}'"
            " does not appear to contain a valid package name"
        )
    return match.group(0)


def _path_is_prefix(prefix: Optional[str], path: str) -> bool:
    try:
        return (
            prefix is not None and os.path.commonpath([prefix, path]) == prefix
        )
    except ValueError:
        return False


def _is_system_package(dist: Distribution) -> bool:
    path = str(dist.locate_file("."))
    return not (
        _path_is_prefix(sysconfig.get_path("purelib"), path)
        or _path_is_prefix(sysconfig.get_path("platlib"), path)
    )


def diagnose(
    dep_spec: str,
    online: bool,
    wheels_dir: Optional[Union[str, Path]],
    prog: Optional[str],
) -> Tuple[str, bool]:
    """
    Offer a summary to the user as to why a package failed to be installed.

    :param dep_spec: The package we tried to ensure, e.g. 'meson>=0.61.5'
    :param online: Did we allow PyPI access?
    :param prog:
        Optionally, a shell program name that can be used as a
        bellwether to detect if this program is installed elsewhere on
        the system. This is used to offer advice when a program is
        detected for a different python version.
    :param wheels_dir:
        Optionally, a directory that was searched for vendored packages.
    """
    # pylint: disable=too-many-branches

    # Some errors are not particularly serious
    bad = False

    pkg_name = pkgname_from_depspec(dep_spec)
    pkg_version: Optional[str] = None
    try:
        pkg_version = version(pkg_name)
    except PackageNotFoundError:
        pass

    lines = []

    if pkg_version:
        lines.append(
            f"Python package '{pkg_name}' version '{pkg_version}' was found,"
            " but isn't suitable."
        )
    else:
        lines.append(
            f"Python package '{pkg_name}' was not found nor installed."
        )

    if wheels_dir:
        lines.append(
            "No suitable version found in, or failed to install from"
            f" '{wheels_dir}'."
        )
        bad = True

    if online:
        lines.append("A suitable version could not be obtained from PyPI.")
        bad = True
    else:
        lines.append(
            "mkvenv was configured to operate offline and did not check PyPI."
        )

    if prog and not pkg_version:
        which = shutil.which(prog)
        if which:
            if sys.base_prefix in site.PREFIXES:
                pypath = Path(sys.executable).resolve()
                lines.append(
                    f"'{prog}' was detected on your system at '{which}', "
                    f"but the Python package '{pkg_name}' was not found by "
                    f"this Python interpreter ('{pypath}'). "
                    f"Typically this means that '{prog}' has been installed "
                    "against a different Python interpreter on your system."
                )
            else:
                lines.append(
                    f"'{prog}' was detected on your system at '{which}', "
                    "but the build is using an isolated virtual environment."
                )
            bad = True

    lines = [f" • {line}" for line in lines]
    if bad:
        lines.insert(0, f"Could not provide build dependency '{dep_spec}':")
    else:
        lines.insert(0, f"'{dep_spec}' not found:")
    return os.linesep.join(lines), bad


def pip_install(
    args: Sequence[str],
    online: bool = False,
    wheels_dir: Optional[Union[str, Path]] = None,
) -> None:
    """
    Use pip to install a package or package(s) as specified in @args.
    """
    loud = bool(
        os.environ.get("DEBUG")
        or os.environ.get("GITLAB_CI")
        or os.environ.get("V")
    )

    full_args = [
        sys.executable,
        "-m",
        "pip",
        "install",
        "--disable-pip-version-check",
        "-v" if loud else "-q",
    ]
    if not online:
        full_args += ["--no-index"]
    if wheels_dir:
        full_args += ["--find-links", f"file://{str(wheels_dir)}"]
    full_args += list(args)
    subprocess.run(
        full_args,
        check=True,
    )


def _make_version_constraint(info: Dict[str, str], install: bool) -> str:
    """
    Construct the version constraint part of a PEP 508 dependency
    specification (for example '>=0.61.5') from the accepted and
    installed keys of the provided dictionary.

    :param info: A dictionary corresponding to a TOML key-value list.
    :param install: True generates install constraints, False generates
        presence constraints
    """
    if install and "installed" in info:
        return "==" + info["installed"]

    dep_spec = info.get("accepted", "")
    dep_spec = dep_spec.strip()
    # Double check that they didn't just use a version number
    if dep_spec and dep_spec[0] not in "!~><=(":
        raise Ouch(
            "invalid dependency specifier " + dep_spec + " in dependency file"
        )

    return dep_spec


def _do_ensure(
    group: Dict[str, Dict[str, str]],
    online: bool = False,
    wheels_dir: Optional[Union[str, Path]] = None,
) -> Optional[Tuple[str, bool]]:
    """
    Use pip to ensure we have the packages specified in @group.

    If the packages are already installed, do nothing. If online and
    wheels_dir are both provided, prefer packages found in wheels_dir
    first before connecting to PyPI.

    :param group: A dictionary of dictionaries, corresponding to a
        section in a pythondeps.toml file.
    :param online: If True, fall back to PyPI.
    :param wheels_dir: If specified, search this path for packages.
    """
    absent = []
    present = []
    canary = None
    for name, info in group.items():
        constraint = _make_version_constraint(info, False)
        matcher = distlib.version.LegacyMatcher(name + constraint)
        print(f"mkvenv: checking for {matcher}", file=sys.stderr)

        dist: Optional[Distribution] = None
        try:
            dist = distribution(matcher.name)
        except PackageNotFoundError:
            pass

        if (
            dist is None
            # Always pass installed package to pip, so that they can be
            # updated if the requested version changes
            or not _is_system_package(dist)
            or not matcher.match(distlib.version.LegacyVersion(dist.version))
        ):
            absent.append(name + _make_version_constraint(info, True))
            if len(absent) == 1:
                canary = info.get("canary", None)
        else:
            logger.info("found %s %s", name, dist.version)
            present.append(name)

    if present:
        generate_console_scripts(present)

    if absent:
        if online or wheels_dir:
            # Some packages are missing or aren't a suitable version,
            # install a suitable (possibly vendored) package.
            print(f"mkvenv: installing {', '.join(absent)}", file=sys.stderr)
            try:
                pip_install(args=absent, online=online, wheels_dir=wheels_dir)
                return None
            except subprocess.CalledProcessError:
                pass

        return diagnose(
            absent[0],
            online,
            wheels_dir,
            canary,
        )

    return None


def ensure(
    dep_specs: Sequence[str],
    online: bool = False,
    wheels_dir: Optional[Union[str, Path]] = None,
    prog: Optional[str] = None,
) -> None:
    """
    Use pip to ensure we have the package specified by @dep_specs.

    If the package is already installed, do nothing. If online and
    wheels_dir are both provided, prefer packages found in wheels_dir
    first before connecting to PyPI.

    :param dep_specs:
        PEP 508 dependency specifications. e.g. ['meson>=0.61.5'].
    :param online: If True, fall back to PyPI.
    :param wheels_dir: If specified, search this path for packages.
    :param prog:
        If specified, use this program name for error diagnostics that will
        be presented to the user. e.g., 'sphinx-build' can be used as a
        bellwether for the presence of 'sphinx'.
    """

    if not HAVE_DISTLIB:
        raise Ouch("a usable distlib could not be found, please install it")

    # Convert the depspecs to a dictionary, as if they came
    # from a section in a pythondeps.toml file
    group: Dict[str, Dict[str, str]] = {}
    for spec in dep_specs:
        name = distlib.version.LegacyMatcher(spec).name
        group[name] = {}

        spec = spec.strip()
        pos = len(name)
        ver = spec[pos:].strip()
        if ver:
            group[name]["accepted"] = ver

        if prog:
            group[name]["canary"] = prog
            prog = None

    result = _do_ensure(group, online, wheels_dir)
    if result:
        # Well, that's not good.
        if result[1]:
            raise Ouch(result[0])
        raise SystemExit(f"\n{result[0]}\n\n")


def _parse_groups(file: str) -> Dict[str, Dict[str, Any]]:
    if not HAVE_TOMLLIB:
        if sys.version_info < (3, 11):
            raise Ouch("found no usable tomli, please install it")

        raise Ouch(
            "Python >=3.11 does not have tomllib... what have you done!?"
        )

    # Use loads() to support both tomli v1.2.x (Ubuntu 22.04,
    # Debian bullseye-backports) and v2.0.x
    with open(file, "r", encoding="ascii") as depfile:
        contents = depfile.read()
        return tomllib.loads(contents)  # type: ignore


def ensure_group(
    file: str,
    groups: Sequence[str],
    online: bool = False,
    wheels_dir: Optional[Union[str, Path]] = None,
) -> None:
    """
    Use pip to ensure we have the package specified by @dep_specs.

    If the package is already installed, do nothing. If online and
    wheels_dir are both provided, prefer packages found in wheels_dir
    first before connecting to PyPI.

    :param dep_specs:
        PEP 508 dependency specifications. e.g. ['meson>=0.61.5'].
    :param online: If True, fall back to PyPI.
    :param wheels_dir: If specified, search this path for packages.
    """

    if not HAVE_DISTLIB:
        raise Ouch("found no usable distlib, please install it")

    parsed_deps = _parse_groups(file)

    to_install: Dict[str, Dict[str, str]] = {}
    for group in groups:
        try:
            to_install.update(parsed_deps[group])
        except KeyError as exc:
            raise Ouch(f"group {group} not defined") from exc

    result = _do_ensure(to_install, online, wheels_dir)
    if result:
        # Well, that's not good.
        if result[1]:
            raise Ouch(result[0])
        raise SystemExit(f"\n{result[0]}\n\n")


def post_venv_setup() -> None:
    """
    This is intended to be run *inside the venv* after it is created.
    """
    logger.debug("post_venv_setup()")
    # Generate a 'pip' script so the venv is usable in a normal
    # way from the CLI. This only happens when we inherited pip from a
    # parent/system-site and haven't run ensurepip in some way.
    generate_console_scripts(["pip"])


def _add_create_subcommand(subparsers: Any) -> None:
    subparser = subparsers.add_parser("create", help="create a venv")
    subparser.add_argument(
        "target",
        type=str,
        action="store",
        help="Target directory to install virtual environment into.",
    )


def _add_post_init_subcommand(subparsers: Any) -> None:
    subparsers.add_parser("post_init", help="post-venv initialization")


def _add_ensuregroup_subcommand(subparsers: Any) -> None:
    subparser = subparsers.add_parser(
        "ensuregroup",
        help="Ensure that the specified package group is installed.",
    )
    subparser.add_argument(
        "--online",
        action="store_true",
        help="Install packages from PyPI, if necessary.",
    )
    subparser.add_argument(
        "--dir",
        type=str,
        action="store",
        help="Path to vendored packages where we may install from.",
    )
    subparser.add_argument(
        "file",
        type=str,
        action="store",
        help=("Path to a TOML file describing package groups"),
    )
    subparser.add_argument(
        "group",
        type=str,
        action="store",
        help="One or more package group names",
        nargs="+",
    )


def _add_ensure_subcommand(subparsers: Any) -> None:
    subparser = subparsers.add_parser(
        "ensure", help="Ensure that the specified package is installed."
    )
    subparser.add_argument(
        "--online",
        action="store_true",
        help="Install packages from PyPI, if necessary.",
    )
    subparser.add_argument(
        "--dir",
        type=str,
        action="store",
        help="Path to vendored packages where we may install from.",
    )
    subparser.add_argument(
        "--diagnose",
        type=str,
        action="store",
        help=(
            "Name of a shell utility to use for "
            "diagnostics if this command fails."
        ),
    )
    subparser.add_argument(
        "dep_specs",
        type=str,
        action="store",
        help="PEP 508 Dependency specification, e.g. 'meson>=0.61.5'",
        nargs="+",
    )


def main() -> int:
    """CLI interface to make_qemu_venv. See module docstring."""
    if os.environ.get("DEBUG") or os.environ.get("GITLAB_CI"):
        # You're welcome.
        logging.basicConfig(level=logging.DEBUG)
    else:
        if os.environ.get("V"):
            logging.basicConfig(level=logging.INFO)

    parser = argparse.ArgumentParser(
        prog="mkvenv",
        description="QEMU pyvenv bootstrapping utility",
    )
    subparsers = parser.add_subparsers(
        title="Commands",
        dest="command",
        required=True,
        metavar="command",
        help="Description",
    )

    _add_create_subcommand(subparsers)
    _add_post_init_subcommand(subparsers)
    _add_ensure_subcommand(subparsers)
    _add_ensuregroup_subcommand(subparsers)

    args = parser.parse_args()
    try:
        if args.command == "create":
            make_venv(
                args.target,
                system_site_packages=True,
                clear=True,
            )
        if args.command == "post_init":
            post_venv_setup()
        if args.command == "ensure":
            ensure(
                dep_specs=args.dep_specs,
                online=args.online,
                wheels_dir=args.dir,
                prog=args.diagnose,
            )
        if args.command == "ensuregroup":
            ensure_group(
                file=args.file,
                groups=args.group,
                online=args.online,
                wheels_dir=args.dir,
            )
        logger.debug("mkvenv.py %s: exiting", args.command)
    except Ouch as exc:
        print("\n*** Ouch! ***\n", file=sys.stderr)
        print(str(exc), "\n\n", file=sys.stderr)
        return 1
    except SystemExit:
        raise
    except:  # pylint: disable=bare-except
        logger.exception("mkvenv did not complete successfully:")
        return 2
    return 0


if __name__ == "__main__":
    sys.exit(main())