Commits

Olemis Lang committed b970a11

BH RPC : Export product.create RPC method in product context

Test results improved ... (errors=7)

Comments (0)

Files changed (4)

trunk/bhrpc/product.py

         yield (None, ((None, str,),), self.delete)
         yield (None, ((None, str, dict),), self.update)
         yield (None, ((None, str, str),), self.chown)
-        if not isinstance(self.env, ProductEnvironment):
-            yield ('PRODUCT_CREATE', ((None, str, str),), self.create)
+        yield (None, ((None, str, str),), self.create)
 
     # Exported methods
     def getAll(self, req, owner=None):
         products, filtered considering user permissions. 
         """
         filter = {'owner' : owner} if owner else None
-        return [p for p in Product.select(global_env, where=filter)
+        return [p.prefix for p in Product.select(self.env, where=filter)
                   if 'PRODUCT_VIEW' in req.perm(Neighborhood('product',
                                                              p.prefix))]
 

trunk/bhrpc/tests/__init__.py

 class BloodhoundRpcTestCase(TracRpcTestCase):
     def setUp(self):
         TracRpcTestCase.setUp(self)
-        rpc_testenv._tracadmin('config', 'set', 
-                               'components', 'bhrpc.*', 'enabled')
+        rpc_testenv.global_env.config.set('components', 'bhrpc.*', 'enabled')
+        rpc_testenv.global_env.config.save()
+        rpc_testenv.env.config.set('components', 'bhrpc.*', 'enabled')
+        rpc_testenv.env.config.save()
 
     def tearDown(self):
-        rpc_testenv._tracadmin('config', 'remove', 
-                               'components', 'bhrpc.*')
+        rpc_testenv.global_env.config.remove('components', 'bhrpc.*')
+        rpc_testenv.global_env.config.save()
+        rpc_testenv.env.config.remove('components', 'bhrpc.*')
+        rpc_testenv.env.config.save()
         TracRpcTestCase.tearDown(self)
 
 def test_suite():

trunk/bhrpc/tests/product.py

         self.admin.product.create('new', 'new', 'new')
 
         check_products = self.admin.product.getAll()
-        
-        self.assertEqual(1, len(check_products))
-        self.assertEqual('new', check_products[0])
+
+        self.assertEqual(['@', 'new'], sorted(check_products))
         self.assertEqual(['new', 'new', 'new', 'admin'],
                          self.admin.product.get('new'))
 
     def test_create_duplicate_key(self):
         """test attempted saving of Product with existing key fails"""
 
-        self.admin.create('tp', 'original', 'first product')
+        self.admin.product.create('tp', 'original', 'first product')
         
-        with self.assertRaises(TracError) as test_cm:
-            self.admin.create('tp', 'dupe', 'dupe primary key')
-        self.assertEquals('', unicode(test_cm.exception))
+        with self.assertRaises(xmlrpclib.Fault) as test_cm:
+            self.admin.product.create('tp', 'dupe', 'dupe primary key')
+        self.assertEqual(1, test_cm.exception.faultCode)
+        self.assertEquals("'Product prefix='tp' already exists "
+                          "{'name': 'dupe', 'prefix': 'tp', 'owner': 'admin', "
+                          "'description': 'dupe primary key'}' while executing "
+                          "'product.create()'", 
+                          test_cm.exception.faultString)
 
-        with self.assertRaises(TracError) as test_cm:
-            self.admin.create('tp1', 'original', 'dupe unique field')
-        self.assertEquals('', unicode(test_cm.exception))
+        with self.assertRaises(xmlrpclib.Fault) as test_cm:
+            self.admin.product.create('tp1', 'original', 'dupe unique field')
+        self.assertEqual(1, test_cm.exception.faultCode)
+        self.assertEquals("'Product name='original' already exists"
+                          " {'name': 'original', 'prefix': 'tp1', "
+                          "'owner': 'admin', 'description': 'dupe unique field'}'"
+                          " while executing 'product.create()'", 
+                          test_cm.exception.faultString)
     
     def test_create_no_owner(self):
         """test saving new Product when owner is missing"""
         self.addCleanup(rpc_testenv._tracadmin, 'permission', 'remove',
                         'anonymous', 'TRAC_ADMIN')
 
-        with self.assertRaises(TracError) as test_cm:
-            self.admin.create('tp', 'test', 'no owner')
-        self.assertEquals('', unicode(test_cm.exception))
+        with self.assertRaises(xmlrpclib.Fault) as test_cm:
+            self.anon.product.create('tp', 'test product', 'no owner')
+        self.assertEqual(1, test_cm.exception.faultCode)
+        self.assertEquals("'Product owner must be specified' while executing "
+                          "'product.create()'",
+                          test_cm.exception.faultString)
 
-        self.admin.create('tp1', 'test', 'with owner', 'user')
+        self.anon.product.create('tp1', 'test product', 'with owner', 'user')
         
-        self.assertEqual(['tp1'], self.user.getAll())
-        self.assertEqual(['tp1', 'test', 'with owner', 'user'],
-                         self.user.get('tp1'))
+        self.assertEqual(['@', 'tp1'], sorted(self.user.product.getAll()))
+        self.assertEqual(['tp1', 'test product', 'with owner', 'user'],
+                         self.user.product.get('tp1'))
 
     def test_update(self):
         """tests that we can use update to push data to the database"""
         self.assertEqual(['tp', 'updated', 'nothing', 'admin'], 
                          comp_product)
 
-    def test_update_key_change(self):
-        """tests that we raise an error for attempting to update key fields"""
-
-        bad_data = {'prefix':'tp0', 
-                    'name':'update', 
-                    'description':'nothing'}
-        product = list(Product.select(self.env, where={'prefix':'tp'}))[0]
-        product._data.update(bad_data)
-        self.assertRaises(TracError, product.update)
-    
-
     def test_field_set(self):
         """tests that we can update a single field data"""
 
                                   {'description' : 'test change of description', 
                                    'whatever' : 'nothing'})
         comp_product = self.admin.product.get('tp')
-        self.assertEqual(['tp', 'updated', 'test change of description', 
+        self.assertEqual(['tp', 'test project', 'test change of description', 
                           'admin'], comp_product)
 
 class RpcGlobalProductTestCase(RpcProductTestCase):

trunk/tracrpc/tests/__init__.py

 
         def _tracadmin(self, *args, **kwargs):
             do_wait = kwargs.pop('wait', False)
-            if self.upgraded and args[0] not in ProductAdminModule.GLOBAL_COMMANDS:
+            global_scope = kwargs.pop('globalenv', False)
+            if not global_scope and self.upgraded and \
+                    args[0] not in ProductAdminModule.GLOBAL_COMMANDS:
                 args = ('product', 'admin', DEFAULT_PRODUCT) + args
             SvnFunctionalTestEnvironment._tracadmin(self, *args, **kwargs)
             if do_wait: # Delay to ensure command executes and caches resets