IDEV-2013: Add tests.

Author: jbabac

domaintools/utils.py

@@ -177,8 +177,12 @@ def validate_feeds_parameters(params):
     after = params.get("after")
     before = params.get("before")
     if not (sessionID or after or before):
-        raise ValueError("sessionID or after or before must be defined")
+        raise ValueError("sessionID or after or before must be provided")
 
     format = params.get("output_format")
-    if params.get("endpoint") == Endpoint.DOWNLOAD.value and format == OutputFormat.CSV.value:
+    endpoint = params.get("endpoint")
+    if endpoint == Endpoint.DOWNLOAD.value and format == OutputFormat.CSV.value:
         raise ValueError(f"{format} format is not available in {Endpoint.DOWNLOAD.value} API.")
+
+    if endpoint == Endpoint.DOWNLOAD.value and params.get("header_authentication", True):
+        raise ValueError(f"{Endpoint.DOWNLOAD.value} API does not support header authentication. Provide api_key in the parameter")

tests/conftest.py

@@ -1 +1,14 @@
 """Configuration for test environment"""
+
+import pytest
+
+
+@pytest.fixture
+def test_feeds_params():
+    return {
+        "sessionID": "test-session-id",
+        "after": -60,
+        "before": -120,
+        "output_format": "csv",
+        "endpoint": "download",
+    }

tests/test_api.py

@@ -643,3 +643,27 @@ def test_verify_response_is_a_generator():
     results = feeds_api.domaindiscovery(after="-60", header_authenticationn=False)
 
     assert isgenerator(results.response())
+
+
+@vcr.use_cassette
+def test_feeds_endpoint_should_raise_error_if_download_api_using_header_auth():
+    with pytest.raises(ValueError) as excinfo:
+        feeds_api.domaindiscovery(after="-60", endpoint="download")
+
+    assert str(excinfo.value) == "download API does not support header authentication. Provide api_key in the parameter"
+
+
+@vcr.use_cassette
+def test_feeds_endpoint_should_raise_error_if_no_required_params():
+    with pytest.raises(ValueError) as excinfo:
+        feeds_api.domaindiscovery()
+
+    assert str(excinfo.value) == "sessionID or after or before must be provided"
+
+
+@vcr.use_cassette
+def test_feeds_endpoint_should_raise_error_if_asked_csv_format_for_download_api():
+    with pytest.raises(ValueError) as excinfo:
+        feeds_api.domaindiscovery(after="-60", output_format="csv", endpoint="download")
+
+    assert str(excinfo.value) == "csv format is not available in download API."

tests/test_utils.py

@@ -1,4 +1,6 @@
 import json
+import pytest
+
 from datetime import datetime, timedelta
 
 from tests.responses import iris_investigate_data
@@ -15,24 +17,18 @@ def test_get_domain_age():
 def test_get_threat_component():
     threat_components = [
         {"name": "threat_profile_malware", "risk_score": 31},
-        {"name": "threat_profile_spam", "risk_score": 73, "threats": ["spam"]}
+        {"name": "threat_profile_spam", "risk_score": 73, "threats": ["spam"]},
     ]
     result = utils.get_threat_component(threat_components, "threat_profile_malware")
     assert result.get("risk_score") == 31
 
 
 def test_investigate_average_risk_score():
-    domains = [
-        {"domain_risk": {"risk_score": 25}},
-        {"domain_risk": {"risk_score": 27}}
-    ]
+    domains = [{"domain_risk": {"risk_score": 25}}, {"domain_risk": {"risk_score": 27}}]
     result = utils.get_average_risk_score(domains)
     assert result == 26
 
-    domains = [
-        {"domain_risk": {"risk_score": 25}},
-        {}
-    ]
+    domains = [{"domain_risk": {"risk_score": 25}}, {}]
     result = utils.get_average_risk_score(domains)
     assert result == 25
 
@@ -42,17 +38,11 @@ def test_investigate_average_risk_score():
 
 
 def test_detect_average_risk_score():
-    domains = [
-        {"risk_score": 25},
-        {"risk_score": 27}
-    ]
+    domains = [{"risk_score": 25}, {"risk_score": 27}]
     result = utils.get_average_risk_score(domains)
     assert result == 26
 
-    domains = [
-        {"risk_score": 25},
-        {"risk_score": None}
-    ]
+    domains = [{"risk_score": 25}, {"risk_score": None}]
     result = utils.get_average_risk_score(domains)
     assert result == 25
 
