Commit 7a7f2673 authored by Chris's avatar Chris
Browse files

Split tests as a preliminary step to splitting checks into modules, get stuff...

Split tests as a preliminary step to splitting checks into modules, get stuff working, test test test.
parent 782401ef
Pipeline #5982 passed with stage
in 1 minute and 4 seconds
""" Implement DNS based checks. """
import re
import urllib
import logging
from enum import Enum
from typing import Union
from unittest import mock
import dns
import dns.resolver as resolver
import dns.resolver
import pydantic
from isitup.checks import CheckErrorException
from isitup.http import CheckUrl
from isitup.validators.domain import DomainName
LOG = logging.getLogger(F"{__name__}-check")
class CheckDomain(pydantic.BaseModel):
# regex to check whether there is a scheme
......@@ -87,7 +90,7 @@ async def check_record_by_type(
) -> list[dict]:
try:
result = []
for record in resolver.query(domain, record_type):
for record in dns.resolver.query(domain, record_type):
data = {}
for attr in record.__slots__:
data[attr] = str(getattr(record, attr))
......@@ -101,7 +104,11 @@ async def check_record_by_type(
raise CheckErrorException(
598, f"Fetching the {record_type} for {domain} timed out."
)
except dns.resolver.NoNameservers as exc:
LOG.fatal("No nameservers configured!")
raise CheckErrorException(
500, "Internal server error."
)
def register(register_callback: callable) -> None:
def wrap_check_record_by_type(record_type: str) -> callable:
......
......@@ -43,4 +43,4 @@ class DomainName(str):
return cls(v)
def __repr__(self):
return f'Domain({super().__repr__()})'
\ No newline at end of file
return f'DomainName({super().__repr__()})'
\ No newline at end of file
......@@ -3,6 +3,8 @@ black==20.8b1
asgi_lifespan
pytest==6.2.1
pytest-asyncio
pytest-mock
pytest-raises
pytest-watch
pytest-cov
httpx[http2]
from unittest import mock
import pytest
from asgi_lifespan import LifespanManager
from fastapi import FastAPI, HTTPException
......@@ -48,3 +49,4 @@ def _get_client(base_url: str) -> AsyncClient:
def mock_client(_get_client: callable, monkeypatch):
"""Remove requests.sessions.Session.request for all tests."""
monkeypatch.setattr("isitup.http.get_client", _get_client)
from unittest import mock
import pytest
from pytest_mock import MockerFixture
from httpx import AsyncClient
from starlette import status
import dns
import dns.resolver
import isitup.dns
@pytest.fixture
def mock_dns_query_response():
def wrapped_mock_query_response(*args, **kwargs):
response = mock.Mock()
response.__slots__ = ["address"]
response.address = "127.0.0.1"
return [
response,
]
return wrapped_mock_query_response
class TestDnsChecks:
@pytest.mark.asyncio
async def test_post_empty(self, api: AsyncClient) -> None:
res = await api.post("/check/dns/a/", json={})
assert res.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
@pytest.mark.asyncio
async def test_post_not_domain(self, api: AsyncClient) -> None:
res = await api.post("/check/dns/a/", json={"domain": "!"})
assert res.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
@pytest.mark.asyncio
async def test_post_query_timeout(self, api: AsyncClient, base_url: str) -> None:
with mock.patch("dns.resolver.query") as query:
query.side_effect = dns.exception.Timeout()
res = await api.post("/check/dns/a/", json={"domain": "example.com"})
# Custom status code used by large organisations unoffically
assert res.status_code == 598
assert "Fetching the A for example.com timed out." in res.json()["detail"]
@pytest.mark.asyncio
async def test_post_non_existent_domain(
self, api: AsyncClient, base_url: str
) -> None:
with mock.patch("dns.resolver.query") as query:
query.side_effect = dns.resolver.NXDOMAIN()
res = await api.post(
"/check/dns/a/", json={"domain": "nonexistentdomain.tld"}
)
assert res.status_code == status.HTTP_404_NOT_FOUND
assert "No A record found for nonexistentdomain.tld" in res.json()["detail"]
query.side_effect = dns.resolver.NoAnswer()
res = await api.post(
"/check/dns/a/", json={"domain": "nonexistentdomain.tld"}
)
assert res.status_code == status.HTTP_404_NOT_FOUND
assert "No A record found for nonexistentdomain.tld" in res.json()["detail"]
@pytest.mark.asyncio
async def test_post_no_nameserver(self, api: AsyncClient, base_url: str, mocker: MockerFixture) -> None:
mocker.patch("isitup.dns.LOG.fatal")
with mock.patch("dns.resolver.query") as query:
query.side_effect = dns.resolver.NoNameservers()
res = await api.post(
"/check/dns/a/", json={"domain": "nonexistentdomain.tld"}
)
isitup.dns.LOG.fatal.assert_called_once_with("No nameservers configured!")
assert res.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR
assert "Internal server error." in res.json()["detail"]
@pytest.mark.asyncio
async def test_post_mocked_existing_domain(
self, api: AsyncClient, base_url: str
) -> None:
def mock_dns_query_response(*args, **kwargs):
response = mock.Mock()
response.__slots__ = ["address"]
response.address = "127.0.0.1"
return [
response,
]
with mock.patch("dns.resolver.query") as query:
query.side_effect = mock_dns_query_response
res = await api.post("/check/dns/a/", json={"domain": "existingdomain.tld"})
query.assert_called_once_with("existingdomain.tld", "A")
assert res.status_code == status.HTTP_200_OK
assert res.json()["record_type"] == "a"
assert res.json()["value"][0]["address"] == "127.0.0.1"
@pytest.mark.asyncio
async def test_post_mocked_existing_url(
self, api: AsyncClient, base_url: str, mock_dns_query_response: callable
) -> None:
with mock.patch("dns.resolver.query") as query:
query.side_effect = mock_dns_query_response
res = await api.post("/check/dns/a/", json={"domain": "https://existingdomain.tld"})
query.assert_called_once_with("existingdomain.tld", "A")
assert res.status_code == status.HTTP_200_OK
assert res.json()["record_type"] == "a"
assert res.json()["value"][0]["address"] == "127.0.0.1"
import pytest
from unittest import mock
from httpx import AsyncClient
from starlette.status import (
HTTP_404_NOT_FOUND,
HTTP_422_UNPROCESSABLE_ENTITY,
HTTP_200_OK,
)
from starlette import status
class TestHttpChecks:
@pytest.mark.asyncio
async def test_post_empty(self, api: AsyncClient) -> None:
res = await api.post("/check/http/", json={})
assert res.status_code == HTTP_422_UNPROCESSABLE_ENTITY
assert res.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
@pytest.mark.asyncio
async def test_post_invalid(self, api: AsyncClient) -> None:
res = await api.post("/check/http/", json={"url": "ftp://greenhost.net"})
assert res.status_code == HTTP_422_UNPROCESSABLE_ENTITY
assert res.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
@pytest.mark.asyncio
async def test_post_to_200_url(self, api: AsyncClient, base_url: str) -> None:
res = await api.post("/check/http/", json={"url": f"{base_url}200"})
assert res.status_code == 200
assert res.json()["status_code"] == 200
assert res.status_code == status.HTTP_200_OK
assert res.json()["status_code"] == status.HTTP_200_OK
@pytest.mark.asyncio
async def test_post_to_404_url(self, api: AsyncClient, base_url: str) -> None:
res = await api.post("/check/http/", json={"url": f"{base_url}404"})
assert res.status_code == 200
assert res.json()["status_code"] == 404
assert res.status_code == status.HTTP_200_OK
assert res.json()["status_code"] == status.HTTP_404_NOT_FOUND
@pytest.mark.asyncio
async def test_post_to_500_url(self, api: AsyncClient, base_url: str) -> None:
res = await api.post("/check/http/", json={"url": f"{base_url}500"})
assert res.status_code == 200
assert res.json()["status_code"] == 500
assert res.status_code == status.HTTP_200_OK
assert res.json()["status_code"] == status.HTTP_500_INTERNAL_SERVER_ERROR
class TestDnsChecks:
@pytest.mark.asyncio
async def test_post_empty(self, api: AsyncClient) -> None:
res = await api.post("/check/dns/a/", json={})
assert res.status_code == HTTP_422_UNPROCESSABLE_ENTITY
@pytest.mark.asyncio
async def test_post_nonexistent_domain(
self, api: AsyncClient, base_url: str
) -> None:
res = await api.post("/check/dns/a/", json={"domain": "nonexistentdomain.lan"})
assert res.status_code == 404
assert "No A record found for nonexistentdomain.lan" in res.json()["detail"]
@pytest.mark.asyncio
async def test_post_existing_domain(self, api: AsyncClient, base_url: str) -> None:
res = await api.post("/check/dns/", json={"domain": base_url})
assert res.status_code == 200
assert res.json()["value"]["address"] == "127.0.0.1"
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment