← Back to branch summary

~mthaddon/codetree/built-location-updates

« back to all changes in this revision

Viewing changes to tests/test_data/cs-nrpe-33/hooks/charmhelpers/fetch/snap.py

  • Committer: Tom Haddon
  • Date: 2017-10-23 15:20:57 UTC
  • Revision ID: tom.haddon@canonical.com-20171023152057-ogr80fajq309imxh
Add cs:nrpe-33 to tests/test_data

Show diffs side-by-side

added added

removed removed

Lines of Context:
tests/test_data/cs-nrpe-33/hooks/charmhelpers/fetch/snap.py
 
1
# Copyright 2014-2017 Canonical Limited.
 
2
#
 
3
# Licensed under the Apache License, Version 2.0 (the "License");
 
4
# you may not use this file except in compliance with the License.
 
5
# You may obtain a copy of the License at
 
6
#
 
7
#  http://www.apache.org/licenses/LICENSE-2.0
 
8
#
 
9
# Unless required by applicable law or agreed to in writing, software
 
10
# distributed under the License is distributed on an "AS IS" BASIS,
 
11
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 
12
# See the License for the specific language governing permissions and
 
13
# limitations under the License.
 
14
"""
 
15
Charm helpers snap for classic charms.
 
16
 
 
17
If writing reactive charms, use the snap layer:
 
18
https://lists.ubuntu.com/archives/snapcraft/2016-September/001114.html
 
19
"""
 
20
import subprocess
 
21
import os
 
22
from time import sleep
 
23
from charmhelpers.core.hookenv import log
 
24
 
 
25
__author__ = 'Joseph Borg <joseph.borg@canonical.com>'
 
26
 
 
27
# The return code for "couldn't acquire lock" in Snap
 
28
# (hopefully this will be improved).
 
29
SNAP_NO_LOCK = 1
 
30
SNAP_NO_LOCK_RETRY_DELAY = 10  # Wait X seconds between Snap lock checks.
 
31
SNAP_NO_LOCK_RETRY_COUNT = 30  # Retry to acquire the lock X times.
 
32
SNAP_CHANNELS = [
 
33
    'edge',
 
34
    'beta',
 
35
    'candidate',
 
36
    'stable',
 
37
]
 
38
 
 
39
 
 
40
class CouldNotAcquireLockException(Exception):
 
41
    pass
 
42
 
 
43
 
 
44
def _snap_exec(commands):
 
45
    """
 
46
    Execute snap commands.
 
47
 
 
48
    :param commands: List commands
 
49
    :return: Integer exit code
 
50
    """
 
51
    assert type(commands) == list
 
52
 
 
53
    retry_count = 0
 
54
    return_code = None
 
55
 
 
56
    while return_code is None or return_code == SNAP_NO_LOCK:
 
57
        try:
 
58
            return_code = subprocess.check_call(['snap'] + commands,
 
59
                                                env=os.environ)
 
60
        except subprocess.CalledProcessError as e:
 
61
            retry_count += + 1
 
62
            if retry_count > SNAP_NO_LOCK_RETRY_COUNT:
 
63
                raise CouldNotAcquireLockException(
 
64
                    'Could not aquire lock after {} attempts'
 
65
                    .format(SNAP_NO_LOCK_RETRY_COUNT))
 
66
            return_code = e.returncode
 
67
            log('Snap failed to acquire lock, trying again in {} seconds.'
 
68
                .format(SNAP_NO_LOCK_RETRY_DELAY, level='WARN'))
 
69
            sleep(SNAP_NO_LOCK_RETRY_DELAY)
 
70
 
 
71
    return return_code
 
72
 
 
73
 
 
74
def snap_install(packages, *flags):
 
75
    """
 
76
    Install a snap package.
 
77
 
 
78
    :param packages: String or List String package name
 
79
    :param flags: List String flags to pass to install command
 
80
    :return: Integer return code from snap
 
81
    """
 
82
    if type(packages) is not list:
 
83
        packages = [packages]
 
84
 
 
85
    flags = list(flags)
 
86
 
 
87
    message = 'Installing snap(s) "%s"' % ', '.join(packages)
 
88
    if flags:
 
89
        message += ' with option(s) "%s"' % ', '.join(flags)
 
90
 
 
91
    log(message, level='INFO')
 
92
    return _snap_exec(['install'] + flags + packages)
 
93
 
 
94
 
 
95
def snap_remove(packages, *flags):
 
96
    """
 
97
    Remove a snap package.
 
98
 
 
99
    :param packages: String or List String package name
 
100
    :param flags: List String flags to pass to remove command
 
101
    :return: Integer return code from snap
 
102
    """
 
103
    if type(packages) is not list:
 
104
        packages = [packages]
 
105
 
 
106
    flags = list(flags)
 
107
 
 
108
    message = 'Removing snap(s) "%s"' % ', '.join(packages)
 
109
    if flags:
 
110
        message += ' with options "%s"' % ', '.join(flags)
 
111
 
 
112
    log(message, level='INFO')
 
113
    return _snap_exec(['remove'] + flags + packages)
 
114
 
 
115
 
 
116
def snap_refresh(packages, *flags):
 
117
    """
 
118
    Refresh / Update snap package.
 
119
 
 
120
    :param packages: String or List String package name
 
121
    :param flags: List String flags to pass to refresh command
 
122
    :return: Integer return code from snap
 
123
    """
 
124
    if type(packages) is not list:
 
125
        packages = [packages]
 
126
 
 
127
    flags = list(flags)
 
128
 
 
129
    message = 'Refreshing snap(s) "%s"' % ', '.join(packages)
 
130
    if flags:
 
131
        message += ' with options "%s"' % ', '.join(flags)
 
132
 
 
133
    log(message, level='INFO')
 
134
    return _snap_exec(['refresh'] + flags + packages)