1# 2# Copyright BitBake Contributors 3# 4# SPDX-License-Identifier: GPL-2.0-only 5# 6""" 7Bitbake "Fetch" implementation for osc (Opensuse build service client). 8Based on the svn "Fetch" implementation. 9 10""" 11 12import logging 13import os 14import re 15import bb 16from bb.fetch2 import FetchMethod 17from bb.fetch2 import FetchError 18from bb.fetch2 import MissingParameterError 19from bb.fetch2 import runfetchcmd 20 21logger = logging.getLogger(__name__) 22 23class Osc(FetchMethod): 24 """Class to fetch a module or modules from Opensuse build server 25 repositories.""" 26 27 def supports(self, ud, d): 28 """ 29 Check to see if a given url can be fetched with osc. 30 """ 31 return ud.type in ['osc'] 32 33 def urldata_init(self, ud, d): 34 if not "module" in ud.parm: 35 raise MissingParameterError('module', ud.url) 36 37 ud.module = ud.parm["module"] 38 39 # Create paths to osc checkouts 40 oscdir = d.getVar("OSCDIR") or (d.getVar("DL_DIR") + "/osc") 41 relpath = self._strip_leading_slashes(ud.path) 42 ud.oscdir = oscdir 43 ud.pkgdir = os.path.join(oscdir, ud.host) 44 ud.moddir = os.path.join(ud.pkgdir, relpath, ud.module) 45 46 if 'rev' in ud.parm: 47 ud.revision = ud.parm['rev'] 48 else: 49 pv = d.getVar("PV", False) 50 rev = bb.fetch2.srcrev_internal_helper(ud, d, '') 51 if rev: 52 ud.revision = rev 53 else: 54 ud.revision = "" 55 56 ud.localfile = d.expand('%s_%s_%s.tar.gz' % (ud.module.replace('/', '.'), relpath.replace('/', '.'), ud.revision)) 57 58 def _buildosccommand(self, ud, d, command): 59 """ 60 Build up an ocs commandline based on ud 61 command is "fetch", "update", "info" 62 """ 63 64 basecmd = d.getVar("FETCHCMD_osc") or "/usr/bin/env osc" 65 66 proto = ud.parm.get('protocol', 'https') 67 68 options = [] 69 70 config = "-c %s" % self.generate_config(ud, d) 71 72 if getattr(ud, 'revision', ''): 73 options.append("-r %s" % ud.revision) 74 75 coroot = self._strip_leading_slashes(ud.path) 76 77 if command == "fetch": 78 osccmd = "%s %s -A %s://%s co %s/%s %s" % (basecmd, config, proto, ud.host, coroot, ud.module, " ".join(options)) 79 elif command == "update": 80 osccmd = "%s %s -A %s://%s up %s" % (basecmd, config, proto, ud.host, " ".join(options)) 81 elif command == "api_source": 82 osccmd = "%s %s -A %s://%s api source/%s/%s" % (basecmd, config, proto, ud.host, coroot, ud.module) 83 else: 84 raise FetchError("Invalid osc command %s" % command, ud.url) 85 86 return osccmd 87 88 def _latest_revision(self, ud, d, name): 89 """ 90 Fetch latest revision for the given package 91 """ 92 api_source_cmd = self._buildosccommand(ud, d, "api_source") 93 94 output = runfetchcmd(api_source_cmd, d) 95 match = re.match(r'<directory ?.* rev="(\d+)".*>', output) 96 if match is None: 97 raise FetchError("Unable to parse osc response", ud.url) 98 return match.groups()[0] 99 100 def _revision_key(self, ud, d, name): 101 """ 102 Return a unique key for the url 103 """ 104 # Collapse adjacent slashes 105 slash_re = re.compile(r"/+") 106 rev = getattr(ud, 'revision', "latest") 107 return "osc:%s%s.%s.%s" % (ud.host, slash_re.sub(".", ud.path), name, rev) 108 109 def download(self, ud, d): 110 """ 111 Fetch url 112 """ 113 114 logger.debug2("Fetch: checking for module directory '" + ud.moddir + "'") 115 116 if os.access(ud.moddir, os.R_OK): 117 oscupdatecmd = self._buildosccommand(ud, d, "update") 118 logger.info("Update "+ ud.url) 119 # update sources there 120 logger.debug("Running %s", oscupdatecmd) 121 bb.fetch2.check_network_access(d, oscupdatecmd, ud.url) 122 runfetchcmd(oscupdatecmd, d, workdir=ud.moddir) 123 else: 124 oscfetchcmd = self._buildosccommand(ud, d, "fetch") 125 logger.info("Fetch " + ud.url) 126 # check out sources there 127 bb.utils.mkdirhier(ud.pkgdir) 128 logger.debug("Running %s", oscfetchcmd) 129 bb.fetch2.check_network_access(d, oscfetchcmd, ud.url) 130 runfetchcmd(oscfetchcmd, d, workdir=ud.pkgdir) 131 132 # tar them up to a defined filename 133 runfetchcmd("tar -czf %s %s" % (ud.localpath, ud.module), d, 134 cleanup=[ud.localpath], workdir=os.path.join(ud.pkgdir + ud.path)) 135 136 def supports_srcrev(self): 137 return False 138 139 def generate_config(self, ud, d): 140 """ 141 Generate a .oscrc to be used for this run. 142 """ 143 144 config_path = os.path.join(ud.oscdir, "oscrc") 145 if not os.path.exists(ud.oscdir): 146 bb.utils.mkdirhier(ud.oscdir) 147 148 if (os.path.exists(config_path)): 149 os.remove(config_path) 150 151 f = open(config_path, 'w') 152 proto = ud.parm.get('protocol', 'https') 153 f.write("[general]\n") 154 f.write("apiurl = %s://%s\n" % (proto, ud.host)) 155 f.write("su-wrapper = su -c\n") 156 f.write("build-root = %s\n" % d.getVar('WORKDIR')) 157 f.write("urllist = %s\n" % d.getVar("OSCURLLIST")) 158 f.write("extra-pkgs = gzip\n") 159 f.write("\n") 160 f.write("[%s://%s]\n" % (proto, ud.host)) 161 f.write("user = %s\n" % ud.parm["user"]) 162 f.write("pass = %s\n" % ud.parm["pswd"]) 163 f.close() 164 165 return config_path 166