Move tests into package, too

1 files changed, 147 insertions, 0 deletions

diff --git a/simplejenkinsapi/api_tests.py b/simplejenkinsapi/api_tests.py
new file mode 100644
index 0000000..6e7c943
--- /dev/null
+++ b/simplejenkinsapi/api_tests.py
@@ -0,0 +1,147 @@
+# simplejenkinsapi/api_tests.py -- unit test for simplejenkinsapi/api.py
+#
+# Copyright 2012  Lars Wirzenius
+# 
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+# 
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+# 
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+
+import httplib
+import json
+import StringIO
+import unittest
+
+import simplejenkinsapi
+
+
+class DummyResponse(object):
+
+    def __init__(self, body, content_type):
+        self.status = httplib.OK
+        self.reason = 'yeah yeah'
+        self.body = body
+        self.content_type = content_type
+        
+    def getheader(self, header):
+        if header == 'Content-Type':
+            return self.content_type
+        else:
+            return ''
+        
+    def read(self):
+        return self.body
+
+
+class DummyServer(object):
+
+    def __init__(self):
+        self.routes = {}
+        self.add_route('GET', '/api/json', self.toplevel)
+        self.add_route('POST', '/job/job-1/doDelete', self.delete_job_1)
+        self.add_route('POST', '/createItem?name=job-1', self.create_job_1)
+        self.add_route('GET', '/job/job-1/config.xml', self.config_job_1)
+        self.add_route('POST', '/job/job-1/config.xml', self.update_job_1)
+
+        self.jobs = {}
+        
+        self.response = None
+
+    def add_route(self, method, path, callback):
+        key = (method, path)
+        assert key not in self.routes
+        self.routes[key] = callback
+
+    def toplevel(self, body, headers):
+        assert body == ''
+        assert headers == {}
+
+        result = []
+        for job_id in sorted(self.jobs.keys()):
+            result.append({
+                'name': job_id,
+            })
+
+        obj = { 'jobs': result }
+        return json.dumps(obj), 'application/json'
+
+    def create_job_1(self, body, headers):
+        assert 'job-1' not in self.jobs
+        self.jobs['job-1'] = body
+        return '', 'text/plain'
+
+    def update_job_1(self, body, headers):
+        assert 'job-1' in self.jobs
+        self.jobs['job-1'] = body
+        return '', 'text/plain'
+
+    def delete_job_1(self, body, headers):
+        del self.jobs['job-1']
+        return '', 'text/plain'
+
+    def config_job_1(self, body, headers):
+        return self.jobs['job-1'], 'application/xml'
+
+    def request(self, method, path, body=None, headers=None):
+        key = (method, path)
+        assert key in self.routes
+        self.response, self.response_type = self.routes[key](body, headers)
+        
+    def getresponse(self):
+        return DummyResponse(self.response, self.response_type)
+
+
+class JenkinsTests(unittest.TestCase):
+
+    def setUp(self):
+        self.url = 'http://does.not.exist:8080/'
+        self.jenkins = simplejenkinsapi.Jenkins(self.url)
+        
+        self.dummy_server = DummyServer()
+        self.jenkins._connect = lambda: self.dummy_server
+        
+        self.config_xml = '''
+            <?xml version="1.0" encoding="UTF-8"?>
+            <project>
+            </project>
+        '''
+        
+        self.config_xml_2 = '''
+            <?xml version="1.0" encoding="UTF-8"?>
+            <project>
+                <foo/>
+            </project>
+        '''
+        
+    def test_lists_no_jobs_when_there_are_none(self):
+        self.assertEqual(self.jenkins.list_jobs(), [])
+        
+    def test_lists_single_job_when_one_exists(self):
+        self.dummy_server.jobs = { 'job-1': '' }
+        self.assertEqual(self.jenkins.list_jobs(), ['job-1'])
+
+    def test_deletes_existing_job(self):
+        self.dummy_server.jobs = { 'job-1': '' }
+        self.jenkins.delete_job('job-1')
+        self.assertEqual(self.jenkins.list_jobs(), [])
+
+    def test_creates_job(self):
+        self.jenkins.create_job('job-1', self.config_xml)
+        self.assertEqual(self.jenkins.list_jobs(), ['job-1'])
+        self.assertEqual(self.jenkins.get_job_config('job-1'), self.config_xml)
+
+    def test_updates_job(self):
+        self.jenkins.create_job('job-1', self.config_xml)
+        self.jenkins.update_job('job-1', self.config_xml_2)
+        self.assertEqual(self.jenkins.get_job_config('job-1'), 
+                         self.config_xml_2)
+