13108386717

Committed 03 Feb 2025 07:19AM UTC coverage: 80.453% (-3.8%) from 84.265%

Build # 13108386717

Build Type

Pull #101

github

Committed by

co-stig

Commit Message

Resolving SonarQube security warnings

Pull Request Pull Request #101: Rc 0.10.0

Run Details

218 of 400 new or added lines in 18 files covered. (54.5%)

35 existing lines in 7 files now uncovered.

3054 of 3796 relevant lines covered (80.45%)

0.8 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

98.09

/src/fk/tests/test_workitems.py

#  Flowkeeper - Pomodoro timer for power users and teams
#  Copyright (c) 2023 Constantine Kulak
#
#  This program is free software: you can redistribute it and/or modify
#  it under the terms of the GNU General Public License as published by
#  the Free Software Foundation; either version 3 of the License, or
#  (at your option) any later version.
#
#  This program is distributed in the hope that it will be useful,
#  but WITHOUT ANY WARRANTY; without even the implied warranty of
#  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
#  GNU General Public License for more details.
#
#  You should have received a copy of the GNU General Public License
#  along with this program.  If not, see <https://www.gnu.org/licenses/>.
import datetime
from unittest import TestCase

from fk.core.abstract_cryptograph import AbstractCryptograph
from fk.core.abstract_settings import AbstractSettings
from fk.core.backlog import Backlog
from fk.core.backlog_strategies import CreateBacklogStrategy
from fk.core.ephemeral_event_source import EphemeralEventSource
from fk.core.fernet_cryptograph import FernetCryptograph
from fk.core.mock_settings import MockSettings
from fk.core.pomodoro_strategies import AddPomodoroStrategy, StartWorkStrategy
from fk.core.tenant import Tenant
from fk.core.user import User
from fk.core.workitem import Workitem
from fk.core.workitem_strategies import CreateWorkitemStrategy, RenameWorkitemStrategy, DeleteWorkitemStrategy, \
    CompleteWorkitemStrategy


