files: import pytest
  import shutil
  
  import tmt
  import tmt.beakerlib
  
  
  @pytest.mark.web
  def test_library():
      """ Fetch a beakerlib library with/without providing a parent """
      parent = tmt.utils.Common(workdir=True)
      library_with_parent = tmt.beakerlib.Library(
          'library(openssl/certgen)', parent=parent)
      library_without_parent = tmt.beakerlib.Library(
          'library(openssl/certgen)')
  
      for library in [library_with_parent, library_without_parent]:
          assert library.format == 'rpm'
          assert library.repo == 'openssl'
          assert library.url == 'https://github.com/beakerlib/openssl'
          assert library.ref == 'master' # The default branch is master
          assert library.dest == tmt.beakerlib.DEFAULT_DESTINATION
          shutil.rmtree(library.parent.workdir)
  
  
  @pytest.mark.web
  @pytest.mark.parametrize(
          'url, name, default_branch', [
          ('https://github.com/beakerlib/httpd', '/http', 'master'),
          ('https://github.com/beakerlib/example', '/file', 'main')
          ])
  def test_library_from_fmf(url, name, default_branch):
      """ Fetch beakerlib library referenced by fmf identifier """
      library = tmt.beakerlib.Library(dict(url=url, name=name))
      assert library.format == 'fmf'
      assert library.ref == default_branch
      assert library.url == url
      assert library.dest == tmt.beakerlib.DEFAULT_DESTINATION
      assert library.repo == url.split('/')[-1]
      assert library.name == name
      shutil.rmtree(library.parent.workdir)
  
  
  @pytest.mark.web
  def test_dependencies():
      """ Check requires for possible libraries """
      parent = tmt.utils.Common(workdir=True)
      requires, recommends, libraries = tmt.beakerlib.dependencies(
          ['library(httpd/http)', 'wget'], ['forest'], parent=parent)
      # Check for correct requires and recommends
      for require in ['httpd', 'lsof', 'mod_ssl']:
          assert require in requires
          assert require in libraries[0].require
      assert 'openssl' in libraries[2].require
      assert 'forest' in recommends
      assert 'wget' in requires
      # Library require should be in httpd requires but not in the final result
      assert 'library(openssl/certgen)' in libraries[0].require
      assert 'library(openssl/certgen)' not in requires
      # Check library attributes for sane values
      assert libraries[0].repo == 'httpd'
      assert libraries[0].name == '/http'
      assert libraries[0].url == 'https://github.com/beakerlib/httpd'
      assert libraries[0].ref == 'master' # The default branch is master
      assert libraries[0].dest == tmt.beakerlib.DEFAULT_DESTINATION
      assert libraries[1].repo == 'openssl'
      assert libraries[1].name == '/certgen'
      shutil.rmtree(parent.workdir)
  
