from debian.changelog import Changelog
import os
import shutil
import subprocess
import tempfile
from ... import errors as bzr_errors
from ...trace import note
from ...transport import NoSuchFile, FileExists
from .util import (
export_with_nested,
extract_orig_tarballs,
get_parent_dir,
)
class SourceDistiller:
def __init__(self, tree, subpath):
self.tree = tree
self.subpath = subpath
def distill(self, target):
raise NotImplementedError(self.distill)
class NativeSourceDistiller(SourceDistiller):
def __init__(self, tree, subpath, use_existing=False):
super().__init__(tree, subpath)
self.use_existing = use_existing
def distill(self, target):
if not self.use_existing:
if os.path.exists(target):
raise FileExists(target)
export_with_nested(self.tree, target, subdir=self.subpath)
class FullSourceDistiller(SourceDistiller):
def __init__(self, tree, subpath, upstream_provider, use_existing=False):
super().__init__(tree, subpath)
self.upstream_provider = upstream_provider
self.use_existing = use_existing
def distill(self, target):
if not self.use_existing:
if os.path.exists(target):
raise FileExists(target)
parent_dir = get_parent_dir(target)
self.upstream_provider.provide(parent_dir)
export_with_nested(self.tree, target, subdir=self.subpath)
class MergeModeDistiller(SourceDistiller):
def __init__(self, tree, subpath, upstream_provider, top_level=False,
use_existing=False):
super().__init__(tree, subpath)
self.upstream_provider = upstream_provider
self.top_level = top_level
self.use_existing = use_existing
def distill(self, target):
if not self.use_existing:
if os.path.exists(target):
raise FileExists(target)
elif self.use_existing:
if not os.path.exists(target):
raise bzr_errors.NotADirectory(target)
parent_dir = get_parent_dir(target)
if parent_dir != '' and not os.path.exists(parent_dir):
os.makedirs(parent_dir)
if not self.use_existing:
tarballs = self.upstream_provider.provide(parent_dir)
if not os.path.exists(target):
os.mkdir(target)
extract_orig_tarballs(tarballs, target)
with tempfile.TemporaryDirectory(
prefix='builddeb-merge-debian-') as basetempdir:
tempdir = os.path.join(basetempdir, "export")
if self.top_level:
os.makedirs(tempdir)
export_dir = os.path.join(tempdir, 'debian')
else:
export_dir = tempdir
export_with_nested(self.tree, export_dir, subdir=self.subpath)
if os.path.exists(os.path.join(target, 'debian')):
shutil.rmtree(os.path.join(target, 'debian'))
shutil.copytree(
tempdir, target, symlinks=True, dirs_exist_ok=True)
class DebcargoError(bzr_errors.BzrError):
_fmt = "Debcargo failed to run."
def cargo_translate_dashes(crate):
output = subprocess.check_output(['cargo', 'search', crate])
for line in output.splitlines(False):
name = line.split(b' = ')[0].decode()
return name
return crate
def unmangle_debcargo_version(version):
return version.replace('~', '-')
class DebcargoDistiller(SourceDistiller):
def __init__(self, tree, subpath, top_level=False, use_existing=False):
super().__init__(tree, subpath)
self.top_level = top_level
self.use_existing = use_existing
def distill(self, target):
from debmutate.debcargo import parse_debcargo_source_name
if os.path.exists(target):
raise FileExists(target)
with self.tree.get_file(os.path.join(
self.subpath,
'debian/changelog' if not self.top_level else
'changelog'), 'r') as f:
cl = Changelog(f, max_blocks=1)
package = cl.package
version = cl.version
if not package.startswith('rust-'):
raise NotImplementedError
debcargo_path = [self.subpath]
if not self.top_level:
debcargo_path.append('debian')
debcargo_path.append('debcargo.toml')
try:
debcargo_text = self.tree.get_file_text(
os.path.join(*debcargo_path))
except NoSuchFile:
semver_suffix = False
else:
from toml.decoder import loads as loads_toml
debcargo = loads_toml(debcargo_text.decode())
semver_suffix = debcargo.get('semver_suffix')
crate, crate_semver_version = parse_debcargo_source_name(
package, semver_suffix)
if '-' in crate:
crate = cargo_translate_dashes(crate)
crate_version = unmangle_debcargo_version(version.upstream_version)
if crate_semver_version is not None:
note('Using crate name: %s, version %s (semver: %s)', crate,
crate_version, crate_semver_version)
else:
note('Using crate name: %s, version %s', crate, crate_version)
try:
subprocess.check_call([
'debcargo', 'package',
'--changelog-ready',
'--config', self.tree.abspath(os.path.join(*debcargo_path)),
'--directory', target, crate]
+ ([crate_version] if crate_version else []))
except subprocess.CalledProcessError as e:
raise DebcargoError() from e