blob: 6fb591b9c314b0dd01d93ff12f3820bf26d4fe79 [file] [log] [blame]
import itertools
import json
import os
from collections import MutableMapping, defaultdict
from six import iteritems, iterkeys, itervalues, string_types, binary_type, text_type
from . import vcs
from .item import (ConformanceCheckerTest, ManifestItem, ManualTest, RefTest, RefTestNode,
SupportFile, TestharnessTest, VisualTest, WebDriverSpecTest)
from .log import get_logger
from .sourcefile import SourceFile
from .utils import from_os_path, to_os_path
MYPY = False
if MYPY:
# MYPY is set to True when run under Mypy.
from logging import Logger
from typing import Any
from typing import Container
from typing import Dict
from typing import IO
from typing import Iterable
from typing import Iterator
from typing import List
from typing import Optional
from typing import Set
from typing import Text
from typing import Tuple
from typing import Type
from typing import Union
from typing import cast
import ujson
fast_json = ujson
except ImportError:
fast_json = json # type: ignore
class ManifestError(Exception):
class ManifestVersionMismatch(ManifestError):
item_classes = {"testharness": TestharnessTest,
"reftest": RefTest,
"reftest_node": RefTestNode,
"manual": ManualTest,
"wdspec": WebDriverSpecTest,
"conformancechecker": ConformanceCheckerTest,
"visual": VisualTest,
"support": SupportFile} # type: Dict[str, Type[ManifestItem]]
if MYPY:
TypeDataType = MutableMapping[Text, Set[ManifestItem]]
TypeDataType = MutableMapping
class TypeData(TypeDataType):
def __init__(self, manifest, type_cls):
# type: (Manifest, Type[ManifestItem]) -> None
"""Dict-like object containing the TestItems for each test type.
Loading an actual Item class for each test is unnecessarily
slow, so this class allows lazy-loading of the test
items. When the manifest is loaded we store the raw json
corresponding to the test type, and only create an Item
subclass when the test is accessed. In order to remain
API-compatible with consumers that depend on getting an Item
from iteration, we do egerly load all items when iterating
over the class."""
self.manifest = manifest
self.type_cls = type_cls
self.json_data = {} # type: Optional[Dict[Text, List[Any]]]
self.tests_root = None # type: Optional[str] = {} # type: Dict[Text, Set[ManifestItem]]
def __getitem__(self, key):
# type: (Text) -> Set[ManifestItem]
if key not in and self.json_data is not None:
def __nonzero__(self):
# type: () -> bool
return bool( or bool(self.json_data)
def __len__(self):
# type: () -> int
rv = len(
if self.json_data is not None:
rv += len(self.json_data)
return rv
def __delitem__(self, key):
# type: (Text) -> None
if key in
elif self.json_data is not None:
del self.json_data[from_os_path(key)]
raise KeyError
def __setitem__(self, key, value):
# type: (Text, Set[ManifestItem]) -> None
if self.json_data is not None:
path = from_os_path(key)
if path in self.json_data:
del self.json_data[path][key] = value
def __contains__(self, key):
# type: (Any) -> bool
return key in
def __iter__(self):
# type: () -> Iterator[Text]
def itervalues(self):
# type: () -> Iterator[Set[ManifestItem]]
return itervalues(
def iteritems(self):
# type: () -> Iterator[Tuple[Text, Set[ManifestItem]]]
return iteritems(
def values(self):
# type: () -> List[Set[ManifestItem]]
return list(self.itervalues())
def items(self):
# type: () -> List[Tuple[Text, Set[ManifestItem]]]
return list(self.iteritems())
def load(self, key):
# type: (Text) -> None
"""Load a specific Item given a path"""
if self.json_data is not None:
data = set()
path = from_os_path(key)
for test in self.json_data.get(path, []):
manifest_item = self.type_cls.from_json(self.manifest, path, test)
del self.json_data[path]
except KeyError:
pass[key] = data
raise ValueError
def load_all(self):
# type: () -> None
"""Load all test items in this class"""
if self.json_data is not None:
for path, value in iteritems(self.json_data):
key = to_os_path(path)
if key in
data = set()
for test in self.json_data.get(path, []):
manifest_item = self.type_cls.from_json(self.manifest, path, test)
data.add(manifest_item)[key] = data
self.json_data = None
def set_json(self, tests_root, data):
# type: (str, Dict[Text, Any]) -> None
if not isinstance(data, dict):
raise ValueError("Got a %s expected a dict" % (type(data)))
self.tests_root = tests_root
self.json_data = data
def to_json(self):
# type: () -> Dict[Text, Any]
data = {
[t for t in sorted(test.to_json() for test in tests)]
for path, tests in iteritems(
if self.json_data is not None:
if not data:
# avoid copying if there's nothing here yet
return self.json_data
return data
def paths(self):
# type: () -> Set[Text]
"""Get a list of all paths containing items of this type,
without actually constructing all the items"""
rv = set(iterkeys(
if self.json_data:
rv |= {to_os_path(item) for item in iterkeys(self.json_data)}
return rv
if MYPY:
ManifestDataType = Dict[Any, TypeData]
ManifestDataType = dict
class ManifestData(ManifestDataType):
def __init__(self, manifest):
# type: (Manifest) -> None
"""Dictionary subclass containing a TypeData instance for each test type,
keyed by type name"""
self.initialized = False # type: bool
for key, value in iteritems(item_classes):
self[key] = TypeData(manifest, value)
self.initialized = True
self.json_obj = None # type: None
def __setitem__(self, key, value):
# type: (str, TypeData) -> None
if self.initialized:
raise AttributeError
dict.__setitem__(self, key, value)
def paths(self):
# type: () -> Set[Text]
"""Get a list of all paths containing test items
without actually constructing all the items"""
rv = set() # type: Set[Text]
for item_data in itervalues(self):
rv |= set(item_data.paths())
return rv
class Manifest(object):
def __init__(self, tests_root=None, url_base="/"):
# type: (Optional[str], Text) -> None
assert url_base is not None
self._path_hash = {} # type: Dict[Text, Tuple[Text, Text]]
self._data = ManifestData(self) # type: ManifestData
self._reftest_nodes_by_url = None # type: Optional[Dict[Text, Union[RefTest, RefTestNode]]]
self.tests_root = tests_root # type: Optional[str]
self.url_base = url_base # type: Text
def __iter__(self):
# type: () -> Iterator[Tuple[str, Text, Set[ManifestItem]]]
return self.itertypes()
def itertypes(self, *types):
# type: (*str) -> Iterator[Tuple[str, Text, Set[ManifestItem]]]
for item_type in (types or sorted(self._data.keys())):
for path in sorted(self._data[item_type]):
tests = self._data[item_type][path]
yield item_type, path, tests
def iterpath(self, path):
# type: (Text) -> Iterator[ManifestItem]
for type_tests in self._data.values():
i = type_tests.get(path, set())
assert i is not None
for test in i:
yield test
def iterdir(self, dir_name):
# type: (Text) -> Iterator[ManifestItem]
if not dir_name.endswith(os.path.sep):
dir_name = dir_name + os.path.sep
for type_tests in self._data.values():
for path, tests in type_tests.iteritems():
if path.startswith(dir_name):
for test in tests:
yield test
def reftest_nodes_by_url(self):
# type: () -> Dict[Text, Union[RefTest, RefTestNode]]
if self._reftest_nodes_by_url is None:
by_url = {}
for path, nodes in itertools.chain(iteritems(self._data["reftest"]),
for node in nodes:
assert isinstance(node, (RefTest, RefTestNode))
by_url[node.url] = node
self._reftest_nodes_by_url = by_url
return self._reftest_nodes_by_url
def get_reference(self, url):
# type: (Text) -> Optional[ManifestItem]
return self.reftest_nodes_by_url.get(url)
def update(self, tree):
# type: (Iterable[Tuple[Union[SourceFile, bytes], bool]]) -> bool
"""Update the manifest given an iterable of items that make up the updated manifest.
The iterable must either generate tuples of the form (SourceFile, True) for paths
that are to be updated, or (path, False) for items that are not to be updated. This
unusual API is designed as an optimistaion meaning that SourceFile items need not be
constructed in the case we are not updating a path, but the absence of an item from
the iterator may be used to remove defunct entries from the manifest."""
all_reftest_nodes = [] # type: List[Tuple[ManifestItem, Text]]
seen_files = set() # type: Set[Text]
changed = False
reftest_changes = False
# Create local variable references to these dicts so we avoid the
# attribute access in the hot loop below
path_hash = self._path_hash # type: Dict[Text, Tuple[Text, Text]]
data = self._data
prev_files = data.paths() # type: Set[Text]
reftest_types = ("reftest", "reftest_node")
for source_file, update in tree:
if not update:
assert isinstance(source_file, (binary_type, text_type))
rel_path = source_file # type: Text
assert rel_path in path_hash
old_hash, old_type = path_hash[rel_path] # type: Tuple[Text, Text]
if old_type in reftest_types:
manifest_items = data[old_type][rel_path] # type: Iterable[ManifestItem]
all_reftest_nodes.extend((item, old_hash) for item in manifest_items)
assert not isinstance(source_file, bytes)
rel_path = source_file.rel_path
file_hash = source_file.hash # type: Text
is_new = rel_path not in path_hash # type: bool
hash_changed = False # type: bool
if not is_new:
old_hash, old_type = path_hash[rel_path]
if old_hash != file_hash:
new_type, manifest_items = source_file.manifest_items()
hash_changed = True
if new_type != old_type:
del data[old_type][rel_path]
if old_type in reftest_types:
reftest_changes = True
new_type = old_type
if old_type in reftest_types:
manifest_items = data[old_type][rel_path]
new_type, manifest_items = source_file.manifest_items()
if new_type in reftest_types:
all_reftest_nodes.extend((item, file_hash) for item in manifest_items)
if is_new or hash_changed:
reftest_changes = True
elif is_new or hash_changed:
data[new_type][rel_path] = set(manifest_items)
if is_new or hash_changed:
path_hash[rel_path] = (file_hash, new_type)
changed = True
deleted = prev_files - seen_files
if deleted:
changed = True
for rel_path in deleted:
if rel_path in path_hash:
_, old_type = path_hash[rel_path]
if old_type in reftest_types:
reftest_changes = True
del path_hash[rel_path]
del data[old_type][rel_path]
except KeyError:
for test_data in itervalues(data):
if rel_path in test_data:
del test_data[rel_path]
if reftest_changes:
reftests, reftest_nodes, changed_hashes = self._compute_reftests(all_reftest_nodes)
reftest_data = data["reftest"]
for path, items in iteritems(reftests):
if MYPY:
reftest_data[path] = cast(Set[ManifestItem], items)
reftest_data[path] = items
reftest_node_data = data["reftest_node"]
for node_path, node_items in iteritems(reftest_nodes):
if MYPY:
reftest_node_data[node_path] = cast(Set[ManifestItem], node_items)
reftest_node_data[node_path] = node_items
return changed
def _compute_reftests(self,
reftest_nodes # type: List[Tuple[ManifestItem, Text]]
# type: (...) -> Tuple[Dict[Text, Set[RefTest]], Dict[Text, Set[RefTestNode]], Dict[Text, Tuple[Text, Text]]]
self._reftest_nodes_by_url = {}
has_inbound = set()
for item, _ in reftest_nodes:
assert isinstance(item, (RefTestNode, RefTest))
for ref_url, ref_type in item.references:
reftests = defaultdict(set) # type: Dict[Text, Set[RefTest]]
references = defaultdict(set) # type: Dict[Text, Set[RefTestNode]]
changed_hashes = {} # type: Dict[Text, Tuple[Text, Text]]
for item, file_hash in reftest_nodes:
assert isinstance(item, (RefTestNode, RefTest))
if item.url in has_inbound:
# This is a reference
if isinstance(item, RefTest):
item = item.to_RefTestNode()
changed_hashes[item.path] = (file_hash,
assert isinstance(item, RefTestNode)
if isinstance(item, RefTestNode):
item = item.to_RefTest()
changed_hashes[item.path] = (file_hash,
assert isinstance(item, RefTest)
assert isinstance(item, (RefTestNode, RefTest))
self._reftest_nodes_by_url[item.url] = item
return reftests, references, changed_hashes
def to_json(self):
# type: () -> Dict[Text, Any]
out_items = {
test_type: type_paths.to_json()
for test_type, type_paths in iteritems(self._data) if type_paths
rv = {"url_base": self.url_base,
"paths": {from_os_path(k): v for k, v in iteritems(self._path_hash)},
"items": out_items,
"version": CURRENT_VERSION} # type: Dict[Text, Any]
return rv
def from_json(cls, tests_root, obj, types=None):
# type: (str, Dict[Text, Any], Optional[Container[Text]]) -> Manifest
version = obj.get("version")
if version != CURRENT_VERSION:
raise ManifestVersionMismatch
self = cls(tests_root, url_base=obj.get("url_base", "/"))
if not hasattr(obj, "items") and hasattr(obj, "paths"):
raise ManifestError
self._path_hash = {to_os_path(k): v for k, v in iteritems(obj["paths"])}
for test_type, type_paths in iteritems(obj["items"]):
# Drop "stub" items, which are no longer supported but may be
# present when doing an incremental manifest update.
# See for background.
# TODO(MANIFESTv7): remove this condition
if test_type == "stub":
if test_type not in item_classes:
raise ManifestError
if types and test_type not in types:
self._data[test_type].set_json(tests_root, type_paths)
return self
def load(tests_root, manifest, types=None):
# type: (str, Union[IO[bytes], str], Optional[Container[Text]]) -> Optional[Manifest]
logger = get_logger()
logger.warning("Prefer load_and_update instead")
return _load(logger, tests_root, manifest, types)
__load_cache = {} # type: Dict[str, Manifest]
def _load(logger, # type: Logger
tests_root, # type: str
manifest, # type: Union[IO[bytes], str]
types=None, # type: Optional[Container[Text]]
allow_cached=True # type: bool
# type: (...) -> Optional[Manifest]
manifest_path = (manifest if isinstance(manifest, string_types)
if allow_cached and manifest_path in __load_cache:
return __load_cache[manifest_path]
if isinstance(manifest, string_types):
if os.path.exists(manifest):
logger.debug("Opening manifest at %s" % manifest)
logger.debug("Creating new manifest at %s" % manifest)
with open(manifest, "rb") as f:
rv = Manifest.from_json(tests_root,
except IOError:
return None
except ValueError:
logger.warning("%r may be corrupted", manifest)
return None
rv = Manifest.from_json(tests_root,
if allow_cached:
__load_cache[manifest_path] = rv
return rv
def load_and_update(tests_root, # type: bytes
manifest_path, # type: bytes
url_base, # type: Text
update=True, # type: bool
rebuild=False, # type: bool
metadata_path=None, # type: Optional[bytes]
cache_root=None, # type: Optional[bytes]
working_copy=True, # type: bool
types=None, # type: Optional[Container[Text]]
write_manifest=True, # type: bool
allow_cached=True # type: bool
# type: (...) -> Manifest
logger = get_logger()
manifest = None
if not rebuild:
manifest = _load(logger,
except ManifestVersionMismatch:"Manifest version changed, rebuilding")
if manifest is not None and manifest.url_base != url_base:"Manifest url base did not match, rebuilding")
manifest = None
if manifest is None:
manifest = Manifest(tests_root, url_base)
rebuild = True
update = True
if rebuild or update:
tree = vcs.get_tree(tests_root, manifest, manifest_path, cache_root,
working_copy, rebuild)
changed = manifest.update(tree)
if write_manifest and changed:
write(manifest, manifest_path)
return manifest
def write(manifest, manifest_path):
# type: (Manifest, bytes) -> None
dir_name = os.path.dirname(manifest_path)
if not os.path.exists(dir_name):
with open(manifest_path, "wb") as f:
# Use ',' instead of the default ', ' separator to prevent trailing
# spaces:
json.dump(manifest.to_json(), f,
sort_keys=True, indent=1, separators=(',', ': '))