"Fossies" - the Fresh Open Source Software Archive

Member "roundup-2.0.0/test/test_xmlrpc.py" (26 Aug 2019, 14226 Bytes) of package /linux/www/roundup-2.0.0.tar.gz:


As a special service "Fossies" has tried to format the requested source page into HTML format using (guessed) Python source code syntax highlighting (style: standard) with prefixed line numbers. Alternatively you can here view or download the uninterpreted source code file. See also the latest Fossies "Diffs" side-by-side code changes report for "test_xmlrpc.py": 1.6.1_vs_2.0.0.

    1 #
    2 # Copyright (C) 2007 Stefan Seefeld
    3 # All rights reserved.
    4 # For license terms see the file COPYING.txt.
    5 #
    6 
    7 from __future__ import print_function
    8 import unittest, os, shutil, errno, sys, difflib, cgi, re
    9 
   10 from roundup.anypy import xmlrpc_
   11 MultiCall = xmlrpc_.client.MultiCall
   12 from roundup.cgi.exceptions import *
   13 from roundup import init, instance, password, hyperdb, date
   14 from roundup.xmlrpc import RoundupInstance, RoundupDispatcher
   15 from roundup.backends import list_backends
   16 from roundup.hyperdb import String
   17 from roundup.cgi import TranslationService
   18 
   19 from . import db_test_base
   20 from .test_mysql import skip_mysql
   21 from .test_postgresql import skip_postgresql
   22 
   23 
   24 class XmlrpcTest(object):
   25 
   26     backend = None
   27 
   28     def setUp(self):
   29         self.dirname = '_test_xmlrpc'
   30         # set up and open a tracker
   31         self.instance = db_test_base.setupTracker(self.dirname, self.backend)
   32 
   33         # open the database
   34         self.db = self.instance.open('admin')
   35 
   36         print("props_only default", self.db.security.get_props_only_default())
   37 
   38         # Get user id (user4 maybe). Used later to get data from db.
   39         self.joeid = 'user' + self.db.user.create(username='joe',
   40             password=password.Password('random'), address='random@home.org',
   41             realname='Joe Random', roles='User')
   42 
   43         self.db.commit()
   44         self.db.close()
   45         self.db = self.instance.open('joe')
   46 
   47         self.db.tx_Source = 'web'
   48 
   49         self.db.issue.addprop(tx_Source=hyperdb.String())
   50         self.db.msg.addprop(tx_Source=hyperdb.String())
   51 
   52         self.db.post_init()
   53 
   54         thisdir = os.path.dirname(__file__)
   55         vars = {}
   56         exec(compile(open(os.path.join(thisdir,
   57                                        "tx_Source_detector.py")).read(),
   58                      os.path.join(thisdir, "tx_Source_detector.py"), 'exec'),
   59              vars)
   60         vars['init'](self.db)
   61 
   62         self.server = RoundupInstance(self.db, self.instance.actions, None)
   63 
   64     def tearDown(self):
   65         self.db.close()
   66         try:
   67             shutil.rmtree(self.dirname)
   68         except OSError as error:
   69             if error.errno not in (errno.ENOENT, errno.ESRCH): raise
   70 
   71     def testAccess(self):
   72         # Retrieve all three users.
   73         results = self.server.list('user', 'id')
   74         self.assertEqual(len(results), 3)
   75 
   76         # Obtain data for 'joe'.
   77         results = self.server.display(self.joeid)
   78         self.assertEqual(results['username'], 'joe')
   79         self.assertEqual(results['realname'], 'Joe Random')
   80 
   81     def testChange(self):
   82         # Reset joe's 'realname'.
   83         results = self.server.set(self.joeid, 'realname=Joe Doe')
   84         results = self.server.display(self.joeid, 'realname')
   85         self.assertEqual(results['realname'], 'Joe Doe')
   86 
   87         # check we can't change admin's details
   88         self.assertRaises(Unauthorised, self.server.set, 'user1', 'realname=Joe Doe')
   89 
   90     def testCreate(self):
   91         results = self.server.create('issue', 'title=foo')
   92         issueid = 'issue' + results
   93         results = self.server.display(issueid, 'title')
   94         self.assertEqual(results['title'], 'foo')
   95         self.assertEqual(self.db.issue.get('1', "tx_Source"), 'web')
   96 
   97     def testFileCreate(self):
   98         results = self.server.create('file', 'content=hello\r\nthere')
   99         fileid = 'file' + results
  100         results = self.server.display(fileid, 'content')
  101         self.assertEqual(results['content'], 'hello\r\nthere')
  102 
  103     def testSchema(self):
  104         schema={'status': [('name', '<roundup.hyperdb.String>'),
  105                            ('order', '<roundup.hyperdb.Number>')],
  106                 'keyword': [('name', '<roundup.hyperdb.String>')],
  107                 'priority': [('name', '<roundup.hyperdb.String>'),
  108                              ('order', '<roundup.hyperdb.Number>')],
  109                 'user': [('address', '<roundup.hyperdb.String>'),
  110                          ('alternate_addresses', '<roundup.hyperdb.String>'),
  111                          ('organisation', '<roundup.hyperdb.String>'),
  112                          ('password', '<roundup.hyperdb.Password>'),
  113                          ('phone', '<roundup.hyperdb.String>'),
  114                          ('queries', '<roundup.hyperdb.Multilink to "query">'),
  115                          ('realname', '<roundup.hyperdb.String>'),
  116                          ('roles', '<roundup.hyperdb.String>'),
  117                          ('timezone', '<roundup.hyperdb.String>'),
  118                          ('username', '<roundup.hyperdb.String>')],
  119                 'file': [('content', '<roundup.hyperdb.String>'),
  120                          ('name', '<roundup.hyperdb.String>'),
  121                          ('type', '<roundup.hyperdb.String>')],
  122                 'msg': [('author', '<roundup.hyperdb.Link to "user">'),
  123                         ('content', '<roundup.hyperdb.String>'),
  124                         ('date', '<roundup.hyperdb.Date>'),
  125                         ('files', '<roundup.hyperdb.Multilink to "file">'),
  126                         ('inreplyto', '<roundup.hyperdb.String>'),
  127                         ('messageid', '<roundup.hyperdb.String>'),
  128                         ('recipients', '<roundup.hyperdb.Multilink to "user">'),
  129                         ('summary', '<roundup.hyperdb.String>'),
  130                         ('tx_Source', '<roundup.hyperdb.String>'),
  131                         ('type', '<roundup.hyperdb.String>')],
  132                 'query': [('klass', '<roundup.hyperdb.String>'),
  133                           ('name', '<roundup.hyperdb.String>'),
  134                           ('private_for', '<roundup.hyperdb.Link to "user">'),
  135                           ('url', '<roundup.hyperdb.String>')],
  136                 'issue': [('assignedto', '<roundup.hyperdb.Link to "user">'),
  137                           ('files', '<roundup.hyperdb.Multilink to "file">'),
  138                           ('keyword', '<roundup.hyperdb.Multilink to "keyword">'),
  139                           ('messages', '<roundup.hyperdb.Multilink to "msg">'), 
  140                           ('nosy', '<roundup.hyperdb.Multilink to "user">'),
  141                           ('priority', '<roundup.hyperdb.Link to "priority">'),
  142                           ('status', '<roundup.hyperdb.Link to "status">'),
  143                           ('superseder', '<roundup.hyperdb.Multilink to "issue">'),
  144                           ('title', '<roundup.hyperdb.String>'),
  145                           ('tx_Source', '<roundup.hyperdb.String>')]}
  146 
  147         results = self.server.schema()
  148         self.assertEqual(results, schema)
  149 
  150     def testLookup(self):
  151         self.assertRaises(KeyError, self.server.lookup, 'user', '1')
  152         results = self.server.lookup('user', 'admin')
  153         self.assertEqual(results, '1')
  154 
  155     def testAction(self):
  156         # As this action requires special previledges, we temporarily switch
  157         # to 'admin'
  158         self.db.setCurrentUser('admin')
  159         users_before = self.server.list('user')
  160         try:
  161             tmp = 'user' + self.db.user.create(username='tmp')
  162             self.server.action('retire', tmp)
  163         finally:
  164             self.db.setCurrentUser('joe')
  165         users_after = self.server.list('user')
  166         self.assertEqual(users_before, users_after)
  167 
  168         # test a bogus action
  169         with self.assertRaises(Exception) as cm:
  170             self.server.action('bogus')
  171         print(cm.exception)
  172         self.assertEqual(cm.exception.args[0],
  173                          'action "bogus" is not supported ')
  174 
  175     def testAuthDeniedEdit(self):
  176         # Wrong permissions (caught by roundup security module).
  177         self.assertRaises(Unauthorised, self.server.set,
  178                           'user1', 'realname=someone')
  179 
  180     def testAuthDeniedCreate(self):
  181         self.assertRaises(Unauthorised, self.server.create,
  182                           'user', {'username': 'blah'})
  183 
  184     def testAuthAllowedEdit(self):
  185         self.db.setCurrentUser('admin')
  186         try:
  187             try:
  188                 self.server.set('user2', 'realname=someone')
  189             except Unauthorised as err:
  190                 self.fail('raised %s'%err)
  191         finally:
  192             self.db.setCurrentUser('joe')
  193 
  194     def testAuthAllowedCreate(self):
  195         self.db.setCurrentUser('admin')
  196         try:
  197             try:
  198                 self.server.create('user', 'username=blah')
  199             except Unauthorised as err:
  200                 self.fail('raised %s'%err)
  201         finally:
  202             self.db.setCurrentUser('joe')
  203 
  204     def testAuthFilter(self):
  205         # this checks if we properly check for search permissions
  206         self.db.security.permissions = {}
  207         # self.db.security.set_props_only_default(props_only=False)
  208         self.db.security.addRole(name='User')
  209         self.db.security.addRole(name='Project')
  210         self.db.security.addPermissionToRole('User', 'Web Access')
  211         self.db.security.addPermissionToRole('Project', 'Web Access')
  212         # Allow viewing keyword
  213         p = self.db.security.addPermission(name='View', klass='keyword')
  214         print("View keyword class: %r"%p)
  215         self.db.security.addPermissionToRole('User', p)
  216         # Allow viewing interesting things (but not keyword) on issue
  217         # But users might only view issues where they are on nosy
  218         # (so in the real world the check method would be better)
  219         p = self.db.security.addPermission(name='View', klass='issue',
  220             properties=("title", "status"), check=lambda x,y,z: True)
  221         print("View keyword class w/ props: %r"%p)
  222         self.db.security.addPermissionToRole('User', p)
  223         # Allow role "Project" access to whole issue
  224         p = self.db.security.addPermission(name='View', klass='issue')
  225         self.db.security.addPermissionToRole('Project', p)
  226         # Allow all access to status:
  227         p = self.db.security.addPermission(name='View', klass='status')
  228         self.db.security.addPermissionToRole('User', p)
  229         self.db.security.addPermissionToRole('Project', p)
  230 
  231         keyword = self.db.keyword
  232         status = self.db.status
  233         issue = self.db.issue
  234 
  235         d1 = keyword.create(name='d1')
  236         d2 = keyword.create(name='d2')
  237         open = status.create(name='open')
  238         closed = status.create(name='closed')
  239         issue.create(title='i1', status=open, keyword=[d2])
  240         issue.create(title='i2', status=open, keyword=[d1])
  241         issue.create(title='i2', status=closed, keyword=[d1])
  242 
  243         chef = self.db.user.create(username = 'chef', roles='User, Project')
  244         joe  = self.db.user.lookup('joe')
  245 
  246         # Conditionally allow view of whole issue (check is False here,
  247         # this might check for keyword owner in the real world)
  248         p = self.db.security.addPermission(name='View', klass='issue',
  249             check=lambda x,y,z: False)
  250         print("View issue class: %r"%p)
  251         self.db.security.addPermissionToRole('User', p)
  252         # Allow user to search for issue.status
  253         p = self.db.security.addPermission(name='Search', klass='issue',
  254             properties=("status",))
  255         print("View Search class w/ props: %r"%p)
  256         self.db.security.addPermissionToRole('User', p)
  257 
  258         keyw = {'keyword':self.db.keyword.lookup('d1')}
  259         stat = {'status':self.db.status.lookup('open')}
  260         keygroup = keysort = [('+', 'keyword')]
  261         self.db.commit()
  262 
  263         # Filter on keyword ignored for role 'User':
  264         r = self.server.filter('issue', None, keyw)
  265         self.assertEqual(r, ['1', '2', '3'])
  266         # Filter on status works for all:
  267         r = self.server.filter('issue', None, stat)
  268         self.assertEqual(r, ['1', '2'])
  269         # Sorting and grouping for class User fails:
  270         r = self.server.filter('issue', None, {}, sort=keysort)
  271         self.assertEqual(r, ['1', '2', '3'])
  272         r = self.server.filter('issue', None, {}, group=keygroup)
  273         self.assertEqual(r, ['1', '2', '3'])
  274 
  275         self.db.close()
  276         self.db = self.instance.open('chef')
  277         self.db.tx_Source = 'web'
  278 
  279         self.db.issue.addprop(tx_Source=hyperdb.String())
  280         self.db.msg.addprop(tx_Source=hyperdb.String())
  281         self.db.post_init()
  282 
  283         self.server = RoundupInstance(self.db, self.instance.actions, None)
  284 
  285         # Filter on keyword works for role 'Project':
  286         r = self.server.filter('issue', None, keyw)
  287         self.assertEqual(r, ['2', '3'])
  288         # Filter on status works for all:
  289         r = self.server.filter('issue', None, stat)
  290         self.assertEqual(r, ['1', '2'])
  291         # Sorting and grouping for class Project works:
  292         r = self.server.filter('issue', None, {}, sort=keysort)
  293         self.assertEqual(r, ['2', '3', '1'])
  294         r = self.server.filter('issue', None, {}, group=keygroup)
  295         self.assertEqual(r, ['2', '3', '1'])
  296 
  297     def testMulticall(self):
  298         translator = TranslationService.get_translation(
  299             language=self.instance.config["TRACKER_LANGUAGE"],
  300             tracker_home=self.instance.config["TRACKER_HOME"])
  301         self.server = RoundupDispatcher(self.db, self.instance.actions,
  302             translator, allow_none = True)
  303         class S:
  304             multicall=self.server.funcs['system.multicall']
  305         self.server.system = S()
  306         self.db.issue.create(title='i1')
  307         self.db.issue.create(title='i2')
  308         m = MultiCall(self.server)
  309         m.display('issue1')
  310         m.display('issue2')
  311         result = m()
  312         results = [
  313             {'files': [], 'status': '1', 'tx_Source': 'web',
  314              'keyword': [], 'title': 'i1', 'nosy': [], 'messages': [],
  315              'priority': None, 'assignedto': None, 'superseder': []},
  316             {'files': [], 'status': '1', 'tx_Source': 'web',
  317              'keyword': [], 'title': 'i2', 'nosy': [], 'messages': [],
  318              'priority': None, 'assignedto': None, 'superseder': []}]
  319         for n, r in enumerate(result):
  320             self.assertEqual(r, results[n])
  321 
  322 
  323 class anydbmXmlrpcTest(XmlrpcTest, unittest.TestCase):
  324     backend = 'anydbm'
  325 
  326 
  327 @skip_mysql
  328 class mysqlXmlrpcTest(XmlrpcTest, unittest.TestCase):
  329     backend = 'mysql'
  330 
  331 
  332 class sqliteXmlrpcTest(XmlrpcTest, unittest.TestCase):
  333     backend = 'sqlite'
  334 
  335 
  336 @skip_postgresql
  337 class postgresqlXmlrpcTest(XmlrpcTest, unittest.TestCase):
  338     backend = 'postgresql'