[1669] Package relationship tests

ckan/logic/auth/publisher/create.py

@@ -24,10 +24,17 @@ def package_relationship_create(context, data_dict):
     model = context['model']
     user = context['user']
 
-    id = data_dict['id']
-    id2 = data_dict['id2']
-    pkg1grps = model.Package.get(id).get_groups('publisher')
-    pkg2grps = model.Package.get(id2).get_groups('publisher')
+    id = data_dict.get('id', '')
+    id2 = data_dict.get('id2', '')
+
+    pkg1 = model.Package.get(id)
+    pkg2 = model.Package.get(id2)
+
+    if not pkg1 or not pkg2:
+        return {'success': False, 'msg': _('Two package IDs are required')}    
+
+    pkg1grps = pkg1.get_groups('publisher')
+    pkg2grps = pkg2.get_groups('publisher')
 
     usergrps = model.User.get( user ).get_groups('publisher')
     authorized = _groups_intersect( usergrps, pkg1grps ) and _groups_intersect( usergrps, pkg2grps )    

ckan/logic/auth/publisher/update.py

@@ -29,6 +29,9 @@ def resource_update(context, data_dict):
     resource = get_resource_object(context, data_dict)
     userobj = model.User.get( user )
 
+    if not userobj:
+        return {'success': False, 'msg': _('User %s not authorized to edit resources in this package') % str(user)}        
+
     if not _groups_intersect( userobj.get_groups('publisher'), resource.resource_group.package.get_groups('publisher') ):
         return {'success': False, 'msg': _('User %s not authorized to edit resources in this package') % str(user)}
 

ckan/tests/functional/test_publisher_auth.py

@@ -192,3 +192,77 @@ def test_delete_anon_fail(self):
     def test_delete_unknown_fail(self):
         self._run_fail_test( 'nosuchuser', 'package_delete' )
 
+
+class TestPublisherPackageRelationships(FunctionalTestCase):
+
+    @classmethod
+    def setup_class(self):                        
+        from ckan.tests.mock_publisher_auth import MockPublisherAuth
+        self.auth = MockPublisherAuth()
+
+        model.Session.remove()
+        CreateTestData.create(auth_profile='publisher')
+        self.groupname = u'david'
+        self.package1name = u'testpkg'
+        self.package2name = u'testpkg2'
+        model.repo.new_revision()
+        model.Session.add(model.Package(name=self.package1name))
+        model.Session.add(model.Package(name=self.package2name))        
+        model.repo.commit_and_remove()
+
+    @classmethod
+    def teardown_class(self):
+        model.Session.remove()
+        model.repo.rebuild_db()
+        model.Session.remove()
+
+    def test_create_fail_user( self):
+        p1 = model.Package.by_name( self.package1name )
+        p2 = model.Package.by_name( self.package2name )        
+
+        context = { 'model': model, 'user': 'russianfan' }
+        try:
+            self.auth.check_access('package_relationship_create', context, {'id': p1.id, 'id2': p2.id})
+            assert False, "The user should not have access."            
+        except NotAuthorized, e:
+            pass
+
+    def test_create_fail_ddict( self):
+        p1 = model.Package.by_name( self.package1name )
+        p2 = model.Package.by_name( self.package2name )        
+
+        context = { 'model': model, 'user': 'russianfan' }
+        try:
+            self.auth.check_access('package_relationship_create', context, {'id': p1.id})
+            assert False, "The user should not have access."            
+        except NotAuthorized, e:
+            pass
+
+        try:
+            self.auth.check_access('package_relationship_create', context, {'id2': p2.id})
+            assert False, "The user should not have access."            
+        except NotAuthorized, e:
+            pass            
+
+    def test_create_success(self):
+        userobj = model.User.get('russianfan')     
+
+        f = model.User.get_groups
+        g = model.Package.get_groups
+        def gg(*args, **kwargs):
+            return ['test_group']
+        model.User.get_groups = gg
+        model.Package.get_groups = gg
+
+        p1 = model.Package.by_name( self.package1name )
+        p2 = model.Package.by_name( self.package2name )        
+
+        context = { 'model': model, 'user': 'russianfan' }
+        try:
+            self.auth.check_access('package_relationship_create', context, {'id': p1.id, 'id2': p2.id})
+        except NotAuthorized, e:
+            assert False, "The user should have %s access: %r." % (action, e.extra_msg)
+        model.User.get_groups = f
+        model.Package.get_groups = g
+
+