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