From 5fbbdbe49909f1c8c3ae70fd77fdec3036af2e2d Mon Sep 17 00:00:00 2001 From: Rushabh Mehta Date: Thu, 28 Feb 2019 17:38:40 +0530 Subject: [PATCH] fix: remove old auto_assign.py files --- .../doctype/auto_assign/auto_assign.py | 108 ----------- .../doctype/auto_assign/test_auto_assign.py | 173 ------------------ 2 files changed, 281 deletions(-) delete mode 100644 frappe/automation/doctype/auto_assign/auto_assign.py delete mode 100644 frappe/automation/doctype/auto_assign/test_auto_assign.py diff --git a/frappe/automation/doctype/auto_assign/auto_assign.py b/frappe/automation/doctype/auto_assign/auto_assign.py deleted file mode 100644 index 7419d5b723..0000000000 --- a/frappe/automation/doctype/auto_assign/auto_assign.py +++ /dev/null @@ -1,108 +0,0 @@ -# -*- coding: utf-8 -*- -# Copyright (c) 2019, Frappe Technologies and contributors -# For license information, please see license.txt - -from __future__ import unicode_literals - -import frappe -from frappe.model.document import Document -from frappe.desk.form import assign_to - -class AutoAssign(Document): - def on_update(self): # pylint: disable=no-self-use - frappe.cache().delete_value('assignment_rule') - - def after_rename(self): # pylint: disable=no-self-use - frappe.cache().delete_value('assignment_rule') - - def apply(self, doc): - if self.safe_eval('assign_condition', doc): - self.do_assignment(doc) - return True - - # try clearing - if self.unassign_condition: - return self.clear_assignment(doc) - - return False - - def do_assignment(self, doc): - user = self.get_user() - - assign_to.add(dict( - assign_to = user, - doctype = doc.get('doctype'), - name = doc.get('name'), - description = frappe.render_template(self.description, doc) - )) - - # set for reference in round robin - self.db_set('last_user', user) - - def clear_assignment(self, doc): - '''Clear assignments''' - if self.safe_eval('unassign_condition', doc): - return assign_to.clear(doc.get('doctype'), doc.get('name')) - - def get_user(self): - ''' - Get the next user for assignment - ''' - if self.rule == 'Round Robin': - return self.get_user_round_robin() - elif self.rule == 'Load Balancing': - return self.get_user_load_balancing() - - def get_user_round_robin(self): - ''' - Get next user based on round robin - ''' - - # first time, or last in list, pick the first - if not self.last_user or self.last_user == self.users[-1].user: - return self.users[0].user - - # find out the next user in the list - for i, d in enumerate(self.users): - if self.last_user == d.user: - return self.users[i+1].user - - def get_user_load_balancing(self): - '''Assign to the user with least number of open assignments''' - counts = [] - for d in self.users: - counts.append(dict( - user = d.user, - count = frappe.db.count('ToDo', dict( - reference_type = self.document_type, - owner = d.user, - status = "Open")) - )) - - # sort by dict value - sorted_counts = sorted(counts, key = lambda k: k['count']) - - # pick the first user - return sorted_counts[0].get('user') - - def safe_eval(self, fieldname, doc): - try: - return frappe.safe_eval(self.get(fieldname), None, doc) - except Exception: - # when assignment fails, don't block the document as it may be - # a part of the email pulling - frappe.msgprint(frappe._('Auto assignment failed'), indicator = 'orange') - -def apply(doc, method): - if frappe.flags.in_patch or frappe.flags.in_install: - return - - assignment_rules = frappe.cache().get_value('assignment_rule', get_assignment_rules) - if doc.doctype in assignment_rules: - # multiple auto assigns - for d in frappe.db.get_all('Assignment Rule', dict(document_type=doc.doctype, disabled = 0), order_by = 'priority desc'): - if frappe.get_doc('Assignment Rule', d.name).apply(doc.as_dict()): - break - -def get_assignment_rules(): - return [d.document_type for d in frappe.db.get_all('Assignment Rule', fields=['document_type'], filters=dict(disabled = 0))] \ No newline at end of file diff --git a/frappe/automation/doctype/auto_assign/test_auto_assign.py b/frappe/automation/doctype/auto_assign/test_auto_assign.py deleted file mode 100644 index 654e372b5c..0000000000 --- a/frappe/automation/doctype/auto_assign/test_auto_assign.py +++ /dev/null @@ -1,173 +0,0 @@ -# -*- coding: utf-8 -*- -# Copyright (c) 2019, Frappe Technologies and Contributors -# See license.txt -from __future__ import unicode_literals - -import frappe -import unittest -from frappe.utils import random_string - -class TestAutoAssign(unittest.TestCase): - def setUp(self): - self.assignment_rule = get_assignment_rule() - clear_assignments() - - def test_round_robin(self): - note = make_note(dict(public=1)) - - # check if auto assigned to first user - self.assertEqual(frappe.db.get_value('ToDo', dict( - reference_type = 'Note', - reference_name = note.name, - status = 'Open' - ), 'owner'), 'test@example.com') - - note = make_note(dict(public=1)) - - # check if auto assigned to second user - self.assertEqual(frappe.db.get_value('ToDo', dict( - reference_type = 'Note', - reference_name = note.name, - status = 'Open' - ), 'owner'), 'test1@example.com') - - clear_assignments() - - note = make_note(dict(public=1)) - - # check if auto assigned to third user, even if - # previous assignments where closed - self.assertEqual(frappe.db.get_value('ToDo', dict( - reference_type = 'Note', - reference_name = note.name, - status = 'Open' - ), 'owner'), 'test2@example.com') - - # check loop back to first user - note = make_note(dict(public=1)) - - self.assertEqual(frappe.db.get_value('ToDo', dict( - reference_type = 'Note', - reference_name = note.name, - status = 'Open' - ), 'owner'), 'test@example.com') - - def test_load_balancing(self): - self.assignment_rule.rule = 'Load Balancing' - self.assignment_rule.save() - - for _ in range(30): - note = make_note(dict(public=1)) - - # check if each user has 10 assignments (?) - for user in ('test@example.com', 'test1@example.com', 'test2@example.com'): - self.assertEqual(len(frappe.get_all('ToDo', dict(owner = user, reference_type = 'Note'))), 10) - - # clear 5 assignments for first user - # can't do a limit in "delete" since postgres does not support it - for d in frappe.get_all('ToDo', dict(reference_type = 'Note', owner = 'test@example.com'), limit=5): - frappe.db.sql("delete from tabToDo where name = %s", d.name) - - # add 5 more assignments - for i in range(5): - make_note(dict(public=1)) - - # check if each user still has 10 assignments - for user in ('test@example.com', 'test1@example.com', 'test2@example.com'): - self.assertEqual(len(frappe.get_all('ToDo', dict(owner = user, reference_type = 'Note'))), 10) - - - def test_assign_condition(self): - # check condition - note = make_note(dict(public=0)) - - self.assertEqual(frappe.db.get_value('ToDo', dict( - reference_type = 'Note', - reference_name = note.name, - status = 'Open' - ), 'owner'), None) - - def test_clear_assignment(self): - note = make_note(dict(public=1)) - - # check if auto assigned to first user - self.assertEqual(frappe.db.get_value('ToDo', dict( - reference_type = 'Note', - reference_name = note.name, - status = 'Open' - ), 'owner'), 'test@example.com') - - # test auto unassign - note.public = 0 - note.save() - - # check if cleared - self.assertEqual(frappe.db.get_value('ToDo', dict( - reference_type = 'Note', - reference_name = note.name, - status = 'Open' - ), 'owner'), None) - - def check_multiple_rules(self): - note = make_note(dict(public=1, notify_on_login=1)) - - # check if auto assigned to test3 (2nd rule is applied, as it has higher priority) - self.assertEqual(frappe.db.get_value('ToDo', dict( - reference_type = 'Note', - reference_name = note.name, - status = 'Open' - ), 'owner'), 'test@example.com') - -def clear_assignments(): - frappe.db.sql("delete from tabToDo where reference_type = 'Note'") - -def get_assignment_rule(): - frappe.delete_doc_if_exists('Assignment Rule', 'For Note 1') - - assignment_rule = frappe.get_doc(dict( - name = 'For Note 1', - doctype = 'Assignment Rule', - priority = 0, - document_type = 'Note', - assign_condition = 'public == 1', - unassign_condition = 'public == 0', - rule = 'Round Robin', - users = [ - dict(user = 'test@example.com'), - dict(user = 'test1@example.com'), - dict(user = 'test2@example.com'), - ] - )).insert() - - frappe.delete_doc_if_exists('Assignment Rule', 'For Note 2') - - # 2nd rule - frappe.get_doc(dict( - name = 'For Note 2', - doctype = 'Assignment Rule', - priority = 1, - document_type = 'Note', - assign_condition = 'notify_on_login == 1', - unassign_condition = 'notify_on_login == 0', - rule = 'Round Robin', - users = [ - dict(user = 'test3@example.com') - ] - )).insert() - - - return assignment_rule - -def make_note(values=None): - note = frappe.get_doc(dict( - doctype = 'Note', - title = random_string(10), - content = random_string(20) - )) - - if values: - note.update(values) - - note.insert() - - return note \ No newline at end of file