fix: remove old auto_assign.py files
This commit is contained in:
Rushabh Mehta
parent
fc5b1d2c3e
commit
5fbbdbe499
2 changed files with 0 additions and 281 deletions
|
|
@ -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))]
|
||||
|
|
@ -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
|
||||
Loading…
Add table
Reference in a new issue