Aerex/taskwarrior-grocy
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

test(grocy): added unit tests for add, find_all, delete, to_grocy,etc

Browse Source 
BREAKING_CHANGE: changed findAll to find_all

style: clean up code
This commit is contained in:
Aerex
2019-01-20 22:53:06 -08:00
parent 77f3b06e11
commit 1b220db744
5 changed files with 269 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
twservices/apps/__init__.py
View File

@@ -9,7 +9,9 @@ class TaskService(object):
DESCRIPTION = 'description'
TAGS = 'tags'
ANNOTATIONS = 'annotations'
STATUS = 'status'
ENTRY = 'entry'
COMPLETED = 'completed'
DONE = 'done'
DUE = 'due'

102
twservices/apps/grocy.py
View File

@@ -13,6 +13,7 @@ class Grocy(TaskService):
# PROPS
ID = 'id'
DONE = 'done'
DESCRIPTION = 'description'
ASSIGNED_TO_USER_ID = 'assigned_to_user_id'
ROW_CREATED_TIMESTAMP = 'row_created_timestamp'
CATEGORY = 'category'
@@ -155,39 +156,38 @@ class Grocy(TaskService):
def to_taskwarrior(self, grocy_task):
taskwarrior_task = {}
if 'id' in grocy_task:
taskwarrior_task[Grocy.UDA_GROCY_ID] = grocy_task['id']
if Grocy.ID in grocy_task:
taskwarrior_task[Grocy.UDA_GROCY_ID] = grocy_task[Grocy.ID]
if Grocy.UDA_GROCY_TW_UUID in grocy_task:
taskwarrior_task['uuid'] = grocy_task[Grocy.UDA_GROCY_TW_UUID]
taskwarrior_task[TaskService.UUID] = grocy_task[Grocy.UDA_GROCY_TW_UUID]
if Grocy.ASSIGNED_TO_USER_ID in grocy_task:
taskwarrior_task[Grocy.UDA_GROCY_ASSIGNED_TO_USER_ID] = grocy_task[Grocy.ASSIGNED_TO_USER_ID]
if 'name' in grocy_task:
taskwarrior_task['description'] = grocy_task['name']
if Grocy.NAME in grocy_task:
taskwarrior_task[TaskService.DESCRIPTION] = grocy_task[Grocy.NAME]
if 'description' in grocy_task:
if '\r' in grocy_task['description']:
grocy_task['description'] = grocy_task['description'].replace('\r', '')
taskwarrior_task['annotations'] = grocy_task['description'].split('\n')
last_element_is_empty_string = taskwarrior_task['annotations'][-1] == ''
if Grocy.DESCRIPTION in grocy_task:
if '\r' in grocy_task[Grocy.DESCRIPTION]:
grocy_task[Grocy.DESCRIPTION] = grocy_task[Grocy.DESCRIPTION].replace('\r', '')
taskwarrior_task[TaskService.ANNOTATIONS] = grocy_task[Grocy.DESCRIPTION].split('\n')
last_element_is_empty_string = taskwarrior_task[TaskService.ANNOTATIONS][-1] == ''
if last_element_is_empty_string:
taskwarrior_task['annotations'] = taskwarrior_task['annotations'][0:-1] # Remove empty string in array
taskwarrior_task[TaskService.ANNOTATIONS] = taskwarrior_task[TaskService.ANNOTATIONS][0:-1] # Remove empty string in array
if 'row_created_timestamp' in grocy_task:
taskwarrior_task['entry'] = grocy_task['row_created_timestamp']
if Grocy.ROW_CREATED_TIMESTAMP in grocy_task:
taskwarrior_task[TaskService.ENTRY] = grocy_task[Grocy.ROW_CREATED_TIMESTAMP]
if 'due_date' in grocy_task:
taskwarrior_task['due'] = grocy_task['due_date']
if Grocy.DUE_DATE in grocy_task:
taskwarrior_task[TaskService.DUE] = grocy_task[Grocy.DUE_DATE]
if 'done' in grocy_task and grocy_task['done'] == '1':
taskwarrior_task['status'] = 'completed'
if Grocy.DONE in grocy_task and grocy_task[Grocy.DONE] == '1':
taskwarrior_task[TaskService.STATUS] = TaskService.COMPLETED
if 'category_id' in grocy_task:
if Grocy.CATEGORY in grocy_task:
category = self.get_category()
print('category {}'.format(category))
if category:
taskwarrior_task['tags'] = [ category['name'] ]
taskwarrior_task[TaskService.TAGS] = [ category['name'] ]
return taskwarrior_task
@@ -200,16 +200,16 @@ class Grocy(TaskService):
rest_service.addHeader(Grocy.API_KEY_HEADER, self.token)
return rest_service
def findAll(self):
def find_all(self):
if not self.rest_service:
raise Exception('rest service for grocy is not defined')
# Get all tasks as json
try:
response = self.rest_service.get(Grocy.FIND_ALL_TASKS_ENDPOINT)
responses = self.rest_service.get(Grocy.FIND_ALL_TASKS_ENDPOINT)
except BaseException as e:
logger.error(e)
raise e
return response
return responses
def delete(self, id):
if not self.rest_service:
@@ -219,49 +219,57 @@ class Grocy(TaskService):
# Delete a task
try:
response = self.rest_service.get(Grocy.DELETE_TASK_ENDPOINT, id)
if response and response['success']:
return response
elif response and not response['success']:
raise Exception(response['error'])
else:
raise Exception('Grocy service unavailable Service')
except Exception as e:
logger.error(e)
raise e
return response
def add(self, task):
def add(self, grocy_task):
if not self.rest_service:
raise Exception('rest service for grocy is not defined')
responses = []
if not task:
raise Exception('task is not defined')
if not type(task) is list:
tasks = [task]
if not grocy_task:
raise Exception('grocy_task is not defined')
if not type(grocy_task) is list:
grocy_tasks = [grocy_task]
else:
tasks = task
grocy_tasks = grocy_task
try:
for next_task in tasks:
self.to_grocy(next_task)
response = self.rest_service.post(Grocy.ADD_TASK_ENDPOINT, task)
for next_grocy_task in grocy_tasks:
response = self.rest_service.post(Grocy.ADD_TASK_ENDPOINT, next_grocy_task)
responses.append(response)
except Exception as e:
logger.error(e)
raise e
return responses
def modify(self, task):
def modify(self, grocy_task):
if not self.rest_service:
raise Exception('rest service for grocy is not defined')
response = []
if not task:
raise Exception('task is not defined')
if not type(task) is list:
tasks = [task]
responses = []
if not grocy_task:
raise Exception('grocy_task is not defined')
if not type(grocy_task) is list:
grocy_tasks = [grocy_task]
else:
tasks = task
grocy_tasks = grocy_task
try:
for next_task in tasks:
modify_endpoint = '{0}/{1}'.format(Grocy.MODIFY_TASK_ENDPOINT, next_task['id'])
response = self.rest_service.post(modify_endpoint, next_task)
for grocy_next_task in grocy_tasks:
modify_endpoint = '{0}/{1}'.format(Grocy.MODIFY_TASK_ENDPOINT, grocy_next_task['id'])
response = self.rest_service.post(modify_endpoint, grocy_next_task)
responses.append(response)
except Exception as e:
logger.exception('Could not send post to modify grocy task %s', e.stderr)
logger.exception('Could not send post to modify grocy grocy_task %s', e.stderr)
raise e
return responses
def _should_merge(self, converted_tw_to_grocy_task, grocy_task):
return True if converted_tw_to_grocy_task['id'] == grocy_task['id'] else False
@@ -380,17 +388,17 @@ class Grocy(TaskService):
taskwarrior_tasks = self.get_grocy_tw_tasks()
try:
grocy_tasks = self.findAll()
grocy_tasks = self.find_all()
except BaseException as e:
logger.error(e)
logger.error('Could not get all grocy tasks')
try:
if self.resolution == 'tw':
grocy_tasks = self.findAll()
grocy_tasks = self.find_all()
self.merge_from_tw(taskwarrior_tasks, grocy_tasks)
elif self.resolution == 'grocy':
grocy_tasks = self.findAll()
grocy_tasks = self.find_all()
self.merge_into_tw(taskwarrior_tasks, grocy_tasks)
else:
raise Exception('Could not determine resolution for grocy')