mirror of
https://framagit.org/framasoft/framaspace/argos.git
synced 2025-04-28 18:02:41 +02:00
parent
0e08bc8be6
commit
33becdd753
4 changed files with 57 additions and 4 deletions
|
@ -47,8 +47,6 @@ class SSLCertificateExpiration(BaseCheck):
|
||||||
async def run(self):
|
async def run(self):
|
||||||
"""Returns the number of days in which the certificate will expire."""
|
"""Returns the number of days in which the certificate will expire."""
|
||||||
response = await self.http_client.get(self.task.url)
|
response = await self.http_client.get(self.task.url)
|
||||||
if response.is_error:
|
|
||||||
raise
|
|
||||||
|
|
||||||
network_stream = response.extensions["network_stream"]
|
network_stream = response.extensions["network_stream"]
|
||||||
ssl_obj = network_stream.get_extra_info("ssl_object")
|
ssl_obj = network_stream.get_extra_info("ssl_object")
|
||||||
|
|
|
@ -37,6 +37,7 @@ dev = [
|
||||||
"isort==5.11.5",
|
"isort==5.11.5",
|
||||||
"pytest>=6.2.5",
|
"pytest>=6.2.5",
|
||||||
"pytest-asyncio>=0.21,<1",
|
"pytest-asyncio>=0.21,<1",
|
||||||
|
"respx>=0.20,<1",
|
||||||
"ipython>=8.16,<9",
|
"ipython>=8.16,<9",
|
||||||
"ipdb>=0.13,<0.14",
|
"ipdb>=0.13,<0.14",
|
||||||
"sphinx-autobuild",
|
"sphinx-autobuild",
|
||||||
|
|
56
tests/test_checks.py
Normal file
56
tests/test_checks.py
Normal file
|
@ -0,0 +1,56 @@
|
||||||
|
import datetime
|
||||||
|
import ssl
|
||||||
|
from unittest.mock import MagicMock
|
||||||
|
|
||||||
|
import httpx
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
from argos.checks import SSLCertificateExpiration
|
||||||
|
from argos.schemas import Task
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def now():
|
||||||
|
return datetime.datetime.now()
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def httpx_extensions_ssl():
|
||||||
|
"""Returns the httpx extension dict used by the SSL verification check,
|
||||||
|
to be used when mocking the client responses"""
|
||||||
|
ssl_context = ssl.create_default_context()
|
||||||
|
ssl_context.check_hostname = False
|
||||||
|
ssl_context.verify_mode = ssl.CERT_NONE
|
||||||
|
|
||||||
|
ssl_obj = MagicMock()
|
||||||
|
ssl_obj.getpeercert.return_value = {"notAfter": "Jan 25 20:35:00 2024 GMT"}
|
||||||
|
network_stream = MagicMock()
|
||||||
|
network_stream.get_extra_info = MagicMock(return_value=ssl_obj)
|
||||||
|
return {"network_stream": network_stream}
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def ssl_task(now):
|
||||||
|
return Task(
|
||||||
|
id=1,
|
||||||
|
url="https://example.org",
|
||||||
|
domain="https://example.org",
|
||||||
|
check="ssl-certificate-expiration",
|
||||||
|
expected="on-check",
|
||||||
|
selected_at=now,
|
||||||
|
selected_by="pytest",
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.parametrize("http_status", ["200", "300", "400", "500"])
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_ssl_check_accepts_statuts(
|
||||||
|
respx_mock, ssl_task, httpx_extensions_ssl, http_status
|
||||||
|
):
|
||||||
|
respx_mock.get("https://example.org").mock(
|
||||||
|
return_value=httpx.Response(http_status, extensions=httpx_extensions_ssl),
|
||||||
|
)
|
||||||
|
async with httpx.AsyncClient() as client:
|
||||||
|
check = SSLCertificateExpiration(client, ssl_task)
|
||||||
|
check_response = await check.run()
|
||||||
|
assert check_response.status == "on-check"
|
|
@ -1,5 +1,3 @@
|
||||||
import pytest
|
|
||||||
|
|
||||||
from argos.checks.base import Response, Status
|
from argos.checks.base import Response, Status
|
||||||
|
|
||||||
|
|
||||||
|
|
Loading…
Reference in a new issue