In a test context, the special list is not required
This commit is contained in:
parent
08943cd8d5
commit
8f6dfb2511
6
.idea/vcs.xml
Normal file
6
.idea/vcs.xml
Normal file
@ -0,0 +1,6 @@
|
|||||||
|
<?xml version="1.0" encoding="UTF-8"?>
|
||||||
|
<project version="4">
|
||||||
|
<component name="VcsDirectoryMappings">
|
||||||
|
<mapping directory="$PROJECT_DIR$" vcs="Git" />
|
||||||
|
</component>
|
||||||
|
</project>
|
@ -2,12 +2,11 @@
|
|||||||
Pytest Mock SMTP Server for testing
|
Pytest Mock SMTP Server for testing
|
||||||
"""
|
"""
|
||||||
from email.message import Message as EmailMessage
|
from email.message import Message as EmailMessage
|
||||||
from typing import Optional
|
from typing import List, Optional
|
||||||
|
|
||||||
from aiosmtpd.handlers import Message
|
from aiosmtpd.handlers import Message
|
||||||
|
|
||||||
from smtp_test_server.exceptions import AlreadyStartedError, NotProperlyInitializedError, NotStartedError
|
from smtp_test_server.exceptions import AlreadyStartedError, NotProperlyInitializedError, NotStartedError
|
||||||
from smtp_test_server.list import ThreadSafeMailList
|
|
||||||
from smtp_test_server.mock import SmtpController
|
from smtp_test_server.mock import SmtpController
|
||||||
from smtp_test_server.net import find_free_port
|
from smtp_test_server.net import find_free_port
|
||||||
|
|
||||||
@ -22,7 +21,7 @@ class MessageKeeper(Message):
|
|||||||
Initialize with an empty message list
|
Initialize with an empty message list
|
||||||
"""
|
"""
|
||||||
super().__init__(*args, **kwargs)
|
super().__init__(*args, **kwargs)
|
||||||
self.__messages: ThreadSafeMailList[EmailMessage] = ThreadSafeMailList()
|
self.__messages: List[EmailMessage] = []
|
||||||
|
|
||||||
def handle_message(self, message: EmailMessage) -> None:
|
def handle_message(self, message: EmailMessage) -> None:
|
||||||
"""
|
"""
|
||||||
@ -31,7 +30,7 @@ class MessageKeeper(Message):
|
|||||||
self.__messages.append(message)
|
self.__messages.append(message)
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def messages(self) -> ThreadSafeMailList[EmailMessage]:
|
def messages(self) -> List[EmailMessage]:
|
||||||
"""
|
"""
|
||||||
Access the messages list
|
Access the messages list
|
||||||
"""
|
"""
|
||||||
@ -93,7 +92,7 @@ class SmtpMockServer:
|
|||||||
return self.__bind_port
|
return self.__bind_port
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def messages(self) -> ThreadSafeMailList[EmailMessage]:
|
def messages(self) -> List[EmailMessage]:
|
||||||
"""
|
"""
|
||||||
Access the messages
|
Access the messages
|
||||||
"""
|
"""
|
||||||
|
@ -1,32 +0,0 @@
|
|||||||
"""
|
|
||||||
Implementation of a thread safe list for mails
|
|
||||||
"""
|
|
||||||
import threading
|
|
||||||
from typing import TypeVar
|
|
||||||
|
|
||||||
T = TypeVar("T")
|
|
||||||
|
|
||||||
|
|
||||||
class ThreadSafeMailList(list[T]):
|
|
||||||
"""
|
|
||||||
Simplified Mail List
|
|
||||||
"""
|
|
||||||
|
|
||||||
def __init__(self, *args, **kwargs):
|
|
||||||
"""
|
|
||||||
List with locking
|
|
||||||
"""
|
|
||||||
super().__init__(*args, **kwargs)
|
|
||||||
self.__lock = threading.Lock()
|
|
||||||
|
|
||||||
def append(self, __object: T):
|
|
||||||
with self.__lock:
|
|
||||||
super().append(__object)
|
|
||||||
|
|
||||||
def __len__(self) -> int:
|
|
||||||
with self.__lock:
|
|
||||||
return super().__len__()
|
|
||||||
|
|
||||||
def __getitem__(self, item) -> T:
|
|
||||||
with self.__lock:
|
|
||||||
return super().__getitem__(item)
|
|
Loading…
Reference in New Issue
Block a user