class TestWorkitems(TestCase):
    settings: AbstractSettings
    cryptograph: AbstractCryptograph
    source: EphemeralEventSource
    data: dict[str, User]

    def setUp(self) -> None:
        self.settings = MockSettings()
        self.cryptograph = FernetCryptograph(self.settings)
        self.source = EphemeralEventSource[Tenant](self.settings, self.cryptograph, Tenant(self.settings))
        self.source.start()
        self.data = self.source.get_data()

    def tearDown(self) -> None:
        self.source.dump()

    def _assert_workitem(self, workitem1: Workitem, user: User, backlog: Backlog):
        self.assertEqual(workitem1.get_name(), 'First workitem')
        self.assertEqual(workitem1.get_uid(), 'w11')
        self.assertEqual(workitem1.get_parent(), backlog)
        self.assertEqual(workitem1.get_owner(), user)
        self.assertFalse(workitem1.is_running())
        self.assertFalse(workitem1.is_sealed())
        self.assertFalse(workitem1.is_startable())
        self.assertFalse(workitem1.has_running_pomodoro())
        self.assertTrue(workitem1.is_planned())
        self.assertEqual(len(workitem1.values()), 0)
        
    def _standard_backlog(self) -> (User, Backlog): 
        self.source.execute(CreateBacklogStrategy, ['b1', 'First backlog'])
        self.source.auto_seal()
        user = self.data['user@local.host']
        backlog = user['b1']
        return user, backlog

    def test_create_workitems(self):
        user, backlog = self._standard_backlog()
        self.source.execute(CreateWorkitemStrategy, ['w11', 'b1', 'First workitem'])
        self.source.execute(CreateWorkitemStrategy, ['w12', 'b1', 'Second workitem'])
        self.source.auto_seal()
        self.assertIn('w11', backlog)
        self.assertIn('w12', backlog)
        workitem1: Workitem = backlog['w11']
        self._assert_workitem(workitem1, user, backlog)
        workitem2 = backlog['w12']
        self.assertEqual(workitem2.get_name(), 'Second workitem')

    def test_create_workitems_with_tags(self):
        user, backlog = self._standard_backlog()
        self.source.execute(CreateWorkitemStrategy, ['w11', 'b1', 'First workitem'])
        self.source.execute(CreateWorkitemStrategy, ['w12', 'b1', '#Second workitem'])
        self.source.execute(CreateWorkitemStrategy, ['w13', 'b1', '#Third #workitem'])
        self.source.execute(CreateWorkitemStrategy, ['w14', 'b1', 'Fourth #workitem and some more #workitem text'])
        self.source.execute(CreateWorkitemStrategy, ['w15', 'b1', 'Fifth #workitem.'])
        self.source.execute(CreateWorkitemStrategy, ['w16', 'b1', 'Six #workitem and #workitems'])
        self.source.auto_seal()
        self.assertIn('w11', backlog)
        self.assertIn('w12', backlog)
        self.assertIn('w13', backlog)
        self.assertIn('w14', backlog)
        self.assertIn('w15', backlog)
        self.assertIn('w16', backlog)
        workitem1: Workitem = backlog['w11']
        self._assert_workitem(workitem1, user, backlog)
        workitem2 = backlog['w12']
        self.assertEqual(workitem2.get_name(), '#Second workitem')
        workitem3 = backlog['w13']
        self.assertEqual(workitem3.get_name(), '#Third #workitem')
        workitem4 = backlog['w14']
        self.assertEqual(workitem4.get_name(), 'Fourth #workitem and some more #workitem text')
        workitem5 = backlog['w15']
        self.assertEqual(workitem5.get_name(), 'Fifth #workitem.')
        workitem6 = backlog['w16']
        self.assertEqual(workitem6.get_name(), 'Six #workitem and #workitems')
        tags = user.get_tags()
        self.assertEqual(len(tags), 4)
        self.assertIn('workitem', tags)
        self.assertIn('second', tags)
        self.assertIn('third', tags)
        self.assertIn('workitems', tags)
        workitems = tags['workitem'].get_workitems()
        self.assertEqual(len(workitems), 4)
        self.assertIn(workitem3, workitems)
        self.assertIn(workitem4, workitems)
        self.assertIn(workitem5, workitems)
        self.assertIn(workitem6, workitems)

    def test_execute_prepared(self):
        user, backlog = self._standard_backlog()
        s = CreateWorkitemStrategy(2,
                                  datetime.datetime.now(datetime.timezone.utc),
                                  user.get_identity(),
                                  ['w11', 'b1', 'First workitem'],
                                  self.settings)
        self.source.execute_prepared_strategy(s)
        self.source.auto_seal()
        self.assertIn('w11', backlog)
        workitem1: Workitem = backlog['w11']
        self._assert_workitem(workitem1, user, backlog)

    def test_create_duplicate_workitem_failure(self):
        self._standard_backlog()
        self.source.execute(CreateWorkitemStrategy, ['w11', 'b1', 'First workitem 1'])
        self.source.auto_seal()
        self.assertRaises(Exception,
                          lambda: self.source.execute(CreateWorkitemStrategy, ['w11', 'b1', 'First workitem 2']))

    def test_rename_nonexistent_workitem_failure(self):
        self._standard_backlog()
        self.source.execute(CreateWorkitemStrategy, ['w11', 'b1', 'First workitem'])
        self.source.execute(CreateWorkitemStrategy, ['w12', 'b1', 'Second workitem'])
        self.source.auto_seal()
        self.assertRaises(Exception,
                          lambda: self.source.execute(RenameWorkitemStrategy, ['w13', 'Renamed workitem']))

    def test_rename_workitem(self):
        user, backlog = self._standard_backlog()
        self.source.execute(CreateWorkitemStrategy, ['w11', 'b1', 'First workitem'])
        self.source.execute(RenameWorkitemStrategy, ['w11', 'Renamed workitem'])
        self.source.auto_seal()
        self.assertEqual(backlog['w11'].get_name(), 'Renamed workitem')

    def test_delete_nonexistent_workitem_failure(self):
        self._standard_backlog()
        self.source.execute(CreateWorkitemStrategy, ['w11', 'b1', 'First workitem'])
        self.source.execute(CreateWorkitemStrategy, ['w12', 'b1', 'Second workitem'])
        self.source.auto_seal()
        self.assertRaises(Exception,
                          lambda: self.source.execute(DeleteWorkitemStrategy, ['w13']))

    def test_delete_workitem(self):
        user, backlog = self._standard_backlog()
        self.source.execute(CreateWorkitemStrategy, ['w11', 'b1', 'First workitem'])
        self.source.execute(CreateWorkitemStrategy, ['w12', 'b1', 'Second workitem'])
        self.source.auto_seal()
        self.assertIn('w11', backlog)
        self.source.execute(DeleteWorkitemStrategy, ['w11'])
        self.source.auto_seal()
        self.assertNotIn('w11', backlog)
        self.assertIn('w12', backlog)

    def test_complete_workitem_basic(self):
        user, backlog = self._standard_backlog()
        self.source.execute(CreateWorkitemStrategy, ['w11', 'b1', 'First workitem'])
        self.source.auto_seal()
        workitem = backlog['w11']
        self.source.execute(CompleteWorkitemStrategy, ['w11', 'finished'])
        self.source.auto_seal()
        self.assertIn('w11', backlog)
        self.assertFalse(workitem.is_startable())
        self.assertTrue(workitem.is_sealed())
        self.assertFalse(workitem.is_running())
        self.assertFalse(workitem.has_running_pomodoro())

    def test_complete_workitem_with_two_pomodoros(self):
        user, backlog = self._standard_backlog()
        self.source.execute(CreateWorkitemStrategy, ['w11', 'b1', 'First workitem'])
        self.source.execute(AddPomodoroStrategy, ['w11', '2'])
        self.source.execute(CompleteWorkitemStrategy, ['w11', 'finished'])
        self.source.auto_seal()
        self.assertFalse(backlog['w11'].is_startable())

    def test_complete_workitem_invalid_state(self):
        self._standard_backlog()
        self.source.execute(CreateWorkitemStrategy, ['w11', 'b1', 'First workitem'])
        self.source.auto_seal()
        self.assertRaises(Exception,
                          lambda: self.source.execute(CompleteWorkitemStrategy, ['w11', 'invalid']))

    def test_complete_workitem_twice(self):
        self._standard_backlog()
        self.source.execute(CreateWorkitemStrategy, ['w11', 'b1', 'First workitem'])
        self.source.execute(CompleteWorkitemStrategy, ['w11', 'finished'])
        self.source.auto_seal()
        self.assertRaises(Exception,
                          lambda: self.source.execute(CompleteWorkitemStrategy, ['w11', 'finished']))

    def test_rename_completed_workitem(self):
        self._standard_backlog()
        self.source.execute(CreateWorkitemStrategy, ['w11', 'b1', 'Before'])
        self.source.execute(CompleteWorkitemStrategy, ['w11', 'finished'])
        self.source.auto_seal()
        self.assertRaises(Exception,
                          lambda: self.source.execute(RenameWorkitemStrategy, ['w11', 'After']))

    def test_add_pomodoro_to_completed_workitem(self):
        self._standard_backlog()
        self.source.execute(CreateWorkitemStrategy, ['w11', 'b1', 'Before'])
        self.source.execute(CompleteWorkitemStrategy, ['w11', 'finished'])
        self.source.auto_seal()
        self.assertRaises(Exception,
                          lambda: self.source.execute(AddPomodoroStrategy, ['w11', '1']))

    def test_delete_completed_workitem(self):
        _, backlog = self._standard_backlog()
        self.source.execute(CreateWorkitemStrategy, ['w11', 'b1', 'Before'])
        self.source.execute(CompleteWorkitemStrategy, ['w11', 'finished'])
        self.source.execute(DeleteWorkitemStrategy, ['w11'])
        self.source.auto_seal()
        self.assertNotIn('w11', backlog)

    def test_start_completed_workitem(self):
        self._standard_backlog()
        self.source.execute(CreateWorkitemStrategy, ['w11', 'b1', 'Before'])
        self.source.execute(AddPomodoroStrategy, ['w11', '1'])
        self.source.execute(CompleteWorkitemStrategy, ['w11', 'finished'])
        self.source.auto_seal()
        self.assertRaises(Exception,
                          lambda: self.source.execute(StartWorkStrategy, ['w11', '1', '1']))

    # Next -- Test all workitem-specific stuff (check coverage)
    # - Lifecycle, including automatic voiding and completion of pomodoros (check all situations)
    # - State -- isStartable based on pomodoros
    # - Isolation between backlogs
    # - That we can find them via the Source
    # - Check update timestamps
    # - Add (2), (3) and (4) to backlogs, too

    def test_events_create_workitem(self):
        fired = list()

        def on_event(event, **kwargs):
            fired.append(event)
            if event == 'BeforeWorkitemCreate':
                self.assertIn('workitem_uid', kwargs)
                self.assertIn('backlog_uid', kwargs)
                self.assertIn('workitem_name', kwargs)
                self.assertEqual(kwargs['workitem_uid'], 'w11')
                self.assertEqual(kwargs['backlog_uid'], 'b1')
                self.assertEqual(kwargs['workitem_name'], 'First workitem')
            elif event == 'AfterWorkitemCreate':
                self.assertIn('workitem', kwargs)
                self.assertTrue(type(kwargs['workitem']) is Workitem)

        self._standard_backlog()
        self.source.on('*', on_event)
        self.source.execute(CreateWorkitemStrategy, ['w11', 'b1', 'First workitem'])
        self.source.auto_seal()
        self.assertEqual(len(fired), 4)
        self.assertEqual(fired[0], 'BeforeMessageProcessed')
        self.assertEqual(fired[1], 'BeforeWorkitemCreate')
        self.assertEqual(fired[2], 'AfterWorkitemCreate')
        self.assertEqual(fired[3], 'AfterMessageProcessed')

    def test_events_delete_workitem(self):
        fired = list()

        def on_event(event, **kwargs):
            fired.append(event)
            if event == 'BeforeWorkitemDelete' or event == 'AfterWorkitemDelete':
                self.assertIn('workitem', kwargs)
                self.assertTrue(type(kwargs['workitem']) is Workitem)
                self.assertEqual(kwargs['workitem'].get_name(), 'First item')
            elif event == 'BeforePomodoroComplete' or event == 'AfterPomodoroComplete':
                self.assertIn('workitem', kwargs)
                self.assertTrue(type(kwargs['workitem']) is Workitem)
                self.assertEqual(kwargs['workitem'].get_name(), 'First item')

        self._standard_backlog()
        self.source.execute(CreateWorkitemStrategy, ['w11', 'b1', 'First item'])
        self.source.execute(AddPomodoroStrategy, ['w11', '2'])
        self.source.execute(CreateWorkitemStrategy, ['w12', 'b1', 'Second item'])
        self.source.execute(AddPomodoroStrategy, ['w11', '2'])
        self.source.execute(StartWorkStrategy, ['w11', '1', '1'])
        self.source.auto_seal()
        self.source.on('*', on_event)  # We only care about delete here
        self.source.execute(DeleteWorkitemStrategy, ['w11'])
        self.source.auto_seal()
        self.assertEqual(len(fired), 10)
        self.assertEqual(fired[0], 'BeforeMessageProcessed')
        self.assertEqual(fired[1], 'BeforeWorkitemDelete')
        self.assertEqual(fired[2], 'BeforeMessageProcessed')  # auto
        self.assertEqual(fired[3], 'BeforePomodoroInterrupted')
        self.assertEqual(fired[4], 'AfterPomodoroInterrupted')
        self.assertEqual(fired[5], 'BeforePomodoroVoided')
        self.assertEqual(fired[6], 'AfterPomodoroVoided')
        self.assertEqual(fired[7], 'AfterMessageProcessed')  # auto
        self.assertEqual(fired[8], 'AfterWorkitemDelete')
        self.assertEqual(fired[9], 'AfterMessageProcessed')

    def test_events_complete_workitem(self):
        fired = list()

        def on_event(event, **kwargs):
            fired.append(event)
            if event == 'BeforeWorkitemComplete' or event == 'AfterWorkitemComplete':
                self.assertIn('workitem', kwargs)
                self.assertTrue(type(kwargs['workitem']) is Workitem)
                self.assertEqual(kwargs['workitem'].get_name(), 'First item')
            elif event == 'BeforePomodoroComplete' or event == 'AfterPomodoroComplete':
                self.assertIn('workitem', kwargs)
                self.assertTrue(type(kwargs['workitem']) is Workitem)
                self.assertEqual(kwargs['workitem'].get_name(), 'First item')

        self._standard_backlog()
        self.source.execute(CreateWorkitemStrategy, ['w11', 'b1', 'First item'])
        self.source.execute(CreateWorkitemStrategy, ['w12', 'b1', 'Second item'])
        self.source.execute(AddPomodoroStrategy, ['w11', '2'])
        self.source.execute(StartWorkStrategy, ['w11', '1', '1'])
        self.source.auto_seal()
        self.source.on('*', on_event)  # We only care about delete here
        self.source.execute(CompleteWorkitemStrategy, ['w11', 'finished'])
        self.source.auto_seal()
        self.assertEqual(len(fired), 10)
        self.assertEqual(fired[0], 'BeforeMessageProcessed')
        self.assertEqual(fired[1], 'BeforeWorkitemComplete')
        self.assertEqual(fired[2], 'BeforeMessageProcessed')  # auto
        self.assertEqual(fired[3], 'BeforePomodoroInterrupted')
        self.assertEqual(fired[4], 'AfterPomodoroInterrupted')
        self.assertEqual(fired[5], 'BeforePomodoroVoided')
        self.assertEqual(fired[6], 'AfterPomodoroVoided')
        self.assertEqual(fired[7], 'AfterMessageProcessed')  # auto
        self.assertEqual(fired[8], 'AfterWorkitemComplete')
        self.assertEqual(fired[9], 'AfterMessageProcessed')

    def test_events_rename_workitem(self):
        fired = list()

        def on_event(event, **kwargs):
            fired.append(event)
            if event == 'BeforeWorkitemRename' or event == 'AfterWorkitemRename':
                self.assertIn('workitem', kwargs)
                self.assertIn('old_name', kwargs)
                self.assertIn('new_name', kwargs)
                self.assertEqual(kwargs['old_name'], 'Before')
                self.assertEqual(kwargs['new_name'], 'After')
                self.assertTrue(type(kwargs['workitem']) is Workitem)

        self._standard_backlog()
        self.source.execute(CreateWorkitemStrategy, ['w11', 'b1', 'Before'])
        self.source.auto_seal()
        self.source.on('*', on_event)
        self.source.execute(RenameWorkitemStrategy, ['w11', 'After'])
        self.source.auto_seal()
        self.assertEqual(len(fired), 4)
        self.assertEqual(fired[0], 'BeforeMessageProcessed')
        self.assertEqual(fired[1], 'BeforeWorkitemRename')
        self.assertEqual(fired[2], 'AfterWorkitemRename')
        self.assertEqual(fired[3], 'AfterMessageProcessed')

    # Reordering tests:
    # - Positive test -- move up and down
    # - Negative index and index > len()
    # - No move -- up and down
    # - Events

    def _create_workitems_for_reorder_tests(self):
        _, backlog = self._standard_backlog()
        self.source.execute(CreateWorkitemStrategy, ['w11', 'b1', 'First workitem'])
        self.source.execute(CreateWorkitemStrategy, ['w12', 'b1', 'Second workitem'])
        self.source.execute(CreateWorkitemStrategy, ['w13', 'b1', 'Third workitem'])
        self.source.execute(CreateWorkitemStrategy, ['w14', 'b1', 'Fourth workitem'])
        self.source.auto_seal()
        return backlog

    def _assert_workitem_order(self, backlog: Backlog, order: str):
        pass

    def test_reorder_workitem_up_normal(self):
        backlog = self._create_workitems_for_reorder_tests()
        self._assert_workitem_order(backlog, 'w11,w12,w13,w14')


