Improve test coverage

Author: dsgnr

backend/api/app/helpers/query.py

@@ -120,6 +120,10 @@ def check_ports(address: str, ports: list[int]) -> list[dict[str, int | bool]]:
         list[dict[str, int | bool]]:
             A list of dictionaries containing the ports checked and their statuses.
     """
+    # Explicit type check for ports
+    for port in ports:
+        if not isinstance(port, int):
+            raise TypeError(f"Port '{port}' is not an integer.")
     results = []
     with ThreadPoolExecutor() as executor:
         futures = {

backend/tests/test_api_routes.py

@@ -10,6 +10,7 @@ def test_route_health_check(client):
     assert response.status_code == HTTP_200_OK
     assert response.text == "true"
 
+
 def test_my_ip_endpoint(client, mocker):
     """Test my_ip endpoint returns correct requester IP."""
     mock_get_requester = mocker.patch(
@@ -89,8 +90,12 @@ def test_query_post_endpoint_invalid_port_v1(client):
     assert ret["extra"][0]["message"] == "Input should be less than or equal to 65535"
 
 
-def test_query_post_endpoint_invalid_hostname_v1(client):
+def test_query_post_endpoint_invalid_hostname_v1(client, mocker):
     """Test v1 query endpoint raises error with invalid hostname."""
+    mocker.patch(
+        "socket.gethostbyname",
+        side_effect=OSError("Hostname does not appear to resolve"),
+    )
     request_data = {"host": INVALID_HOST, "ports": [80]}  # Invalid host
     path = "/api/v1/query"
     response = client.post(path, json=request_data)
@@ -134,8 +139,12 @@ def test_query_post_endpoint_invalid_port_v2(client):
     assert ret["extra"][0]["message"] == "Input should be less than or equal to 65535"
 
 
-def test_query_post_endpoint_invalid_hostname_v2(client):
+def test_query_post_endpoint_invalid_hostname_v2(client, mocker):
     """Test v2 query endpoint raises error with invalid hostname."""
+    mocker.patch(
+        "socket.gethostbyname",
+        side_effect=OSError("Hostname does not appear to resolve"),
+    )
     request_data = {"host": INVALID_HOST, "ports": [80]}  # Invalid host
     path = "/api/query"
     response = client.post(path, json=request_data)

backend/tests/test_query_address.py

@@ -20,37 +20,84 @@
 )
 
 
-def test_query_address_single_closed_port():
-    """Mock socket connection to return non-zero, indicating the port is closed"""
-    result = query_address(VALID_PUBLIC_IPV4, [CLOSED_PORTS[0]])
-    assert result == [{"port": CLOSED_PORTS[0], "status": False}]
+@pytest.mark.parametrize(
+    "host,ports,expected",
+    [
+        (
+            VALID_PUBLIC_IPV4,
+            [CLOSED_PORTS[0]],
+            [{"port": CLOSED_PORTS[0], "status": False}],
+        ),
+        (
+            VALID_PUBLIC_IPV4,
+            OPEN_PORTS,
+            [{"port": port, "status": True} for port in OPEN_PORTS],
+        ),
+        (
+            VALID_PUBLIC_IPV4,
+            CLOSED_PORTS,
+            [{"port": port, "status": False} for port in CLOSED_PORTS],
+        ),
+        (VALID_PUBLIC_IPV4, [], []),
+    ],
+)
+def test_query_address_various(host, ports, expected):
+    """Test query_address for various valid scenarios."""
+    assert query_address(host, ports) == expected
 
 
 def test_query_address_multiple_ports_mixed_status():
     """Test when some ports are open and some are closed."""
     result = query_address(VALID_PUBLIC_IPV4, PORTS)
     expected = [
-        {
-            "port": port,
-            "status": mock_connect((VALID_PUBLIC_IPV4, port)) == SOCKET_OPEN,
-        }
+        {"port": port, "status": mock_connect((VALID_PUBLIC_IPV4, port)) == SOCKET_OPEN}
         for port in PORTS
     ]
     assert result == expected
 
 
-def test_query_address_empty_ports_list():
-    """Test query_address returns empty list when ports list is empty."""
-    ports = expected_result = []
-    assert query_address(VALID_PUBLIC_IPV4, ports) == expected_result
-
-
-def test_query_address_invalid_address():
-    """Test query_address raises JsonAPIException for an invalid hostname."""
+def test_query_address_invalid_hostname(mocker):
+    """Test query_address raises JsonAPIException for an invalid hostname (mocked DNS failure)."""
+    mocker.patch(
+        "socket.gethostbyname",
+        side_effect=OSError("Hostname does not appear to resolve"),
+    )
     with pytest.raises(JsonAPIException, match=".*Hostname does not appear to resolve"):
         query_address(INVALID_HOST, [OPEN_PORTS[0]])
 
 
+def test_query_address_empty_host():
+    """Test query_address raises JsonAPIException for empty host."""
+    with pytest.raises(JsonAPIException, match="A hostname must be provided"):
+        query_address("", [80])
+
+
+def test_query_address_none_host():
+    """Test query_address raises JsonAPIException for None as host."""
+    with pytest.raises(JsonAPIException):
+        query_address(None, [80])
+
+
+def test_query_address_invalid_port_type():
+    """Test query_address raises TypeError for non-integer port."""
+    with pytest.raises(Exception):
+        query_address(VALID_PUBLIC_IPV4, ["notaport"])
+
+
+def test_query_address_ipv6():
+    """Test query_address raises JsonAPIException for IPv6 address."""
+    with pytest.raises(JsonAPIException, match="IPv6 is not currently supported"):
+        query_address("2001:4860:4860::8888", [80])
+
+
+def test_query_address_duplicate_ports():
+    """Test query_address handles duplicate ports gracefully."""
+    ports = [80, 80, 443]
+    result = query_address(VALID_PUBLIC_IPV4, ports)
+    assert result.count({"port": 80, "status": True}) == 2
+    assert {"port": 443, "status": True} in result
+
+
 def test_check_ports_all_open():
     """Test when all ports are open."""
     result = check_ports(VALID_PUBLIC_IPV4, OPEN_PORTS)

backend/tests/test_schemas_api.py

@@ -0,0 +1,73 @@
+"""Tests for schemas"""
+import pytest
+from pydantic import ValidationError
+
+from api.app.schemas.api import (
+    APICheckSchema,
+    APIErrorResponseSchema,
+    APIResponseSchema,
+    APISchema,
+)
+
+# Dummy values for annotations
+HOST = "example.com"
+PORT = 443
+
+
+def test_apischema_valid():
+    """Test that APISchema validates and parses correct input data."""
+    data = {"host": HOST, "ports": [PORT, 80]}
+    schema = APISchema(**data)
+    assert schema.host == HOST
+    assert schema.ports == [PORT, 80]
+
+
+def test_apischema_missing_host():
+    """Test that APISchema raises ValidationError if 'host' is missing."""
+    data = {"ports": [PORT]}
+    with pytest.raises(ValidationError):
+        APISchema(**data)
+
+
+def test_apicheckschema_valid():
+    """Test that APICheckSchema validates and parses correct input data."""
+    data = {"port": PORT, "status": True}
+    schema = APICheckSchema(**data)
+    assert schema.port == PORT
+    assert schema.status is True
+
+
+def test_apiresponseschema_valid():
+    """Test that APIResponseSchema validates and parses correct input data."""
+    check = {"port": PORT, "status": True}
+    data = {
+        "error": False,
+        "msg": "OK",
+        "check": [check],
+        "host": HOST,
+    }
+    schema = APIResponseSchema(**data)
+    assert schema.error is False
+    assert schema.msg == "OK"
+    assert schema.host == HOST
+    assert schema.check[0].port == PORT
+
+
+def test_apierrorresponseschema_valid():
+    """Test that APIErrorResponseSchema validates and parses correct input data."""
+    data = {
+        "error": True,
+        "detail": "fail",
+        "extra": [{"param": "host", "error": "invalid"}],
+    }
+    schema = APIErrorResponseSchema(**data)
+    assert schema.error is True
+    assert schema.detail == "fail"
+    assert schema.extra[0]["param"] == "host"
+
+
+def test_apierrorresponseschema_missing_fields():
+    """Test that APIErrorResponseSchema raises ValidationError if required fields are missing."""
+    data = {"error": True}
+    with pytest.raises(ValidationError):
+        APIErrorResponseSchema(**data)