/usr/lib/python3/dist-packages/checkbox/lib/resolver.py is in python3-checkbox 0.17.6-0ubuntu6.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 | #
# This file is part of Checkbox.
#
# Copyright 2008 Canonical Ltd.
#
# Checkbox is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License version 3,
# as published by the Free Software Foundation.
#
# Checkbox is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with Checkbox. If not, see <http://www.gnu.org/licenses/>.
#
from collections import defaultdict, OrderedDict
class Resolver:
"""
Main class. Instantiate with the root directory of your items.
"""
def __init__(self, key_func=None):
if key_func is None:
key_func = lambda k: k
self.key_func = key_func
self.items_added = OrderedDict()
self.depends = {} # Dependencies
self.rdepends = defaultdict(list) # Reverse dependencies
# Data used in _resolve method
self.items = None
self.items_blocked = None
self.resolved = False
def add(self, item, *dependencies):
"""
Add item as pending
"""
key = self.key_func(item)
if key in self.items_added:
raise Exception("%s: key already exists" % key)
# Dependencies bookkeeping
self.items_added[key] = item
self.depends[key] = list(dependencies)
for dependency in dependencies:
self.rdepends[dependency].append(key)
# Circular dependencies check
def circular_dependencies(node):
seen = circular_dependencies.seen
for dependency in self.depends.get(node, []):
if key == dependency:
raise Exception("circular dependency involving "
"%s and %s" % (key, node))
if dependency in seen:
continue
seen.add(dependency)
circular_dependencies(dependency)
circular_dependencies.seen = set((key,))
circular_dependencies(key)
# Resolve on next call to get_dependencies/get_dependents
self.resolved = False
def _resolve(self):
"""
Work through the pending items and reorder them properly
"""
if self.resolved:
return
# Initialize internal ordering data
self.items = OrderedDict()
self.items_blocked = {}
def add_unblocked(key):
"""Add items that have been unblocked"""
for dependent in self.rdepends[key]:
if not dependent in self.items_blocked:
continue
unblocked = all(dependency in self.items
for dependency in self.depends[dependent]
if dependency in self.items_added)
if unblocked:
item = self.items_blocked[dependent]
self.items[dependent] = item
del self.items_blocked[dependent]
add_unblocked(dependent)
for key, item in self.items_added.items():
# Avoid adding an item until all dependencies have been met
blocked = any(dependency not in self.items
for dependency in self.depends[key]
if dependency in self.items_added)
if blocked:
self.items_blocked[key] = item
else:
self.items[key] = item
add_unblocked(key)
if self.items_blocked:
raise Exception('There are {} items blocked: {}'
.format(len(self.items_blocked),
', '.join(self.items_blocked.keys())))
# Don't resolve again on next call to get_dependencies/get_dependents
# unless a new item is added
self.resolved = True
def get_dependencies(self, item):
"""
Return a list of the dependencies for a given item
"""
self._resolve()
key = self.key_func(item)
if key not in self.depends:
msg = "no dependencies found for %s" % key
raise Exception(msg)
dependencies = self.depends[key] + [key]
return dependencies
def get_dependents(self, item=None):
"""
Return a list of the items that depend on the given one
or the whole list of items topologically ordered
"""
self._resolve()
items = list(self.items.values())
if item is not None:
index = items.index(item)
items = items[index + 1:]
return items
|