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
|
# Copyright (C) 2009 Canonical
#
# Authors:
# Michael Vogt
#
# This program is free software; you can redistribute it and/or modify it under
# the terms of the GNU General Public License as published by the Free Software
# Foundation; version 3.
#
# This program is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
# FOR A PARTICULAR PURPOSE. See the GNU General Public License for more
# details.
#
# You should have received a copy of the GNU General Public License along with
# this program; if not, write to the Free Software Foundation, Inc.,
# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
import aptsources.distro
from datetime import datetime
import distro_info
from functools import wraps
import gi
gi.require_version("Gtk", "3.0")
gi.require_version("Handy", "1")
from gi.repository import Gio, Gtk, GLib, Handy
import json
import os
import subprocess
from softwareproperties.distro import is_current_distro_ubuntu
import logging
LOG=logging.getLogger(__name__)
import time
UA_STATUS_JSON = "/var/lib/ubuntu-advantage/status.json"
def setup_ui(self, path, domain):
# setup ui
self.builder = Gtk.Builder()
self.builder.set_translation_domain(domain)
self.builder.add_from_file(path)
self.builder.connect_signals(self)
for o in self.builder.get_objects():
if issubclass(type(o), Gtk.Buildable):
name = Gtk.Buildable.get_name(o)
setattr(self, name, o)
else:
logging.debug("can not get name for object '%s'" % o)
def has_gnome_online_accounts():
try:
d = Gio.DesktopAppInfo.new('gnome-online-accounts-panel.desktop')
return d != None
except Exception:
return False
def get_distro_info(distro):
if isinstance(distro, aptsources.distro.UbuntuDistribution):
return distro_info.UbuntuDistroInfo()
return distro_info.DebianDistroInfo()
def is_current_distro_lts():
if not is_current_distro_ubuntu():
return False
distro = aptsources.distro.get_distro()
di = get_distro_info(distro)
return di.is_lts(distro.codename)
def is_current_distro_supported():
distro = aptsources.distro.get_distro()
di = get_distro_info(distro)
return distro.codename in di.supported(datetime.now().date())
def current_distro():
distro = aptsources.distro.get_distro()
di = get_distro_info(distro)
releases = di.get_all(result="object")
for release in releases:
if release.series == distro.codename:
return release
def get_ua_status():
"""Return a dict of all UA status information or empty dict on error."""
# status.json will exist on any attached system. It will also be created
# by the systemd timer ua-timer which will update UA_STATUS_JSON every 12
# hours to reflect current status of UA subscription services.
# Invoking `pro status` with subp will result in a network call to
# contracts.canonical.com which could raise Timeouts on network limited
# machines. So, prefer the status.json file when possible.
if not is_current_distro_ubuntu():
return {}
status_json = ""
if os.path.exists(UA_STATUS_JSON):
with open(UA_STATUS_JSON) as stream:
status_json = stream.read()
else:
try:
# Success writes UA_STATUS_JSON
result = subprocess.run(
['pro', 'status', '--format=json'], stdout=subprocess.PIPE
)
except Exception as e:
print("Failed to run `pro status`:\n%s" % e)
return {}
if result.returncode != 0:
print(
"Ubuntu Pro client returned code %d" % result.returncode
)
return {}
status_json = result.stdout
if not status_json:
print(
"Warning: no Ubuntu Pro Client status found."
" Is ubuntu-pro-client installed?"
)
return {}
try:
status = json.loads(status_json)
except json.JSONDecodeError as e:
print("Failed to parse ubuntu advantage client JSON:\n%s" % e)
return {}
if status.get("_schema_version", "0.1") != "0.1":
print(
"Pro status schema version change: %s" % status["_schema_version"]
)
return status
def get_ua_service_status(service_name='esm-infra', status=None):
"""Get service availability and status for a specific UA service.
Return a tuple (available, service_status).
:boolean available: set True when either:
- attached contract is entitled to the service
- unattached machine reports service "availability" as "yes"
:str service_status: will be one of the following:
- "disabled" when the service is available and applicable but not
active
- "enabled" when the service is available and active
- "n/a" when the service is not applicable to the environment or not
entitled for the attached contract
"""
if not status:
status = get_ua_status()
# Assume unattached on empty status dict
available = False
service_status = "n/a"
for service in status.get("services", []):
if service.get("name") != service_name:
continue
if "available" in service:
available = bool("yes" == service["available"])
if "status" in service:
service_status = service["status"] # enabled, disabled or n/a
return (available, service_status)
def retry(exceptions, tries=10, delay=0.1, backoff=2):
"""
Retry calling the decorated function using an exponential backoff.
Args:
exceptions: The exception to check. may be a tuple of
exceptions to check.
tries: Number of times to try (not retry) before giving up.
delay: Initial delay between retries in seconds.
backoff: Backoff multiplier (e.g. value of 2 will double the delay
each retry).
"""
def deco_retry(f):
@wraps(f)
def f_retry(*args, **kwargs):
mtries, mdelay = tries, delay
while mtries > 1:
try:
return f(*args, **kwargs)
except exceptions as e:
msg = '{}, Retrying in {} seconds...'.format(e, mdelay)
logging.warning(msg)
time.sleep(mdelay)
mtries -= 1
mdelay *= backoff
return f(*args, **kwargs)
return f_retry # true decorator
return deco_retry
def is_dark_theme(widget):
env_gtk_theme = GLib.getenv("GTK_THEME")
if env_gtk_theme != None:
return GLib.str_has_suffix(env_gtk_theme, "dark")
return Handy.StyleManager.get_default().get_dark()
|