files: tests/unit/test_beakerlib.py
files: import pytest
  import shutil
  
  import tmt
  import tmt.beakerlib
  
  
  @pytest.mark.web
  def test_library():
      """ Fetch a beakerlib library with/without providing a parent """
      parent = tmt.utils.Common(workdir=True)
      library_with_parent = tmt.beakerlib.Library(
          'library(openssl/certgen)', parent=parent)
      library_without_parent = tmt.beakerlib.Library(
          'library(openssl/certgen)')
  
      for library in [library_with_parent, library_without_parent]:
          assert library.format == 'rpm'
          assert library.repo == 'openssl'
          assert library.url == 'https://github.com/beakerlib/openssl'
          assert library.ref == 'master' # default branch is called master
          assert library.dest == tmt.beakerlib.DEFAULT_DESTINATION
          shutil.rmtree(library.parent.workdir)
  
  
  @pytest.mark.web
  def test_library_from_fmf():
      """ Fetch beakerlib library referenced by fmf identifier """
      library = tmt.beakerlib.Library(
          {
              'url': 'https://github.com/beakerlib/httpd',
              'name': '/http'
          }
      )
      assert library.format == 'fmf'
      assert library.ref == 'master' # default branch is called master
      assert library.url == 'https://github.com/beakerlib/httpd'
      assert library.dest == tmt.beakerlib.DEFAULT_DESTINATION
      assert library.repo == 'httpd'
      assert library.name == '/http'
      shutil.rmtree(library.parent.workdir)
  
  
  @pytest.mark.web
  def test_dependencies():
      """ Check requires for possible libraries """
      parent = tmt.utils.Common(workdir=True)
      requires, recommends, libraries = tmt.beakerlib.dependencies(
          ['library(httpd/http)', 'wget'], ['forest'], parent=parent)
      # Check for correct requires and recommends
      for require in ['httpd', 'lsof', 'mod_ssl']:
          assert require in requires
          assert require in libraries[0].require
      assert 'openssl' in libraries[2].require
      assert 'forest' in recommends
      assert 'wget' in requires
      # Library require should be in httpd requires but not in the final result
      assert 'library(openssl/certgen)' in libraries[0].require
      assert 'library(openssl/certgen)' not in requires
      # Check library attributes for sane values
      assert libraries[0].repo == 'httpd'
      assert libraries[0].name == '/http'
      assert libraries[0].url == 'https://github.com/beakerlib/httpd'
      assert libraries[0].ref == 'master' # default branch is called master
      assert libraries[0].dest == tmt.beakerlib.DEFAULT_DESTINATION
      assert libraries[1].repo == 'openssl'
      assert libraries[1].name == '/certgen'
      shutil.rmtree(parent.workdir)
  
