from mock import Mock
from tornado.escape import json_encode
from tornado.testing import AsyncHTTPTestCase
hooksmod = __import__('hooks-queue')
rconn = Mock()
class HookTestCase(AsyncHTTPTestCase):
    def get_app(self):
        return hooksmod.CandolintHooks(rconn, debug=True)
    def post_json(self, url, data, extra_headers):
        body = json_encode(data)
        headers = {'Content-Type': 'application/json'}
        headers.update(extra_headers)
        return self.fetch(url, method='POST', body=body, headers=headers)
class BitbucketHookTestCase(HookTestCase):
    url = '/bitbucket'
    def test_invalid_payload(self):
        response = self.fetch(self.url, method='POST', body='invalid')
        assert response.code == 400
    def test_push_hg(self):
        data = {
            'repository': {
                'scm': 'hg',
                'name': 'test',
                'links': {
                    'html': {
                        'href': 'https://bitbucket.org/example/test'
                    }
                }
            },
            'push': {
                'changes': [{
                    'new': {
                        'heads': [
                            {'hash': 'foo'},
                            {'hash': 'bar'},
                            {'hash': 'baz'}
                        ],
                        'target': {'hash': 'bar'}
                    }
                }]
            }
        }
        response = self.post_json(self.url, data, {'X-Event-Key': 'repo:push'})
        assert response.code == 200
        assert b'"queued": 3' in response.body
    def test_push_git(self):
        data = {
            'repository': {
                'scm': 'git',
                'name': 'test',
                'links': {
                    'html': {
                        'href': 'https://bitbucket.org/example/test'
                    }
                }
            },
            'push': {
                'changes': [{
                    'new': {'type': 'branch', 'name': 'master'}
                }]
            }
        }
        response = self.post_json(self.url, data, {'X-Event-Key': 'repo:push'})
        assert response.code == 200
        assert b'"queued": 1' in response.body
class GithubHookTestCase(HookTestCase):
    url = '/github'
    def test_no_event(self):
        response = self.post_json(self.url, {}, {})
        assert response.code == 200
        assert response.body == b'{"result": "unknown event"}'
    def test_ping(self):
        response = self.post_json(self.url, {}, {'X-GitHub-Event': 'ping'})
        assert response.code == 200
        assert response.body == b'{"result": "pong"}'
    def test_push(self):
        data = {
            'repository': {
                'name': 'test',
                'html_url': 'https://github.com/example/test'
            },
            'ref': 'refs/heads/master'
        }
        response = self.post_json(self.url, data, {'X-GitHub-Event': 'push'})
        assert response.code == 200
        assert b'"queued": 1' in response.body