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
|
"""Eclass specific support and addon."""
import os
from functools import total_ordering
from pkgcore.ebuild.eclass import EclassDoc
from snakeoil.klass import jit_attr_none
from snakeoil.mappings import ImmutableDict
from snakeoil.osutils import pjoin
from .. import base
from . import caches
@total_ordering
class Eclass:
"""Generic eclass object."""
def __init__(self, name, path):
self.name = name
self.path = os.path.realpath(path)
def __str__(self):
return self.name
@property
def lines(self):
try:
with open(self.path) as f:
return tuple(f)
except FileNotFoundError:
return ()
def __lt__(self, other):
if isinstance(other, Eclass):
return self.name < other.name
return self.name < other
def __hash__(self):
return hash(self.path)
def __eq__(self, other):
if isinstance(other, Eclass):
return self.path == other.path
return self.path == other
class EclassAddon(caches.CachedAddon):
"""Eclass support for various checks."""
# cache registry
cache = caches.CacheData(type="eclass", file="eclass.pickle", version=EclassDoc.ABI_VERSION)
def __init__(self, *args):
super().__init__(*args)
# mapping of repo locations to their corresponding eclass caches
self._eclass_repos = {}
@jit_attr_none
def eclasses(self, repo=None):
"""Mapping of available eclasses to eclass doc info."""
d = {}
for r in self.options.target_repo.trees:
d.update(self._eclass_repos.get(r.location, ()))
return ImmutableDict(d)
@jit_attr_none
def deprecated(self):
"""Mapping of deprecated eclasses to their replacements (if any)."""
d = {}
for r in self.options.target_repo.trees:
try:
for name, eclass in self._eclass_repos[r.location].items():
if eclass.deprecated:
d[name] = eclass.deprecated
except KeyError:
continue
return ImmutableDict(d)
def update_cache(self, force=False):
"""Update related cache and push updates to disk."""
for repo in self.options.target_repo.trees:
eclass_dir = pjoin(repo.location, "eclass")
cache_file = self.cache_file(repo)
cache_eclasses = False
eclasses = {}
if not force:
eclasses = self.load_cache(cache_file, fallback={})
# check for eclass removals
for name in list(eclasses):
if not os.path.exists(pjoin(eclass_dir, f"{name}.eclass")):
del eclasses[name]
cache_eclasses = True
# verify the repo has eclasses
try:
repo_eclasses = sorted(
(x[:-7], pjoin(eclass_dir, x))
for x in os.listdir(eclass_dir)
if x.endswith(".eclass")
)
except FileNotFoundError:
repo_eclasses = []
if repo_eclasses:
# padding for progress output
padding = max(len(x[0]) for x in repo_eclasses)
# check for eclass additions and updates
with base.ProgressManager(verbosity=self.options.verbosity) as progress:
for name, path in repo_eclasses:
try:
if os.path.getmtime(path) != eclasses[name].mtime:
raise KeyError
except (KeyError, AttributeError):
try:
progress(f"{repo} -- updating eclass cache: {name:<{padding}}")
eclasses[name] = EclassDoc(path, sourced=True, repo=repo)
cache_eclasses = True
except IOError:
continue
if cache_eclasses:
# reset jit attrs
self._eclasses = None
self._deprecated = None
# push cache updates to disk
data = caches.DictCache(eclasses, self.cache)
self.save_cache(data, cache_file)
self._eclass_repos[repo.location] = eclasses
|