files: tests/unit/test_beakerlib.py
files: """ Handle BeakerLib Libraries """
  
  import re
  import os
  
  import fmf
  import tmt
  
  # Regular expressions for beakerlib libraries
  LIBRARY_REGEXP = re.compile(r'^library\(([^/]+)(/[^)]+)\)$')
  
  # Default beakerlib libraries location and destination directory
  DEFAULT_REPOSITORY = 'https://github.com/beakerlib'
  DEFAULT_DESTINATION = 'libs'
  
  
  class LibraryError(Exception):
      """ Used when library cannot be parsed from the identifier """
  
  
  class Library(object):
      """
      A beakerlib library
  
      Takes care of fetching beakerlib libraries from remote repositories
      based on provided library identifier described in detail here:
      https://tmt.readthedocs.io/en/latest/spec/tests.html#require
  
      Optional 'parent' object inheriting from tmt.utils.Common can be
      provided in order to share the cache of already fetched libraries.
  
      The following attributes are available in the object:
  
      repo ........ library prefix (git repository name or nick if provided)
      name ........ library suffix (folder containing the library code)
  
      url ......... full git repository url
      ref ......... git revision (branch, tag or commit)
      dest ........ target folder into which the library repo is cloned
  
      tree ........ fmf tree holding library metadata
      require ..... list of required packages
      recommend ... list of recommended packages
  
      Libraries are fetched into the 'libs' directory under parent's
      workdir or into 'destination' if provided in the identifier.
      """
  
      def __init__(self, identifier, parent=None):
          """ Process the library identifier and fetch the library """
          # Use an empty common class if parent not provided (for logging, cache)
          self.parent = parent or tmt.utils.Common(workdir=True)
  
          # Default branch is detected from the origin after cloning
          self.default_branch = None
  
          # The 'library(repo/lib)' format
          if isinstance(identifier, str):
              identifier = identifier.strip()
              matched = LIBRARY_REGEXP.search(identifier)
              if not matched:
                  raise LibraryError
              self.parent.debug(f"Detected library '{identifier}'.", level=3)
              self.format = 'rpm'
              self.repo, self.name = matched.groups()
              self.url = os.path.join(DEFAULT_REPOSITORY, self.repo)
              self.ref = None
              self.dest = DEFAULT_DESTINATION
  
          # The fmf identifier
          elif isinstance(identifier, dict):
              self.parent.debug(f"Detected library '{identifier}'.", level=3)
              self.format = 'fmf'
              self.url = identifier.get('url')
              self.ref = identifier.get('ref', None)
              self.dest = identifier.get(
                  'destination', DEFAULT_DESTINATION).lstrip('/')
              self.name = identifier.get('name', '/')
              if not self.name.startswith('/'):
                  raise tmt.utils.SpecificationError(
                      f"Library name '{self.name}' does not start with a '/'.")
              # Use provided repository nick name or parse it from the url
              try:
                  self.repo = identifier.get('nick') or re.search(
                      r'/([^/]+?)(/|\.git)?$', self.url).group(1)
              except AttributeError:
                  raise tmt.utils.GeneralError(
                      f"Unable to parse repository name from '{self.url}'.")
          # Something weird
          else:
              raise LibraryError
  
          # Fetch the library
          try:
              self.fetch()
          except fmf.utils.RootError:
              raise tmt.utils.SpecificationError(
                  f"Repository '{self.url}' does not contain fmf metadata.")
  
      def __str__(self):
          """ Use repo/name for string representation """
          return f"{self.repo}{self.name}"
  
      def fetch(self):
          """ Fetch the library (unless already fetched) """
          # Initialize library cache (indexed by the repository name)
          if not hasattr(self.parent, '_library_cache'):
              self.parent._library_cache = dict()
  
          # Check if the library was already fetched
          try:
              library = self.parent._library_cache[self.repo]
              # The url must be identical
              if library.url != self.url:
                  raise tmt.utils.GeneralError(
                      f"Library '{self.repo}' with url '{self.url}' conflicts "
                      f"with already fetched library from '{library.url}'.")
              # Use the default branch if no ref provided
              if self.ref is None:
                  self.ref = library.default_branch
              # The same ref has to be used
              if library.ref != self.ref:
                  raise tmt.utils.GeneralError(
                      f"Library '{self.repo}' using ref '{self.ref}' conflicts "
                      f"with already fetched library using ref '{library.ref}'.")
              self.parent.debug(f"Library '{self}' already fetched.", level=3)
              # Reuse the existing metadata tree
              self.tree = library.tree
          # Fetch the library and add it to the index
          except KeyError:
              self.parent.debug(f"Fetch library '{self}'.", level=3)
              # Prepare path, clone the repository, checkout ref
              directory = os.path.join(self.parent.workdir, self.dest, self.repo)
              # Clone repo with disabled prompt to ignore missing/private repos
              try:
                  self.parent.run(
                      ['git', 'clone', self.url, directory],
                      shell=False, env={"GIT_ASKPASS": "echo"})
                  # Detect the default branch from the origin
                  # The ref format is 'ref: refs/remotes/origin/master'
                  head = os.path.join(directory, '.git/refs/remotes/origin/HEAD')
                  with open(head) as ref:
                      self.default_branch = ref.read().strip().split('/')[-1]
                  # Use the default branch if no ref provided
                  if self.ref is None:
                      self.ref = self.default_branch
              except tmt.utils.RunError as error:
                  # Fallback to install during the prepare step if in rpm format
                  if self.format == 'rpm':
                      self.parent.debug(f"Repository '{self.url}' not found.")
                      raise LibraryError
                  self.parent.fail(
                      f"Failed to fetch library '{self}' from '{self.url}'.")
                  raise
              # Check out the requested branch
              try:
                  self.parent.run(
                      ['git', 'checkout', self.ref], shell=False, cwd=directory)
              except tmt.utils.RunError as error:
                  # Fallback to install during the prepare step if in rpm format
                  if self.format == 'rpm':
                      self.parent.debug(f"Invalid reference '{self.ref}'.")
                      raise LibraryError
                  self.parent.fail(
                      f"Reference '{self.ref}' for library '{self}' not found.")
                  raise
              # Initialize metadata tree, add self into the library index
              self.tree = fmf.Tree(directory)
              self.parent._library_cache[self.repo] = self
  
          # Get the library node, check require and recommend
          library = self.tree.find(self.name)
          if not library:
              # Fallback to install during the prepare step if in rpm format
              if self.format == 'rpm':
                  self.parent.debug(
                      f"Library '{self.name.lstrip('/')}' not found "
                      f"in the '{self.url}' repo.")
                  raise LibraryError
              raise tmt.utils.GeneralError(
                  f"Library '{self.name}' not found in '{self.repo}'.")
          self.require = tmt.utils.listify(library.get('require', []))
          self.recommend = tmt.utils.listify(library.get('recommend', []))
  
          # Create a symlink if the library is deep in the structure
          # FIXME: hot fix for https://github.com/beakerlib/beakerlib/pull/72
          # Covers also cases when library is stored more than 2 levels deep
          if os.path.dirname(self.name).lstrip('/'):
              link = self.name.lstrip('/')
              path = os.path.join(self.tree.root, os.path.basename(self.name))
              self.parent.debug(
                  f"Create a '{link}' symlink as the library is stored "
                  f"deep in the directory structure.")
              try:
                  os.symlink(link, path)
              except OSError as error:
                  self.parent.warn(
                      f"Unable to create a '{link}' symlink "
                      f"for a deep library ({error}).")
  
  
  def dependencies(original_require, original_recommend=None, parent=None):
      """
      Check dependencies for possible beakerlib libraries
  
      Fetch all identified libraries, check their required and recommended
      packages. Return tuple (requires, recommends, libraries) containing
      list of regular rpm package names aggregated from all fetched
      libraries, list of aggregated recommended packages and a list of
      gathered libraries (instances of the Library class).
      """
      # Initialize lists, use set for require & recommend
      processed_require = set()
      processed_recommend = set()
      gathered_libraries = []
      if not original_require:
          original_require = []
      if not original_recommend:
          original_recommend = []
  
      for dependency in original_require + original_recommend:
          # Library require/recommend
          try:
              library = Library(dependency, parent=parent)
              gathered_libraries.append(library)
              # Recursively check for possible dependent libraries
              requires, recommends, libraries = dependencies(
                  library.require, library.recommend, parent)
              processed_require.update(set(requires))
              processed_recommend.update(set(recommends))
              gathered_libraries.extend(libraries)
          # Regular package require/recommend
          except LibraryError:
              if dependency in original_require:
                  processed_require.add(dependency)
              if dependency in original_recommend:
                  processed_recommend.add(dependency)
  
      # Convert to list and return the results
      processed_require = list(processed_require)
      processed_recommend = list(processed_recommend)
      return processed_require, processed_recommend, gathered_libraries
  
