sin/samba-mirror
1
0
Fork 0
mirror of https://github.com/samba-team/samba.git synced 2024-12-22 13:34:15 +03:00
Code

tests/auth_log: Factor out isRemote()

Browse Source 
This makes waitForMessages() easier to read.

View with ‘git show -b’.

Signed-off-by: Joseph Sutton <josephsutton@catalyst.net.nz>
Reviewed-by: Andrew Bartlett <abartlet@samba.org>
This commit is contained in:
Joseph Sutton 2023-05-24 10:31:53 +12:00 committed by Andrew Bartlett
parent 1f74f9f366
commit 97a5ee4bbb
1 changed files with 22 additions and 22 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

44
python/samba/tests/auth_log_base.py
View File

@ -83,6 +83,26 @@ class AuthLogTestBase(samba.tests.TestCase):
super(AuthLogTestBase, self).setUp()
type(self).discardMessages()
def isRemote(self, message):
if self.remoteAddress is None:
return True
supported_types = {
"Authentication",
"Authorization",
}
message_type = message["type"]
if message_type in supported_types:
remote = message[message_type]["remoteAddress"]
else:
return False
try:
addr = remote.split(":")
return addr[1] == self.remoteAddress
except IndexError:
return False
def waitForMessages(self, isLastExpectedMessage, connection=None):
"""Wait for all the expected messages to arrive
The connection is passed through to keep the connection alive
@ -91,30 +111,10 @@ class AuthLogTestBase(samba.tests.TestCase):
def completed(messages):
for message in messages:
if isRemote(message) and isLastExpectedMessage(message):
if self.isRemote(message) and isLastExpectedMessage(message):
return True
return False
def isRemote(message):
if self.remoteAddress is None:
return True
supported_types = {
"Authentication",
"Authorization",
}
message_type = message["type"]
if message_type in supported_types:
remote = message[message_type]["remoteAddress"]
else:
return False
try:
addr = remote.split(":")
return addr[1] == self.remoteAddress
except IndexError:
return False
self.connection = connection
start_time = time.time()
@ -125,7 +125,7 @@ class AuthLogTestBase(samba.tests.TestCase):
return []
self.connection = None
return list(filter(isRemote, self.context["messages"]))
return list(filter(self.isRemote, self.context["messages"]))
# Discard any previously queued messages.
@classmethod