Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 17 additions & 4 deletions azure/functions/timer.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,19 +5,29 @@
import typing

from azure.functions import _abc as azf_abc

from . import meta


class TimerRequest(azf_abc.TimerRequest):
Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@YunchuWang - Getting your input here - If we plan to add PyDocs here - what could we add here?

@pdthummar No need to worry in this PR (mostly). But we should explore adding more PyDoc comments as they eventually show up in azure.functions.timerrequest

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@vrdmr would suggest giving a overview what this class does, adding definition of each field in init, and link to https://docs.microsoft.com/en-us/azure/azure-functions/functions-bindings-timer?tabs=in-process&pivots=programming-language-java#attributes


def __init__(self, *, past_due: bool) -> None:
def __init__(self, *, past_due: bool = False, schedule_status: dict = {},
schedule: dict = {}) -> None:
self.__past_due = past_due
self.__schedule_status = schedule_status
self.__schedule = schedule

@property
def past_due(self):
def past_due(self) -> bool:
Comment thread
vrdmr marked this conversation as resolved.
return self.__past_due

@property
def schedule_status(self) -> dict:
return self.__schedule_status

@property
def schedule(self) -> dict:
return self.__schedule


class TimerRequestConverter(meta.InConverter,
binding='timerTrigger', trigger=True):
Expand All @@ -32,5 +42,8 @@ def decode(cls, data: meta.Datum, *, trigger_metadata) -> typing.Any:
raise NotImplementedError

info = json.loads(data.value)

return TimerRequest(
past_due=info.get('IsPastDue', False))
past_due=info.get('IsPastDue', False),
schedule_status=info.get('ScheduleStatus', {}),
schedule=info.get('Schedule', {}))
66 changes: 66 additions & 0 deletions tests/test_timer.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,66 @@
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License.
Comment thread
vrdmr marked this conversation as resolved.

import json
import unittest

import azure.functions.timer as timer
from azure.functions.meta import Datum


class TestTimer(unittest.TestCase):
def test_timer_decode(self):
# given
Comment thread
vrdmr marked this conversation as resolved.
data = '''{"Schedule":{"AdjustForDST":true},
"ScheduleStatus":{
"Last":"2022-03-28T15:40:00.0105419-05:00",
"Next":"2022-03-28T15:45:00-05:00",
"LastUpdated":"2022-03-28T15:40:00.0105419-05:00"},
"IsPastDue":false}'''
datum: Datum = Datum(value=data, type='json')
data_dict = json.loads(data)

# when
timer_request: timer.TimerRequest = \
timer.TimerRequestConverter.decode(datum, trigger_metadata={})

# then
self.assertEqual(timer_request.schedule, data_dict["Schedule"])
self.assertEqual(timer_request.schedule_status,
data_dict["ScheduleStatus"])
self.assertEqual(timer_request.past_due, data_dict["IsPastDue"])

def test_timer_initialize_without_args(self):
# given
past_due = False
schedule_status = {}
schedule = {}

# when
test_timer = timer.TimerRequest()

# then
self.assertEqual(past_due, test_timer.past_due)
self.assertEqual(schedule_status, test_timer.schedule_status)
self.assertEqual(schedule, test_timer.schedule)

def test_timer_no_implementation_exception(self):
# given
datum: Datum = Datum(value="test", type='string')
is_exception_raised = False

# when
try:
timer.TimerRequestConverter.decode(datum, trigger_metadata={})
except NotImplementedError:
is_exception_raised = True

# then
self.assertTrue(is_exception_raised)

def test_timer_input_type(self):
check_input_type = (
timer.TimerRequestConverter.check_input_type_annotation
)
self.assertTrue(check_input_type(timer.TimerRequest))
self.assertFalse(check_input_type(str))