blob: 495ac8a30aa082dabb8baaddce974973bee9211a [file] [log] [blame]
#
# Copyright BitBake Contributors
#
# SPDX-License-Identifier: GPL-2.0-only
#
"""
Bitbake "Fetch" implementation for osc (Opensuse build service client).
Based on the svn "Fetch" implementation.
"""
import logging
import os
import re
import bb
from bb.fetch2 import FetchMethod
from bb.fetch2 import FetchError
from bb.fetch2 import MissingParameterError
from bb.fetch2 import runfetchcmd
logger = logging.getLogger(__name__)
class Osc(FetchMethod):
"""Class to fetch a module or modules from Opensuse build server
repositories."""
def supports(self, ud, d):
"""
Check to see if a given url can be fetched with osc.
"""
return ud.type in ['osc']
def urldata_init(self, ud, d):
if not "module" in ud.parm:
raise MissingParameterError('module', ud.url)
ud.module = ud.parm["module"]
# Create paths to osc checkouts
oscdir = d.getVar("OSCDIR") or (d.getVar("DL_DIR") + "/osc")
relpath = self._strip_leading_slashes(ud.path)
ud.oscdir = oscdir
ud.pkgdir = os.path.join(oscdir, ud.host)
ud.moddir = os.path.join(ud.pkgdir, relpath, ud.module)
if 'rev' in ud.parm:
ud.revision = ud.parm['rev']
else:
pv = d.getVar("PV", False)
rev = bb.fetch2.srcrev_internal_helper(ud, d, '')
if rev:
ud.revision = rev
else:
ud.revision = ""
ud.localfile = d.expand('%s_%s_%s.tar.gz' % (ud.module.replace('/', '.'), relpath.replace('/', '.'), ud.revision))
def _buildosccommand(self, ud, d, command):
"""
Build up an ocs commandline based on ud
command is "fetch", "update", "info"
"""
basecmd = d.getVar("FETCHCMD_osc") or "/usr/bin/env osc"
proto = ud.parm.get('protocol', 'https')
options = []
config = "-c %s" % self.generate_config(ud, d)
if getattr(ud, 'revision', ''):
options.append("-r %s" % ud.revision)
coroot = self._strip_leading_slashes(ud.path)
if command == "fetch":
osccmd = "%s %s -A %s://%s co %s/%s %s" % (basecmd, config, proto, ud.host, coroot, ud.module, " ".join(options))
elif command == "update":
osccmd = "%s %s -A %s://%s up %s" % (basecmd, config, proto, ud.host, " ".join(options))
elif command == "api_source":
osccmd = "%s %s -A %s://%s api source/%s/%s" % (basecmd, config, proto, ud.host, coroot, ud.module)
else:
raise FetchError("Invalid osc command %s" % command, ud.url)
return osccmd
def _latest_revision(self, ud, d, name):
"""
Fetch latest revision for the given package
"""
api_source_cmd = self._buildosccommand(ud, d, "api_source")
output = runfetchcmd(api_source_cmd, d)
match = re.match(r'<directory ?.* rev="(\d+)".*>', output)
if match is None:
raise FetchError("Unable to parse osc response", ud.url)
return match.groups()[0]
def _revision_key(self, ud, d, name):
"""
Return a unique key for the url
"""
# Collapse adjacent slashes
slash_re = re.compile(r"/+")
rev = getattr(ud, 'revision', "latest")
return "osc:%s%s.%s.%s" % (ud.host, slash_re.sub(".", ud.path), name, rev)
def download(self, ud, d):
"""
Fetch url
"""
logger.debug2("Fetch: checking for module directory '" + ud.moddir + "'")
if os.access(ud.moddir, os.R_OK):
oscupdatecmd = self._buildosccommand(ud, d, "update")
logger.info("Update "+ ud.url)
# update sources there
logger.debug("Running %s", oscupdatecmd)
bb.fetch2.check_network_access(d, oscupdatecmd, ud.url)
runfetchcmd(oscupdatecmd, d, workdir=ud.moddir)
else:
oscfetchcmd = self._buildosccommand(ud, d, "fetch")
logger.info("Fetch " + ud.url)
# check out sources there
bb.utils.mkdirhier(ud.pkgdir)
logger.debug("Running %s", oscfetchcmd)
bb.fetch2.check_network_access(d, oscfetchcmd, ud.url)
runfetchcmd(oscfetchcmd, d, workdir=ud.pkgdir)
# tar them up to a defined filename
runfetchcmd("tar -czf %s %s" % (ud.localpath, ud.module), d,
cleanup=[ud.localpath], workdir=os.path.join(ud.pkgdir + ud.path))
def supports_srcrev(self):
return False
def generate_config(self, ud, d):
"""
Generate a .oscrc to be used for this run.
"""
config_path = os.path.join(ud.oscdir, "oscrc")
if not os.path.exists(ud.oscdir):
bb.utils.mkdirhier(ud.oscdir)
if (os.path.exists(config_path)):
os.remove(config_path)
f = open(config_path, 'w')
proto = ud.parm.get('protocol', 'https')
f.write("[general]\n")
f.write("apiurl = %s://%s\n" % (proto, ud.host))
f.write("su-wrapper = su -c\n")
f.write("build-root = %s\n" % d.getVar('WORKDIR'))
f.write("urllist = %s\n" % d.getVar("OSCURLLIST"))
f.write("extra-pkgs = gzip\n")
f.write("\n")
f.write("[%s://%s]\n" % (proto, ud.host))
f.write("user = %s\n" % ud.parm["user"])
f.write("pass = %s\n" % ud.parm["pswd"])
f.close()
return config_path