measurement/src/sources.py

89 lines
1.7 KiB
Python

import socket
import time
from collections import namedtuple
from imaplib import IMAP4_SSL
from poplib import POP3_SSL
from smtplib import SMTP
import requests
Result = namedtuple("Result", ['time', 'status', 'protocol'])
def measure_http(url, exlog, timeout=256):
protocol = url.strip().split(":")[0]
try:
response = requests.get(url, allow_redirects=False, timeout=256)
result = Result(
time=response.elapsed.total_seconds(),
status=response.status_code,
protocol=protocol
)
except requests.exceptions.ConnectionError as e:
exlog.exception(e)
result = Result(
time=timeout,
status="-1",
protocol=protocol
)
return result
def measure_smtp(host, exlog, timeout=256):
smtp = SMTP(timeout=timeout)
try:
before = time.time()
status, text = smtp.connect(host)
smtp.starttls()
smtp.noop()
smtp.quit()
after = time.time()
duration = after - before
except Exception as e:
exlog.exception(e)
duration = timeout
status = "-1"
return Result(
time=duration,
status=status,
protocol="smtp"
)
def measure_pop(host, exlog, timeout=256):
status = 0
try:
before = time.time()
pop = POP3_SSL(host, timeout=timeout)
pop.quit()
after = time.time()
duration = after - before
except Exception as e:
exlog.exception(e)
duration = timeout
status = "-1"
return Result(
time=duration,
status=status,
protocol="pop3"
)
def measure_imap(host, exlog, timeout=256):
socket.setdefaulttimeout(timeout)
status = 0
try:
before = time.time()
imap = IMAP4_SSL(host)
imap.noop()
imap.shutdown()
after = time.time()
duration = after - before
except Exception as e:
exlog.exception(e)
duration = timeout
status = "-1"
return Result(
time=duration,
status=status,
protocol="imap"
)