| 1 | # -*- coding: utf-8 -*- |
---|
| 2 | # |
---|
| 3 | # Copyright (C) 2005-2007 Christopher Lenz <cmlenz@gmx.de> |
---|
| 4 | # Copyright (C) 2007-2010 Edgewall Software |
---|
| 5 | # All rights reserved. |
---|
| 6 | # |
---|
| 7 | # This software is licensed as described in the file COPYING, which |
---|
| 8 | # you should have received as part of this distribution. The terms |
---|
| 9 | # are also available at http://bitten.edgewall.org/wiki/License. |
---|
| 10 | |
---|
1 | 11 | import os |
---|
1 | 12 | import sys |
---|
1 | 13 | import shutil |
---|
1 | 14 | import tempfile |
---|
1 | 15 | import unittest |
---|
| 16 | |
---|
1 | 17 | from bitten.slave import BuildSlave, ExitSlave |
---|
1 | 18 | from bitten.util import xmlio |
---|
1 | 19 | from bitten.slave import encode_multipart_formdata |
---|
| 20 | |
---|
2 | 21 | class DummyResponse(object): |
---|
1 | 22 | def __init__(self, code): |
---|
2 | 23 | self.code = code |
---|
| 24 | |
---|
2 | 25 | class TestSlave(BuildSlave): |
---|
| 26 | |
---|
1 | 27 | def __init__(self, filename, work_dir): |
---|
2 | 28 | BuildSlave.__init__(self, [filename], work_dir=work_dir) |
---|
2 | 29 | self.results = [] |
---|
| 30 | |
---|
1 | 31 | def _gather(self, method, url, body=None, headers=None): |
---|
2 | 32 | self.results.append(xmlio.parse(body)) |
---|
2 | 33 | return DummyResponse(201) |
---|
| 34 | |
---|
1 | 35 | def _execute_step(self, _build_url, recipe, step): |
---|
2 | 36 | old_local, old_request = self.local, self.request |
---|
2 | 37 | try: |
---|
2 | 38 | self.local, self.request = False, self._gather |
---|
2 | 39 | return BuildSlave._execute_step(self, 'dummy_build', recipe, step) |
---|
2 | 40 | finally: |
---|
2 | 41 | self.local, self.request = old_local, old_request |
---|
| 42 | |
---|
2 | 43 | class BuildSlaveTestCase(unittest.TestCase): |
---|
| 44 | |
---|
1 | 45 | def setUp(self): |
---|
3 | 46 | self.work_dir = tempfile.mkdtemp(prefix='bitten_test') |
---|
3 | 47 | self.python_path = xmlio._escape_attr(sys.executable) |
---|
| 48 | |
---|
1 | 49 | def tearDown(self): |
---|
3 | 50 | shutil.rmtree(self.work_dir) |
---|
| 51 | |
---|
1 | 52 | def _create_file(self, *path): |
---|
2 | 53 | filename = os.path.join(self.work_dir, *path) |
---|
2 | 54 | fd = file(filename, 'w') |
---|
2 | 55 | fd.close() |
---|
2 | 56 | return filename |
---|
| 57 | |
---|
1 | 58 | def _run_slave(self, recipe): |
---|
2 | 59 | results = [] |
---|
2 | 60 | filename = self._create_file("recipe.xml") |
---|
2 | 61 | recipe_file = file(filename, "wb") |
---|
2 | 62 | recipe_file.write(recipe) |
---|
2 | 63 | recipe_file.close() |
---|
2 | 64 | slave = TestSlave(filename, self.work_dir) |
---|
2 | 65 | slave.run() |
---|
2 | 66 | return slave.results |
---|
| 67 | |
---|
1 | 68 | def test_quit_raises(self): |
---|
1 | 69 | self.slave = BuildSlave([], work_dir=self.work_dir) |
---|
1 | 70 | self.assertRaises(ExitSlave, self.slave.quit) |
---|
| 71 | |
---|
1 | 72 | def test_simple_recipe(self): |
---|
1 | 73 | results = self._run_slave(""" |
---|
1 | 74 | <build xmlns:sh="http://bitten.edgewall.org/tools/sh" |
---|
1 | 75 | > |
---|
1 | 76 | <step id="print"> |
---|
1 | 77 | <sh:exec executable="%s" args='-c "print (\\"Hello\\")"' /> |
---|
1 | 78 | </step> |
---|
1 | 79 | </build>""" % self.python_path) |
---|
| 80 | |
---|
1 | 81 | result = results[0] |
---|
1 | 82 | self.assertEqual(result.attr["step"], "print") |
---|
1 | 83 | self.assertEqual(result.attr["status"], "success") |
---|
1 | 84 | log = list(result)[0] |
---|
1 | 85 | msg = list(log)[0] |
---|
1 | 86 | self.assertEqual(str(msg), '<message level="info">Hello</message>') |
---|
| 87 | |
---|
1 | 88 | def test_non_utf8(self): |
---|
1 | 89 | results = self._run_slave(""" |
---|
1 | 90 | <build xmlns:sh="http://bitten.edgewall.org/tools/sh" |
---|
1 | 91 | > |
---|
1 | 92 | <step id="print"> |
---|
1 | 93 | <sh:exec executable="%s" args='-c "print (\\"\\xe9\\")"' /> |
---|
1 | 94 | </step> |
---|
1 | 95 | </build>""" % self.python_path) |
---|
| 96 | |
---|
1 | 97 | result = results[0] |
---|
1 | 98 | self.assertEqual(result.attr["step"], "print") |
---|
1 | 99 | self.assertEqual(result.attr["status"], "success") |
---|
1 | 100 | log = list(result)[0] |
---|
1 | 101 | msg = list(log)[0] |
---|
| 102 | # check replacement character (\uFFFD) was generated correctly |
---|
1 | 103 | self.assertEqual(str(msg).decode("utf-8"), |
---|
1 | 104 | u'<message level="info">\uFFFD</message>') |
---|
| 105 | |
---|
2 | 106 | class MultiPartEncodeTestCase(unittest.TestCase): |
---|
| 107 | |
---|
1 | 108 | def setUp(self): |
---|
1 | 109 | self.work_dir = tempfile.mkdtemp(prefix='bitten_test') |
---|
| 110 | |
---|
1 | 111 | def tearDown(self): |
---|
1 | 112 | shutil.rmtree(self.work_dir) |
---|
| 113 | |
---|
1 | 114 | def test_mutlipart_encode_one(self): |
---|
1 | 115 | fields = { |
---|
1 | 116 | 'foo': 'bar', |
---|
1 | 117 | 'foofile': ('test.txt', 'contents of foofile'), |
---|
1 | 118 | } |
---|
1 | 119 | body, content_type = encode_multipart_formdata(fields) |
---|
1 | 120 | boundary = content_type.split(';')[1].strip().split('=')[1] |
---|
1 | 121 | self.assertEquals('multipart/form-data; boundary=%s' % boundary, |
---|
1 | 122 | content_type) |
---|
1 | 123 | self.assertEquals('--%s\r\nContent-Disposition: form-data; ' \ |
---|
1 | 124 | 'name="foo"\r\n\r\nbar\r\n--%s\r\nContent-Disposition: ' \ |
---|
1 | 125 | 'form-data; name="foofile"; filename="test.txt"\r\n' \ |
---|
1 | 126 | 'Content-Type: application/octet-stream\r\n\r\n' \ |
---|
1 | 127 | 'contents of foofile\r\n--%s--\r\n' % ( |
---|
1 | 128 | boundary,boundary,boundary), body) |
---|
| 129 | |
---|
1 | 130 | def suite(): |
---|
1 | 131 | suite = unittest.TestSuite() |
---|
1 | 132 | suite.addTest(unittest.makeSuite(BuildSlaveTestCase, 'test')) |
---|
1 | 133 | suite.addTest(unittest.makeSuite(MultiPartEncodeTestCase, 'test')) |
---|
1 | 134 | return suite |
---|
| 135 | |
---|
1 | 136 | if __name__ == '__main__': |
---|
0 | 137 | unittest.main(defaultTest='suite') |
---|