← Back to branch summary

~oddbloke/pqm/xmlrpc

« back to all changes in this revision

Viewing changes to pqm/ui/tests/test_xmlrpc.py

  • Committer: Daniel Watkins
  • Date: 2008-08-07 02:56:09 UTC
  • mfrom: (206.1.10 xmlrpc)
  • Revision ID: daniel@daniel-watkins.co.uk-20080807025609-57is5lmna7789as1
Merged changes to xmlrpc thread from jml's review.

Show diffs side-by-side

added added

removed removed

Lines of Context:
pqm/ui/tests/test_xmlrpc.py
17
17
 
18
18
import os
19
19
 
 
20
from twisted.web.xmlrpc import Fault
 
21
 
20
22
from pqm.errors import PQMCmdFailure
21
23
from pqm.tests import (sample_message_bad,
22
24
                       sample_message_template,
27
29
 
28
30
class TestPQM_XMLRPC(TestCaseWithQueue):
29
31
 
 
32
    def assertQueueLength(self, queuedir, length):
 
33
        patches = [f for f in os.listdir(queuedir) if f.startswith('patch.')]
 
34
        self.assertEqual(length, len(patches))
 
35
        return patches
 
36
 
30
37
    def get_xmlrpc(self, empty=True, verify_sigs=False):
31
38
        queue = self.getQueue(empty=empty, verify_sigs=verify_sigs)
32
39
        pqminfo = PQMInfo([queue.configFileName])
34
41
 
35
42
    def test_empty_submission(self):
36
43
        xmlrpc = self.get_xmlrpc()
37
 
        out = xmlrpc.xmlrpc_submit('')
38
 
        self.assertEqual("Error: 'No From specified'", out)
 
44
        out = self.assertRaises(Fault, xmlrpc.xmlrpc_submit, '')
 
45
        self.assertEqual("'No From specified'", out.faultString)
39
46
 
40
47
    def test_valid_submission(self):
 
48
        # Tests that a valid message (i.e. one with a From and Subject line)
 
49
        # receives the appropriate message from XMLRPC.
41
50
        xmlrpc = self.get_xmlrpc()
42
51
        msg = (sample_message_template %
43
52
                ('star-merge baz baz',))
44
 
        out = xmlrpc.xmlrpc_submit(msg)
45
 
        self.assertEqual('Success!', out)
 
53
        self.assertTrue(xmlrpc.xmlrpc_submit(msg))
46
54
        queuedir = xmlrpc.pqminfo.queuedir
47
 
        patches = [f for f in os.listdir(queuedir) if f.startswith('patch.')]
48
 
        self.assertEqual(1, len(patches)) # We should have only one patch
 
55
        patches = self.assertQueueLength(queuedir, 1)
49
56
        self.assertFileEqual(msg, os.path.join(queuedir, patches[0]))
50
57
 
51
58
    def test_invalid_submission(self):
 
59
        # Tests that a meaningless message receives an appropriate response
 
60
        # from XMLRPC.
52
61
        xmlrpc = self.get_xmlrpc()
53
 
        out = xmlrpc.xmlrpc_submit(sample_message_bad)
54
 
        self.assertEqual("Error: 'Sender not authorised to commit to branch"
 
62
        out = self.assertRaises(Fault, xmlrpc.xmlrpc_submit, sample_message_bad)
 
63
        self.assertEqual("'Sender not authorised to commit to branch"
55
64
                         " http://www.example.com/foo/unregistered'",
56
 
                         out)
 
65
                         out.faultString)
57
66
        queuedir = xmlrpc.pqminfo.queuedir
58
 
        patches = [f for f in os.listdir(queuedir) if f.startswith('patch.')]
59
 
        self.assertEqual(0, len(patches))
 
67
        self.assertQueueLength(queuedir, 0)
60
68
 
61
69
    def test_invalid_target(self):
62
70
        xmlrpc = self.get_xmlrpc()
63
71
        msg = (sample_message_template %
64
72
                ('star-merge baz http://www.example.com/foo/unregistered',))
65
 
        out = xmlrpc.xmlrpc_submit(msg)
66
 
        self.assertEqual("Error: 'Sender not authorised to commit to branch"
 
73
        out = self.assertRaises(Fault, xmlrpc.xmlrpc_submit, msg)
 
74
        self.assertEqual("'Sender not authorised to commit to branch"
67
75
                         " http://www.example.com/foo/unregistered'",
68
 
                         out)
 
76
                         out.faultString)
69
77
        queuedir = xmlrpc.pqminfo.queuedir
70
 
        patches = [f for f in os.listdir(queuedir) if f.startswith('patch.')]
71
 
        self.assertEqual(0, len(patches))
 
78
        self.assertQueueLength(queuedir, 0)
72
79
 
73
80
    def test_invalid_source(self):
74
81
        xmlrpc = self.get_xmlrpc()
75
 
        msg = (sample_message_template %
76
 
                ('star-merge fluff baz',))
77
 
        out = xmlrpc.xmlrpc_submit(msg)
78
 
        self.assertEqual("Error: 'fluff is not a branch'", out)
 
82
        msg = (sample_message_template % ('star-merge fluff baz',))
 
83
        out = self.assertRaises(Fault, xmlrpc.xmlrpc_submit, msg)
 
84
        self.assertEqual("'fluff is not a branch'", out.faultString)
79
85
        queuedir = xmlrpc.pqminfo.queuedir
80
 
        patches = [f for f in os.listdir(queuedir) if f.startswith('patch.')]
81
 
        self.assertEqual(0, len(patches))
 
86
        self.assertQueueLength(queuedir, 0)