Source code for insights.parsers.subscription_manager_list

"""
Subscription manager list outputs - command ``subscription-manager list``
=========================================================================

This module provides parsers for various list outputs of ``subscription-manager``.

Parsers provided by this module are:

SubscriptionManagerListConsumed - command ``subscription-manager list --consumed``
----------------------------------------------------------------------------------

SubscriptionManagerListInstalled - command ``subscription-manager list --installed``
------------------------------------------------------------------------------------
"""
import re
from datetime import datetime
import six
from insights.specs import Specs
from .. import parser, CommandParser
from . import keyword_search


[docs]class SubscriptionManagerList(CommandParser): """ A general object for parsing the output of ``subscription-manager list``. This should be subclassed to read the specific output - e.g. ``--consumed`` or ``--installed``. Attributes: records (list): A list of dict with the output info, it's empty when the ``error`` occurs error (str): The raised exception when there is traceback """
[docs] def parse_content(self, content): self.records = [] current_record = {} record_start_key = '' key = '' # The key currently in use key_val_re = re.compile('^(?P<key>\w[\w\s]+\w):\s+(?P<value>\w.*)$') cont_val_re = re.compile('^\s+(?P<value>\w.*)$') # The first line that matches the key_val_re is treated as the start # of a record. The header doesn't match because it doesn't have a # colon. Once that's read, if the record start key is seen again # it starts a new record. for line in content: if 'Traceback' in line: self.error = content[-1] break # Check for match of key/value line match = key_val_re.search(line) if match: key, value = match.group('key', 'value') if not record_start_key: record_start_key = key # Have we started a new record if key == record_start_key: # If we have an existing record, save it if current_record: self.records.append(current_record) current_record = {} current_record[key] = value # Do some type conversions and add-ons if key == 'Active' and value in ('True', 'False'): current_record[key] = (value == 'True') elif key in ('Starts', 'Ends'): try: current_record[key + ' timestamp'] = datetime.strptime( value, '%m/%d/%y' ) except ValueError: pass elif not record_start_key: # Ignore any lines before the first record key. continue # Check for value continuations match = cont_val_re.search(line) if match: # Add this value to the current key: if isinstance(current_record[key], six.string_types): # Convert the single string into a list current_record[key] = [ current_record[key], match.group('value') ] else: current_record[key].append(match.group('value')) # Save the last read record if we had one. if current_record: self.records.append(current_record)
[docs] def search(self, *args, **kwargs): """ Search for records that match the given keys and values. See the :func:`insights.parsers.keyword_search` function for more details on usage. Arguments: **kwargs: Key-value pairs of search parameters. Returns: (list): A list of records that matched the search criteria. Examples: >>> len(consumed.search(Service_Level='PREMIUM')) 1 >>> consumed.search(Provides__contains='Red Hat Enterprise Virtualization') [] """ return keyword_search(self.records, **kwargs)
[docs]@parser(Specs.subscription_manager_list_consumed) class SubscriptionManagerListConsumed(SubscriptionManagerList): """ Read the output of ``subscription-manager list --consumed``. Sample input file:: +-------------------------------------------+ Consumed Subscriptions +-------------------------------------------+ Subscription Name: Red Hat Enterprise Linux Server, Premium (1-2 sockets) (Up to 1 guest) Provides: Oracle Java (for RHEL Server) Red Hat Software Collections Beta (for RHEL Server) Red Hat Enterprise Linux Server Red Hat Beta SKU: RH0155783S Contract: 12345678 Account: 1000001 Serial: 0102030405060708090 Pool ID: 8a85f981477e5284014783abaf5d4dcd Active: True Quantity Used: 1 Service Level: PREMIUM Service Type: L1-L3 Status Details: Subscription is current Subscription Type: Standard Starts: 11/14/14 Ends: 07/06/15 System Type: Physical Examples: >>> type(consumed) <class 'insights.parsers.subscription_manager_list.SubscriptionManagerListConsumed'> >>> len(consumed.records) 1 >>> sub1 = consumed.records[0] >>> sub1['SKU'] 'RH0155783S' >>> sub1['Active'] # Type conversion on Active field True >>> sub1['Status Details'] # Keys appear as given 'Subscription is current' >>> sub1['Provides'][1] 'Red Hat Software Collections Beta (for RHEL Server)' >>> sub1['Starts'] # Basic field as text - note month/day/year '11/14/14' >>> sub1['Starts timestamp'].year 2014 >>> consumed.all_current # Are all subscriptions listed as current? True """ @property def all_current(self): """ (bool) Does every subscription record have the Status Details value set to 'Subscription is current'? """ return all( sub['Status Details'] == 'Subscription is current' for sub in self.records )
[docs]@parser(Specs.subscription_manager_list_installed) class SubscriptionManagerListInstalled(SubscriptionManagerList): """ Read the output of ``subscription-manager list --installed``. Sample input file:: +-------------------------------------------+ Installed Product Status +-------------------------------------------+ Product Name: Red Hat Software Collections (for RHEL Server) Product ID: 201 Version: 2 Arch: x86_64 Status: Subscribed Status Details: Starts: 04/27/15 Ends: 04/27/16 Product Name: Red Hat Enterprise Linux Server Product ID: 69 Version: 7.1 Arch: x86_64 Status: Subscribed Status Details: Starts: 04/27/15 Ends: 04/27/16 Examples: >>> type(installed) <class 'insights.parsers.subscription_manager_list.SubscriptionManagerListInstalled'> >>> len(installed.records) 2 >>> prod1 = installed.records[0] >>> prod1['Product ID'] # Note - not converted to number '201' >>> prod1['Starts'] # String date as is '04/27/15' >>> prod1['Starts timestamp'].year # Extra converted to date 2015 >>> installed.all_subscribed True """ @property def all_subscribed(self): """ (bool) Does every product record have the Status value set to 'Subscribed'? """ return all( sub['Status'] == 'Subscribed' for sub in self.records )