1	# Flowkeeper - Pomodoro timer for power users and teams
2	# Copyright (c) 2023 Constantine Kulak
3	#
4	# This program is free software: you can redistribute it and/or modify
5	# it under the terms of the GNU General Public License as published by
6	# the Free Software Foundation; either version 3 of the License, or
7	# (at your option) any later version.
8	#
9	# This program is distributed in the hope that it will be useful,
10	# but WITHOUT ANY WARRANTY; without even the implied warranty of
11	# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
12	# GNU General Public License for more details.
13	#
14	# You should have received a copy of the GNU General Public License
15	# along with this program. If not, see <https://www.gnu.org/licenses/>.
16	import datetime	1✔
17	from unittest import TestCase	1✔
18
19	from fk.core.abstract_cryptograph import AbstractCryptograph	1✔
20	from fk.core.abstract_settings import AbstractSettings	1✔
21	from fk.core.backlog import Backlog	1✔
22	from fk.core.backlog_strategies import CreateBacklogStrategy	1✔
23	from fk.core.ephemeral_event_source import EphemeralEventSource	1✔
24	from fk.core.fernet_cryptograph import FernetCryptograph	1✔
25	from fk.core.mock_settings import MockSettings	1✔
26	from fk.core.pomodoro_strategies import AddPomodoroStrategy, StartWorkStrategy	1✔
27	from fk.core.tenant import Tenant	1✔
28	from fk.core.user import User	1✔
29	from fk.core.workitem import Workitem	1✔
30	from fk.core.workitem_strategies import CreateWorkitemStrategy, RenameWorkitemStrategy, DeleteWorkitemStrategy, \	1✔
31	CompleteWorkitemStrategy
32
33
34	class TestWorkitems(TestCase):	1✔
35	settings: AbstractSettings	1✔
36	cryptograph: AbstractCryptograph	1✔
37	source: EphemeralEventSource	1✔
38	data: dict[str, User]	1✔
39
40	def setUp(self) -> None:	1✔
41	self.settings = MockSettings()	1✔
42	self.cryptograph = FernetCryptograph(self.settings)	1✔
43	self.source = EphemeralEventSource[Tenant](self.settings, self.cryptograph, Tenant(self.settings))	1✔
44	self.source.start()	1✔
45	self.data = self.source.get_data()	1✔
46
47	def tearDown(self) -> None:	1✔
48	self.source.dump()	1✔
49
50	def _assert_workitem(self, workitem1: Workitem, user: User, backlog: Backlog):	1✔
51	self.assertEqual(workitem1.get_name(), 'First workitem')	1✔
52	self.assertEqual(workitem1.get_uid(), 'w11')	1✔
53	self.assertEqual(workitem1.get_parent(), backlog)	1✔
54	self.assertEqual(workitem1.get_owner(), user)	1✔
55	self.assertFalse(workitem1.is_running())	1✔
56	self.assertFalse(workitem1.is_sealed())	1✔
57	self.assertFalse(workitem1.is_startable())	1✔
58	self.assertFalse(workitem1.has_running_pomodoro())	1✔
59	self.assertTrue(workitem1.is_planned())	1✔
60	self.assertEqual(len(workitem1.values()), 0)	1✔
61
62	def _standard_backlog(self) -> (User, Backlog):	1✔
63	self.source.execute(CreateBacklogStrategy, ['b1', 'First backlog'])	1✔
64	self.source.auto_seal()	1✔
65	user = self.data['user@local.host']	1✔
66	backlog = user['b1']	1✔
67	return user, backlog	1✔
68
69	def test_create_workitems(self):	1✔
70	user, backlog = self._standard_backlog()	1✔
71	self.source.execute(CreateWorkitemStrategy, ['w11', 'b1', 'First workitem'])	1✔
72	self.source.execute(CreateWorkitemStrategy, ['w12', 'b1', 'Second workitem'])	1✔
73	self.source.auto_seal()	1✔
74	self.assertIn('w11', backlog)	1✔
75	self.assertIn('w12', backlog)	1✔
76	workitem1: Workitem = backlog['w11']	1✔
77	self._assert_workitem(workitem1, user, backlog)	1✔
78	workitem2 = backlog['w12']	1✔
79	self.assertEqual(workitem2.get_name(), 'Second workitem')	1✔
80
81	def test_create_workitems_with_tags(self):	1✔
82	user, backlog = self._standard_backlog()	1✔
83	self.source.execute(CreateWorkitemStrategy, ['w11', 'b1', 'First workitem'])	1✔
84	self.source.execute(CreateWorkitemStrategy, ['w12', 'b1', '#Second workitem'])	1✔
85	self.source.execute(CreateWorkitemStrategy, ['w13', 'b1', '#Third #workitem'])	1✔
86	self.source.execute(CreateWorkitemStrategy, ['w14', 'b1', 'Fourth #workitem and some more #workitem text'])	1✔
87	self.source.execute(CreateWorkitemStrategy, ['w15', 'b1', 'Fifth #workitem.'])	1✔
88	self.source.execute(CreateWorkitemStrategy, ['w16', 'b1', 'Six #workitem and #workitems'])	1✔
89	self.source.auto_seal()	1✔
90	self.assertIn('w11', backlog)	1✔
91	self.assertIn('w12', backlog)	1✔
92	self.assertIn('w13', backlog)	1✔
93	self.assertIn('w14', backlog)	1✔
94	self.assertIn('w15', backlog)	1✔
95	self.assertIn('w16', backlog)	1✔
96	workitem1: Workitem = backlog['w11']	1✔
97	self._assert_workitem(workitem1, user, backlog)	1✔
98	workitem2 = backlog['w12']	1✔
99	self.assertEqual(workitem2.get_name(), '#Second workitem')	1✔
100	workitem3 = backlog['w13']	1✔
101	self.assertEqual(workitem3.get_name(), '#Third #workitem')	1✔
102	workitem4 = backlog['w14']	1✔
103	self.assertEqual(workitem4.get_name(), 'Fourth #workitem and some more #workitem text')	1✔
104	workitem5 = backlog['w15']	1✔
105	self.assertEqual(workitem5.get_name(), 'Fifth #workitem.')	1✔
106	workitem6 = backlog['w16']	1✔
107	self.assertEqual(workitem6.get_name(), 'Six #workitem and #workitems')	1✔
108	tags = user.get_tags()	1✔
109	self.assertEqual(len(tags), 4)	1✔
110	self.assertIn('workitem', tags)	1✔
111	self.assertIn('second', tags)	1✔
112	self.assertIn('third', tags)	1✔
113	self.assertIn('workitems', tags)	1✔
114	workitems = tags['workitem'].get_workitems()	1✔
115	self.assertEqual(len(workitems), 4)	1✔
116	self.assertIn(workitem3, workitems)	1✔
117	self.assertIn(workitem4, workitems)	1✔
118	self.assertIn(workitem5, workitems)	1✔
119	self.assertIn(workitem6, workitems)	1✔
120
121	def test_execute_prepared(self):	1✔
122	user, backlog = self._standard_backlog()	1✔
123	s = CreateWorkitemStrategy(2,	1✔
124	datetime.datetime.now(datetime.timezone.utc),
125	user.get_identity(),
126	['w11', 'b1', 'First workitem'],
127	self.settings)
128	self.source.execute_prepared_strategy(s)	1✔
129	self.source.auto_seal()	1✔
130	self.assertIn('w11', backlog)	1✔
131	workitem1: Workitem = backlog['w11']	1✔
132	self._assert_workitem(workitem1, user, backlog)	1✔
133
134	def test_create_duplicate_workitem_failure(self):	1✔
135	self._standard_backlog()	1✔
136	self.source.execute(CreateWorkitemStrategy, ['w11', 'b1', 'First workitem 1'])	1✔
137	self.source.auto_seal()	1✔
138	self.assertRaises(Exception,	1✔
139	lambda: self.source.execute(CreateWorkitemStrategy, ['w11', 'b1', 'First workitem 2']))
140
141	def test_rename_nonexistent_workitem_failure(self):	1✔
142	self._standard_backlog()	1✔
143	self.source.execute(CreateWorkitemStrategy, ['w11', 'b1', 'First workitem'])	1✔
144	self.source.execute(CreateWorkitemStrategy, ['w12', 'b1', 'Second workitem'])	1✔
145	self.source.auto_seal()	1✔
146	self.assertRaises(Exception,	1✔
147	lambda: self.source.execute(RenameWorkitemStrategy, ['w13', 'Renamed workitem']))
148
149	def test_rename_workitem(self):	1✔
150	user, backlog = self._standard_backlog()	1✔
151	self.source.execute(CreateWorkitemStrategy, ['w11', 'b1', 'First workitem'])	1✔
152	self.source.execute(RenameWorkitemStrategy, ['w11', 'Renamed workitem'])	1✔
153	self.source.auto_seal()	1✔
154	self.assertEqual(backlog['w11'].get_name(), 'Renamed workitem')	1✔
155
156	def test_delete_nonexistent_workitem_failure(self):	1✔
157	self._standard_backlog()	1✔
158	self.source.execute(CreateWorkitemStrategy, ['w11', 'b1', 'First workitem'])	1✔
159	self.source.execute(CreateWorkitemStrategy, ['w12', 'b1', 'Second workitem'])	1✔
160	self.source.auto_seal()	1✔
161	self.assertRaises(Exception,	1✔
162	lambda: self.source.execute(DeleteWorkitemStrategy, ['w13']))
163
164	def test_delete_workitem(self):	1✔
165	user, backlog = self._standard_backlog()	1✔
166	self.source.execute(CreateWorkitemStrategy, ['w11', 'b1', 'First workitem'])	1✔
167	self.source.execute(CreateWorkitemStrategy, ['w12', 'b1', 'Second workitem'])	1✔
168	self.source.auto_seal()	1✔
169	self.assertIn('w11', backlog)	1✔
170	self.source.execute(DeleteWorkitemStrategy, ['w11'])	1✔
171	self.source.auto_seal()	1✔
172	self.assertNotIn('w11', backlog)	1✔
173	self.assertIn('w12', backlog)	1✔
174
175	def test_complete_workitem_basic(self):	1✔
176	user, backlog = self._standard_backlog()	1✔
177	self.source.execute(CreateWorkitemStrategy, ['w11', 'b1', 'First workitem'])	1✔
178	self.source.auto_seal()	1✔
179	workitem = backlog['w11']	1✔
180	self.source.execute(CompleteWorkitemStrategy, ['w11', 'finished'])	1✔
181	self.source.auto_seal()	1✔
182	self.assertIn('w11', backlog)	1✔
183	self.assertFalse(workitem.is_startable())	1✔
184	self.assertTrue(workitem.is_sealed())	1✔
185	self.assertFalse(workitem.is_running())	1✔
186	self.assertFalse(workitem.has_running_pomodoro())	1✔
187
188	def test_complete_workitem_with_two_pomodoros(self):	1✔
189	user, backlog = self._standard_backlog()	1✔
190	self.source.execute(CreateWorkitemStrategy, ['w11', 'b1', 'First workitem'])	1✔
191	self.source.execute(AddPomodoroStrategy, ['w11', '2'])	1✔
192	self.source.execute(CompleteWorkitemStrategy, ['w11', 'finished'])	1✔
193	self.source.auto_seal()	1✔
194	self.assertFalse(backlog['w11'].is_startable())	1✔
195
196	def test_complete_workitem_invalid_state(self):	1✔
197	self._standard_backlog()	1✔
198	self.source.execute(CreateWorkitemStrategy, ['w11', 'b1', 'First workitem'])	1✔
199	self.source.auto_seal()	1✔
200	self.assertRaises(Exception,	1✔
201	lambda: self.source.execute(CompleteWorkitemStrategy, ['w11', 'invalid']))
202
203	def test_complete_workitem_twice(self):	1✔
204	self._standard_backlog()	1✔
205	self.source.execute(CreateWorkitemStrategy, ['w11', 'b1', 'First workitem'])	1✔
206	self.source.execute(CompleteWorkitemStrategy, ['w11', 'finished'])	1✔
207	self.source.auto_seal()	1✔
208	self.assertRaises(Exception,	1✔
209	lambda: self.source.execute(CompleteWorkitemStrategy, ['w11', 'finished']))
210
211	def test_rename_completed_workitem(self):	1✔
212	self._standard_backlog()	1✔
213	self.source.execute(CreateWorkitemStrategy, ['w11', 'b1', 'Before'])	1✔
214	self.source.execute(CompleteWorkitemStrategy, ['w11', 'finished'])	1✔
215	self.source.auto_seal()	1✔
216	self.assertRaises(Exception,	1✔
217	lambda: self.source.execute(RenameWorkitemStrategy, ['w11', 'After']))
218
219	def test_add_pomodoro_to_completed_workitem(self):	1✔
220	self._standard_backlog()	1✔
221	self.source.execute(CreateWorkitemStrategy, ['w11', 'b1', 'Before'])	1✔
222	self.source.execute(CompleteWorkitemStrategy, ['w11', 'finished'])	1✔
223	self.source.auto_seal()	1✔
224	self.assertRaises(Exception,	1✔
225	lambda: self.source.execute(AddPomodoroStrategy, ['w11', '1']))
226
227	def test_delete_completed_workitem(self):	1✔
228	_, backlog = self._standard_backlog()	1✔
229	self.source.execute(CreateWorkitemStrategy, ['w11', 'b1', 'Before'])	1✔
230	self.source.execute(CompleteWorkitemStrategy, ['w11', 'finished'])	1✔
231	self.source.execute(DeleteWorkitemStrategy, ['w11'])	1✔
232	self.source.auto_seal()	1✔
233	self.assertNotIn('w11', backlog)	1✔
234
235	def test_start_completed_workitem(self):	1✔
236	self._standard_backlog()	1✔
237	self.source.execute(CreateWorkitemStrategy, ['w11', 'b1', 'Before'])	1✔
238	self.source.execute(AddPomodoroStrategy, ['w11', '1'])	1✔
239	self.source.execute(CompleteWorkitemStrategy, ['w11', 'finished'])	1✔
240	self.source.auto_seal()	1✔
241	self.assertRaises(Exception,	1✔
242	lambda: self.source.execute(StartWorkStrategy, ['w11', '1', '1']))
243
244	# Next -- Test all workitem-specific stuff (check coverage)
245	# - Lifecycle, including automatic voiding and completion of pomodoros (check all situations)
246	# - State -- isStartable based on pomodoros
247	# - Isolation between backlogs
248	# - That we can find them via the Source
249	# - Check update timestamps
250	# - Add (2), (3) and (4) to backlogs, too
251
252	def test_events_create_workitem(self):	1✔
253	fired = list()	1✔
254
255	def on_event(event, **kwargs):	1✔
256	fired.append(event)	1✔
257	if event == 'BeforeWorkitemCreate':	1✔
258	self.assertIn('workitem_uid', kwargs)	1✔
259	self.assertIn('backlog_uid', kwargs)	1✔
260	self.assertIn('workitem_name', kwargs)	1✔
261	self.assertEqual(kwargs['workitem_uid'], 'w11')	1✔
262	self.assertEqual(kwargs['backlog_uid'], 'b1')	1✔
263	self.assertEqual(kwargs['workitem_name'], 'First workitem')	1✔
264	elif event == 'AfterWorkitemCreate':	1✔
265	self.assertIn('workitem', kwargs)	1✔
266	self.assertTrue(type(kwargs['workitem']) is Workitem)	1✔
267
268	self._standard_backlog()	1✔
269	self.source.on('*', on_event)	1✔
270	self.source.execute(CreateWorkitemStrategy, ['w11', 'b1', 'First workitem'])	1✔
271	self.source.auto_seal()	1✔
272	self.assertEqual(len(fired), 4)	1✔
273	self.assertEqual(fired[0], 'BeforeMessageProcessed')	1✔
274	self.assertEqual(fired[1], 'BeforeWorkitemCreate')	1✔
275	self.assertEqual(fired[2], 'AfterWorkitemCreate')	1✔
276	self.assertEqual(fired[3], 'AfterMessageProcessed')	1✔
277
278	def test_events_delete_workitem(self):	1✔
279	fired = list()	1✔
280
281	def on_event(event, **kwargs):	1✔
282	fired.append(event)	1✔
283	if event == 'BeforeWorkitemDelete' or event == 'AfterWorkitemDelete':	1✔
284	self.assertIn('workitem', kwargs)	1✔
285	self.assertTrue(type(kwargs['workitem']) is Workitem)	1✔
286	self.assertEqual(kwargs['workitem'].get_name(), 'First item')	1✔
287	elif event == 'BeforePomodoroComplete' or event == 'AfterPomodoroComplete':	1✔
UNCOV 288	self.assertIn('workitem', kwargs)	×
UNCOV 289	self.assertTrue(type(kwargs['workitem']) is Workitem)	×
UNCOV 290	self.assertEqual(kwargs['workitem'].get_name(), 'First item')	×
291
292	self._standard_backlog()	1✔
293	self.source.execute(CreateWorkitemStrategy, ['w11', 'b1', 'First item'])	1✔
294	self.source.execute(AddPomodoroStrategy, ['w11', '2'])	1✔
295	self.source.execute(CreateWorkitemStrategy, ['w12', 'b1', 'Second item'])	1✔
296	self.source.execute(AddPomodoroStrategy, ['w11', '2'])	1✔
297	self.source.execute(StartWorkStrategy, ['w11', '1', '1'])	1✔
298	self.source.auto_seal()	1✔
299	self.source.on('*', on_event) # We only care about delete here	1✔
300	self.source.execute(DeleteWorkitemStrategy, ['w11'])	1✔
301	self.source.auto_seal()	1✔
302	self.assertEqual(len(fired), 10)	1✔
303	self.assertEqual(fired[0], 'BeforeMessageProcessed')	1✔
304	self.assertEqual(fired[1], 'BeforeWorkitemDelete')	1✔
305	self.assertEqual(fired[2], 'BeforeMessageProcessed') # auto	1✔
306	self.assertEqual(fired[3], 'BeforePomodoroInterrupted')	1✔
307	self.assertEqual(fired[4], 'AfterPomodoroInterrupted')	1✔
308	self.assertEqual(fired[5], 'BeforePomodoroVoided')	1✔
309	self.assertEqual(fired[6], 'AfterPomodoroVoided')	1✔
310	self.assertEqual(fired[7], 'AfterMessageProcessed') # auto	1✔
311	self.assertEqual(fired[8], 'AfterWorkitemDelete')	1✔
312	self.assertEqual(fired[9], 'AfterMessageProcessed')	1✔
313
314	def test_events_complete_workitem(self):	1✔
315	fired = list()	1✔
316
317	def on_event(event, **kwargs):	1✔
318	fired.append(event)	1✔
319	if event == 'BeforeWorkitemComplete' or event == 'AfterWorkitemComplete':	1✔
320	self.assertIn('workitem', kwargs)	1✔
321	self.assertTrue(type(kwargs['workitem']) is Workitem)	1✔
322	self.assertEqual(kwargs['workitem'].get_name(), 'First item')	1✔
323	elif event == 'BeforePomodoroComplete' or event == 'AfterPomodoroComplete':	1✔
UNCOV 324	self.assertIn('workitem', kwargs)	×
UNCOV 325	self.assertTrue(type(kwargs['workitem']) is Workitem)	×
UNCOV 326	self.assertEqual(kwargs['workitem'].get_name(), 'First item')	×
327
328	self._standard_backlog()	1✔
329	self.source.execute(CreateWorkitemStrategy, ['w11', 'b1', 'First item'])	1✔
330	self.source.execute(CreateWorkitemStrategy, ['w12', 'b1', 'Second item'])	1✔
331	self.source.execute(AddPomodoroStrategy, ['w11', '2'])	1✔
332	self.source.execute(StartWorkStrategy, ['w11', '1', '1'])	1✔
333	self.source.auto_seal()	1✔
334	self.source.on('*', on_event) # We only care about delete here	1✔
335	self.source.execute(CompleteWorkitemStrategy, ['w11', 'finished'])	1✔
336	self.source.auto_seal()	1✔
337	self.assertEqual(len(fired), 10)	1✔
338	self.assertEqual(fired[0], 'BeforeMessageProcessed')	1✔
339	self.assertEqual(fired[1], 'BeforeWorkitemComplete')	1✔
340	self.assertEqual(fired[2], 'BeforeMessageProcessed') # auto	1✔
341	self.assertEqual(fired[3], 'BeforePomodoroInterrupted')	1✔
342	self.assertEqual(fired[4], 'AfterPomodoroInterrupted')	1✔
343	self.assertEqual(fired[5], 'BeforePomodoroVoided')	1✔
344	self.assertEqual(fired[6], 'AfterPomodoroVoided')	1✔
345	self.assertEqual(fired[7], 'AfterMessageProcessed') # auto	1✔
346	self.assertEqual(fired[8], 'AfterWorkitemComplete')	1✔
347	self.assertEqual(fired[9], 'AfterMessageProcessed')	1✔
348
349	def test_events_rename_workitem(self):	1✔
350	fired = list()	1✔
351
352	def on_event(event, **kwargs):	1✔
353	fired.append(event)	1✔
354	if event == 'BeforeWorkitemRename' or event == 'AfterWorkitemRename':	1✔
355	self.assertIn('workitem', kwargs)	1✔
356	self.assertIn('old_name', kwargs)	1✔
357	self.assertIn('new_name', kwargs)	1✔
358	self.assertEqual(kwargs['old_name'], 'Before')	1✔
359	self.assertEqual(kwargs['new_name'], 'After')	1✔
360	self.assertTrue(type(kwargs['workitem']) is Workitem)	1✔
361
362	self._standard_backlog()	1✔
363	self.source.execute(CreateWorkitemStrategy, ['w11', 'b1', 'Before'])	1✔
364	self.source.auto_seal()	1✔
365	self.source.on('*', on_event)	1✔
366	self.source.execute(RenameWorkitemStrategy, ['w11', 'After'])	1✔
367	self.source.auto_seal()	1✔
368	self.assertEqual(len(fired), 4)	1✔
369	self.assertEqual(fired[0], 'BeforeMessageProcessed')	1✔
370	self.assertEqual(fired[1], 'BeforeWorkitemRename')	1✔
371	self.assertEqual(fired[2], 'AfterWorkitemRename')	1✔
372	self.assertEqual(fired[3], 'AfterMessageProcessed')	1✔
373
374	# Reordering tests:
375	# - Positive test -- move up and down
376	# - Negative index and index > len()
377	# - No move -- up and down
378	# - Events
379
380	def _create_workitems_for_reorder_tests(self):	1✔
381	_, backlog = self._standard_backlog()	1✔
382	self.source.execute(CreateWorkitemStrategy, ['w11', 'b1', 'First workitem'])	1✔
383	self.source.execute(CreateWorkitemStrategy, ['w12', 'b1', 'Second workitem'])	1✔
384	self.source.execute(CreateWorkitemStrategy, ['w13', 'b1', 'Third workitem'])	1✔
385	self.source.execute(CreateWorkitemStrategy, ['w14', 'b1', 'Fourth workitem'])	1✔
386	self.source.auto_seal()	1✔
387	return backlog	1✔
388
389	def _assert_workitem_order(self, backlog: Backlog, order: str):	1✔
390	pass	1✔
391
392	def test_reorder_workitem_up_normal(self):	1✔
393	backlog = self._create_workitems_for_reorder_tests()	1✔
394	self._assert_workitem_order(backlog, 'w11,w12,w13,w14')	1✔
395

flowkeeper-org / fk-desktop / 13108386717

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous