mirror of https://github.com/caronc/apprise
Chris Caron
6 months ago
committed by
GitHub
15 changed files with 2864 additions and 42 deletions
@ -0,0 +1,212 @@
|
||||
# -*- coding: utf-8 -*- |
||||
# BSD 2-Clause License |
||||
# |
||||
# Apprise - Push Notification Library. |
||||
# Copyright (c) 2024, Chris Caron <lead2gold@gmail.com> |
||||
# |
||||
# Redistribution and use in source and binary forms, with or without |
||||
# modification, are permitted provided that the following conditions are met: |
||||
# |
||||
# 1. Redistributions of source code must retain the above copyright notice, |
||||
# this list of conditions and the following disclaimer. |
||||
# |
||||
# 2. Redistributions in binary form must reproduce the above copyright notice, |
||||
# this list of conditions and the following disclaimer in the documentation |
||||
# and/or other materials provided with the distribution. |
||||
# |
||||
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" |
||||
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE |
||||
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE |
||||
# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE |
||||
# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR |
||||
# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF |
||||
# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS |
||||
# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN |
||||
# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) |
||||
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE |
||||
# POSSIBILITY OF SUCH DAMAGE. |
||||
|
||||
import re |
||||
import os |
||||
import io |
||||
from .base import AttachBase |
||||
from ..common import ContentLocation |
||||
from ..locale import gettext_lazy as _ |
||||
import uuid |
||||
|
||||
|
||||
class AttachMemory(AttachBase): |
||||
""" |
||||
A wrapper for Memory based attachment sources |
||||
""" |
||||
|
||||
# The default descriptive name associated with the service |
||||
service_name = _('Memory') |
||||
|
||||
# The default protocol |
||||
protocol = 'memory' |
||||
|
||||
# Content is local to the same location as the apprise instance |
||||
# being called (server-side) |
||||
location = ContentLocation.LOCAL |
||||
|
||||
def __init__(self, content=None, name=None, mimetype=None, |
||||
encoding='utf-8', **kwargs): |
||||
""" |
||||
Initialize Memory Based Attachment Object |
||||
|
||||
""" |
||||
# Create our BytesIO object |
||||
self._data = io.BytesIO() |
||||
|
||||
if content is None: |
||||
# Empty; do nothing |
||||
pass |
||||
|
||||
elif isinstance(content, str): |
||||
content = content.encode(encoding) |
||||
if mimetype is None: |
||||
mimetype = 'text/plain' |
||||
|
||||
if not name: |
||||
# Generate a unique filename |
||||
name = str(uuid.uuid4()) + '.txt' |
||||
|
||||
elif not isinstance(content, bytes): |
||||
raise TypeError( |
||||
'Provided content for memory attachment is invalid') |
||||
|
||||
# Store our content |
||||
if content: |
||||
self._data.write(content) |
||||
|
||||
if mimetype is None: |
||||
# Default mimetype |
||||
mimetype = 'application/octet-stream' |
||||
|
||||
if not name: |
||||
# Generate a unique filename |
||||
name = str(uuid.uuid4()) + '.dat' |
||||
|
||||
# Initialize our base object |
||||
super().__init__(name=name, mimetype=mimetype, **kwargs) |
||||
|
||||
return |
||||
|
||||
def url(self, privacy=False, *args, **kwargs): |
||||
""" |
||||
Returns the URL built dynamically based on specified arguments. |
||||
""" |
||||
|
||||
# Define any URL parameters |
||||
params = { |
||||
'mime': self._mimetype, |
||||
} |
||||
|
||||
return 'memory://{name}?{params}'.format( |
||||
name=self.quote(self._name), |
||||
params=self.urlencode(params, safe='/') |
||||
) |
||||
|
||||
def open(self, *args, **kwargs): |
||||
""" |
||||
return our memory object |
||||
""" |
||||
# Return our object |
||||
self._data.seek(0, 0) |
||||
return self._data |
||||
|
||||
def __enter__(self): |
||||
""" |
||||
support with clause |
||||
""" |
||||
# Return our object |
||||
self._data.seek(0, 0) |
||||
return self._data |
||||
|
||||
def download(self, **kwargs): |
||||
""" |
||||
Handle memory download() call |
||||
""" |
||||
|
||||
if self.location == ContentLocation.INACCESSIBLE: |
||||
# our content is inaccessible |
||||
return False |
||||
|
||||
if self.max_file_size > 0 and len(self) > self.max_file_size: |
||||
# The content to attach is to large |
||||
self.logger.error( |
||||
'Content exceeds allowable maximum memory size ' |
||||
'({}KB): {}'.format( |
||||
int(self.max_file_size / 1024), self.url(privacy=True))) |
||||
|
||||
# Return False (signifying a failure) |
||||
return False |
||||
|
||||
return True |
||||
|
||||
def invalidate(self): |
||||
""" |
||||
Removes data |
||||
""" |
||||
self._data.truncate(0) |
||||
return |
||||
|
||||
def exists(self): |
||||
""" |
||||
over-ride exists() call |
||||
""" |
||||
size = len(self) |
||||
return True if self.location != ContentLocation.INACCESSIBLE \ |
||||
and size > 0 and ( |
||||
self.max_file_size <= 0 or |
||||
(self.max_file_size > 0 and size <= self.max_file_size)) \ |
||||
else False |
||||
|
||||
@staticmethod |
||||
def parse_url(url): |
||||
""" |
||||
Parses the URL so that we can handle all different file paths |
||||
and return it as our path object |
||||
|
||||
""" |
||||
|
||||
results = AttachBase.parse_url(url, verify_host=False) |
||||
if not results: |
||||
# We're done early; it's not a good URL |
||||
return results |
||||
|
||||
if 'name' not in results: |
||||
# Allow fall-back to be from URL |
||||
match = re.match(r'memory://(?P<path>[^?]+)(\?.*)?', url, re.I) |
||||
if match: |
||||
# Store our filename only (ignore any defined paths) |
||||
results['name'] = \ |
||||
os.path.basename(AttachMemory.unquote(match.group('path'))) |
||||
return results |
||||
|
||||
@property |
||||
def path(self): |
||||
""" |
||||
return the filename |
||||
""" |
||||
if not self.exists(): |
||||
# we could not obtain our path |
||||
return None |
||||
|
||||
return self._name |
||||
|
||||
def __len__(self): |
||||
""" |
||||
Returns the size of he memory attachment |
||||
|
||||
""" |
||||
return self._data.getbuffer().nbytes |
||||
|
||||
def __bool__(self): |
||||
""" |
||||
Allows the Apprise object to be wrapped in an based 'if statement'. |
||||
True is returned if our content was downloaded correctly. |
||||
""" |
||||
|
||||
return self.exists() |
||||
@ -0,0 +1,205 @@
|
||||
# -*- coding: utf-8 -*- |
||||
# BSD 2-Clause License |
||||
# |
||||
# Apprise - Push Notification Library. |
||||
# Copyright (c) 2024, Chris Caron <lead2gold@gmail.com> |
||||
# |
||||
# Redistribution and use in source and binary forms, with or without |
||||
# modification, are permitted provided that the following conditions are met: |
||||
# |
||||
# 1. Redistributions of source code must retain the above copyright notice, |
||||
# this list of conditions and the following disclaimer. |
||||
# |
||||
# 2. Redistributions in binary form must reproduce the above copyright notice, |
||||
# this list of conditions and the following disclaimer in the documentation |
||||
# and/or other materials provided with the distribution. |
||||
# |
||||
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" |
||||
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE |
||||
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE |
||||
# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE |
||||
# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR |
||||
# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF |
||||
# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS |
||||
# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN |
||||
# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) |
||||
# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE |
||||
# POSSIBILITY OF SUCH DAMAGE. |
||||
|
||||
import re |
||||
import urllib |
||||
import pytest |
||||
|
||||
from apprise.attachment.base import AttachBase |
||||
from apprise.attachment.memory import AttachMemory |
||||
from apprise import AppriseAttachment |
||||
from apprise.common import ContentLocation |
||||
|
||||
# Disable logging for a cleaner testing output |
||||
import logging |
||||
logging.disable(logging.CRITICAL) |
||||
|
||||
|
||||
def test_attach_memory_parse_url(): |
||||
""" |
||||
API: AttachMemory().parse_url() |
||||
|
||||
""" |
||||
|
||||
# Bad Entry |
||||
assert AttachMemory.parse_url(object) is None |
||||
|
||||
# Our filename is detected automatically |
||||
assert AttachMemory.parse_url('memory://') |
||||
|
||||
# pass our content in as a string |
||||
mem = AttachMemory(content='string') |
||||
# it loads a string type by default |
||||
mem.mimetype == 'text/plain' |
||||
# Our filename is automatically generated (with .txt) |
||||
assert re.match(r'^[a-z0-9-]+\.txt$', mem.name, re.I) |
||||
|
||||
# open our file |
||||
with mem as fp: |
||||
assert fp.getbuffer().nbytes == len(mem) |
||||
|
||||
# pass our content in as a string |
||||
mem = AttachMemory( |
||||
content='<html/>', name='test.html', mimetype='text/html') |
||||
# it loads a string type by default |
||||
mem.mimetype == 'text/html' |
||||
mem.name == 'test.html' |
||||
|
||||
# Stub function |
||||
assert mem.download() |
||||
|
||||
with pytest.raises(TypeError): |
||||
# garbage in, garbage out |
||||
AttachMemory(content=3) |
||||
|
||||
# pointer to our data |
||||
pointer = mem.open() |
||||
assert pointer.read() == b'<html/>' |
||||
|
||||
# pass our content in as a string |
||||
mem = AttachMemory(content=b'binary-data', name='raw.dat') |
||||
# it loads a string type by default |
||||
assert mem.mimetype == 'application/octet-stream' |
||||
mem.name == 'raw' |
||||
|
||||
# pass our content in as a string |
||||
mem = AttachMemory(content=b'binary-data') |
||||
# it loads a string type by default |
||||
assert mem.mimetype == 'application/octet-stream' |
||||
# Our filename is automatically generated (with .dat) |
||||
assert re.match(r'^[a-z0-9-]+\.dat$', mem.name, re.I) |
||||
|
||||
|
||||
def test_attach_memory(): |
||||
""" |
||||
API: AttachMemory() |
||||
|
||||
""" |
||||
# A url we can test with |
||||
fname = 'testfile' |
||||
url = 'memory:///ignored/path/{fname}'.format(fname=fname) |
||||
|
||||
# Simple gif test |
||||
response = AppriseAttachment.instantiate(url) |
||||
assert isinstance(response, AttachMemory) |
||||
|
||||
# There is no path yet as we haven't written anything to our memory object |
||||
# yet |
||||
assert response.path is None |
||||
assert bool(response) is False |
||||
|
||||
with response as memobj: |
||||
memobj.write(b'content') |
||||
|
||||
# Memory object defaults |
||||
assert response.name == fname |
||||
assert response.path == response.name |
||||
assert response.mimetype == 'application/octet-stream' |
||||
assert bool(response) is True |
||||
|
||||
# |
||||
fname_in_url = urllib.parse.quote(response.name) |
||||
assert response.url().startswith('memory://{}'.format(fname_in_url)) |
||||
|
||||
# Mime is always part of url |
||||
assert re.search(r'[?&]mime=', response.url()) is not None |
||||
|
||||
# Test case where location is simply set to INACCESSIBLE |
||||
# Below is a bad example, but it proves the section of code properly works. |
||||
# Ideally a server admin may wish to just disable all File based |
||||
# attachments entirely. In this case, they simply just need to change the |
||||
# global singleton at the start of their program like: |
||||
# |
||||
# import apprise |
||||
# apprise.attachment.AttachMemory.location = \ |
||||
# apprise.ContentLocation.INACCESSIBLE |
||||
# |
||||
response = AppriseAttachment.instantiate(url) |
||||
assert isinstance(response, AttachMemory) |
||||
with response as memobj: |
||||
memobj.write(b'content') |
||||
|
||||
response.location = ContentLocation.INACCESSIBLE |
||||
assert response.path is None |
||||
# Downloads just don't work period |
||||
assert response.download() is False |
||||
|
||||
# File handling (even if image is set to maxium allowable) |
||||
response = AppriseAttachment.instantiate(url) |
||||
assert isinstance(response, AttachMemory) |
||||
with response as memobj: |
||||
memobj.write(b'content') |
||||
|
||||
# Memory handling when size is to large |
||||
response = AppriseAttachment.instantiate(url) |
||||
assert isinstance(response, AttachMemory) |
||||
with response as memobj: |
||||
memobj.write(b'content') |
||||
|
||||
# Test case where we exceed our defined max_file_size in memory |
||||
prev_value = AttachBase.max_file_size |
||||
AttachBase.max_file_size = len(response) - 1 |
||||
# We can't work in this case |
||||
assert response.path is None |
||||
assert response.download() is False |
||||
|
||||
# Restore our file_size |
||||
AttachBase.max_file_size = prev_value |
||||
|
||||
response = AppriseAttachment.instantiate( |
||||
'memory://apprise-file.gif?mime=image/gif') |
||||
assert isinstance(response, AttachMemory) |
||||
with response as memobj: |
||||
memobj.write(b'content') |
||||
|
||||
assert response.name == 'apprise-file.gif' |
||||
assert response.path == response.name |
||||
assert response.mimetype == 'image/gif' |
||||
# No mime-type and/or filename over-ride was specified, so therefore it |
||||
# won't show up in the generated URL |
||||
assert re.search(r'[?&]mime=', response.url()) is not None |
||||
assert 'image/gif' in response.url() |
||||
|
||||
# Force a mime-type and new name |
||||
response = AppriseAttachment.instantiate( |
||||
'memory://{}?mime={}&name={}'.format( |
||||
'ignored.gif', 'image/jpeg', 'test.jpeg')) |
||||
assert isinstance(response, AttachMemory) |
||||
with response as memobj: |
||||
memobj.write(b'content') |
||||
|
||||
assert response.name == 'test.jpeg' |
||||
assert response.path == response.name |
||||
assert response.mimetype == 'image/jpeg' |
||||
# We will match on mime type now (%2F = /) |
||||
assert re.search(r'[?&]mime=image/jpeg', response.url(), re.I) |
||||
assert response.url().startswith('memory://test.jpeg') |
||||
|
||||
# Test hosted configuration and that we can't add a valid memory file |
||||
aa = AppriseAttachment(location=ContentLocation.HOSTED) |
||||
assert aa.add(response) is False |
||||
Loading…
Reference in new issue