test: started adding tests for grocy
This commit is contained in:
parent
ae132c2484
commit
77f3b06e11
1
.gitignore
vendored
1
.gitignore
vendored
@ -3,4 +3,5 @@ log
|
||||
include/
|
||||
bin/
|
||||
lib/
|
||||
.tox/*
|
||||
*.egg-info
|
||||
|
9
setup.py
9
setup.py
@ -20,8 +20,15 @@ setup(
|
||||
keywords='taskwarrior, grocy',
|
||||
url='http://packages.python.org/an_example_pypi_project',
|
||||
packages=find_packages(),
|
||||
install_requires=['Click', 'pyyaml', 'requests', 'taskw'],
|
||||
include_package_data=True,
|
||||
zip_safe=False,
|
||||
install_requires=['Click', 'pyyaml', 'requests', 'taskw', 'lockfile', 'tox'],
|
||||
long_description=read('README.rst'),
|
||||
tests_require=[
|
||||
"pytest_mock",
|
||||
"pytest",
|
||||
"mock"
|
||||
],
|
||||
classifiers=[
|
||||
"Development Status :: 3 - Alpha",
|
||||
"Topic :: Utilities",
|
||||
|
0
tests/__init__.py
Normal file
0
tests/__init__.py
Normal file
109
tests/test_grocy.py
Normal file
109
tests/test_grocy.py
Normal file
@ -0,0 +1,109 @@
|
||||
import uuid
|
||||
import datetime
|
||||
import pytest
|
||||
import sys
|
||||
import mock
|
||||
from twservices.apps import RestService, TaskService
|
||||
from pytest_mock import mocker
|
||||
|
||||
|
||||
sys.path.append(".")
|
||||
|
||||
from twservices.apps.grocy import Grocy
|
||||
|
||||
#def test_get_grocy_tw_tasks(mocker, test_mock_tw_instance):
|
||||
# expect_filters = {
|
||||
# 'and': [('%s.any' % key, none) for key in list(grocy.udas.keys())],
|
||||
# 'or': [('status', 'pending'), ('status', 'waiting')]
|
||||
# }
|
||||
# mocker.spy(test_mock_tw_instance, 'filter_tasks')
|
||||
#
|
||||
# grocy = grocy()
|
||||
# grocy.set_taskwarrior(test_mock_tw_instance)
|
||||
# actual_tasks = grocy.get_grocy_tw_tasks()
|
||||
#
|
||||
# assert actual_tasks is not none
|
||||
# assert grocy.get_grocy_tw_tasks.called.assert_called_with(expect_filters) is true
|
||||
|
||||
@pytest.fixture(scope='function')
|
||||
def setup_grocy_args():
|
||||
grocy_args = {
|
||||
'token': 'e96a89a5',
|
||||
'api': 'http://example.grocy.com',
|
||||
'category': 'grocy'
|
||||
}
|
||||
return grocy_args
|
||||
|
||||
|
||||
def test_get_category(mocker, setup_grocy_args):
|
||||
expected_category = setup_grocy_args['category']
|
||||
expected_token = setup_grocy_args['token']
|
||||
expected_grocy_categories = [{'name': expected_category}, {'name': 'test'}]
|
||||
|
||||
mocker.patch.object(RestService, '__init__')
|
||||
mocker.patch.object(RestService, 'get')
|
||||
mocker.patch.object(RestService, 'addHeader')
|
||||
RestService.__init__.return_value = None
|
||||
RestService.get.return_value = expected_grocy_categories
|
||||
|
||||
grocy = Grocy(**setup_grocy_args)
|
||||
actual_category = grocy.get_category()
|
||||
RestService.addHeader.assert_called_with(Grocy.API_KEY_HEADER, expected_token)
|
||||
assert actual_category == {'name': expected_category }
|
||||
|
||||
def test_create_category(mocker, setup_grocy_args):
|
||||
expected_category = setup_grocy_args['category']
|
||||
expected_post_response = {'success': True, 'error': ''}
|
||||
|
||||
mocker.patch.object(RestService, '__init__')
|
||||
mocker.patch.object(RestService, 'post')
|
||||
mocker.patch.object(RestService, 'addHeader')
|
||||
RestService.__init__.return_value = None
|
||||
RestService.post.return_value = expected_post_response
|
||||
mocker.patch.object(RestService, 'addHeader')
|
||||
|
||||
grocy = Grocy(**setup_grocy_args)
|
||||
grocy.create_category()
|
||||
RestService.post.assert_called_with(Grocy.ADD_CATEGORY_ENDPOINT, expected_category)
|
||||
|
||||
def test_to_grocy(mocker, setup_grocy_args):
|
||||
mocker.patch.object(Grocy, 'get_category')
|
||||
notes_arr = ['This is line1', 'This is line2']
|
||||
notes_string = ''
|
||||
entry_time = datetime.time(18,22,00)
|
||||
entry_date = datetime.date(2019,1,16)
|
||||
due_time = datetime.time(18,26,00)
|
||||
due_date = datetime.date(2019,1,20)
|
||||
for note in notes_arr:
|
||||
notes_string+= '{}\n'.format(note)
|
||||
|
||||
expected_task = {
|
||||
Grocy.UDA_GROCY_ID: '1337',
|
||||
Grocy.UDA_GROCY_ASSIGNED_TO_USER_ID : '1',
|
||||
TaskService.UUID: '1a727026-9611-4d8e-ba4d-d5dc6eaf7054',
|
||||
TaskService.DESCRIPTION: 'This is a task from taskwarrior',
|
||||
TaskService.TAGS : setup_grocy_args['category'],
|
||||
TaskService.ANNOTATIONS: notes_arr,
|
||||
TaskService.ENTRY: datetime.datetime.combine(entry_date,entry_time),
|
||||
TaskService.DUE: datetime.datetime.combine(entry_date,entry_time),
|
||||
TaskService.DONE: '1'
|
||||
}
|
||||
expected_grocy_task = {
|
||||
Grocy.ID : expected_task[Grocy.UDA_GROCY_ID],
|
||||
Grocy.UDA_GROCY_TW_UUID: expected_task[TaskService.UUID],
|
||||
Grocy.ASSIGNED_TO_USER_ID: expected_task[Grocy.UDA_GROCY_ASSIGNED_TO_USER_ID],
|
||||
Grocy.NAME: expected_task[TaskService.DESCRIPTION],
|
||||
Grocy.CATEGORY: expected_task[TaskService.TAGS],
|
||||
Grocy.DESCRIPTION: notes_string,
|
||||
Grocy.ROW_CREATED_TIMESTAMP: expected_task[TaskService.ENTRY].strftime('%Y-%m-%d %H:%M:%S'),
|
||||
Grocy.DUE_DATE: expected_task[TaskService.DUE].strftime('%Y-%m-%d'),
|
||||
Grocy.DONE: expected_task[TaskService.DONE]
|
||||
}
|
||||
print('expected_grocy_task {}'.format(expected_grocy_task))
|
||||
|
||||
Grocy.get_category.return_value = {'name': setup_grocy_args['category']}
|
||||
grocy = Grocy(**setup_grocy_args)
|
||||
actual_grocy_task = grocy.to_grocy(expected_task)
|
||||
print('actual_grocy_task {}'.format(actual_grocy_task))
|
||||
assert expected_grocy_task == actual_grocy_task
|
||||
|
14
tox.ini
Normal file
14
tox.ini
Normal file
@ -0,0 +1,14 @@
|
||||
[tox]
|
||||
envlist = py37
|
||||
|
||||
[testenv]
|
||||
deps=
|
||||
mock
|
||||
pytest
|
||||
pytest_mock
|
||||
setenv=
|
||||
XDG_CACHE_HOME={envtmpdir}/
|
||||
commands=
|
||||
py.test \
|
||||
--verbose \
|
||||
{posargs:./tests}
|
@ -5,6 +5,16 @@ logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class TaskService(object):
|
||||
UUID = 'uuid'
|
||||
DESCRIPTION = 'description'
|
||||
TAGS = 'tags'
|
||||
ANNOTATIONS = 'annotations'
|
||||
ENTRY = 'entry'
|
||||
DONE = 'done'
|
||||
DUE = 'due'
|
||||
|
||||
|
||||
|
||||
def __init__(self, **entries):
|
||||
self.__dict__.update(entries)
|
||||
|
||||
|
@ -10,6 +10,15 @@ logger = logging.getLogger(__name__)
|
||||
class Grocy(TaskService):
|
||||
API_KEY_HEADER = 'GROCY-API-KEY'
|
||||
|
||||
# PROPS
|
||||
ID = 'id'
|
||||
DONE = 'done'
|
||||
ASSIGNED_TO_USER_ID = 'assigned_to_user_id'
|
||||
ROW_CREATED_TIMESTAMP = 'row_created_timestamp'
|
||||
CATEGORY = 'category'
|
||||
DUE_DATE = 'due_date'
|
||||
NAME = 'name'
|
||||
|
||||
# Endpoints
|
||||
FIND_ALL_TASKS_ENDPOINT = '/get-objects/tasks'
|
||||
FIND_ALL_CATEGORIES_ENDPOINT = '/get-objects/task_categories'
|
||||
@ -17,21 +26,23 @@ class Grocy(TaskService):
|
||||
ADD_CATEGORY_ENDPOINT = '/add-object/task_categories'
|
||||
ADD_TASK_ENDPOINT = '/add-object/tasks'
|
||||
MODIFY_TASK_ENDPOINT = '/edit-object/tasks'
|
||||
|
||||
# UDA
|
||||
GROCY_ID = 'grocy_id'
|
||||
GROCY_ASSIGNED_TO_USER_ID = 'grocy_assigned_to'
|
||||
GROCY_CATEGORY_ID = 'grocy_category_id'
|
||||
GROCY_CATEGORY_NAME = 'grocy_category_name'
|
||||
UDA_GROCY_ID = 'grocy_id'
|
||||
UDA_GROCY_TW_UUID = '_uuid'
|
||||
UDA_GROCY_ASSIGNED_TO_USER_ID = 'grocy_assigned_to'
|
||||
UDA_GROCY_CATEGORY_ID = 'grocy_category_id'
|
||||
UDA_GROCY_CATEGORY_NAME = 'grocy_category_name'
|
||||
UDAS = {
|
||||
GROCY_ID: {
|
||||
UDA_GROCY_ID: {
|
||||
'type': 'string',
|
||||
'label': 'Grocy Task ID'
|
||||
},
|
||||
GROCY_ASSIGNED_TO_USER_ID: {
|
||||
UDA_GROCY_ASSIGNED_TO_USER_ID: {
|
||||
'type': 'string',
|
||||
'label': 'Grocy Assigned User ID'
|
||||
},
|
||||
GROCY_CATEGORY_ID: {
|
||||
UDA_GROCY_CATEGORY_ID: {
|
||||
'type': 'string',
|
||||
'label': 'Grocy Category ID'
|
||||
}
|
||||
@ -76,10 +87,8 @@ class Grocy(TaskService):
|
||||
if responses:
|
||||
for category in responses:
|
||||
if category['name'] == self.category:
|
||||
print('not sure')
|
||||
print(category)
|
||||
return category
|
||||
logger.warn('Could not find category {} in list'.format(self.category))
|
||||
logger.warning('Could not find category {} in list'.format(self.category))
|
||||
|
||||
return None
|
||||
|
||||
@ -88,50 +97,58 @@ class Grocy(TaskService):
|
||||
raise Exception('rest service for grocy is not defined')
|
||||
# Add new category to list
|
||||
try:
|
||||
responses = self.rest_service.get(Grocy.ADD_CATEGORY_ENDPOINT, self.category)
|
||||
except BaseException as e:
|
||||
response = self.rest_service.post(Grocy.ADD_CATEGORY_ENDPOINT, self.category)
|
||||
if response and response['success']:
|
||||
return
|
||||
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 responses
|
||||
|
||||
def to_grocy(self, task):
|
||||
grocy_task = {}
|
||||
|
||||
if Grocy.GROCY_ID in task:
|
||||
grocy_task['id'] = task[Grocy.GROCY_ID]
|
||||
if Grocy.UDA_GROCY_ID in task:
|
||||
grocy_task[Grocy.ID] = task[Grocy.UDA_GROCY_ID]
|
||||
|
||||
if 'uuid' in task:
|
||||
grocy_task['_uuid'] = task['uuid']
|
||||
if TaskService.UUID in task:
|
||||
grocy_task[Grocy.UDA_GROCY_TW_UUID] = task[TaskService.UUID]
|
||||
|
||||
if Grocy.GROCY_ASSIGNED_TO_USER_ID in task:
|
||||
grocy_task['assigned_to_user_id'] = task[Grocy.GROCY_ASSIGNED_TO_USER_ID]
|
||||
if Grocy.UDA_GROCY_ASSIGNED_TO_USER_ID in task:
|
||||
grocy_task[Grocy.ASSIGNED_TO_USER_ID] = task[Grocy.UDA_GROCY_ASSIGNED_TO_USER_ID]
|
||||
|
||||
if 'description' in task:
|
||||
grocy_task['name'] = task['description']
|
||||
if TaskService.DESCRIPTION in task:
|
||||
grocy_task[Grocy.NAME] = task[TaskService.DESCRIPTION]
|
||||
|
||||
if 'tags' in task and self.category in task['tags']:
|
||||
category_id = self.get_category()
|
||||
category_does_not_exist = not category_id
|
||||
if TaskService.TAGS in task and self.category in task[TaskService.TAGS]:
|
||||
category = self.get_category()
|
||||
category_does_not_exist = not category
|
||||
|
||||
if category_does_not_exist:
|
||||
category_id = self.create_category()
|
||||
category = self.create_category()
|
||||
elif category and 'name' in category:
|
||||
grocy_task[Grocy.CATEGORY] = category['name']
|
||||
else:
|
||||
logger.error('Grocy category does not have a name property or is undefined')
|
||||
raise Exception('Grocy category does not have a name property or is undefined')
|
||||
|
||||
grocy_task['category'] = category_id
|
||||
if TaskService.ANNOTATIONS in task:
|
||||
grocy_task[Grocy.DESCRIPTION] = ''
|
||||
for note in task[TaskService.ANNOTATIONS]:
|
||||
grocy_task[Grocy.DESCRIPTION] += '{}\n'.format(note)
|
||||
|
||||
if 'annotations' in task:
|
||||
grocy_task['description'] = ''
|
||||
for note in task['annotations']:
|
||||
grocy_task['description'] += '{}\n'.format(note)
|
||||
if TaskService.ENTRY in task and type(task[TaskService.ENTRY]) is datetime:
|
||||
grocy_task[Grocy.ROW_CREATED_TIMESTAMP] = task[TaskService.ENTRY].strftime('%Y-%m-%d %H:%M:%S')
|
||||
|
||||
if 'entry' in task and type(task['entry']) is datetime:
|
||||
grocy_task['row_created_timestamp'] = task['entry'].strftime('%Y-%m-%d %H:%M:%S')
|
||||
if TaskService.DUE in task and type(task[TaskService.DUE]) is datetime:
|
||||
grocy_task[Grocy.DUE_DATE] = task[TaskService.DUE].strftime('%Y-%m-%d')
|
||||
|
||||
if 'due' in task and type(task['due']) is datetime:
|
||||
grocy_task['due_date'] = task['due'].strftime('%Y-%m-%d')
|
||||
|
||||
if 'done' in task:
|
||||
grocy_task['done'] = task['done']
|
||||
if TaskService.DONE in task:
|
||||
grocy_task[Grocy.DONE] = task[TaskService.DONE]
|
||||
|
||||
return grocy_task
|
||||
|
||||
@ -139,12 +156,12 @@ class Grocy(TaskService):
|
||||
taskwarrior_task = {}
|
||||
|
||||
if 'id' in grocy_task:
|
||||
taskwarrior_task[Grocy.GROCY_ID] = grocy_task['id']
|
||||
if '_uuid' in grocy_task:
|
||||
taskwarrior_task['uuid'] = grocy_task['_uuid']
|
||||
taskwarrior_task[Grocy.UDA_GROCY_ID] = grocy_task['id']
|
||||
if Grocy.UDA_GROCY_TW_UUID in grocy_task:
|
||||
taskwarrior_task['uuid'] = grocy_task[Grocy.UDA_GROCY_TW_UUID]
|
||||
|
||||
if 'assigned_to_user_id' in grocy_task:
|
||||
taskwarrior_task[Grocy.GROCY_ASSIGNED_TO_USER_ID] = grocy_task['assigned_to_user_id']
|
||||
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']
|
||||
@ -157,7 +174,6 @@ class Grocy(TaskService):
|
||||
if last_element_is_empty_string:
|
||||
taskwarrior_task['annotations'] = taskwarrior_task['annotations'][0:-1] # Remove empty string in array
|
||||
|
||||
|
||||
if 'row_created_timestamp' in grocy_task:
|
||||
taskwarrior_task['entry'] = grocy_task['row_created_timestamp']
|
||||
|
||||
@ -169,8 +185,9 @@ class Grocy(TaskService):
|
||||
|
||||
if 'category_id' in grocy_task:
|
||||
category = self.get_category()
|
||||
taskwarrior_task[Grocy.GRO
|
||||
|
||||
print('category {}'.format(category))
|
||||
if category:
|
||||
taskwarrior_task['tags'] = [ category['name'] ]
|
||||
|
||||
return taskwarrior_task
|
||||
|
||||
@ -178,7 +195,7 @@ class Grocy(TaskService):
|
||||
if self.api.startswith == '/':
|
||||
self.api = self.api[1:]
|
||||
if self.api.endswith == '/':
|
||||
self.api = self.api[0:-1]
|
||||
self.api = self.api[1:-1]
|
||||
rest_service = RestService(self.api, json=True)
|
||||
rest_service.addHeader(Grocy.API_KEY_HEADER, self.token)
|
||||
return rest_service
|
||||
@ -259,7 +276,7 @@ class Grocy(TaskService):
|
||||
tasks_to_add_to_grocy = []
|
||||
converted_tw_to_grocy_tasks = [self.to_grocy(task) for task in taskwarrior_tasks]
|
||||
|
||||
# Sort grocy tasks by id and converted taskwarrior tasks by GROCY_ID
|
||||
# Sort grocy tasks by id and converted taskwarrior tasks by UDA_GROCY_ID
|
||||
grocy_tasks.sort(key=lambda x: x['id'])
|
||||
converted_tw_to_grocy_tasks.sort(key=lambda x: x['id'])
|
||||
|
||||
@ -306,13 +323,11 @@ class Grocy(TaskService):
|
||||
logger.debug(added_responses)
|
||||
|
||||
def merge_into_tw(self, taskwarrior_tasks, grocy_tasks):
|
||||
print(taskwarrior_tasks)
|
||||
print(grocy_tasks)
|
||||
# Convert to taskwarrior tasks into grocy tasks
|
||||
converted_tw_to_grocy_tasks = []
|
||||
converted_tw_to_grocy_tasks = [self.to_grocy(task) for task in taskwarrior_tasks]
|
||||
|
||||
# Sort grocy tasks by id and converted taskwarrior tasks by GROCY_ID
|
||||
# Sort grocy tasks by id and converted taskwarrior tasks by UDA_GROCY_ID
|
||||
grocy_tasks.sort(key=lambda x: x['id'])
|
||||
converted_tw_to_grocy_tasks.sort(key=lambda x: x['id'])
|
||||
|
||||
@ -350,9 +365,7 @@ class Grocy(TaskService):
|
||||
if self._should_merge(converted_tw_to_grocy_task, grocy_task):
|
||||
try:
|
||||
converted_grocy_task_to_tw = self.to_taskwarrior(grocy_task)
|
||||
converted_grocy_task_to_tw['uuid'] = converted_tw_to_grocy_task['_uuid']
|
||||
print('convert fool')
|
||||
print(converted_grocy_task_to_tw)
|
||||
converted_grocy_task_to_tw['uuid'] = converted_tw_to_grocy_task[Grocy.UDA_GROCY_TW_UUID]
|
||||
self.tw.task_update(converted_grocy_task_to_tw)
|
||||
logger.debug('Merged grocy task %s to taskwarrior task %s', grocy_task['id'], converted_grocy_task_to_tw['uuid'])
|
||||
del converted_tw_to_grocy_tasks[index]
|
||||
@ -374,8 +387,10 @@ class Grocy(TaskService):
|
||||
|
||||
try:
|
||||
if self.resolution == 'tw':
|
||||
grocy_tasks = self.findAll()
|
||||
self.merge_from_tw(taskwarrior_tasks, grocy_tasks)
|
||||
elif self.resolution == 'grocy':
|
||||
grocy_tasks = self.findAll()
|
||||
self.merge_into_tw(taskwarrior_tasks, grocy_tasks)
|
||||
else:
|
||||
raise Exception('Could not determine resolution for grocy')
|
||||
|
Loading…
Reference in New Issue
Block a user