Last active
December 14, 2015 18:29
-
-
Save holmboe/5129722 to your computer and use it in GitHub Desktop.
A WIP to test the Salt (http://www.saltstack.org) REST API. The ambition is to have a framework/implementation independent test suite for the API. This will allow easy verification of API implementations such as salt-api (http://github.com/saltstack/salt-api) and django-saltapi (http://github.com/holmboe/django-saltapi).
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
#!/usr/bin/env python | |
# -*- coding: utf-8 -*- | |
# Usage examples: | |
# | |
# test-saltapi | |
# test-saltapi testPing | |
# VERBOSITY=9 test-saltapi | |
# VALID_TARGETS="salt min1 min2" test-saltapi testMinions | |
import os | |
import sys | |
import json | |
import re | |
try: | |
# check the system path first | |
import unittest2 as unittest | |
except ImportError: | |
if sys.version_info >= (2,7): | |
# unittest2 features are native in Python 2.7 | |
import unittest | |
else: | |
raise | |
import requests | |
CONFIG = {'API_ROOT_URL': 'http://salt/api/salt/', | |
'VALID_TARGETS': ['salt'], | |
'INVALID_TARGETS': ['nonexist'], | |
'VERBOSITY': 1} | |
for (k, v) in CONFIG.iteritems(): | |
if k in os.environ: | |
if re.match('.+ .+', os.environ[k]): | |
setattr(sys.modules[__name__], k, os.environ[k].split(' ')) | |
elif re.match('\d+', os.environ[k]): | |
setattr(sys.modules[__name__], k, int(os.environ[k])) | |
else: | |
setattr(sys.modules[__name__], k, os.environ[k]) | |
else: | |
setattr(sys.modules[__name__], k, v) | |
def url(*paths, **kwargs): | |
# TODO: if there is a '/' prefix, remove it, on all paths | |
url = API_ROOT_URL + '/'.join(paths) | |
url += '/' if not url.endswith('/') else '' | |
# TODO: add urlencoded arguments through **kwargs | |
return url | |
class testMinions(unittest.TestCase): | |
def test_list_minions(self): | |
r = requests.get(url('minions')) | |
self.assertEqual(r.status_code, 200) | |
self.assertTrue(len(json.loads(r.content).keys())) | |
def test_minions(self): | |
for target in VALID_TARGETS: | |
r = requests.get(url('minions', target)) | |
self.assertEqual(r.status_code, 200) | |
self.assertTrue(len(json.loads(r.content).keys()) == 1) | |
class testJobs(unittest.TestCase): | |
def test_list_jobs(self): | |
r = requests.get(url('jobs')) | |
self.assertEqual(r.status_code, 200) | |
self.assertTrue(re.match('\d+', json.loads(r.content).keys()[0])) | |
@unittest.skip("not implemented yet") | |
def test_lookup_jid(self): | |
pass | |
@unittest.skip("does not return 404 yet") | |
def test_lookup_invalid_jid(self): | |
r = requests.get(url('jobs', '20999999999999999999')) | |
self.assertEqual(r.status_code, 404) | |
class testPing(unittest.TestCase): | |
def test_ping(self): | |
for target in VALID_TARGETS: | |
r = requests.get(url('ping', target)) | |
self.assertEqual(r.status_code, 200) | |
self.assertTrue(json.loads(r.content)[target]) | |
@unittest.skip("does not return 404 yet") | |
def test_invalid_ping(self): | |
for target in INVALID_TARGETS: | |
r = requests.get(url('ping', target)) | |
self.assertEqual(r.status_code, 404) | |
#self.assertEqual(json.loads(r.content), {}) | |
class testApi(unittest.TestCase): | |
def test_ping(self): | |
for target in VALID_TARGETS: | |
r = requests.post(url(), | |
data={'client': 'local', | |
'tgt': target, | |
'fun': 'test.ping'}) | |
self.assertEqual(r.status_code, 200) | |
self.assertTrue(json.loads(r.content)[target]) | |
@unittest.skip("does not return 404 yet") | |
def test_invalid_ping(self): | |
for target in INVALID_TARGETS: | |
r = requests.post(url(), | |
data={'client': 'local', | |
'tgt': target, | |
'fun': 'test.ping'}) | |
self.assertEqual(r.status_code, 404) | |
#self.assertEqual(json.loads(r.content), {}) | |
def test_grains_items(self): | |
for target in VALID_TARGETS: | |
r = requests.post(url(), | |
data={'client': 'local', | |
'tgt': target, | |
'fun': 'grains.items'}) | |
self.assertEqual(r.status_code, 200) | |
self.assertTrue(target in json.loads(r.content)[target]['nodename']) | |
if __name__ == '__main__': | |
loader = unittest.TestLoader() | |
suite = unittest.TestSuite() | |
all_tests = [ | |
testApi, | |
testMinions, | |
testJobs, | |
testPing, | |
] | |
run_tests = [] | |
# called without arguments | |
if len(sys.argv) == 1: | |
for test in all_tests: | |
suite.addTests(loader.loadTestsFromTestCase(test)) | |
run_tests.append(str(test)) | |
# called with arguments, run only specified classes | |
else: | |
sys.argv.pop(0) | |
for name in sys.argv: | |
test = getattr(sys.modules[__name__], name) | |
suite.addTests(loader.loadTestsFromTestCase(test)) | |
run_tests.append(name) | |
unittest.TextTestRunner(verbosity=VERBOSITY).run(suite) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment