blob: aed2634f4f790813a2d20ba4640c39388689a068 [file] [log] [blame] [edit]
# -*- coding: utf-8 -*-
# Copyright 2018 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
import logging
LOGGER = logging.getLogger(__name__)
from diagnostic_checks.cloud_storage_speedtest import CloudStorageSpeedTest
from diagnostic_checks.disk_info import DiskInfo
from diagnostic_checks.diagnostic_error import DiagnosticError
class DiagnosticCheckManager:
"""
Manages access to diagnostic checks. A diagnostic check
is an on demand test that gathers some information
on the system and returns a free-form text response back
to the user. They shouldn't perform any changes to the
system, just gather information.
Diagnostic checks are organized by a category (such as
network, cloud storage, disk), and then specific diagnostics
under those categories (network - speed, disk - free space).
Diagnostic check interface is defined by AbstractDiagnosticCheck,
and any new diagnostics must inherit from this class.
"""
# The list of diagnostic checks to provide
# All checks must inherit from AbstractDiagnosticCheck
CHECKS = [
CloudStorageSpeedTest,
DiskInfo,
]
def __init__(self):
self.diagnostic_checks = {}
def init_checks(self):
"""
Creates instances of all the supported checks
"""
for check in self.CHECKS:
self._add_check(check())
def _add_check(self, check):
if check.category not in self.diagnostic_checks:
self.diagnostic_checks[check.category] = {}
self.diagnostic_checks[check.category][check.name] = check
def list_diagnostic_checks(self):
"""
Lists the available diagnostic checks
Return a list of checks organized by category, then check:
[
{
'category': 'category 1',
'checks': [{
'name': 'a test under category 1',
'description': '...'
}]
},
{
'category': 'category 2',
'checks': [{
'name': 'a test under category 2',
'description': '...'
}]
}
]
"""
check_list = []
for category in self.diagnostic_checks:
category_entry = {"category": category}
category_checks = []
for name in self.diagnostic_checks[category]:
description = self.diagnostic_checks[category][
name
].description
category_checks.append(
{"name": name, "description": description}
)
category_entry["checks"] = category_checks
check_list.append(category_entry)
return check_list
def run_diagnostic_check(self, category, name):
"""
Runs the check specified by it's category and name
Args:
category: the category of the check to run
name: the name of the check to run
Return:
A freeform text response containing the results of the diagnostic
Raises:
DiagnosticError if the check is not found, or encounters an error
"""
try:
LOGGER.info("running diagnostic %s %s" % (category, name))
check = self.diagnostic_checks[category][name]
if check is None:
raise DiagnosticError("Check not found")
return check.run()
except KeyError:
raise DiagnosticError("Check not found")
except DiagnosticError as e:
LOGGER.error("failed to run diagnostic check: %s", str(e))
raise e