Hera lives!

README.rst

@@ -0,0 +1,40 @@
+Hera
+====
+
+In mythology Hera was Zeus's wife (and sister!).  In this directory, Hera is an
+interface to interact with a `Zeus Traffic Manager`_ (formerly ZXTM) via it's
+SOAP interface.  This library was written for Traffic Manager version 6.
+
+You can do pretty much anything you want to from the API, however, this
+particular library currently only interacts with the ``System.Cache.wsdl`` since
+that's the only part I'm interested in.
+
+Zeus manuals and the complete set of WSDL files are available on each Traffic
+Manager through a maze of clicking:
+
+    1) Log in
+    2) Click "Diagnose"
+    3) Click "Technical Support"
+    4) The manuals are all on this page as well as a link to "Control API WSDL Files"
+
+
+Basic Use
+---------
+Just copy settings.py-dist to settings.py and fill in appropriate values.  Once
+setup, import into your project.  A simple example:
+
+       >>> from interface import Zeus
+
+       >>> z = zeus()
+
+       # Empties the entire cache
+       >>> z.flushAll()
+
+
+Tests
+-----
+
+After configuring the ``settings.py`` file, simply run ``fab test``.
+
+
+.. _Zeus Traffic Manager: http://www.zeus.com/

fabfile.py

@@ -0,0 +1,14 @@
+# Used just to set the PYTHONPATH so the tests can load your settings file.
+import functools
+import os
+
+from fabric.api import local
+
+ROOT = os.path.abspath(os.path.dirname(__file__))
+
+os.environ['PYTHONPATH'] = ROOT
+
+local = functools.partial(local, capture=False)
+
+def test():
+    local('nosetests')

interface.py

@@ -0,0 +1,79 @@
+import os
+from urlparse import urlparse
+
+from suds.client import Client
+from suds.transport.http import HttpAuthenticated
+from suds.xsd.doctor import ImportDoctor, Import
+
+import settings
+
+if settings.LOG_LEVEL:
+    import logging
+    logging.basicConfig(level=settings.LOG_LEVEL)
+
+
+class Zeus:
+
+    def __init__(self, wsdl="System.Cache.wsdl"):
+
+        # Sorry windows
+        url = "file://%s" % os.path.abspath(os.path.join('wsdl', wsdl))
+
+        # Apparently Zeus's wsdl is broken and we have to jimmy this thing in
+        # manually.  See https://fedorahosted.org/suds/ticket/220 for details.
+        # Also, you'll be happy to know that the zillion .wsdl files that Zeus
+        # includes apparently have different targetNamespace's.  This one
+        # happens to be 1.1, but if you load something else you'll probably
+        # need to adjust it.
+        imp = Import('http://schemas.xmlsoap.org/soap/encoding/')
+        imp.filter.add('http://soap.zeus.com/zxtm/1.1/')
+        doctor = ImportDoctor(imp)
+
+        transporter = HttpAuthenticated(username=settings.USERNAME,
+                                        password=settings.PASSWORD)
+
+        self.client = Client(url, doctor=doctor,
+                                  location=settings.LOCATION,
+                                  transport=transporter)
+
+    def flushAll(self):
+        """Flushes everything in the system: all objects across all virtual
+        servers."""
+        return self.client.service.clearWebCache()
+
+
+    def getGlobalCacheInfo(self):
+        """Returns a small object of statistics."""
+        return self.client.service.getGlobalCacheInfo()
+
+
+    def flushObjectsByPattern(self, url, return_list=False):
+        """Flush objects out of the cache.  This accepts simple wildcards (*)
+        in the host and/or path.  If return_list is True we'll return a list of
+        URLs that matched the pattern.  There is a performance hit when
+        returning the list since we have to request it, build it, and return
+        it.  """
+        if return_list:
+            objects = self.getObjectsByPattern(url)
+
+        o = urlparse(url)
+        r = self.client.service.clearMatchingCacheContent(o.scheme,
+                                                          o.netloc,
+                                                          o.path)
+        if return_list:
+            return ["%s://%s%s" % (o.protocol, o.host, o.path)
+                                                        for o in objects]
+
+
+    def getObjectByPattern(self, url):
+        """A simple convenience function.  If you have a full URL and you want
+        a single object back, this is the one."""
+        return self.getObjectsByPattern(url, 1)
+
+    def getObjectsByPattern(self, url, limit=None):
+        o = urlparse(url)
+        r = self.client.service.getCacheContent(o.scheme, o.netloc,
+                                                o.path, limit)
+
+        if r.number_matching_items:
+            return r.matching_items

requirements.txt

@@ -0,0 +1,3 @@
+fabric
+nose
+suds

settings.py-dist

@@ -0,0 +1,12 @@
+# https://fedorahosted.org/suds/wiki/Documentation#LOGGING
+import logging
+LOG_LEVEL = None
+
+LOCATION = 'https://zeus:9090/soap'
+USERNAME = ''
+PASSWORD = ''
+
+# If you want to run the tests, you'll need to add at least 2 complete URLs
+# here that are behind your Zeus box.  We'll load/flush them to make sure
+# things are working.
+TEST_URLS = []

tests/test_interface.py

@@ -0,0 +1,65 @@
+import unittest
+import urllib
+from urlparse import urlparse
+
+import settings
+from interface import Zeus
+
+
+class TestInterface(unittest.TestCase):
+
+    def setUp(self):
+        if len(settings.TEST_URLS) < 2:
+            self.fail("Please add at least 2 URLs we can test.")
+        self.zeus = Zeus()
+
+    def _loadTestURLs(self):
+        for url in settings.TEST_URLS:
+            urllib.urlopen(url)
+
+    def test_flushAll(self):
+        self._loadTestURLs()
+
+        x = self.zeus.getObjectsByPattern('*')
+
+        # If this is a used box, there is the potential that there are URLs
+        # besides the ones we load, so just make sure there are at least that
+        # many
+        assert len(self.zeus.getObjectsByPattern('*')) >= len(settings.TEST_URLS)
+
+        self.zeus.flushAll()
+
+        assert self.zeus.getObjectsByPattern('*') == None
+
+    def test_flushObjectByPattern(self):
+        self._loadTestURLs()
+        r = self.zeus.getObjectByPattern(settings.TEST_URLS[0])
+        assert len(r) == 1
+        self.zeus.flushObjectsByPattern(settings.TEST_URLS[0])
+        r = self.zeus.getObjectByPattern(settings.TEST_URLS[0])
+        assert r is None
+
+        # Verify a list of flushed URLs is returned
+        self._loadTestURLs()
+        r = self.zeus.getObjectByPattern(settings.TEST_URLS[0])
+        assert len(r) == 1
+        f = self.zeus.flushObjectsByPattern(settings.TEST_URLS[0], True)
+        assert len(f) == 1
+        r = self.zeus.getObjectByPattern(settings.TEST_URLS[0])
+        assert r is None
+
+    def test_getObjectByPattern(self):
+        self._loadTestURLs()
+        r = self.zeus.getObjectByPattern(settings.TEST_URLS[0])
+        o = urlparse(settings.TEST_URLS[0])
+
+        assert len(r) == 1
+        assert r[0].host == o.netloc
+        assert r[0].path == o.path
+
+    def test_getObjectsByPattern(self):
+        self.zeus.flushAll()
+        self._loadTestURLs()
+        r = self.zeus.getObjectsByPattern('*')
+
+        assert len(r) == len(settings.TEST_URLS)