Permalink
Browse files

Hera lives!

  • Loading branch information...
0 parents commit 9b94ce51cf179fea30fb4b2f0d2728445974632b @clouserw committed Mar 23, 2010
Showing with 627 additions and 0 deletions.
  1. +40 −0 README.rst
  2. 0 __init__.py
  3. +14 −0 fabfile.py
  4. +79 −0 interface.py
  5. +3 −0 requirements.txt
  6. +12 −0 settings.py-dist
  7. 0 tests/__init__.py
  8. +65 −0 tests/test_interface.py
  9. +414 −0 wsdl/System.Cache.wsdl
@@ -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/
No changes.
@@ -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')
@@ -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
@@ -0,0 +1,3 @@
+fabric
+nose
+suds
@@ -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 = []
No changes.
@@ -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)
Oops, something went wrong.

0 comments on commit 9b94ce5

Please sign in to comment.