-
Notifications
You must be signed in to change notification settings - Fork 110
/
manifest.py
2346 lines (1943 loc) · 91 KB
/
manifest.py
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
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
# Copyright (c) 2018, 2019, 2020 Nordic Semiconductor ASA
# Copyright 2018, 2019 Foundries.io Ltd
#
# SPDX-License-Identifier: Apache-2.0
'''
Parser and abstract data types for west manifests.
'''
import configparser
import enum
import errno
import logging
import os
from pathlib import PurePosixPath, Path
import re
import shlex
import subprocess
import sys
from typing import Any, Callable, Dict, Iterable, List, NoReturn, \
NamedTuple, Optional, Set, Tuple, TYPE_CHECKING, Union
from packaging.version import parse as parse_version
import pykwalify.core
import yaml
from west import util
from west.util import PathType
import west.configuration as cfg
#
# Public constants
#
#: Index in a Manifest.projects attribute where the `ManifestProject`
#: instance for the workspace is stored.
MANIFEST_PROJECT_INDEX = 0
#: A git revision which points to the most recent `Project` update.
MANIFEST_REV_BRANCH = 'manifest-rev'
#: A fully qualified reference to `MANIFEST_REV_BRANCH`.
QUAL_MANIFEST_REV_BRANCH = 'refs/heads/' + MANIFEST_REV_BRANCH
#: Git ref space used by west for internal purposes.
QUAL_REFS_WEST = 'refs/west/'
#: The latest manifest schema version supported by this west program.
#:
#: This value changes when a new version of west includes new manifest
#: file features not supported by earlier versions of west.
SCHEMA_VERSION = '0.10'
# MAINTAINERS:
#
# If you want to update the schema version, you need to make sure that
# it has the exact same value as west.version.__version__ when the
# next release is cut.
#
# Internal helpers
#
# Type aliases
# The value of a west-commands as passed around during manifest
# resolution. It can become a list due to resolving imports, even
# though it's just a str in each individual file right now.
WestCommandsType = Union[str, List[str]]
# Type for the importer callback passed to the manifest constructor.
# (ImportedContentType is just an alias for what it gives back.)
ImportedContentType = Optional[Union[str, List[str]]]
ImporterType = Callable[['Project', str], ImportedContentType]
# Type for an import map filter function, which takes a Project and
# returns a bool. The various allowlists and blocklists are used to
# create these filter functions. A None value is treated as a function
# which always returns True.
ImapFilterFnType = Optional[Callable[['Project'], bool]]
# A list of group names to enable and disable, like ['+foo', '-bar'].
GroupFilterType = List[str]
# A list of group names belonging to a project, like ['foo', 'bar']
GroupsType = List[str]
# The parsed contents of a manifest YAML file as returned by _load(),
# after sanitychecking with validate().
ManifestDataType = Union[str, Dict]
# Logging
_logger = logging.getLogger(__name__)
# Type for the submodule value passed through the manifest file.
class Submodule(NamedTuple):
'''Represents a Git submodule within a project.'''
path: str
name: Optional[str] = None
# Submodules may be a list of values or a bool.
SubmodulesType = Union[List[Submodule], bool]
# Manifest locating, parsing, loading, etc.
class _defaults(NamedTuple):
remote: Optional[str]
revision: str
_DEFAULT_REV = 'master'
_WEST_YML = 'west.yml'
_SCHEMA_PATH = os.path.join(os.path.dirname(__file__), "manifest-schema.yml")
_SCHEMA_VER = parse_version(SCHEMA_VERSION)
_EARLIEST_VER_STR = '0.6.99' # we introduced the version feature after 0.6
_VALID_SCHEMA_VERS = [_EARLIEST_VER_STR, '0.7', '0.8', '0.9', SCHEMA_VERSION]
def _is_yml(path: PathType) -> bool:
return Path(path).suffix in ['.yml', '.yaml']
def _load(data: str) -> Any:
try:
return yaml.safe_load(data)
except yaml.scanner.ScannerError as e:
raise MalformedManifest(data) from e
def _west_commands_list(west_commands: Optional[WestCommandsType]) -> \
List[str]:
# Convert the raw data from a manifest file to a list of
# west_commands locations. (If it's already a list, make a
# defensive copy.)
if west_commands is None:
return []
elif isinstance(west_commands, str):
return [west_commands]
else:
return list(west_commands)
def _west_commands_maybe_delist(west_commands: List[str]) -> WestCommandsType:
# Convert a west_commands list to a string if there's
# just one element, otherwise return the list itself.
if len(west_commands) == 1:
return west_commands[0]
else:
return west_commands
def _west_commands_merge(wc1: List[str], wc2: List[str]) -> List[str]:
# Merge two west_commands lists, filtering out duplicates.
if wc1 and wc2:
return wc1 + [wc for wc in wc2 if wc not in wc1]
else:
return wc1 or wc2
def _mpath(cp: Optional[configparser.ConfigParser] = None,
topdir: Optional[PathType] = None) -> Tuple[str, str]:
# Return the value of the manifest.path configuration option
# in *cp*, a ConfigParser. If not given, create a new one and
# load configuration options with the given *topdir* as west
# workspace root.
#
# TODO: write a cfg.get(section, key)
# wrapper, with friends for update and delete, to avoid
# requiring this boilerplate.
if cp is None:
cp = cfg._configparser()
cfg.read_config(configfile=cfg.ConfigFile.LOCAL, config=cp, topdir=topdir)
try:
path = cp.get('manifest', 'path')
filename = cp.get('manifest', 'file', fallback=_WEST_YML)
return (path, filename)
except (configparser.NoOptionError, configparser.NoSectionError) as e:
raise MalformedConfig('no "manifest.path" config option is set') from e
# Manifest import handling
def _default_importer(project: 'Project', file: str) -> NoReturn:
raise ManifestImportFailed(project, file)
def _manifest_content_at(project: 'Project', path: PathType,
rev: str = QUAL_MANIFEST_REV_BRANCH) \
-> ImportedContentType:
# Get a list of manifest data from project at path
#
# The data are loaded from Git at ref QUAL_MANIFEST_REV_BRANCH,
# *NOT* the file system.
#
# If path is a tree at that ref, the contents of the YAML files
# inside path are returned, as strings. If it's a file at that
# ref, it's a string with its contents.
#
# Though this module and the "west update" implementation share
# this code, it's an implementation detail, not API.
path = os.fspath(path)
_logger.debug(f'{project.name}: looking up path {path} type at {rev}')
# Returns 'blob', 'tree', etc. for path at revision, if it exists.
out = project.git(['ls-tree', rev, path], capture_stdout=True,
capture_stderr=True).stdout
if not out:
# It's a bit inaccurate to raise FileNotFoundError for
# something that isn't actually file, but this is internal
# API, and git is a content addressable file system, so close
# enough!
raise OSError(errno.ENOENT, os.strerror(errno.ENOENT), path)
ptype = out.decode('utf-8').split()[1]
if ptype == 'blob':
# Importing a file: just return its content.
return project.read_at(path, rev=rev).decode('utf-8')
elif ptype == 'tree':
# Importing a tree: return the content of the YAML files inside it.
ret = []
# Use a PurePosixPath because that's the form git seems to
# store internally, even on Windows.
pathobj = PurePosixPath(path)
for f in filter(_is_yml, project.listdir_at(path, rev=rev)):
ret.append(project.read_at(pathobj / f, rev=rev).decode('utf-8'))
return ret
else:
raise MalformedManifest(f"can't decipher project {project.name} "
f'path {path} revision {rev} '
f'(git type: {ptype})')
class _import_map(NamedTuple):
file: str
name_allowlist: List[str]
path_allowlist: List[str]
name_blocklist: List[str]
path_blocklist: List[str]
path_prefix: str
def _is_imap_list(value: Any) -> bool:
# Return True if the value is a valid import map 'blocklist' or
# 'allowlist'. Empty strings and lists are OK, and list nothing.
return (isinstance(value, str) or
(isinstance(value, list) and
all(isinstance(item, str) for item in value)))
def _imap_filter(imap: _import_map) -> ImapFilterFnType:
# Returns either None (if no filter is necessary) or a
# filter function for the given import map.
if any([imap.name_allowlist, imap.path_allowlist,
imap.name_blocklist, imap.path_blocklist]):
return lambda project: _is_imap_ok(imap, project)
else:
return None
def _ensure_list(item: Union[str, List[str]]) -> List[str]:
# Converts item to a list containing it if item is a string, or
# returns item.
if isinstance(item, str):
return [item]
return item
def _is_imap_ok(imap: _import_map, project: 'Project') -> bool:
# Return True if a project passes an import map's filters,
# and False otherwise.
nwl, pwl, nbl, pbl = [_ensure_list(lst) for lst in
(imap.name_allowlist, imap.path_allowlist,
imap.name_blocklist, imap.path_blocklist)]
name = project.name
path = Path(project.path)
blocked = (name in nbl) or any(path.match(p) for p in pbl)
allowed = (name in nwl) or any(path.match(p) for p in pwl)
no_allowlists = not (nwl or pwl)
if blocked:
return allowed
else:
return allowed or no_allowlists
class _import_ctx(NamedTuple):
# Holds state that changes as we recurse down the manifest import tree.
# The current map from already-defined project names to Projects.
#
# This is shared, mutable state between Manifest() constructor
# calls that happen during resolution. We mutate this directly
# when handling 'manifest: projects:' lists. Manifests which are
# imported earlier get higher precedence: if a 'projects:' list
# contains a name which is already present here, we ignore that
# element.
projects: Dict[str, 'Project']
# The current shared group filter. This is mutable state in the
# same way 'projects' is. Manifests which are imported earlier get
# higher precedence here too.
#
# This is done by prepending (NOT appending) any 'manifest:
# group-filter:' lists we encounter during import resolution onto
# this list. Since group-filter lists have "last entry wins"
# semantics, earlier manifests take precedence.
group_filter: GroupFilterType
# The current restrictions on which projects the importing
# manifest is interested in.
#
# These accumulate as we pick up additional allowlists and
# blocklists in 'import: <map>' values. We handle this composition
# using _compose_ctx_and_imap().
imap_filter: ImapFilterFnType
# The current prefix which should be added to any project paths
# as defined by all the importing manifests up to this point.
# These accumulate as we pick up 'import: path-prefix: ...' values,
# also using _compose_ctx_and_imap().
path_prefix: Path
def _compose_ctx_and_imap(ctx: _import_ctx, imap: _import_map) -> _import_ctx:
# Combine the map data from "some-map" in a manifest's
# "import: some-map" into an existing import context type,
# returning the new context.
return _import_ctx(projects=ctx.projects,
group_filter=ctx.group_filter,
imap_filter=_compose_imap_filters(ctx.imap_filter,
_imap_filter(imap)),
path_prefix=ctx.path_prefix / imap.path_prefix)
def _imap_filter_allows(imap_filter: ImapFilterFnType,
project: 'Project') -> bool:
# imap_filter(project) if imap_filter is not None; True otherwise.
return (imap_filter is None) or imap_filter(project)
def _compose_imap_filters(imap_filter1: ImapFilterFnType,
imap_filter2: ImapFilterFnType) -> ImapFilterFnType:
# Return an import map filter which gives back the logical AND of
# what the two argument filter functions would return.
if imap_filter1 and imap_filter2:
# These type annotated versions silence mypy warnings.
fn1: Callable[['Project'], bool] = imap_filter1
fn2: Callable[['Project'], bool] = imap_filter2
return lambda project: (fn1(project) and fn2(project))
else:
return imap_filter1 or imap_filter2
_RESERVED_GROUP_RE = re.compile(r'(^[+-]|[\s,:])')
_INVALID_PROJECT_NAME_RE = re.compile(r'([/\\])')
def _update_disabled_groups(disabled_groups: Set[str],
group_filter: GroupFilterType):
# Update a set of disabled groups in place based on
# 'group_filter'.
for item in group_filter:
if item.startswith('-'):
disabled_groups.add(item[1:])
elif item.startswith('+'):
group = item[1:]
if group in disabled_groups:
disabled_groups.remove(group)
else:
# We should never get here. This private helper is only
# meant to be invoked on valid data.
assert False, \
(f"Unexpected group filter item {item}. "
"This is a west bug. Please report it to the developers "
"along with as much information as you can, such as the "
"stack trace that preceded this message.")
def _is_submodule_dict_ok(subm: Any) -> bool:
# Check whether subm is a dict that contains the expected
# submodule fields of proper types.
class _failed(Exception):
pass
def _assert(cond):
if not cond:
raise _failed()
try:
_assert(isinstance(subm, dict))
# Required key
_assert('path' in subm)
# Allowed keys
for k in subm:
_assert(k in ['path', 'name'])
_assert(isinstance(subm[k], str))
except _failed:
return False
return True
#
# Public functions
#
def manifest_path() -> str:
'''Absolute path of the manifest file in the current workspace.
Exceptions raised:
- `west.util.WestNotFound` if called from outside of a west
workspace
- `MalformedConfig` if the configuration file has no
``manifest.path`` key
- ``FileNotFoundError`` if no manifest file exists as determined by
``manifest.path`` and ``manifest.file``
'''
(mpath, mname) = _mpath()
ret = os.path.join(util.west_topdir(), mpath, mname)
# It's kind of annoying to manually instantiate a FileNotFoundError.
# This seems to be the best way.
if not os.path.isfile(ret):
raise OSError(errno.ENOENT, os.strerror(errno.ENOENT), ret)
return ret
def validate(data: Any) -> None:
'''Validate manifest data
Raises an exception if the manifest data is not valid for loading
by this version of west. (Actually attempting to load the data may
still fail if the it contains imports which cannot be resolved.)
:param data: YAML manifest data as a string or object
'''
if isinstance(data, str):
as_str = data
data = _load(data)
if not isinstance(data, dict):
raise MalformedManifest(f'{as_str} is not a YAML dictionary')
elif not isinstance(data, dict):
raise TypeError(f'{data} has type {type(data)}, '
'expected valid manifest data')
if 'manifest' not in data:
raise MalformedManifest('manifest data contains no "manifest" key')
data = data['manifest']
# Make sure this version of west can load this manifest data.
# This has to happen before the schema check -- later schemas
# may incompatibly extend this one.
if 'version' in data:
# As a convenience for the user, convert floats to strings.
# This avoids forcing them to write:
#
# version: "0.8"
#
# by explicitly allowing:
#
# version: 0.8
if not isinstance(data['version'], str):
min_version_str = str(data['version'])
casted_to_str = True
else:
min_version_str = data['version']
casted_to_str = False
min_version = parse_version(min_version_str)
if min_version > _SCHEMA_VER:
raise ManifestVersionError(min_version_str)
if min_version_str not in _VALID_SCHEMA_VERS:
msg = (f'invalid version {min_version_str}; must be one of: ' +
', '.join(_VALID_SCHEMA_VERS))
if casted_to_str:
msg += ('. Do you need to quote the value '
'(e.g. "0.10" instead of 0.10)?')
raise MalformedManifest(msg)
try:
pykwalify.core.Core(source_data=data,
schema_files=[_SCHEMA_PATH]).validate()
except pykwalify.errors.SchemaError as se:
raise MalformedManifest(se.msg) from se
# A 'raw' element in a project 'groups:' or manifest 'group-filter:' list,
# as it is parsed from YAML, before conversion to string.
RawGroupType = Union[str, int, float]
def is_group(raw_group: RawGroupType) -> bool:
'''Is a 'raw' project group value 'raw_group' valid?
Valid groups are strings that don't contain whitespace, commas
(","), or colons (":"), and do not start with "-" or "+".
As a special case, groups may also be nonnegative numbers, to
avoid forcing users to quote these values in YAML files.
:param raw_group: the group value to check
'''
# Implementation notes:
#
# - not starting with "-" because "-foo" means "disable group
# foo", and not starting with "+" because "+foo" means
# "enable group foo".
#
# - no commas because that's a separator character in
# manifest.group-filter and 'west update --group-filter'
#
# - no whitespace mostly to guarantee that printing
# comma-separated lists of groups won't cause 'word' breaks
# in 'west list' pipelines to cut(1) or similar
#
# - no colons to reserve some namespace for potential future
# use; we might want to do something like
# "--group-filter=path-prefix:foo" to create additional logical
# groups based on the workspace layout or other metadata
return ((raw_group >= 0) if isinstance(raw_group, (float, int)) else
bool(raw_group and not _RESERVED_GROUP_RE.search(raw_group)))
#
# Exception types
#
class MalformedManifest(Exception):
'''Manifest parsing failed due to invalid data.
'''
class MalformedConfig(Exception):
'''The west configuration was malformed in a way that made a
manifest operation fail.
'''
class ManifestImportFailed(Exception):
'''An operation required to resolve a manifest failed.
Attributes:
- ``project``: the Project instance with the missing manifest data
- ``filename``: the missing file, as a str
'''
def __init__(self, project: 'Project', filename: PathType):
super().__init__(project, filename)
self.project = project
self.filename = os.fspath(filename)
def __str__(self):
return (f'ManifestImportFailed: project {self.project} '
f'file {self.filename}')
class ManifestVersionError(Exception):
'''The manifest required a version of west more recent than the
current version.
'''
def __init__(self, version: str, file: Optional[PathType] = None):
super().__init__(version, file)
self.version = version
'''The minimum version of west that was required.'''
self.file = os.fspath(file) if file else None
'''The file that required this version of west, if any.'''
class _ManifestImportDepth(ManifestImportFailed):
# A hack to signal to main.py what happened.
pass
#
# The main Manifest class and its public helper types, like Project
# and ImportFlag.
#
class ImportFlag(enum.IntFlag):
'''Bit flags for handling imports when resolving a manifest.
Note that any "path-prefix:" values set in an "import:" still take
effect for the project itself even when IGNORE or IGNORE_PROJECTS are
given. For example, in this manifest::
manifest:
projects:
- name: foo
import:
path-prefix: bar
Project 'foo' has path 'bar/foo' regardless of whether IGNORE or
IGNORE_PROJECTS is given. This ensures the Project has the same path
attribute as it normally would if imported projects weren't being
ignored.
'''
#: The default value, 0, reads the file system to resolve
#: "self: import:", and runs git to resolve a "projects:" import.
DEFAULT = 0
#: Ignore projects added via "import:" in "self:" and "projects:"
IGNORE = 1
#: Always invoke importer callback for "projects:" imports
FORCE_PROJECTS = 2
#: Ignore projects added via "import:" : in "projects:" only;
#: including any projects added via "import:" : in "self:"
IGNORE_PROJECTS = 4
def _flags_ok(flags: ImportFlag) -> bool:
# Sanity-check the combination of flags.
F_I = ImportFlag.IGNORE
F_FP = ImportFlag.FORCE_PROJECTS
F_IP = ImportFlag.IGNORE_PROJECTS
if (flags & F_I) or (flags & F_IP):
return not flags & F_FP
elif flags & (F_FP | F_IP):
return bool((flags & F_FP) ^ (flags & F_IP))
else:
return True
class Project:
'''Represents a project defined in a west manifest.
Attributes:
- ``name``: project's unique name
- ``url``: project fetch URL
- ``revision``: revision to fetch from ``url`` when the
project is updated
- ``path``: relative path to the project within the workspace
(i.e. from ``topdir`` if that is set)
- ``abspath``: absolute path to the project in the native path name
format (or ``None`` if ``topdir`` is)
- ``posixpath``: like ``abspath``, but with slashes (``/``) as
path separators
- ``clone_depth``: clone depth to fetch when first cloning the
project, or ``None`` (the revision should not be a SHA
if this is used)
- ``west_commands``: list of YAML files where extension commands in
the project are declared
- ``topdir``: the top level directory of the west workspace
the project is part of, or ``None``
- ``remote_name``: the name of the remote which should be set up
when the project is being cloned (default: 'origin')
- ``groups``: the project's groups (as a list) as given in the manifest.
If the manifest data contains no groups for the project, this is
an empty list.
- ``submodules``: the project's submodules configuration; either
a list of Submodule objects, or a boolean.
- ``userdata``: the parsed 'userdata' field in the manifest, or None
'''
def __eq__(self, other):
return NotImplemented
def __repr__(self):
return (f'Project("{self.name}", "{self.url}", '
f'revision="{self.revision}", path={repr(self.path)}, '
f'clone_depth={self.clone_depth}, '
f'west_commands={self.west_commands}, '
f'topdir={repr(self.topdir)}, '
f'groups={repr(self.groups)}, '
f'userdata={repr(self.userdata)})')
def __str__(self):
path_repr = repr(self.abspath or self.path)
return f'<Project {self.name} ({path_repr}) at {self.revision}>'
def __init__(self, name: str, url: str,
revision: Optional[str] = None,
path: Optional[PathType] = None,
submodules: SubmodulesType = False,
clone_depth: Optional[int] = None,
west_commands: Optional[WestCommandsType] = None,
topdir: Optional[PathType] = None,
remote_name: Optional[str] = None,
groups: Optional[GroupsType] = None,
userdata: Optional[Any] = None):
'''Project constructor.
If *topdir* is ``None``, then absolute path attributes
(``abspath`` and ``posixpath``) will also be ``None``.
:param name: project's ``name:`` attribute in the manifest
:param url: fetch URL
:param revision: fetch revision
:param path: path (relative to topdir), or None for *name*
:param submodules: submodules to pull within the project
:param clone_depth: depth to use for initial clone
:param west_commands: path to a west commands specification YAML
file in the project, relative to its base directory,
or list of these
:param topdir: the west workspace's top level directory
:param remote_name: the name of the remote which should be
set up if the project is being cloned (default: 'origin')
:param groups: a list of groups found in the manifest data for
the project, after conversion to str and validation.
'''
self.name = name
self.url = url
self.submodules = submodules
self.revision = revision or _DEFAULT_REV
self.clone_depth = clone_depth
self.path = os.fspath(path or name)
self.west_commands = _west_commands_list(west_commands)
self.topdir = os.fspath(topdir) if topdir else None
self.remote_name = remote_name or 'origin'
self.groups: GroupsType = groups or []
self.userdata: Any = userdata
@property
def path(self) -> str:
return self._path
@path.setter
def path(self, path: PathType) -> None:
self._path: str = os.fspath(path)
# Invalidate the absolute path attributes. They'll get
# computed again next time they're accessed.
self._abspath: Optional[str] = None
self._posixpath: Optional[str] = None
@property
def abspath(self) -> Optional[str]:
if self._abspath is None and self.topdir:
self._abspath = os.path.abspath(Path(self.topdir) /
self.path)
return self._abspath
@property
def posixpath(self) -> Optional[str]:
if self._posixpath is None and self.abspath is not None:
self._posixpath = Path(self.abspath).as_posix()
return self._posixpath
@property
def name_and_path(self) -> str:
return f'{self.name} ({self.path})'
def as_dict(self) -> Dict:
'''Return a representation of this object as a dict, as it
would be parsed from an equivalent YAML manifest.
'''
ret: Dict = {}
ret['name'] = self.name
ret['url'] = self.url
ret['revision'] = self.revision
if self.path != self.name:
ret['path'] = self.path
if self.clone_depth:
ret['clone-depth'] = self.clone_depth
if self.west_commands:
ret['west-commands'] = \
_west_commands_maybe_delist(self.west_commands)
if self.groups:
ret['groups'] = self.groups
if self.userdata:
ret['userdata'] = self.userdata
return ret
#
# Git helpers
#
def git(self, cmd: Union[str, List[str]],
extra_args: Iterable[str] = (),
capture_stdout: bool = False,
capture_stderr: bool = False,
check: bool = True,
cwd: Optional[PathType] = None) -> subprocess.CompletedProcess:
'''Run a git command in the project repository.
:param cmd: git command as a string (or list of strings)
:param extra_args: sequence of additional arguments to pass to
the git command (useful mostly if *cmd* is a string).
:param capture_stdout: if True, git's standard output is
captured in the ``CompletedProcess`` instead of being
printed.
:param capture_stderr: Like *capture_stdout*, but for standard
error. Use with caution: this may prevent error messages
from being shown to the user.
:param check: if given, ``subprocess.CalledProcessError`` is
raised if git finishes with a non-zero return code
:param cwd: directory to run git in (default: ``self.abspath``)
'''
if isinstance(cmd, str):
cmd_list = shlex.split(cmd)
else:
cmd_list = list(cmd)
extra_args = list(extra_args)
if cwd is None:
if self.abspath is not None:
cwd = self.abspath
else:
raise ValueError('no abspath; cwd must be given')
elif sys.version_info < (3, 6, 1) and not isinstance(cwd, str):
# Popen didn't accept a PathLike cwd on Windows until
# python v3.7; this was backported onto cpython v3.6.1,
# though. West currently supports "python 3.6", though, so
# in the unlikely event someone is running 3.6.0 on
# Windows, do the right thing.
cwd = os.fspath(cwd)
args = ['git'] + cmd_list + extra_args
cmd_str = util.quote_sh_list(args)
_logger.debug(f"running '{cmd_str}' in {cwd}")
popen = subprocess.Popen(
args, cwd=cwd,
stdout=subprocess.PIPE if capture_stdout else None,
stderr=subprocess.PIPE if capture_stderr else None)
stdout, stderr = popen.communicate()
# We use logger style % formatting here to avoid the
# potentially expensive overhead of formatting long
# stdout/stderr strings if the current log level isn't DEBUG,
# which is the usual case.
_logger.debug('"%s" exit code: %d stdout: %r stderr: %r',
cmd_str, popen.returncode, stdout, stderr)
if check and popen.returncode:
raise subprocess.CalledProcessError(popen.returncode, cmd_list,
output=stdout, stderr=stderr)
else:
return subprocess.CompletedProcess(popen.args, popen.returncode,
stdout, stderr)
def sha(self, rev: str, cwd: Optional[PathType] = None) -> str:
'''Get the SHA for a project revision.
:param rev: git revision (HEAD, v2.0.0, etc.) as a string
:param cwd: directory to run command in (default:
self.abspath)
'''
# Though we capture stderr, it will be available as the stderr
# attribute in the CalledProcessError raised by git() in
# Python 3.5 and above if this call fails.
cp = self.git(f'rev-parse {rev}^{{commit}}', capture_stdout=True,
cwd=cwd, capture_stderr=True)
# Assumption: SHAs are hex values and thus safe to decode in ASCII.
# It'll be fun when we find out that was wrong and how...
return cp.stdout.decode('ascii').strip()
def is_ancestor_of(self, rev1: str, rev2: str,
cwd: Optional[PathType] = None) -> bool:
'''Check if 'rev1' is an ancestor of 'rev2' in this project.
Returns True if rev1 is an ancestor commit of rev2 in the
given project; rev1 and rev2 can be anything that resolves to
a commit. (If rev1 and rev2 refer to the same commit, the
return value is True, i.e. a commit is considered an ancestor
of itself.) Returns False otherwise.
:param rev1: commit that could be the ancestor of *rev2*
:param rev2: commit that could be a descendant or *rev1*
:param cwd: directory to run command in (default:
``self.abspath``)
'''
rc = self.git(f'merge-base --is-ancestor {rev1} {rev2}',
check=False, cwd=cwd).returncode
if rc == 0:
return True
elif rc == 1:
return False
else:
raise RuntimeError(f'unexpected git merge-base result {rc}')
def is_up_to_date_with(self, rev: str,
cwd: Optional[PathType] = None) -> bool:
'''Check if the project is up to date with *rev*, returning
``True`` if so.
This is equivalent to ``is_ancestor_of(rev, 'HEAD',
cwd=cwd)``.
:param rev: base revision to check if project is up to date
with.
:param cwd: directory to run command in (default:
``self.abspath``)
'''
return self.is_ancestor_of(rev, 'HEAD', cwd=cwd)
def is_up_to_date(self, cwd: Optional[PathType] = None) -> bool:
'''Check if the project HEAD is up to date with the manifest.
This is equivalent to ``is_up_to_date_with(self.revision,
cwd=cwd)``.
:param cwd: directory to run command in (default:
``self.abspath``)
'''
return self.is_up_to_date_with(self.revision, cwd=cwd)
def is_cloned(self, cwd: Optional[PathType] = None) -> bool:
'''Returns ``True`` if ``self.abspath`` looks like a git
repository's top-level directory, and ``False`` otherwise.
:param cwd: directory to run command in (default:
``self.abspath``)
'''
if not self.abspath or not os.path.isdir(self.abspath):
return False
# --is-inside-work-tree doesn't require that the directory is
# the top-level directory of a Git repository. Use --show-cdup
# instead, which prints an empty string (i.e., just a newline,
# which we strip) for the top-level directory.
_logger.debug(f'{self.name}: checking if cloned')
res = self.git('rev-parse --show-cdup', check=False, cwd=cwd,
capture_stderr=True, capture_stdout=True)
return not (res.returncode or res.stdout.strip())
def read_at(self, path: PathType, rev: Optional[str] = None,
cwd: Optional[PathType] = None) -> bytes:
'''Read file contents in the project at a specific revision.
:param path: relative path to file in this project
:param rev: revision to read *path* from (default: ``self.revision``)
:param cwd: directory to run command in (default: ``self.abspath``)
'''
if rev is None:
rev = self.revision
cp = self.git(['show', f'{rev}:{os.fspath(path)}'],
capture_stdout=True, capture_stderr=True, cwd=cwd)
return cp.stdout
def listdir_at(self, path: PathType, rev: Optional[str] = None,
cwd: Optional[PathType] = None,
encoding: Optional[str] = None) -> List[str]:
'''List of directory contents in the project at a specific revision.
The return value is the directory contents as a list of files and
subdirectories.
:param path: relative path to file in this project
:param rev: revision to read *path* from (default: ``self.revision``)
:param cwd: directory to run command in (default: ``self.abspath``)
:param encoding: directory contents encoding (default: 'utf-8')
'''
if rev is None:
rev = self.revision
if encoding is None:
encoding = 'utf-8'
# git-ls-tree -z means we get NUL-separated output with no quoting
# of the file names. Using 'git-show' or 'git-cat-file -p'
# wouldn't work for files with special characters in their names.
out = self.git(['ls-tree', '-z', f'{rev}:{os.fspath(path)}'], cwd=cwd,
capture_stdout=True, capture_stderr=True).stdout
# A tab character separates the SHA from the file name in each
# NUL-separated entry.
return [f.decode(encoding).split('\t', 1)[1]
for f in out.split(b'\x00') if f]
# FIXME: this whole class should just go away. See #327.
class ManifestProject(Project):
'''Represents the manifest repository as a `Project`.
Meaningful attributes:
- ``name``: the string ``"manifest"``
- ``topdir``: the top level directory of the west workspace
the manifest project controls, or ``None``
- ``path``: relative path to the manifest repository within the
workspace, or ``None`` (i.e. from ``topdir`` if that is set)
- ``abspath``: absolute path to the manifest repository in the
native path name format (or ``None`` if ``topdir`` is)
- ``posixpath``: like ``abspath``, but with slashes (``/``) as
path separators
- ``west_commands``:``west_commands:`` key in the manifest's
``self:`` map. This may be a list of such if the self
section imports multiple additional files with west commands.
Other readable attributes included for Project compatibility:
- ``url``: the empty string; the west manifest is not
version-controlled by west itself, even though 'west init'
can fetch a manifest repository from a Git remote
- ``revision``: ``"HEAD"``
- ``clone_depth``: ``None``, because there's no URL
- ``groups``: the empty list
'''
def __repr__(self):
return (f'ManifestProject({self.name}, path={repr(self.path)}, '
f'west_commands={self.west_commands}, '
f'topdir={repr(self.topdir)})')
def __init__(self, path: Optional[PathType] = None,
west_commands: Optional[WestCommandsType] = None,
topdir: Optional[PathType] = None):
'''
:param path: Relative path to the manifest repository in the