from __future__ import absolute_import, unicode_literals
from contextlib import contextmanager
import json
from .files import (
AbsoluteSymlinkFile,
ExistingFile,
File,
FileFinder,
GeneratedFile,
HardlinkFile,
PreprocessedFile,
)
import mozpack.path as mozpath
@contextmanager
def _auto_fileobj(path, fileobj, mode='r'):
if path and fileobj:
raise AssertionError('Only 1 of path or fileobj may be defined.')
if not path and not fileobj:
raise AssertionError('Must specified 1 of path or fileobj.')
if path:
fileobj = open(path, mode)
try:
yield fileobj
finally:
if path:
fileobj.close()
class UnreadableInstallManifest(Exception):
class InstallManifest(object):
CURRENT_VERSION = 5
FIELD_SEPARATOR = '\x1f'
LINK = 1
COPY = 2
REQUIRED_EXISTS = 3
OPTIONAL_EXISTS = 4
PATTERN_LINK = 5
PATTERN_COPY = 6
PREPROCESS = 7
CONTENT = 8
def __init__(self, path=None, fileobj=None):
self._dests = {}
self._source_files = set()
if path or fileobj:
with _auto_fileobj(path, fileobj, 'rb') as fh:
self._source_files.add(fh.name)
self._load_from_fileobj(fh)
def _load_from_fileobj(self, fileobj):
version = fileobj.readline().rstrip()
if version not in ('1', '2', '3', '4', '5'):
raise UnreadableInstallManifest('Unknown manifest version: %s' %
version)
for line in fileobj:
line = line.rstrip()
fields = line.split(self.FIELD_SEPARATOR)
record_type = int(fields[0])
if record_type == self.LINK:
dest, source = fields[1:]
self.add_link(source, dest)
continue
if record_type == self.COPY:
dest, source = fields[1:]
self.add_copy(source, dest)
continue
if record_type == self.REQUIRED_EXISTS:
_, path = fields
self.add_required_exists(path)
continue
if record_type == self.OPTIONAL_EXISTS:
_, path = fields
self.add_optional_exists(path)
continue
if record_type == self.PATTERN_LINK:
_, base, pattern, dest = fields[1:]
self.add_pattern_link(base, pattern, dest)
continue
if record_type == self.PATTERN_COPY:
_, base, pattern, dest = fields[1:]
self.add_pattern_copy(base, pattern, dest)
continue
if record_type == self.PREPROCESS:
dest, source, deps, marker, defines, warnings = fields[1:]
self.add_preprocess(source, dest, deps, marker,
self._decode_field_entry(defines),
silence_missing_directive_warnings=bool(int(warnings)))
continue
if record_type == self.CONTENT:
dest, content = fields[1:]
self.add_content(
self._decode_field_entry(content).encode('utf-8'), dest)
continue
if record_type >= 0:
raise UnreadableInstallManifest('Unknown record type: %d' %
record_type)
def __len__(self):
return len(self._dests)
def __contains__(self, item):
return item in self._dests
def __eq__(self, other):
return isinstance(other, InstallManifest) and self._dests == other._dests
def __neq__(self, other):
return not self.__eq__(other)
def __ior__(self, other):
if not isinstance(other, InstallManifest):
raise ValueError('Can only | with another instance of InstallManifest.')
self.add_entries_from(other)
return self
def _encode_field_entry(self, data):
return json.dumps(data, sort_keys=True)
def _decode_field_entry(self, data):
return json.loads(data)
def write(self, path=None, fileobj=None, expand_pattern=False):
with _auto_fileobj(path, fileobj, 'wb') as fh:
fh.write('%d\n' % self.CURRENT_VERSION)
for dest in sorted(self._dests):
entry = self._dests[dest]
if expand_pattern and entry[0] in (self.PATTERN_LINK, self.PATTERN_COPY):
type, base, pattern, dest = entry
type = self.LINK if type == self.PATTERN_LINK else self.COPY
finder = FileFinder(base)
paths = [f[0] for f in finder.find(pattern)]
for path in paths:
source = mozpath.join(base, path)
parts = ['%d' % type, mozpath.join(dest, path), source]
fh.write('%s\n' % self.FIELD_SEPARATOR.join(
p.encode('utf-8') for p in parts))
else:
parts = ['%d' % entry[0], dest]
parts.extend(entry[1:])
fh.write('%s\n' % self.FIELD_SEPARATOR.join(
p.encode('utf-8') for p in parts))
def add_link(self, source, dest):
self._add_entry(dest, (self.LINK, source))
def add_copy(self, source, dest):
self._add_entry(dest, (self.COPY, source))
def add_required_exists(self, dest):
self._add_entry(dest, (self.REQUIRED_EXISTS,))
def add_optional_exists(self, dest):
self._add_entry(dest, (self.OPTIONAL_EXISTS,))
def add_pattern_link(self, base, pattern, dest):
self._add_entry(mozpath.join(dest, pattern),
(self.PATTERN_LINK, base, pattern, dest))
def add_pattern_copy(self, base, pattern, dest):
self._add_entry(mozpath.join(dest, pattern),
(self.PATTERN_COPY, base, pattern, dest))
def add_preprocess(self, source, dest, deps, marker='#', defines={},
silence_missing_directive_warnings=False):
self._add_entry(dest, (
self.PREPROCESS,
source,
deps,
marker,
self._encode_field_entry(defines),
'1' if silence_missing_directive_warnings else '0',
))
def add_content(self, content, dest):
self._add_entry(dest, (
self.CONTENT,
self._encode_field_entry(content),
))
def _add_entry(self, dest, entry):
if dest in self._dests:
raise ValueError('Item already in manifest: %s' % dest)
self._dests[dest] = entry
def add_entries_from(self, other, base=''):
self._source_files |= other._source_files
for dest in sorted(other._dests):
new_dest = mozpath.join(base, dest) if base else dest
entry = other._dests[dest]
if entry[0] in (self.PATTERN_LINK, self.PATTERN_COPY):
entry_type, entry_base, entry_pattern, entry_dest = entry
new_entry_dest = mozpath.join(base, entry_dest) if base else entry_dest
new_entry = (entry_type, entry_base, entry_pattern, new_entry_dest)
else:
new_entry = tuple(entry)
self._add_entry(new_dest, new_entry)
def populate_registry(self, registry, defines_override={},
link_policy='symlink'):
assert link_policy in ("symlink", "hardlink", "copy")
for dest in sorted(self._dests):
entry = self._dests[dest]
install_type = entry[0]
if install_type == self.LINK:
if link_policy == "symlink":
cls = AbsoluteSymlinkFile
elif link_policy == "hardlink":
cls = HardlinkFile
else:
cls = File
registry.add(dest, cls(entry[1]))
continue
if install_type == self.COPY:
registry.add(dest, File(entry[1]))
continue
if install_type == self.REQUIRED_EXISTS:
registry.add(dest, ExistingFile(required=True))
continue
if install_type == self.OPTIONAL_EXISTS:
registry.add(dest, ExistingFile(required=False))
continue
if install_type in (self.PATTERN_LINK, self.PATTERN_COPY):
_, base, pattern, dest = entry
finder = FileFinder(base)
paths = [f[0] for f in finder.find(pattern)]
if install_type == self.PATTERN_LINK:
if link_policy == "symlink":
cls = AbsoluteSymlinkFile
elif link_policy == "hardlink":
cls = HardlinkFile
else:
cls = File
else:
cls = File
for path in paths:
source = mozpath.join(base, path)
registry.add(mozpath.join(dest, path), cls(source))
continue
if install_type == self.PREPROCESS:
defines = self._decode_field_entry(entry[4])
if defines_override:
defines.update(defines_override)
registry.add(dest, PreprocessedFile(
entry[1],
depfile_path=entry[2],
marker=entry[3],
defines=defines,
extra_depends=self._source_files,
silence_missing_directive_warnings=bool(int(entry[5])),
))
continue
if install_type == self.CONTENT:
content = self._decode_field_entry(entry[1]).encode('utf-8')
registry.add(dest, GeneratedFile(content))
continue
raise Exception('Unknown install type defined in manifest: %d' %
install_type)