Start testing some features and changes for the cli in Conan V2

conans/cli/command.py

@@ -0,0 +1,276 @@
+import argparse
+import inspect
+import signal
+import sys
+import textwrap
+from difflib import get_close_matches
+
+
+from conans import __version__ as client_version
+from conans.client.conan_api_v2 import Conan
+from conans.client.output import Color
+from conans.errors import ConanException, ConanInvalidConfiguration,  ConanMigrationError
+from conans.util.files import exception_message_safe
+from conans.util.log import logger
+from conans.client.outputers import OutputerFormats
+
+
+# Exit codes for conan command:
+SUCCESS = 0                         # 0: Success (done)
+ERROR_GENERAL = 1                   # 1: General ConanException error (done)
+ERROR_MIGRATION = 2                 # 2: Migration error
+USER_CTRL_C = 3                     # 3: Ctrl+C
+USER_CTRL_BREAK = 4                 # 4: Ctrl+Break
+ERROR_SIGTERM = 5                   # 5: SIGTERM
+ERROR_INVALID_CONFIGURATION = 6     # 6: Invalid configuration (done)
+
+
+class OnceArgument(argparse.Action):
+    """Allows declaring a parameter that can have only one value, by default argparse takes the
+    latest declared and it's very confusing.
+    """
+    def __call__(self, parser, namespace, values, option_string=None):
+        if getattr(namespace, self.dest) is not None and self.default is None:
+            msg = '{o} can only be specified once'.format(o=option_string)
+            raise argparse.ArgumentError(None, msg)
+        setattr(namespace, self.dest, values)
+
+
+class SmartFormatter(argparse.HelpFormatter):
+
+    def _fill_text(self, text, width, indent):
+        text = textwrap.dedent(text)
+        return ''.join(indent + line for line in text.splitlines(True))
+
+
+class Command(object):
+    """A single command of the conan application, with all the first level commands. Manages the
+    parsing of parameters and delegates functionality to the conan python api. It can also show the
+    help of the tool.
+    """
+    def __init__(self, conan_api):
+        assert isinstance(conan_api, Conan)
+        self._conan = conan_api
+        self._out = conan_api.out
+
+    def help(self, *args):
+        """
+        Shows help for a specific command.
+        """
+        parser = argparse.ArgumentParser(description=self.help.__doc__, prog="conan help",
+                                         formatter_class=SmartFormatter)
+        parser.add_argument("command", help='command', nargs="?")
+        args = parser.parse_args(*args)
+        if not args.command:
+            self._show_help()
+            return
+        try:
+            commands = self._commands()
+            method = commands[args.command]
+            method(["--help"])
+        except KeyError:
+            raise ConanException("Unknown command '%s'" % args.command)
+
+    # conan v2 search:
+    # conan search "*" will search in the default remote
+    # to search in the local cache: conan search "*" --cache explicitly
+
+    def search(self, *args):
+        """
+        Searches for package recipes whose name contain <query> in a remote or in the local cache
+        """
+        parser = argparse.ArgumentParser(description=self.search.__doc__, prog="conan search",
+                                         formatter_class=SmartFormatter)
+        parser.add_argument('query',
+                            help="Search query to find package recipe reference, e.g., 'boost', 'lib*'")
+
+        exclusive_args = parser.add_mutually_exclusive_group()
+        exclusive_args.add_argument('-r', '--remote', default=None, action=OnceArgument, nargs='?',
+                                    help="Remote to search. Accepts wildcards. To search in all remotes use *")
+        exclusive_args.add_argument('-c', '--cache', action="store_true", help="Search in the local cache")
+        parser.add_argument('-o', '--output', default="cli", action=OnceArgument,
+                            help="Select the output format: json, html,...")
+        args = parser.parse_args(*args)
+
+        out_kwargs = {'out': self._out, 'f': 'search'}
+        info = self._conan.search_recipes(args.query, remote_pattern=args.remote)
+        OutputerFormats.get(args.output).out(info=info, **out_kwargs)
+
+    def _show_help(self):
+        """
+        Prints a summary of all commands.
+        """
+        grps = [("Consumer commands", ("search", )),
+                ("Misc commands", ("help", ))]
+
+        def check_all_commands_listed():
+            """Keep updated the main directory, raise if don't"""
+            all_commands = self._commands()
+            all_in_grps = [command for _, command_list in grps for command in command_list]
+            if set(all_in_grps) != set(all_commands):
+                diff = set(all_commands) - set(all_in_grps)
+                raise Exception("Some command is missing in the main help: %s" % ",".join(diff))
+            return all_commands
+
+        commands = check_all_commands_listed()
+        max_len = max((len(c) for c in commands)) + 1
+        fmt = '  %-{}s'.format(max_len)
+
+        for group_name, comm_names in grps:
+            self._out.writeln(group_name, Color.BRIGHT_MAGENTA)
+            for name in comm_names:
+                # future-proof way to ensure tabular formatting
+                self._out.write(fmt % name, Color.GREEN)
+
+                # Help will be all the lines up to the first empty one
+                docstring_lines = commands[name].__doc__.split('\n')
+                start = False
+                data = []
+                for line in docstring_lines:
+                    line = line.strip()
+                    if not line:
+                        if start:
+                            break
+                        start = True
+                        continue
+                    data.append(line)
+
+                txt = textwrap.fill(' '.join(data), 80, subsequent_indent=" "*(max_len+2))
+                self._out.writeln(txt)
+
+        self._out.writeln("")
+        self._out.writeln('Conan commands. Type "conan <command> -h" for help', Color.BRIGHT_YELLOW)
+
+    def _commands(self):
+        """ Returns a list of available commands.
+        """
+        result = {}
+        for m in inspect.getmembers(self, predicate=inspect.ismethod):
+            method_name = m[0]
+            if not method_name.startswith('_'):
+                if "export_pkg" == method_name:
+                    method_name = "export-pkg"
+                method = m[1]
+                if method.__doc__ and not method.__doc__.startswith('HIDDEN'):
+                    result[method_name] = method
+        return result
+
+    def _print_similar(self, command):
+        """ Looks for similar commands and prints them if found.
+        """
+        matches = get_close_matches(
+            word=command, possibilities=self._commands().keys(), n=5, cutoff=0.75)
+
+        if len(matches) == 0:
+            return
+
+        if len(matches) > 1:
+            self._out.writeln("The most similar commands are")
+        else:
+            self._out.writeln("The most similar command is")
+
+        for match in matches:
+            self._out.writeln("    %s" % match)
+
+        self._out.writeln("")
+
+    def run(self, *args):
+        """HIDDEN: entry point for executing commands, dispatcher to class
+        methods
+        """
+        version = sys.version_info
+        if version.major == 2 or version.minor <= 4:
+            raise ConanException("Unsupported Python version")
+
+        ret_code = SUCCESS
+        try:
+            try:
+                command = args[0][0]
+            except IndexError:  # No parameters
+                self._show_help()
+                return False
+            try:
+                commands = self._commands()
+                method = commands[command]
+            except KeyError as exc:
+                if command in ["-v", "--version"]:
+                    self._out.success("Conan version %s" % client_version)
+                    return False
+
+                if command in ["-h", "--help"]:
+                    self._show_help()
+                    return False
+
+                self._out.writeln("'%s' is not a Conan command. See 'conan --help'." % command)
+                self._out.writeln("")
+                self._print_similar(command)
+                raise ConanException("Unknown command %s" % str(exc))
+
+            method(args[0][1:])
+        except KeyboardInterrupt as exc:
+            logger.error(exc)
+            ret_code = SUCCESS
+        except SystemExit as exc:
+            if exc.code != 0:
+                logger.error(exc)
+                self._out.error("Exiting with code: %d" % exc.code)
+            ret_code = exc.code
+        except ConanInvalidConfiguration as exc:
+            ret_code = ERROR_INVALID_CONFIGURATION
+            self._out.error(exc)
+        except ConanException as exc:
+            ret_code = ERROR_GENERAL
+            self._out.error(exc)
+        except Exception as exc:
+            import traceback
+            print(traceback.format_exc())
+            ret_code = ERROR_GENERAL
+            msg = exception_message_safe(exc)
+            self._out.error(msg)
+
+        return ret_code
+
+
+def main(args):
+    """ main entry point of the conan application, using a Command to
+    parse parameters
+
+    Exit codes for conan command:
+
+        0: Success (done)
+        1: General ConanException error (done)
+        2: Migration error
+        3: Ctrl+C
+        4: Ctrl+Break
+        5: SIGTERM
+        6: Invalid configuration (done)
+    """
+    try:
+        conan_api, _, _ = Conan.factory()
+    except ConanMigrationError:  # Error migrating
+        sys.exit(ERROR_MIGRATION)
+    except ConanException as e:
+        sys.stderr.write("Error in Conan initialization: {}".format(e))
+        sys.exit(ERROR_GENERAL)
+
+    def ctrl_c_handler(_, __):
+        print('You pressed Ctrl+C!')
+        sys.exit(USER_CTRL_C)
+
+    def sigterm_handler(_, __):
+        print('Received SIGTERM!')
+        sys.exit(ERROR_SIGTERM)
+
+    def ctrl_break_handler(_, __):
+        print('You pressed Ctrl+Break!')
+        sys.exit(USER_CTRL_BREAK)
+
+    signal.signal(signal.SIGINT, ctrl_c_handler)
+    signal.signal(signal.SIGTERM, sigterm_handler)
+
+    if sys.platform == 'win32':
+        signal.signal(signal.SIGBREAK, ctrl_break_handler)
+
+    command = Command(conan_api)
+    error = command.run(args)
+    sys.exit(error)

