Coverage for ckanext/udc/licenses/logic/action.py: 22%

1from ckan import model, authz, logic 

2from ckan.model.license import DefaultLicense, License as CKANLicense 

3from ckan.common import _ 

4import logging 

5 

6from ckanext.udc.licenses.model import CustomLicense, init_tables 

7from ckanext.udc.licenses.utils import ( 

8 bump_license_version, 

9 refresh_custom_licenses, 

10 refresh_license_register_if_needed, 

11) 

12from psycopg2.errors import UndefinedTable 

13 

14log = logging.getLogger(__name__) 

15 

16 

17def license_create(context, data_dict): 

18 """ 

19 Any user can create a custom license. 

20 """ 

21 

22 user = context.get("user") 

23 id = data_dict.get('id') 

24 title = data_dict.get('title') 

25 url = data_dict.get('url') 

26 

27 if not user: 

28 raise logic.ValidationError(_("You are not logged in")) 

29 

30 # Validate that we have the required fields. 

31 if not id: 

32 raise logic.ValidationError(_("license id is required")) 

33 

34 if not title: 

35 raise logic.ValidationError(_("license title is required")) 

36 

37 if not url: 

38 raise logic.ValidationError(_("license url is required")) 

39 

40 refresh_license_register_if_needed() 

41 license_register = model.Package.get_license_register() 

42 registered_ids = set([license.id for license in license_register.licenses]) 

43 

44 if data_dict['id'] in registered_ids: 

45 raise logic.ValidationError(_("license id already exists")) 

46 

47 # Create the object 

48 userobj = model.User.get(user) 

49 model.Session.add(CustomLicense(id=id, title=title, url=url, user_id=userobj.id)) 

50 

51 model.Session.commit() 

52 refresh_custom_licenses() 

53 bump_license_version() 

54 return {"success": True} 

55 

56 

57def license_delete(context, data_dict): 

58 """ 

59 The user who created the license, or the admin can delete the license. 

60 The license can only be deleted when it is not used by other packages. 

61 """ 

62 if not data_dict.get('id'): 

63 raise logic.ValidationError("license id is required") 

64 

65 user = context.get("user") 

66 if not user: 

67 raise logic.NotAuthorized("You are not logged in") 

68 

69 userobj = model.User.get(user) 

70 license = CustomLicense.get(data_dict["id"]) 

71 

72 if not authz.is_sysadmin(user) and not license.user_id == userobj.id: 

73 raise logic.NotAuthorized(_("You are not authorized to delete this license")) 

74 

75 # Check if any package uses the license 

76 cnt = model.Session.query(model.Package) \ 

77 .filter(model.Package.license_id == data_dict['id']) \ 

78 .count() 

79 

80 if cnt > 0: 

81 raise logic.ValidationError(_("The license is in use and cannot be deleted")) 

82 

83 model.Session.delete(license) 

84 model.Session.commit() 

85 

86 refresh_custom_licenses() 

87 bump_license_version() 

88 return {'success': True} 

89 

90 

91@logic.side_effect_free 

92def licenses_get(context, data_dict): 

93 """ 

94 Get all custom licenses and predefined CKAN licenses. 

95 """ 

96 licenses = [] 

97 

98 custom_licenses = model.Session.query(CustomLicense) 

99 for custom_license in custom_licenses: 

100 custom_license = custom_license.as_dict() 

101 userObj = model.User.get(custom_license.get('user_id')) 

102 user = {'id': custom_license.get('user_id'), 'name': userObj.name, 'fullname': userObj.fullname} 

103 

104 licenses.append({ 

105 **custom_license, 

106 "user": user 

107 }) 

108 

109 return licenses 

110 

111def license_update(context, data_dict): 

112 """ 

113 The user who created the license, or the admin can update the license. 

114 """ 

115 if not data_dict.get('id'): 

116 raise logic.ValidationError("license id is required") 

117 

118 user = context.get("user") 

119 if not user: 

120 raise logic.NotAuthorized("You are not logged in") 

121 

122 userobj = model.User.get(user) 

123 license = CustomLicense.get(data_dict["id"]) 

124 

125 if not authz.is_sysadmin(user) and not license.user_id == userobj.id: 

126 raise logic.NotAuthorized(_("You are not authorized to update this license")) 

127 

128 if data_dict.get('title'): 

129 license.title = data_dict['title'] 

130 

131 if data_dict.get('url'): 

132 license.url = data_dict['url'] 

133 

134 model.Session.commit() 

135 

136 refresh_custom_licenses() 

137 bump_license_version() 

138 return {'success': True} 

139 

140 

141@logic.side_effect_free 

142def test_long_task(context, data_dict): 

143 """ 

144 A test action to simulate a long running task. 

145 """ 

146 import time 

147 time.sleep(10) 

148 return {"success": True} 

149 

150 

151 

152def init_licenses(): 

153 """ 

154 This is not an action call, this initialize all custom licenses from the custom_license table into CKAN. 

155 """ 

156 if model.meta.engine.dialect.has_table(model.meta.engine.connect(), 'custom_license'): 

157 refresh_custom_licenses() 

158 log.info("Loaded custom licenses")