Source

Dad / dad / tests / unit / web / test_dad_api.py

import httplib2
import py.test
import simplejson

from pprint import pprint


class TestDadConfigAPI(object):

    @property
    def base(self):
        return 'http://%s:%s' % (py.test.dev_server.host,
                                 py.test.dev_server.port)

    def add_service(self):
        h = httplib2.Http()
        kw = dict(
            method='POST',
            headers={'Content-Type': 'application/json'},
            body=simplejson.dumps({'listdir': {'command': 'ls -la'}})
            )
        resp, content = h.request(self.base + '/services/', **kw)
        return resp, content
        
    def test_add_service(self):
        resp, content = self.add_service()
        assert resp['status'] == '201'
        result = simplejson.loads(content)
        assert result['Location'] == '/services/listdir'

    def test_get_service(self):
        h = httplib2.Http()
        resp, content = h.request(self.base + '/services/hello_world')
        assert resp['status'] == '200'
        assert content
        result = simplejson.loads(content)
        actions = ['start', 'stop', 'start']
        for a in actions:
            assert result[a] == '/services/hello_world/%s' % (a)

    def test_start_service(self):
        resp, content = self.add_service()
        result = simplejson.loads(content)

        h = httplib2.Http()
        resp, content = h.request(self.base + '/services/listdir/start',
                                  method='POST',
                                  headers={'Content-Length': '0'})

        assert resp['status'] == '200'
        result = simplejson.loads(content)
        assert result
        assert result['started'] == 'listdir'
        assert result['num_instances'] == 1

    def test_stop_service(self):
        '''Stop our long running process started when the test dad
        server started'''
        resp, content = self.add_service()
        result = simplejson.loads(content)

        h = httplib2.Http()
        resp, content = h.request(self.base + '/services/hello_world/stop',
                                  method='POST',
                                  headers={'Content-Length': '0'})
        assert resp['status'] == '200'
        result = simplejson.loads(content)
        assert result
        pprint(result)
        for port, output in result.iteritems():
            assert output['returncode'] == -1
        
        
Tip: Filter by directory path e.g. /media app.js to search for public/media/app.js.
Tip: Use camelCasing e.g. ProjME to search for ProjectModifiedEvent.java.
Tip: Filter by extension type e.g. /repo .js to search for all .js files in the /repo directory.
Tip: Separate your search with spaces e.g. /ssh pom.xml to search for src/ssh/pom.xml.
Tip: Use ↑ and ↓ arrow keys to navigate and return to view the file.
Tip: You can also navigate files with Ctrl+j (next) and Ctrl+k (previous) and view the file with Ctrl+o.
Tip: You can also navigate files with Alt+j (next) and Alt+k (previous) and view the file with Alt+o.