files: tmt/beakerlib.py
files: """ Handle BeakerLib Libraries """
  
  import re
  import os
  import shutil
  
  import fmf
  import tmt
  
  # Regular expressions for beakerlib libraries
  LIBRARY_REGEXP = re.compile(r'^library\(([^/]+)(/[^)]+)\)$')
  
  # Default beakerlib libraries location and destination directory
  DEFAULT_REPOSITORY = 'https://github.com/beakerlib'
  DEFAULT_DESTINATION = 'libs'
  
  
  class LibraryError(Exception):
      """ Used when library cannot be parsed from the identifier """
  
  
  class Library(object):
      """
      A beakerlib library
  
      Takes care of fetching beakerlib libraries from remote repositories
      based on provided library identifier described in detail here:
      https://tmt.readthedocs.io/en/latest/spec/tests.html#require
  
      Optional 'parent' object inheriting from tmt.utils.Common can be
      provided in order to share the cache of already fetched libraries.
  
      The following attributes are available in the object:
  
      repo ........ library prefix (git repository name or nick if provided)
      name ........ library suffix (folder containing the library code)
  
      url ......... full git repository url
      ref ......... git revision (branch, tag or commit)
      dest ........ target folder into which the library repo is cloned
  
      tree ........ fmf tree holding library metadata
      require ..... list of required packages
      recommend ... list of recommended packages
  
      Libraries are fetched into the 'libs' directory under parent's
      workdir or into 'destination' if provided in the identifier.
      """
  
      def __init__(self, identifier, parent=None):
          """ Process the library identifier and fetch the library """
          # Use an empty common class if parent not provided (for logging, cache)
          self.parent = parent or tmt.utils.Common(workdir=True)
  
          # The 'library(repo/lib)' format
          if isinstance(identifier, str):
              identifier = identifier.strip()
              matched = LIBRARY_REGEXP.search(identifier)
              if not matched:
                  raise LibraryError
              self.parent.debug(f"Detected library '{identifier}'.", level=3)
              self.format = 'rpm'
              self.repo, self.name = matched.groups()
              self.url = os.path.join(DEFAULT_REPOSITORY, self.repo)
              self.ref = None   # final value
              self.__ref = None # original value
              self.dest = DEFAULT_DESTINATION
  
          # The fmf identifier
          elif isinstance(identifier, dict):
              self.parent.debug(f"Detected library '{identifier}'.", level=3)
              self.format = 'fmf'
              self.url = identifier.get('url')
              self.ref = identifier.get('ref', None) # final value
              self.__ref = None # original value
              self.dest = identifier.get(
                  'destination', DEFAULT_DESTINATION).lstrip('/')
              self.name = identifier.get('name', '/')
              if not self.name.startswith('/'):
                  raise tmt.utils.SpecificationError(
                      f"Library name '{self.name}' does not start with a '/'.")
              # Use provided repository nick name or parse it from the url
              try:
                  self.repo = identifier.get('nick') or re.search(
                      r'/([^/]+?)(/|\.git)?$', self.url).group(1)
              except AttributeError:
                  raise tmt.utils.GeneralError(
                      f"Unable to parse repository name from '{self.url}'.")
          # Something weird
          else:
              raise LibraryError
  
          # Fetch the library
          try:
              self.fetch()
          except fmf.utils.RootError:
              raise tmt.utils.SpecificationError(
                  f"Repository '{self.url}' does not contain fmf metadata.")
  
      def __str__(self):
          """ Use repo/name for string representation """
          return f"{self.repo}{self.name}"
  
      def fetch(self):
          """ Fetch the library (unless already fetched) """
          # Initialize library cache (indexed by the repository name)
          if not hasattr(self.parent, '_library_cache'):
              self.parent._library_cache = dict()
  
          # Check if the library was already fetched
          try:
              library = self.parent._library_cache[self.repo]
              if library.url != self.url:
                  raise tmt.utils.GeneralError(
                      f"Library '{self.repo}' with url '{self.url}' conflicts "
                      f"with already fetched library from '{library.url}'.")
              if library.__ref != self.__ref:
                  # .__ref can be None, indicating we want default branch
                  # .ref is always a brach/commit/tag string
                  lib_ref = library.__ref if library.__ref else '<default branch>'
                  self_ref = self.__ref if self.__ref else '<default branch>'
                  raise tmt.utils.GeneralError(
                      f"Library '{self.repo}' using ref '{self_ref}' conflicts "
                      f"with already fetched library using ref '{lib_ref}'.")
              self.parent.debug(f"Library '{self}' already fetched.", level=3)
              # Reuse the existing metadata tree
              self.tree = library.tree
          # Fetch the library and add it to the index
          except KeyError:
              self.parent.debug(f"Fetch library '{self}'.", level=3)
              # Prepare path, clone the repository, checkout ref
              directory = os.path.join(
                  self.parent.workdir, self.dest, self.repo)
              # Clone repo with disabled prompt to ignore missing/private repos
              try:
                  self.parent.run(
                      ['git', 'clone', self.url, directory],
                      shell=False, env={"GIT_ASKPASS": "echo"})
                  # Store the default_branch
                  head = os.path.join(directory, '.git/refs/remotes/origin/HEAD')
                  default = os.path.join(directory, '.git/refs/heads/__DEFAULT__')
                  shutil.copyfile(head, default)
              except tmt.utils.RunError as error:
                  # Fallback to install during the prepare step if in rpm format
                  if self.format == 'rpm':
                      self.parent.debug(f"Repository '{self.url}' not found.")
                      raise LibraryError
                  self.parent.fail(
                      f"Failed to fetch library '{self}' from '{self.url}'.")
                  raise
              # Check out the requested branch (sets real name of default branch)
              try:
                  # wants default branch -> replace with the name of real default branch
                  if self.ref is None:
                      with open(os.path.join(directory, '.git/refs/heads/__DEFAULT__')) as f_head:
                          # content should be `ref: refs/remotes/origin/master`
                          self.ref = f_head.read().strip().split('/')[-1]
                  self.parent.run(
                      ['git', 'checkout', self.ref], shell=False, cwd=directory)
              except tmt.utils.RunError as error:
                  # Fallback to install during the prepare step if in rpm format
                  if self.format == 'rpm':
                      self.parent.debug(f"Invalid reference '{self.ref}'.")
                      raise LibraryError
                  self.parent.fail(
                      f"Reference '{self.ref}' for library '{self}' not found.")
                  raise
              # Initialize metadata tree, add self into the library index
              self.tree = fmf.Tree(directory)
              self.parent._library_cache[self.repo] = self
  
          # Get the library node, check require and recommend
          library = self.tree.find(self.name)
          if not library:
              # Fallback to install during the prepare step if in rpm format
              if self.format == 'rpm':
                  self.parent.debug(
                      f"Library '{self.name.lstrip('/')}' not found "
                      f"in the '{self.url}' repo.")
                  raise LibraryError
              raise tmt.utils.GeneralError(
                  f"Library '{self.name}' not found in '{self.repo}'.")
          self.require = tmt.utils.listify(library.get('require', []))
          self.recommend = tmt.utils.listify(library.get('recommend', []))
  
          # Create a symlink if the library is deep in the structure
          # FIXME: hot fix for https://github.com/beakerlib/beakerlib/pull/72
          # Covers also cases when library is stored more than 2 levels deep
          if os.path.dirname(self.name).lstrip('/'):
              link = self.name.lstrip('/')
              path = os.path.join(self.tree.root, os.path.basename(self.name))
              self.parent.debug(
                  f"Create a '{link}' symlink as the library is stored "
                  f"deep in the directory structure.")
              try:
                  os.symlink(link, path)
              except OSError as error:
                  self.parent.warn(
                      f"Unable to create a '{link}' symlink "
                      f"for a deep library ({error}).")
  
  
  def dependencies(original_require, original_recommend=None, parent=None):
      """
      Check dependencies for possible beakerlib libraries
  
      Fetch all identified libraries, check their required and recommended
      packages. Return tuple (requires, recommends, libraries) containing
      list of regular rpm package names aggregated from all fetched
      libraries, list of aggregated recommended packages and a list of
      gathered libraries (instances of the Library class).
      """
      # Initialize lists, use set for require & recommend
      processed_require = set()
      processed_recommend = set()
      gathered_libraries = []
      if not original_require:
          original_require = []
      if not original_recommend:
          original_recommend = []
  
      for dependency in original_require + original_recommend:
          # Library require/recommend
          try:
              library = Library(dependency, parent=parent)
              gathered_libraries.append(library)
              # Recursively check for possible dependent libraries
              requires, recommends, libraries = dependencies(
                  library.require, library.recommend, parent)
              processed_require.update(set(requires))
              processed_recommend.update(set(recommends))
              gathered_libraries.extend(libraries)
          # Regular package require/recommend
          except LibraryError:
              if dependency in original_require:
                  processed_require.add(dependency)
              if dependency in original_recommend:
                  processed_recommend.add(dependency)
  
      # Convert to list and return the results
      processed_require = list(processed_require)
      processed_recommend = list(processed_recommend)
      return processed_require, processed_recommend, gathered_libraries
  
files: tmt/beakerlib.py