Add models_search and devices_by_name methods

Author: orgito

.gitignore

@@ -102,3 +102,6 @@ venv.bak/
 
 # mypy
 .mypy_cache/
+
+# Editors
+.vscode

setup.py

@@ -6,7 +6,7 @@
 
 setup(
     name='spectrum-client',
-    version='0.0.8',
+    version='0.0.9',
     description='CA Spectrum Web Services API wrapper',
     long_description=README,
     long_description_content_type='text/markdown',

spectrum_client/spectrum.py

@@ -22,57 +22,137 @@ class SpectrumClientParameterError(SpectrumClientException):
 
 
 class Spectrum(object):
-
-    _namespace = {'ca': 'http://www.ca.com/spectrum/restful/schema/response'}
+    xml_namespace = {'ca': 'http://www.ca.com/spectrum/restful/schema/response'}
+    default_attributes = '''
+    <rs:requested-attribute id="0x1006e"/> <!-- Model Name -->
+    <rs:requested-attribute id="0x1000a"/> <!-- Condition -->
+    <rs:requested-attribute id="0x11ee8"/> <!-- Model Class -->
+    <rs:requested-attribute id="0x129e7"/> <!-- Site ID -->
+    <rs:requested-attribute id="0x12d7f"/> <!-- IP Address -->
+    <rs:requested-attribute id="0x1290c"/> <!-- Criticality -->
+    <rs:requested-attribute id="0x10000"/> <!-- Model Type Name -->
+    <rs:requested-attribute id="0x23000e"/> <!-- Device Type -->
+    <rs:requested-attribute id="0x11d42"/> <!-- Landscape Name-->
+    '''
+    models_search_template = '''<?xml version="1.0" encoding="UTF-8"?>
+    <rs:model-request throttlesize="9999"
+    xmlns:rs="http://www.ca.com/spectrum/restful/schema/request"
+    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+    xsi:schemaLocation="http://www.ca.com/spectrum/restful/schema/request ../../../xsd/Request.xsd ">
+        <rs:target-models>
+            <rs:models-search>
+                <rs:search-criteria xmlns="http://www.ca.com/spectrum/restful/schema/filter">
+                    <filtered-models>
+                    {models_filter}
+                    </filtered-models>
+                </rs:search-criteria>
+            </rs:models-search>
+        </rs:target-models>
+    ''' + default_attributes + '</rs:model-request>'
 
     def __init__(self, url=SPECTRUM_URL, username=SPECTRUM_USERNAME, password=SPECTRUM_PASSWORD):
         if url is None:
             raise ValueError('Spectrum (OneClick) url must be provided either in the constructor or as an environment variable')
         self.url = url if not url.endswith('/') else url[:-1]
         self.auth = HTTPBasicAuth(username, password)
 
-    def _check_http(self, res):
-        if res.status_code == 401:
-            raise SpectrumClientAuthException('Authorization Failure. Invalid user name or password.')
-        res.raise_for_status()
-
     def _parse_get(self, res):
-        self._check_http(res)
+        self.check_http_response(res)
 
         root = ET.fromstring(res.text)
-        model_error = root.find('.//ca:model', self._namespace).get('error')
+        model_error = root.find('.//ca:model', self.xml_namespace).get('error')
         if model_error:
             raise SpectrumClientParameterError('Model Error: ' + model_error)
-        attr_error = root.find('.//ca:attribute', self._namespace).get('error')
+        attr_error = root.find('.//ca:attribute', self.xml_namespace).get('error')
         if attr_error:
             raise SpectrumClientParameterError(attr_error)
 
     def _parse_update(self, res):
-        self._check_http(res)
+        self.check_http_response(res)
 
         root = ET.fromstring(res.text)
-        if root.find('.//ca:model', self._namespace).get('error') == 'Success':
+        if root.find('.//ca:model', self.xml_namespace).get('error') == 'Success':
             return
 
-        if root.find('.//ca:model', self._namespace).get('error') == 'PartialFailure':
-            msg = root.find('.//ca:attribute', self._namespace).get('error-message')
+        if root.find('.//ca:model', self.xml_namespace).get('error') == 'PartialFailure':
+            msg = root.find('.//ca:attribute', self.xml_namespace).get('error-message')
         else:
-            msg = root.find('.//ca:model', self._namespace).get('error-message')
+            msg = root.find('.//ca:model', self.xml_namespace).get('error-message')
         raise SpectrumClientParameterError(msg)
 
+    @staticmethod
+    def check_http_response(res):
+        if res.status_code == 401:
+            raise SpectrumClientAuthException('Authorization Failure. Invalid user name or password.')
+        res.raise_for_status()
+
+    @staticmethod
+    def xml_landscape_filter(landscape):
+        xml = '''
+        <greater-than>
+            <attribute id="0x129fa">
+                <value>{}</value>
+            </attribute>
+        </greater-than>
+        <less-than>
+            <attribute id="0x129fa">
+                <value>{}</value>
+            </attribute>
+        </less-than>'''
+        landscape_start = hex(landscape)
+        landscape_end = hex(landscape + 0xfffff)
+        return xml.format(landscape_start, landscape_end).strip()
+
     def get_attribute(self, model_handle, attr_id):
         """Get an attribute of a Spectrum model.
 
         Arguments:
             model_handle {int} -- Model Handle of the model being queried.
             attr_id {int} -- Attribute ID of the attribute being queried.
         """
-        url = f'{self.url}/spectrum/restful/model/{hex(model_handle)}'
+        url = '{}/spectrum/restful/model/{}'.format(self.url, hex(model_handle))
         params = {'attr': hex(attr_id)}
         res = requests.get(url, params=params, auth=self.auth)
         self._parse_get(res)
         root = ET.fromstring(res.text)
-        return root.find('.//ca:attribute', self._namespace).text
+        return root.find('.//ca:attribute', self.xml_namespace).text
+
+    def devices_by_name(self, regex, landscape=None):
+        if landscape:
+            landscape_filter = self.xml_landscape_filter(landscape)
+        else:
+            landscape_filter = ''
+
+        models_filter = '''
+        <and>
+            <is-derived-from>
+                <model-type>0x1004b</model-type> <!-- Device -->
+            </is-derived-from>
+            {landscape_filter}
+            <has-pcre>
+                <attribute id="0x1006e">
+                    <value>{regex}</value>
+                </attribute>
+            </has-pcre>
+        </and>'''.format(regex=regex, landscape_filter=landscape_filter)
+        xml = self.models_search_template.format(models_filter=models_filter)
+        root = ET.fromstring(self.search_models(xml))
+        models = root.findall('.//ca:model', self.xml_namespace)
+        devices = {
+            model.get('mh'): {
+                attr.get('id'): attr.text for attr in model.getchildren()
+            } for model in models
+        }
+        return devices
+
+    def search_models(self, xml):
+        url = '{}/spectrum/restful/models'.format(self.url)
+        res = requests.post(url, xml, auth=self.auth)
+        self.check_http_response(res)
+        return res.text
+
+    def set_maintenance(self, model_handle, on=True):
+        return self.update_attribute(model_handle, 0x1295d, str(not on))
 
     def update_attribute(self, model_handle, attr_id, value):
         """Update an attribute of a Spectrum model.
@@ -82,11 +162,7 @@ def update_attribute(self, model_handle, attr_id, value):
             attr_id {int} -- Attribute ID of the attribute being updated.
             value {int or str} -- Value to set the attribute to.
         """
-
-        url = f'{self.url}/spectrum/restful/model/{hex(model_handle)}'
+        url = '{}/spectrum/restful/model/{}'.format(self.url, hex(model_handle))
         params = {'attr': hex(attr_id), 'val': value}
         res = requests.put(url, params=params, auth=self.auth)
         self._parse_update(res)
-
-    def set_maintenance(self, model_handle, on=True):
-        return self.update_attribute(model_handle, 0x1295d, str(not on))