« back to all changes in this revision
Viewing changes to kernel_series.py
- browse files at revision 1426
- compare with another revision
- download diff
- download tarball
- view history from revision 1426
- Committer: Balint Reczey
- Date: 2020-12-18 12:26:22 UTC
- Revision ID: balint.reczey@canonical.com-20201218122622-nq4zud81cqg40bl4
Moved to git at https://git.launchpad.net/ubuntu-archive-tools
- files added:
- THIS_REPOSITORY_HAS_BEEN_MOVED_TO_GIT
- files removed:
- .bzrignore
- edit_acl.py
added
removed
kernel_series.py
1
#!/usr/bin/python3
2
#
3
4
from urllib.request import urlopen
5
6
import os
7
import yaml
8
9
class KernelRoutingEntry:
10
def __init__(self, ks, source, data):
11
name = "{}:{}".format(source.series.codename, source.name)
12
if isinstance(data, str):
13
name = data
14
table = source.series.routing_table
15
if table is None:
16
raise ValueError("unable to map routing alias {}, "
17
"no series routing table".format(data))
18
if data not in table:
19
raise ValueError("unable to map routing alias {}, "
20
"not listed in series routing table".format(data))
21
data = table[data]
22
23
# Clear out any entries that have been overriden to None.
24
for entry in dict(data):
25
if data[entry] is None:
26
del data[entry]
27
28
self._ks = ks
29
self._source = source
30
self._name = name
31
self._data = data if data else {}
32
33
@property
34
def source(self):
35
return self._source
36
37
@property
38
def name(self):
39
return self._name
40
41
def __eq__(self, other):
42
if isinstance(self, other.__class__):
43
return list(self) == list(other)
44
return False
45
46
def __ne__(self, other):
47
return not self.__eq__(other)
48
49
def __iter__(self):
50
return iter(list(self._data.items()))
51
52
def __getitem__(self, which):
53
return self._data[which]
54
55
def lookup_destination(self, dest, primary=False):
56
data = self._data.get(dest, None)
57
if primary is False or data is None:
58
return data
59
return data[0]
60
61
def __str__(self):
62
return str(self._data)
63
64
65
class KernelRepoEntry:
66
def __init__(self, ks, owner, data):
67
if isinstance(data, list):
68
new_data = {'url': data[0]}
69
if len(data) == 1:
70
new_data['branch'] = 'master'
71
elif len(data) == 2:
72
new_data['branch'] = data[1]
73
data = new_data
74
75
self._ks = ks
76
self._owner = owner
77
self._data = data if data else {}
78
79
@property
80
def owner(self):
81
return self._owner
82
83
# XXX: should this object have a name ?
84
85
def __eq__(self, other):
86
if isinstance(self, other.__class__):
87
return self.url == other.url and self.branch == other.branch
88
return False
89
90
def __ne__(self, other):
91
return not self.__eq__(other)
92
93
@property
94
def url(self):
95
return self._data['url']
96
97
@property
98
def branch(self):
99
return self._data.get('branch', None)
100
101
def __str__(self):
102
return "{} {}".format(self.url, self.branch)
103
104
105
class KernelSnapEntry:
106
def __init__(self, ks, source, name, data):
107
self._ks = ks
108
self._source = source
109
self._name = name
110
self._data = data if data else {}
111
112
# Convert arches/track to publish-to form.
113
if 'publish-to' not in self._data:
114
if 'arches' in self._data:
115
publish_to = {}
116
for arch in self._data['arches']:
117
publish_to[arch] = [self._data.get('track', 'latest')]
118
self._data['publish-to'] = publish_to
119
120
# Convert stable to promote-to form.
121
if 'promote-to' not in self._data and 'stable' in self._data:
122
if self._data['stable'] is True:
123
self._data['promote-to'] = 'stable'
124
else:
125
self._data['promote-to'] = 'candidate'
126
# Assume no promote-to data to mean just to edge.
127
promote_to = self._data.get('promote-to', 'edge')
128
if isinstance(promote_to, str):
129
expand_promote_to = []
130
for risk in ('edge', 'beta', 'candidate', 'stable'):
131
expand_promote_to.append(risk)
132
if risk == promote_to:
133
break
134
self._data['promote-to'] = expand_promote_to
135
# Ensure we have stable when promote-to is present.
136
if 'promote-to' in self._data and 'stable' not in self._data:
137
if 'stable' in self._data['promote-to']:
138
self._data['stable'] = True
139
else:
140
self._data['stable'] = False
141
142
def __eq__(self, other):
143
if isinstance(self, other.__class__):
144
return self.name == other.name and self.source == other.source
145
return False
146
147
def __ne__(self, other):
148
return not self.__eq__(other)
149
150
@property
151
def series(self):
152
return self._source.series
153
154
@property
155
def source(self):
156
return self._source
157
158
@property
159
def name(self):
160
return self._name
161
162
@property
163
def repo(self):
164
data = self._data.get('repo', None)
165
if not data:
166
return None
167
return KernelRepoEntry(self._ks, self, data)
168
169
@property
170
def primary(self):
171
return self._data.get('primary', False)
172
173
@property
174
def gated(self):
175
return self._data.get('gated', False)
176
177
@property
178
def stable(self):
179
return self._data.get('stable', False)
180
181
@property
182
def qa(self):
183
return self._data.get('qa', False)
184
185
@property
186
def hw_cert(self):
187
return self._data.get('hw-cert', False)
188
189
@property
190
def arches(self):
191
# XXX: should this be []
192
return self._data.get('arches', None)
193
194
@property
195
def track(self):
196
return self._data.get('track', None)
197
198
@property
199
def publish_to(self):
200
return self._data.get('publish-to', None)
201
202
@property
203
def promote_to(self):
204
return self._data.get('promote-to', None)
205
206
def promote_to_risk(self, risk):
207
return risk in self._data.get('promote-to', [])
208
209
def __str__(self):
210
return "{} {}".format(str(self.source), self.name)
211
212
213
class KernelPackageEntry:
214
def __init__(self, ks, source, name, data):
215
self._ks = ks
216
self._source = source
217
self._name = name
218
self._data = data if data else {}
219
220
def __eq__(self, other):
221
if isinstance(self, other.__class__):
222
return self.name == other.name and self.source == other.source
223
return False
224
225
def __ne__(self, other):
226
return not self.__eq__(other)
227
228
@property
229
def series(self):
230
return self._source.series
231
232
@property
233
def source(self):
234
return self._source
235
236
@property
237
def name(self):
238
return self._name
239
240
@property
241
def type(self):
242
return self._data.get('type', None)
243
244
@property
245
def repo(self):
246
data = self._data.get('repo', None)
247
if not data:
248
return None
249
return KernelRepoEntry(self._ks, self, data)
250
251
def __str__(self):
252
return "{} {} {}".format(str(self.source), self.name, self.type)
253
254
255
class KernelSourceEntry:
256
def __init__(self, ks, series, name, data):
257
self._ks = ks
258
self._series = series
259
self._name = name
260
self._data = data if data else {}
261
262
def __eq__(self, other):
263
if isinstance(self, other.__class__):
264
return self.name == other.name and self.series == other.series
265
return False
266
267
def __ne__(self, other):
268
return not self.__eq__(other)
269
270
@property
271
def name(self):
272
return self._name
273
274
@property
275
def series(self):
276
return self._series
277
278
@property
279
def versions(self):
280
if 'versions' in self._data:
281
return self._data['versions']
282
283
derived_from = self.derived_from
284
if derived_from is not None:
285
return derived_from.versions
286
287
copy_forward = self.copy_forward
288
if copy_forward is not None:
289
return copy_forward.versions
290
291
# XXX: should this be []
292
return None
293
294
@property
295
def version(self):
296
versions = self.versions
297
if not versions:
298
return None
299
return versions[-1]
300
301
@property
302
def development(self):
303
return self._data.get('development', self.series.development)
304
305
@property
306
def supported(self):
307
return self._data.get('supported', self.series.supported)
308
309
@property
310
def severe_only(self):
311
return self._data.get('severe-only', False)
312
313
@property
314
def stakeholder(self):
315
return self._data.get('stakeholder', None)
316
317
@property
318
def packages(self):
319
# XXX: should this return None when empty
320
result = []
321
packages = self._data.get('packages')
322
if packages:
323
for package_key, package in list(packages.items()):
324
result.append(KernelPackageEntry(self._ks, self, package_key, package))
325
return result
326
327
def lookup_package(self, package_key):
328
packages = self._data.get('packages')
329
if not packages or package_key not in packages:
330
return None
331
return KernelPackageEntry(self._ks, self, package_key, packages[package_key])
332
333
@property
334
def snaps(self):
335
# XXX: should this return None when empty
336
result = []
337
snaps = self._data.get('snaps')
338
if snaps:
339
for snap_key, snap in list(snaps.items()):
340
result.append(KernelSnapEntry(self._ks, self, snap_key, snap))
341
return result
342
343
def lookup_snap(self, snap_key):
344
snaps = self._data.get('snaps')
345
if not snaps or snap_key not in snaps:
346
return None
347
return KernelSnapEntry(self._ks, self, snap_key, snaps[snap_key])
348
349
@property
350
def derived_from(self):
351
if 'derived-from' not in self._data:
352
return None
353
354
(series_key, source_key) = self._data['derived-from']
355
356
series = self._ks.lookup_series(series_key)
357
source = series.lookup_source(source_key)
358
359
return source
360
361
@property
362
def testable_flavours(self):
363
retval = []
364
if (self._data.get('testing') is not None and
365
self._data['testing'].get('flavours') is not None
366
):
367
for flavour in self._data['testing']['flavours'].keys():
368
fdata = self._data['testing']['flavours'][flavour]
369
# If we have neither arches nor clouds we represent a noop
370
if not fdata:
371
continue
372
arches = fdata.get('arches', None)
373
arches = arches if arches is not None else []
374
clouds = fdata.get('clouds', None)
375
clouds = clouds if clouds is not None else []
376
retval.append(KernelSourceTestingFlavourEntry(flavour, arches, clouds))
377
return retval
378
379
@property
380
def invalid_tasks(self):
381
retval = self._data.get('invalid-tasks', [])
382
if retval is None:
383
retval = []
384
return retval
385
386
@property
387
def copy_forward(self):
388
if 'copy-forward' not in self._data:
389
return None
390
391
# XXX: backwards compatibility.
392
if self._data['copy-forward'] is False:
393
return None
394
if self._data['copy-forward'] is True:
395
derived_from = self.derived_from
396
if derived_from is None:
397
return True
398
return self.derived_from
399
400
(series_key, source_key) = self._data['copy-forward']
401
402
series = self._ks.lookup_series(series_key)
403
source = series.lookup_source(source_key)
404
405
return source
406
407
@property
408
def backport(self):
409
return self._data.get('backport', False)
410
411
@property
412
def routing(self):
413
default = 'default'
414
if self.series.development:
415
default = 'devel'
416
if self.series.esm:
417
default = 'esm'
418
data = self._data.get('routing', default)
419
if data is None:
420
return data
421
return KernelRoutingEntry(self._ks, self, data)
422
423
@property
424
def swm_data(self):
425
return self._data.get('swm')
426
427
@property
428
def private(self):
429
return self._data.get('private', False)
430
431
def __str__(self):
432
return "{} {}".format(self.series.name, self.name)
433
434
class KernelSourceTestingFlavourEntry:
435
def __init__(self, name, arches, clouds):
436
self._name = name
437
self._arches = arches
438
self._clouds = clouds
439
440
@property
441
def name(self):
442
return self._name
443
444
@property
445
def arches(self):
446
return self._arches
447
448
@property
449
def clouds(self):
450
return self._clouds
451
452
class KernelSeriesEntry:
453
def __init__(self, ks, name, data, defaults=None):
454
self._ks = ks
455
self._name = name
456
self._data = {}
457
if defaults is not None:
458
self._data.update(defaults)
459
if data is not None:
460
self._data.update(data)
461
462
def __eq__(self, other):
463
if isinstance(self, other.__class__):
464
return self.name == other.name
465
return False
466
467
def __ne__(self, other):
468
return not self.__eq__(other)
469
470
@property
471
def name(self):
472
return self._name
473
474
@property
475
def codename(self):
476
return self._data.get('codename', None)
477
478
@property
479
def opening(self):
480
if 'opening' in self._data:
481
if self._data['opening'] is not False:
482
return True
483
return False
484
485
def opening_ready(self, *flags):
486
if 'opening' not in self._data:
487
return True
488
allow = self._data['opening']
489
if allow is None:
490
return False
491
if allow in (True, False):
492
return not allow
493
for flag in flags:
494
flag_allow = allow.get(flag, False)
495
if flag_allow is None or flag_allow is False:
496
return False
497
return True
498
opening_allow = opening_ready
499
500
@property
501
def development(self):
502
return self._data.get('development', False)
503
504
@property
505
def supported(self):
506
return self._data.get('supported', False)
507
508
@property
509
def lts(self):
510
return self._data.get('lts', False)
511
512
@property
513
def esm(self):
514
return self._data.get('esm', False)
515
516
def __str__(self):
517
return "{} ({})".format(self.name, self.codename)
518
519
@property
520
def sources(self):
521
result = []
522
sources = self._data.get('sources')
523
if sources:
524
for source_key, source in list(sources.items()):
525
result.append(KernelSourceEntry(
526
self._ks, self, source_key, source))
527
return result
528
529
@property
530
def routing_table(self):
531
return self._data.get('routing-table', None)
532
533
def lookup_source(self, source_key):
534
sources = self._data.get('sources')
535
if not sources or source_key not in sources:
536
return None
537
return KernelSourceEntry(self._ks, self, source_key, sources[source_key])
538
539
540
# KernelSeries
541
#
542
class KernelSeries:
543
_url = 'https://git.launchpad.net/~canonical-kernel/' \
544
'+git/kteam-tools/plain/info/kernel-series.yaml'
545
_url_local = 'file://' + os.path.realpath(os.path.join(os.path.dirname(__file__),
546
'..', 'info', 'kernel-series.yaml'))
547
#_url = 'file:///home/apw/git2/kteam-tools/info/kernel-series.yaml'
548
#_url = 'file:///home/work/kteam-tools/info/kernel-series.yaml'
549
_data_txt = {}
550
551
@classmethod
552
def __load_once(cls, url):
553
if url not in cls._data_txt:
554
response = urlopen(url)
555
data = response.read()
556
if not isinstance(data, str):
557
data = data.decode('utf-8')
558
cls._data_txt[url] = data
559
return cls._data_txt[url]
560
561
def __init__(self, url=None, data=None, use_local=os.getenv("USE_LOCAL_KERNEL_SERIES_YAML", False)):
562
if data or url:
563
if url:
564
response = urlopen(url)
565
data = response.read()
566
if not isinstance(data, str):
567
data = data.decode('utf-8')
568
else:
569
data = self.__load_once(self._url_local if use_local else self._url)
570
self._data = yaml.safe_load(data)
571
572
self._development_series = None
573
self._codename_to_series = {}
574
for series_key, series in list(self._data.items()):
575
if not series:
576
continue
577
if series.get('development', False):
578
self._development_series = series_key
579
if 'codename' in series:
580
self._codename_to_series[series['codename']] = series_key
581
582
# Pull out the defaults.
583
self._defaults_series = {}
584
if 'defaults' in self._data:
585
self._defaults_series = self._data['defaults']
586
del self._data['defaults']
587
588
@staticmethod
589
def key_series_name(series):
590
return [int(x) for x in series.name.split('.')]
591
592
@property
593
def series(self):
594
return [KernelSeriesEntry(self, series_key, series,
595
defaults=self._defaults_series)
596
for series_key, series in list(self._data.items())]
597
598
def lookup_series(self, series=None, codename=None, development=False):
599
if not series and not codename and not development:
600
raise ValueError("series/codename/development required")
601
if not series and codename:
602
if codename not in self._codename_to_series:
603
return None
604
series = self._codename_to_series[codename]
605
if not series and development:
606
if not self._development_series:
607
return None
608
series = self._development_series
609
if series and series not in self._data:
610
return None
611
return KernelSeriesEntry(self, series, self._data[series],
612
defaults=self._defaults_series)
613
614
615
if __name__ == '__main__':
616
db = KernelSeries()
617
618
series = db.lookup_series('16.04')
619
if series.name != '16.04':
620
print('series.name != 16.04')
621
if series.codename != 'xenial':
622
print('series.codename != xenial')
623
624
series2 = db.lookup_series(codename='xenial')
625
if series2.name != '16.04':
626
print('series2.name != 16.04')
627
if series2.codename != 'xenial':
628
print('series2.codename != xenial')
629
630
series3 = db.lookup_series(development=True)
631
if series3.name != '18.04':
632
print('series3.name != 18.04')
633
if series3.codename != 'bionic':
634
print('series3.codename != bionic')
635
636
print(str(series), str(series2), str(series3))
637
638
for series2 in sorted(db.series, key=db.key_series_name):
639
print(series2)
640
641
for source in series.sources:
642
print(str(source), source.series.name, source.name)
643
644
print(source.derived_from)
645
print(source.versions)
646
647
for package in source.packages:
648
print("PACKAGE", str(package))
649
650
for snap in source.snaps:
651
print("SNAP", str(snap), snap.arches)
652
653
654
# vi:set ts=4 sw=4 expandtab:
Loggerhead is a web-based interface for Breezy
Version: 2.0.1