/
post_rpmqa.py
72 lines (54 loc) · 2.38 KB
/
post_rpmqa.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
"""
Copyright (c) 2015 Red Hat, Inc
All rights reserved.
This software may be modified and distributed under the terms
of the BSD license. See the LICENSE file for details.
"""
from atomic_reactor.plugin import PostBuildPlugin
from atomic_reactor.utils.rpm import rpm_qf_args, parse_rpm_output
__all__ = ('PostBuildRPMqaPlugin', )
class PostBuildRPMqaPlugin(PostBuildPlugin):
key = "all_rpm_packages"
is_allowed_to_fail = False
sep = ';'
def __init__(self, tasker, workflow, image_id, ignore_autogenerated_gpg_keys=True):
"""
constructor
:param tasker: ContainerTasker instance
:param workflow: DockerBuildWorkflow instance
"""
# call parent constructor
super(PostBuildRPMqaPlugin, self).__init__(tasker, workflow)
self.image_id = image_id
self.ignore_autogenerated_gpg_keys = ignore_autogenerated_gpg_keys
self._container_ids = []
def run(self):
# If another component has already filled in the image component list, skip
if self.workflow.image_components is not None:
return None
if self.workflow.builder.dockerfile_images.base_from_scratch:
self.log.info("from scratch can't run rpmqa")
return None
plugin_output = self.gather_output()
# gpg-pubkey are autogenerated packages by rpm when you import a gpg key
# these are of course not signed, let's ignore those by default
if self.ignore_autogenerated_gpg_keys:
self.log.debug("ignore rpms 'gpg-pubkey'")
plugin_output = [x for x in plugin_output if not x.startswith("gpg-pubkey" + self.sep)]
self.tasker.cleanup_containers(*self._container_ids)
self.workflow.image_components = parse_rpm_output(plugin_output)
return plugin_output
def gather_output(self):
for _ in range(5):
container_id = self.tasker.run(
self.image_id,
command=rpm_qf_args(),
create_kwargs={"entrypoint": "/bin/rpm", "user": "root"},
start_kwargs={},
)
self._container_ids.append(container_id)
self.tasker.wait(container_id)
output = self.tasker.logs(container_id, stream=False)
if output:
return output
raise RuntimeError('Unable to gather list of installed packages in container')