Skip to content
Permalink
Branch: master
Find file Copy path
Find file Copy path
Fetching contributors…
Cannot retrieve contributors at this time
93 lines (71 sloc) 2.29 KB
import unittest
from enum import Enum
from typing import Callable, List, Optional
class TaskType(Enum):
PROVISION = "provision"
DEPROVISION = "deprovision"
class Task:
type: TaskType
customer_id: int
def __init__(self, type: TaskType, customer_id: int):
self.type = type
self.customer_id = customer_id
def create_process_one(
fetch_task: Callable[[], Optional[Task]],
complete_task: Callable[[Task], None],
fail_task: Callable[[Task], None],
provision_user: Callable[[int], None],
deprovision_user: Callable[[int], None]
):
def process_one():
task = fetch_task()
if task is None:
return
success: bool
if task.type == TaskType.PROVISION:
success = provision_user(task.customer_id)
else:
success = deprovision_user(task.customer_id)
if success:
complete_task(task)
else:
fail_task(task)
return process_one
class TestDataProcessor(unittest.TestCase):
def test_success(self):
# Setup
new_tasks: List[Task] = []
completed_tasks: List[Task] = []
failed_tasks: List[Task] = []
customer_ids: List[int] = []
def fetch_task() -> Optional[Task]:
if len(new_tasks) > 0:
item = new_tasks[0]
new_tasks.remove(item)
return item
return None
def complete_task(task: Task) -> None:
completed_tasks.append(task)
def fail_task(task: Task) -> None:
failed_tasks.append(task)
def provision_user(customer_id: int) -> None:
if customer_id not in customer_ids:
customer_ids.append(customer_id)
def deprovision_user(customer_id: int) -> None:
if customer_id in customer_ids:
customer_ids.remove(customer_id)
process_one = create_process_one(
fetch_task,
complete_task,
fail_task,
provision_user,
deprovision_user
)
task = Task(TaskType.PROVISION, 3)
new_tasks.append(task)
# Execute
process_one()
# Assert
self.assertEqual([3], customer_ids)
if __name__ == '__main__':
unittest.main()
You can’t perform that action at this time.