conans/client/cmd/search_v2.py

@@ -0,0 +1,113 @@
+import fnmatch
+from collections import OrderedDict, namedtuple
+
+from conans.errors import NotFoundException, ConanException, NoRemoteAvailable
+from conans.search.search import (filter_outdated, search_packages, search_recipes,
+                                  filter_by_revision)
+
+
+class Search(object):
+    def __init__(self, cache, remote_manager, remotes):
+        self._cache = cache
+        self._remote_manager = remote_manager
+        self._remotes = remotes
+
+    def search_recipes(self, query, remote_pattern=None):
+        ignorecase = True
+
+        references = OrderedDict()
+        if remote_pattern is None:
+            references[None] = search_recipes(self._cache, query, ignorecase)
+            return references
+
+        for remote in self._remotes.values():
+            if fnmatch.fnmatch(remote.name, remote_pattern):
+                refs = self._remote_manager.search_recipes(remote, query, ignorecase)
+                if refs:
+                    references[remote.name] = sorted(refs)
+        return references
+
+    remote_ref = namedtuple('remote_ref', 'ordered_packages recipe_hash')
+
+    def search_packages(self, ref=None, remote_name=None, query=None, outdated=False):
+        """ Return the single information saved in conan.vars about all the packages
+            or the packages which match with a pattern
+
+            Attributes:
+                pattern = string to match packages
+                remote_name = search on another origin to get packages info
+                packages_pattern = String query with binary
+                                   packages properties: "arch=x86 AND os=Windows"
+        """
+        if not remote_name:
+            return self._search_packages_in_local(ref, query, outdated)
+
+        if ref.revision and not self._cache.config.revisions_enabled:
+            raise ConanException("Revisions not enabled in the client, specify a "
+                                 "reference without revision")
+
+        if remote_name == 'all':
+            return self._search_packages_in_all(ref, query, outdated)
+
+        return self._search_packages_in(remote_name, ref, query, outdated)
+
+    def _search_packages_in_local(self, ref=None, query=None, outdated=False):
+        package_layout = self._cache.package_layout(ref, short_paths=None)
+        packages_props = search_packages(package_layout, query)
+        ordered_packages = OrderedDict(sorted(packages_props.items()))
+
+        try:
+            recipe_hash = package_layout.recipe_manifest().summary_hash
+        except IOError:  # It could not exist in local
+            recipe_hash = None
+
+        if outdated:
+            ordered_packages = filter_outdated(ordered_packages, recipe_hash)
+        elif self._cache.config.revisions_enabled:
+            # With revisions, by default filter the packages not belonging to the recipe
+            # unless outdated is specified.
+            metadata = package_layout.load_metadata()
+            ordered_packages = filter_by_revision(metadata, ordered_packages)
+
+        references = OrderedDict()
+        references[None] = self.remote_ref(ordered_packages, recipe_hash)
+        return references
+
+    def _search_packages_in_all(self, ref=None, query=None, outdated=False):
+        references = OrderedDict()
+        # We have to check if there is a remote called "all"
+        # Deprecate: 2.0 can remove this check
+        if 'all' not in self._remotes:
+            for remote in self._remotes.values():
+                try:
+                    packages_props = self._remote_manager.search_packages(remote, ref, query)
+                    if packages_props:
+                        ordered_packages = OrderedDict(sorted(packages_props.items()))
+                        manifest, _ = self._remote_manager.get_recipe_manifest(ref, remote)
+
+                        recipe_hash = manifest.summary_hash
+
+                        if outdated and recipe_hash:
+                            ordered_packages = filter_outdated(ordered_packages, recipe_hash)
+
+                        references[remote.name] = self.remote_ref(ordered_packages, recipe_hash)
+                except NotFoundException:
+                    continue
+            return references
+
+        return self._search_packages_in('all', ref, query, outdated)
+
+    def _search_packages_in(self, remote_name, ref=None, query=None, outdated=False):
+        remote = self._remotes[remote_name]
+        packages_props = self._remote_manager.search_packages(remote, ref, query)
+        ordered_packages = OrderedDict(sorted(packages_props.items()))
+        manifest, ref = self._remote_manager.get_recipe_manifest(ref, remote)
+
+        recipe_hash = manifest.summary_hash
+
+        if outdated and recipe_hash:
+            ordered_packages = filter_outdated(ordered_packages, recipe_hash)
+
+        references = OrderedDict()
+        references[remote.name] = self.remote_ref(ordered_packages, recipe_hash)
+        return references