extern/apprise
1
0
Fork 0
mirror of https://github.com/caronc/apprise.git synced 2025-01-07 14:39:51 +01:00
Code Issues Packages Projects Releases Wiki Activity

Use aware UTC datetimes internally (#887)

Browse Source 
In lieu of Python v3.12 deprecation warnings
This commit is contained in:
Ville Skyttä 2023-06-14 23:39:34 +03:00 committed by GitHub
parent 3346977297
commit 3d36108446
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
12 changed files with 84 additions and 54 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
apprise/plugins/NotifyEmail.py
View File

@ -43,6 +43,7 @@ from email import charset
from socket import error as SocketError
from datetime import datetime
from datetime import timezone
from .NotifyBase import NotifyBase
from ..URLBase import PrivacyMode
@ -805,7 +806,8 @@ class NotifyEmail(NotifyBase):
base['To'] = formataddr((to_name, to_addr), charset='utf-8')
base['Message-ID'] = make_msgid(domain=self.smtp_host)
base['Date'] = \
datetime.utcnow().strftime("%a, %d %b %Y %H:%M:%S +0000")
datetime.now(timezone.utc)\
.strftime("%a, %d %b %Y %H:%M:%S +0000")
base['X-Application'] = self.app_id
if cc:

11
apprise/plugins/NotifyFCM/oauth.py
View File

@ -47,6 +47,7 @@ from cryptography.hazmat.primitives import asymmetric
from cryptography.exceptions import UnsupportedAlgorithm
from datetime import datetime
from datetime import timedelta
from datetime import timezone
from json.decoder import JSONDecodeError
from urllib.parse import urlencode as _urlencode
@ -106,7 +107,7 @@ class GoogleOAuth:
# Our keys we build using the provided content
self.__refresh_token = None
self.__access_token = None
self.__access_token_expiry = datetime.utcnow()
self.__access_token_expiry = datetime.now(timezone.utc)
def load(self, path):
"""
@ -117,7 +118,7 @@ class GoogleOAuth:
self.content = None
self.private_key = None
self.__access_token = None
self.__access_token_expiry = datetime.utcnow()
self.__access_token_expiry = datetime.now(timezone.utc)
try:
with open(path, mode="r", encoding=self.encoding) as fp:
@ -199,7 +200,7 @@ class GoogleOAuth:
'token with.')
return None
if self.__access_token_expiry > datetime.utcnow():
if self.__access_token_expiry > datetime.now(timezone.utc):
# Return our no-expired key
return self.__access_token
@ -209,7 +210,7 @@ class GoogleOAuth:
key_identifier = self.content.get('private_key_id')
# Generate our Assertion
now = datetime.utcnow()
now = datetime.now(timezone.utc)
expiry = now + self.access_token_lifetime_sec
payload = {
@ -301,7 +302,7 @@ class GoogleOAuth:
if 'expires_in' in response:
delta = timedelta(seconds=int(response['expires_in']))
self.__access_token_expiry = \
delta + datetime.utcnow() - self.clock_skew
delta + datetime.now(timezone.utc) - self.clock_skew
else:
# Allow some grace before we expire

10
apprise/plugins/NotifyGitter.py
View File

@ -50,6 +50,7 @@ import requests
from json import loads
from json import dumps
from datetime import datetime
from datetime import timezone
from .NotifyBase import NotifyBase
from ..common import NotifyImageSize
@ -99,7 +100,7 @@ class NotifyGitter(NotifyBase):
request_rate_per_sec = 0
# For Tracking Purposes
ratelimit_reset = datetime.utcnow()
ratelimit_reset = datetime.now(timezone.utc).replace(tzinfo=None)
# Default to 1
ratelimit_remaining = 1
@ -298,7 +299,7 @@ class NotifyGitter(NotifyBase):
# Gitter server. One would hope we're on NTP and our clocks are
# the same allowing this to role smoothly:
now = datetime.utcnow()
now = datetime.now(timezone.utc).replace(tzinfo=None)
if now < self.ratelimit_reset:
# We need to throttle for the difference in seconds
# We add 0.5 seconds to the end just to allow a grace
@ -350,8 +351,9 @@ class NotifyGitter(NotifyBase):
try:
self.ratelimit_remaining = \
int(r.headers.get('X-RateLimit-Remaining'))
self.ratelimit_reset = datetime.utcfromtimestamp(
int(r.headers.get('X-RateLimit-Reset')))
self.ratelimit_reset = datetime.fromtimestamp(
int(r.headers.get('X-RateLimit-Reset')), timezone.utc
).replace(tzinfo=None)
except (TypeError, ValueError):
# This is returned if we could not retrieve this information

10
apprise/plugins/NotifyMastodon.py
View File

@ -35,6 +35,7 @@ import requests
from copy import deepcopy
from json import dumps, loads
from datetime import datetime
from datetime import timezone
from .NotifyBase import NotifyBase
from ..URLBase import PrivacyMode
@ -150,7 +151,7 @@ class NotifyMastodon(NotifyBase):
request_rate_per_sec = 0
# For Tracking Purposes
ratelimit_reset = datetime.utcnow()
ratelimit_reset = datetime.now(timezone.utc).replace(tzinfo=None)
# Default to 1000; users can send up to 1000 DM's and 2400 toot a day
# This value only get's adjusted if the server sets it that way
@ -834,7 +835,7 @@ class NotifyMastodon(NotifyBase):
# Mastodon server. One would hope we're on NTP and our clocks are
# the same allowing this to role smoothly:
now = datetime.utcnow()
now = datetime.now(timezone.utc).replace(tzinfo=None)
if now < self.ratelimit_reset:
# We need to throttle for the difference in seconds
# We add 0.5 seconds to the end just to allow a grace
@ -892,8 +893,9 @@ class NotifyMastodon(NotifyBase):
# Capture rate limiting if possible
self.ratelimit_remaining = \
int(r.headers.get('X-RateLimit-Remaining'))
self.ratelimit_reset = datetime.utcfromtimestamp(
int(r.headers.get('X-RateLimit-Limit')))
self.ratelimit_reset = datetime.fromtimestamp(
int(r.headers.get('X-RateLimit-Limit')), timezone.utc
).replace(tzinfo=None)
except (TypeError, ValueError):
# This is returned if we could not retrieve this information

16
apprise/plugins/NotifyReddit.py
View File

@ -56,6 +56,7 @@ import requests
from json import loads
from datetime import timedelta
from datetime import datetime
from datetime import timezone
from .NotifyBase import NotifyBase
from ..URLBase import PrivacyMode
@ -134,7 +135,7 @@ class NotifyReddit(NotifyBase):
request_rate_per_sec = 0
# For Tracking Purposes
ratelimit_reset = datetime.utcnow()
ratelimit_reset = datetime.now(timezone.utc).replace(tzinfo=None)
# Default to 1.0
ratelimit_remaining = 1.0
@ -275,7 +276,7 @@ class NotifyReddit(NotifyBase):
# Our keys we build using the provided content
self.__refresh_token = None
self.__access_token = None
self.__access_token_expiry = datetime.utcnow()
self.__access_token_expiry = datetime.now(timezone.utc)
self.kind = kind.strip().lower() \
if isinstance(kind, str) \
@ -417,10 +418,10 @@ class NotifyReddit(NotifyBase):
if 'expires_in' in response:
delta = timedelta(seconds=int(response['expires_in']))
self.__access_token_expiry = \
delta + datetime.utcnow() - self.clock_skew
delta + datetime.now(timezone.utc) - self.clock_skew
else:
self.__access_token_expiry = self.access_token_lifetime_sec + \
datetime.utcnow() - self.clock_skew
datetime.now(timezone.utc) - self.clock_skew
# The Refresh Token
self.__refresh_token = response.get(
@ -547,7 +548,7 @@ class NotifyReddit(NotifyBase):
# Gitter server. One would hope we're on NTP and our clocks are
# the same allowing this to role smoothly:
now = datetime.utcnow()
now = datetime.now(timezone.utc).replace(tzinfo=None)
if now < self.ratelimit_reset:
# We need to throttle for the difference in seconds
wait = abs(
@ -671,8 +672,9 @@ class NotifyReddit(NotifyBase):
self.ratelimit_remaining = \
float(r.headers.get(
'X-RateLimit-Remaining'))
self.ratelimit_reset = datetime.utcfromtimestamp(
int(r.headers.get('X-RateLimit-Reset')))
self.ratelimit_reset = datetime.fromtimestamp(
int(r.headers.get('X-RateLimit-Reset')), timezone.utc
).replace(tzinfo=None)
except (TypeError, ValueError):
# This is returned if we could not retrieve this information

6
apprise/plugins/NotifySES.py
View File

@ -89,6 +89,7 @@ import base64
import requests
from hashlib import sha256
from datetime import datetime
from datetime import timezone
from collections import OrderedDict
from xml.etree import ElementTree
from email.mime.text import MIMEText
@ -436,7 +437,8 @@ class NotifySES(NotifyBase):
base['Reply-To'] = formataddr(reply_to, charset='utf-8')
base['Cc'] = ','.join(cc)
base['Date'] = \
datetime.utcnow().strftime("%a, %d %b %Y %H:%M:%S +0000")
datetime.now(
timezone.utc).strftime("%a, %d %b %Y %H:%M:%S +0000")
base['X-Application'] = self.app_id
if attach:
@ -585,7 +587,7 @@ class NotifySES(NotifyBase):
}
# Get a reference time (used for header construction)
reference = datetime.utcnow()
reference = datetime.now(timezone.utc)
# Provide Content-Length
headers['Content-Length'] = str(len(payload))

3
apprise/plugins/NotifySNS.py
View File

@ -35,6 +35,7 @@ import hmac
import requests
from hashlib import sha256
from datetime import datetime
from datetime import timezone
from collections import OrderedDict
from xml.etree import ElementTree
from itertools import chain
@ -396,7 +397,7 @@ class NotifySNS(NotifyBase):
}
# Get a reference time (used for header construction)
reference = datetime.utcnow()
reference = datetime.now(timezone.utc)
# Provide Content-Length
headers['Content-Length'] = str(len(payload))

10
apprise/plugins/NotifyTwitter.py
View File

@ -36,6 +36,7 @@ import re
import requests
from copy import deepcopy
from datetime import datetime
from datetime import timezone
from requests_oauthlib import OAuth1
from json import dumps
from json import loads
@ -124,7 +125,7 @@ class NotifyTwitter(NotifyBase):
request_rate_per_sec = 0
# For Tracking Purposes
ratelimit_reset = datetime.utcnow()
ratelimit_reset = datetime.now(timezone.utc).replace(tzinfo=None)
# Default to 1000; users can send up to 1000 DM's and 2400 tweets a day
# This value only get's adjusted if the server sets it that way
@ -678,7 +679,7 @@ class NotifyTwitter(NotifyBase):
# Twitter server. One would hope we're on NTP and our clocks are
# the same allowing this to role smoothly:
now = datetime.utcnow()
now = datetime.now(timezone.utc).replace(tzinfo=None)
if now < self.ratelimit_reset:
# We need to throttle for the difference in seconds
# We add 0.5 seconds to the end just to allow a grace
@ -736,8 +737,9 @@ class NotifyTwitter(NotifyBase):
# Capture rate limiting if possible
self.ratelimit_remaining = \
int(r.headers.get('x-rate-limit-remaining'))
self.ratelimit_reset = datetime.utcfromtimestamp(
int(r.headers.get('x-rate-limit-reset')))
self.ratelimit_reset = datetime.fromtimestamp(
int(r.headers.get('x-rate-limit-reset')), timezone.utc
).replace(tzinfo=None)
except (TypeError, ValueError):
# This is returned if we could not retrieve this information

12
test/test_plugin_gitter.py
View File

@ -40,6 +40,7 @@ from helpers import AppriseURLTester
from json import dumps
from datetime import datetime
from datetime import timezone
# Disable logging for a cleaner testing output
import logging
@ -156,13 +157,14 @@ def test_plugin_gitter_general(mock_post, mock_get):
]
# Epoch time:
epoch = datetime.utcfromtimestamp(0)
epoch = datetime.fromtimestamp(0, timezone.utc)
request = mock.Mock()
request.content = dumps(response_obj)
request.status_code = requests.codes.ok
request.headers = {
'X-RateLimit-Reset': (datetime.utcnow() - epoch).total_seconds(),
'X-RateLimit-Reset': (
datetime.now(timezone.utc) - epoch).total_seconds(),
'X-RateLimit-Remaining': 1,
}
@ -211,21 +213,21 @@ def test_plugin_gitter_general(mock_post, mock_get):
# Return our object, but place it in the future forcing us to block
request.headers['X-RateLimit-Reset'] = \
(datetime.utcnow() - epoch).total_seconds() + 1
(datetime.now(timezone.utc) - epoch).total_seconds() + 1
request.headers['X-RateLimit-Remaining'] = 0
obj.ratelimit_remaining = 0
assert obj.send(body="test") is True
# Return our object, but place it in the future forcing us to block
request.headers['X-RateLimit-Reset'] = \
(datetime.utcnow() - epoch).total_seconds() - 1
(datetime.now(timezone.utc) - epoch).total_seconds() - 1
request.headers['X-RateLimit-Remaining'] = 0
obj.ratelimit_remaining = 0
assert obj.send(body="test") is True
# Return our limits to always work
request.headers['X-RateLimit-Reset'] = \
(datetime.utcnow() - epoch).total_seconds()
(datetime.now(timezone.utc) - epoch).total_seconds()
request.headers['X-RateLimit-Remaining'] = 1
obj.ratelimit_remaining = 1

12
test/test_plugin_mastodon.py
View File

@ -36,6 +36,7 @@ from unittest import mock
import requests
from json import dumps, loads
from datetime import datetime
from datetime import timezone
from apprise import Apprise
from apprise import NotifyType
from apprise import AppriseAttachment
@ -175,13 +176,14 @@ def test_plugin_mastodon_general(mock_post, mock_get):
}
# Epoch time:
epoch = datetime.utcfromtimestamp(0)
epoch = datetime.fromtimestamp(0, timezone.utc)
request = mock.Mock()
request.content = dumps(response_obj)
request.status_code = requests.codes.ok
request.headers = {
'X-RateLimit-Limit': (datetime.utcnow() - epoch).total_seconds(),
'X-RateLimit-Limit': (
datetime.now(timezone.utc) - epoch).total_seconds(),
'X-RateLimit-Remaining': 1,
}
@ -231,21 +233,21 @@ def test_plugin_mastodon_general(mock_post, mock_get):
# Return our object, but place it in the future forcing us to block
request.headers['X-RateLimit-Limit'] = \
(datetime.utcnow() - epoch).total_seconds() + 1
(datetime.now(timezone.utc) - epoch).total_seconds() + 1
request.headers['X-RateLimit-Remaining'] = 0
obj.ratelimit_remaining = 0
assert obj.send(body="test") is True
# Return our object, but place it in the future forcing us to block
request.headers['X-RateLimit-Limit'] = \
(datetime.utcnow() - epoch).total_seconds() - 1
(datetime.now(timezone.utc) - epoch).total_seconds() - 1
request.headers['X-RateLimit-Remaining'] = 0
obj.ratelimit_remaining = 0
assert obj.send(body="test") is True
# Return our limits to always work
request.headers['X-RateLimit-Limit'] = \
(datetime.utcnow() - epoch).total_seconds()
(datetime.now(timezone.utc) - epoch).total_seconds()
request.headers['X-RateLimit-Remaining'] = 1
obj.ratelimit_remaining = 1

27
test/test_plugin_reddit.py
View File

@ -39,6 +39,7 @@ from unittest import mock
from json import dumps
from datetime import datetime
from datetime import timedelta
from datetime import timezone
# Disable logging for a cleaner testing output
import logging
@ -250,7 +251,7 @@ def test_plugin_reddit_general(mock_post):
}
# Epoch time:
epoch = datetime.utcfromtimestamp(0)
epoch = datetime.fromtimestamp(0, timezone.utc)
good_response = mock.Mock()
good_response.content = dumps({
@ -267,7 +268,8 @@ def test_plugin_reddit_general(mock_post):
})
good_response.status_code = requests.codes.ok
good_response.headers = {
'X-RateLimit-Reset': (datetime.utcnow() - epoch).total_seconds(),
'X-RateLimit-Reset': (
datetime.now(timezone.utc) - epoch).total_seconds(),
'X-RateLimit-Remaining': 1,
}
@ -296,7 +298,8 @@ def test_plugin_reddit_general(mock_post):
# Force a case where there are no more remaining posts allowed
good_response.headers = {
'X-RateLimit-Reset': (datetime.utcnow() - epoch).total_seconds(),
'X-RateLimit-Reset': (
datetime.now(timezone.utc) - epoch).total_seconds(),
'X-RateLimit-Remaining': 0,
}
# behind the scenes, it should cause us to update our rate limit
@ -305,7 +308,8 @@ def test_plugin_reddit_general(mock_post):
# This should cause us to block
good_response.headers = {
'X-RateLimit-Reset': (datetime.utcnow() - epoch).total_seconds(),
'X-RateLimit-Reset': (
datetime.now(timezone.utc) - epoch).total_seconds(),
'X-RateLimit-Remaining': 10,
}
assert obj.send(body="test") is True
@ -319,7 +323,8 @@ def test_plugin_reddit_general(mock_post):
# Reset our variable back to 1
good_response.headers = {
'X-RateLimit-Reset': (datetime.utcnow() - epoch).total_seconds(),
'X-RateLimit-Reset': (
datetime.now(timezone.utc) - epoch).total_seconds(),
'X-RateLimit-Remaining': 1,
}
# Handle cases where our epoch time is wrong
@ -328,7 +333,8 @@ def test_plugin_reddit_general(mock_post):
# Return our object, but place it in the future forcing us to block
good_response.headers = {
'X-RateLimit-Reset': (datetime.utcnow() - epoch).total_seconds() + 1,
'X-RateLimit-Reset': (
datetime.now(timezone.utc) - epoch).total_seconds() + 1,
'X-RateLimit-Remaining': 0,
}
@ -337,7 +343,8 @@ def test_plugin_reddit_general(mock_post):
# Return our object, but place it in the future forcing us to block
good_response.headers = {
'X-RateLimit-Reset': (datetime.utcnow() - epoch).total_seconds() - 1,
'X-RateLimit-Reset': (
datetime.now(timezone.utc) - epoch).total_seconds() - 1,
'X-RateLimit-Remaining': 0,
}
assert obj.send(body="test") is True
@ -348,7 +355,8 @@ def test_plugin_reddit_general(mock_post):
# Invalid JSON
response = mock.Mock()
response.headers = {
'X-RateLimit-Reset': (datetime.utcnow() - epoch).total_seconds(),
'X-RateLimit-Reset': (
datetime.now(timezone.utc) - epoch).total_seconds(),
'X-RateLimit-Remaining': 1,
}
response.content = '{'
@ -393,7 +401,8 @@ def test_plugin_reddit_general(mock_post):
})
good_response.status_code = requests.codes.ok
good_response.headers = {
'X-RateLimit-Reset': (datetime.utcnow() - epoch).total_seconds(),
'X-RateLimit-Reset': (
datetime.now(timezone.utc) - epoch).total_seconds(),
'X-RateLimit-Remaining': 1,
}

17
test/test_plugin_twitter.py
View File

@ -34,6 +34,7 @@ import json
import logging
import os
from datetime import datetime
from datetime import timezone
from unittest.mock import Mock, patch
import pytest
@ -341,13 +342,14 @@ def test_plugin_twitter_general(mocker):
}]
# Epoch time:
epoch = datetime.utcfromtimestamp(0)
epoch = datetime.fromtimestamp(0, timezone.utc)
request = Mock()
request.content = json.dumps(response_obj)
request.status_code = requests.codes.ok
request.headers = {
'x-rate-limit-reset': (datetime.utcnow() - epoch).total_seconds(),
'x-rate-limit-reset': (
datetime.now(timezone.utc) - epoch).total_seconds(),
'x-rate-limit-remaining': 1,
}
@ -402,21 +404,21 @@ def test_plugin_twitter_general(mocker):
# Return our object, but place it in the future forcing us to block
request.headers['x-rate-limit-reset'] = \
(datetime.utcnow() - epoch).total_seconds() + 1
(datetime.now(timezone.utc) - epoch).total_seconds() + 1
request.headers['x-rate-limit-remaining'] = 0
obj.ratelimit_remaining = 0
assert obj.send(body="test") is True
# Return our object, but place it in the future forcing us to block
request.headers['x-rate-limit-reset'] = \
(datetime.utcnow() - epoch).total_seconds() - 1
(datetime.now(timezone.utc) - epoch).total_seconds() - 1
request.headers['x-rate-limit-remaining'] = 0
obj.ratelimit_remaining = 0
assert obj.send(body="test") is True
# Return our limits to always work
request.headers['x-rate-limit-reset'] = \
(datetime.utcnow() - epoch).total_seconds()
(datetime.now(timezone.utc) - epoch).total_seconds()
request.headers['x-rate-limit-remaining'] = 1
obj.ratelimit_remaining = 1
@ -595,10 +597,11 @@ def test_plugin_twitter_dm_attachments_basic(
mock_post = mocker.patch("requests.post")
# Epoch time:
epoch = datetime.utcfromtimestamp(0)
epoch = datetime.fromtimestamp(0, timezone.utc)
mock_get.return_value = good_message_response
mock_post.return_value.headers = {
'x-rate-limit-reset': (datetime.utcnow() - epoch).total_seconds(),
'x-rate-limit-reset': (
datetime.now(timezone.utc) - epoch).total_seconds(),
'x-rate-limit-remaining': 1,
}