@@ -65,24 +55,15 @@ def test_investigate_average_age():
     two_days_ago = (datetime.now() - timedelta(days=2)).strftime("%Y-%m-%d")
     five_days_ago = (datetime.now() - timedelta(days=5)).strftime("%Y-%m-%d")
 
-    domains = [
-        {"create_date": {"value": two_days_ago}},
-        {"create_date": {"value": five_days_ago}}
-    ]
+    domains = [{"create_date": {"value": two_days_ago}}, {"create_date": {"value": five_days_ago}}]
     result = utils.get_average_age(domains)
     assert result == 3
 
-    domains = [
-        {"create_date": {"value": two_days_ago}},
-        {}
-    ]
+    domains = [{"create_date": {"value": two_days_ago}}, {}]
     result = utils.get_average_age(domains)
     assert result == 2
 
-    domains = [
-        {"create_date": {"value": two_days_ago}},
-        {"create_date": {"value": ""}}
-    ]
+    domains = [{"create_date": {"value": two_days_ago}}, {"create_date": {"value": ""}}]
     result = utils.get_average_age(domains)
     assert result == 2
 
@@ -94,17 +75,11 @@ def test_investigate_average_age():
 def test_detect_average_age():
     two_days_ago = int((datetime.now() - timedelta(days=2)).strftime("%Y%m%d"))
     five_days_ago = int((datetime.now() - timedelta(days=5)).strftime("%Y%m%d"))
-    domains = [
-        {"create_date": two_days_ago},
-        {"create_date": five_days_ago}
-    ]
+    domains = [{"create_date": two_days_ago}, {"create_date": five_days_ago}]
     result = utils.get_average_age(domains)
     assert result == 3
 
-    domains = [
-        {"create_date": two_days_ago},
-        {"create_date": None}
-    ]
+    domains = [{"create_date": two_days_ago}, {"create_date": None}]
     result = utils.get_average_age(domains)
     assert result == 2
 
@@ -121,26 +96,51 @@ def test_data_prune():
 
 def test_find_emails():
     emails = utils.find_emails(json.dumps(iris_investigate_data.domaintools()))
-    assert emails == {'[email protected]', '[email protected]'}
+    assert emails == {"[email protected]", "[email protected]"}
 
 
 def test_find_ips():
     ips = utils.find_ips(json.dumps(iris_investigate_data.domaintools()))
-    assert ips == {'142.250.115.26',
-                   '142.250.141.27',
-                   '198.51.44.4',
-                   '198.51.44.68',
-                   '198.51.45.4',
-                   '198.51.45.68',
-                   '199.30.228.112',
-                   '64.233.171.26',
-                   '74.125.142.26'}
+    assert ips == {
+        "142.250.115.26",
+        "142.250.141.27",
+        "198.51.44.4",
+        "198.51.44.68",
+        "198.51.45.4",
+        "198.51.45.68",
+        "199.30.228.112",
+        "64.233.171.26",
+        "74.125.142.26",
+    }
+
 
 def test_get_pivots():
     pivots = utils.get_pivots(iris_investigate_data.domaintools().get("results"), "")
-    assert pivots == [
-        ['IP ADDRESS', ('199.30.228.112', 4)],
-        ['IP ASN', (17318, 111)],
-        ['IP ISP', ('DomainTools LLC', 222)]
-    ]
+    assert pivots == [["IP ADDRESS", ("199.30.228.112", 4)], ["IP ASN", (17318, 111)], ["IP ISP", ("DomainTools LLC", 222)]]
+
+
+def test_validate_feeds_parameters_should_raise_error_if_download_api_using_header_auth(test_feeds_params):
+    test_feeds_params["output_format"] = "jsonl"
+
+    with pytest.raises(ValueError) as excinfo:
+        utils.validate_feeds_parameters(test_feeds_params)
+
+    assert str(excinfo.value) == "download API does not support header authentication. Provide api_key in the parameter"
+
+
+def test_validate_feeds_parameters_should_raise_error_if_no_required_params(test_feeds_params):
+    test_feeds_params.pop("sessionID", None)
+    test_feeds_params.pop("after", None)
+    test_feeds_params.pop("before", None)
+
+    with pytest.raises(ValueError) as excinfo:
+        utils.validate_feeds_parameters(test_feeds_params)
+
+    assert str(excinfo.value) == "sessionID or after or before must be provided"
+
+
+def test_validate_feeds_parameters_should_raise_error_if_asked_csv_format_for_download_api(test_feeds_params):
+    with pytest.raises(ValueError) as excinfo:
+        utils.validate_feeds_parameters(test_feeds_params)
 
+    assert str(excinfo.value) == "csv format is not available in download API."