zoomeye

Signed-off-by: pranjalg1331 <[email protected]>

Author: pranjalg1331

tests/api_app/analyzers_manager/unit_tests/observable_analyzers/base_test_class.py

@@ -157,48 +157,3 @@ def test_analyzer_on_supported_observables(self):
                         self.fail(
                             f"Unexpected exception for {observable_type}: {type(e).__name__}: {str(e)}"
                         )
-
-    def test_analyzer_error_handling(self):
-        if self.analyzer_class is None:
-            self.skipTest("analyzer_class is not set")
-
-        configs = AnalyzerConfig.objects.filter(
-            python_module=self.analyzer_class.python_module
-        )
-
-        if not configs.exists():
-            self.skipTest(
-                f"No AnalyzerConfig found for {self.analyzer_class.python_module}"
-            )
-
-        config = configs.first()
-
-        invalid_observables = {
-            "domain": "invalid..domain",
-            "ip": "999.999.999.999",
-            "url": "not-a-url",
-            "hash": "tooshort",
-        }
-
-        for observable_type in config.observable_supported:
-            if observable_type in invalid_observables:
-                with self.subTest(observable_type=f"{observable_type}_invalid"):
-                    patches = self.get_mocked_response()
-                    with self._apply_patches(patches):
-                        invalid_value = invalid_observables[observable_type]
-                        analyzer = self._setup_analyzer(
-                            config, observable_type, invalid_value
-                        )
-
-                        try:
-                            response = analyzer.run()
-                            logger.warning(
-                                "Analyzer ran with invalid input: %s = %s",
-                                observable_type,
-                                invalid_value,
-                            )
-                            self.assertTrue(response)
-                        except (AnalyzerRunException, ValueError) as e:
-                            logger.info(
-                                "Expected failure for %s: %s", observable_type, e
-                            )

tests/api_app/analyzers_manager/unit_tests/observable_analyzers/test_tor.py

@@ -0,0 +1,37 @@
+from unittest.mock import mock_open, patch
+
+from api_app.analyzers_manager.observable_analyzers.tor import Tor
+from tests.api_app.analyzers_manager.unit_tests.observable_analyzers.base_test_class import (
+    BaseAnalyzerTest,
+)
+from tests.mock_utils import MockUpResponse
+
+
+class TorTestCase(BaseAnalyzerTest):
+    analyzer_class = Tor
+
+    @staticmethod
+    def get_mocked_response():
+        tor_db_content = "93.95.230.253\n1.2.3.4\n8.8.8.8\n"
+        file_mock = mock_open(read_data=tor_db_content)
+
+        return [
+            patch(
+                "requests.get",
+                return_value=MockUpResponse(
+                    {},
+                    200,
+                    content=b"""ExitNode D2A4BEE6754A9711EB0FAC47F3059BE6FC0D72C7
+Published 2022-08-17 18:11:11
+LastStatus 2022-08-18 14:00:00
+ExitAddress 93.95.230.253 2022-08-18 14:44:33""",
+                ),
+            ),
+            patch("builtins.open", file_mock),
+            patch("os.path.exists", return_value=True),
+            patch("os.path.isfile", return_value=True),
+        ]
+
+    @classmethod
+    def get_extra_config(cls) -> dict:
+        return {}

tests/api_app/analyzers_manager/unit_tests/observable_analyzers/test_tor_nodes_danmeuk.py

@@ -0,0 +1,30 @@
+from unittest.mock import mock_open, patch
+
+from api_app.analyzers_manager.observable_analyzers.tor_nodes_danmeuk import (
+    TorNodesDanMeUK,
+)
+from tests.api_app.analyzers_manager.unit_tests.observable_analyzers.base_test_class import (
+    BaseAnalyzerTest,
+)
+
+
+class TorNodesDanMeUKTestCase(BaseAnalyzerTest):
+    analyzer_class = TorNodesDanMeUK
+
+    @staticmethod
+    def get_mocked_response():
+        mock_db_content = (
+            "100.10.37.131\n" "100.14.156.183\n" "8.8.8.8\n" "100.16.153.149\n"
+        )
+
+        return [
+            patch(
+                "api_app.analyzers_manager.observable_analyzers.tor_nodes_danmeuk.os.path.isfile",
+                return_value=True,
+            ),
+            patch(
+                "api_app.analyzers_manager.observable_analyzers.tor_nodes_danmeuk.os.path.exists",
+                return_value=True,
+            ),
+            patch("builtins.open", mock_open(read_data=mock_db_content)),
+        ]

tests/api_app/analyzers_manager/unit_tests/observable_analyzers/test_tranco.py

@@ -0,0 +1,21 @@
+from unittest.mock import patch
+
+from api_app.analyzers_manager.observable_analyzers.tranco import Tranco
+from tests.api_app.analyzers_manager.unit_tests.observable_analyzers.base_test_class import (
+    BaseAnalyzerTest,
+)
+from tests.mock_utils import MockUpResponse
+
+
+class TrancoTestCase(BaseAnalyzerTest):
+    analyzer_class = Tranco
+
+    @staticmethod
+    def get_mocked_response():
+        mock_data = {"rank": 12345, "domain": "example.com", "date": "2025-07-18"}
+
+        return patch("requests.get", return_value=MockUpResponse(mock_data, 200))
+
+    @classmethod
+    def get_extra_config(cls) -> dict:
+        return {}

tests/api_app/analyzers_manager/unit_tests/observable_analyzers/test_tweet_feeds.py

@@ -0,0 +1,35 @@
+import json
+from unittest.mock import mock_open, patch
+
+from api_app.analyzers_manager.observable_analyzers.tweetfeeds import TweetFeeds
+from tests.api_app.analyzers_manager.unit_tests.observable_analyzers.base_test_class import (
+    BaseAnalyzerTest,
+)
+from tests.mock_utils import MockUpResponse
+
+
+class TweetFeedsTestCase(BaseAnalyzerTest):
+    analyzer_class = TweetFeeds
+
+    @staticmethod
+    def get_mocked_response():
+        mock_data = [
+            {
+                "date": "2024-03-19 00:31:36",
+                "user": "Metemcyber",
+                "type": "url",
+                "value": "http://210.56.49.214",
+                "tags": ["#phishing"],
+                "tweet": "https://twitter.com/Metemcyber/status/1769884392477077774",
+            }
+        ]
+
+        return [
+            patch("builtins.open", mock_open(read_data=json.dumps(mock_data))),
+            patch("os.path.exists", return_value=True),
+            patch("requests.get", return_value=MockUpResponse(mock_data, 200)),
+        ]
+
+    @classmethod
+    def get_extra_config(cls) -> dict:
+        return {"filter1": "Metemcyber", "time": "month"}

tests/api_app/analyzers_manager/unit_tests/observable_analyzers/test_url_haus.py

@@ -0,0 +1,22 @@
+from unittest.mock import patch
+
+from api_app.analyzers_manager.observable_analyzers.urlhaus import URLHaus
+from tests.api_app.analyzers_manager.unit_tests.observable_analyzers.base_test_class import (
+    BaseAnalyzerTest,
+)
+from tests.mock_utils import MockUpResponse
+
+
+class URLHausTestCase(BaseAnalyzerTest):
+    analyzer_class = URLHaus
+
+    @staticmethod
+    def get_mocked_response():
+        return patch(
+            "requests.post",
+            return_value=MockUpResponse({"query_status": "ok", "data": []}, 200),
+        )
+
+    @classmethod
+    def get_extra_config(cls):
+        return {"_api_key_name": "dummy_key"}  # Used in authentication_header

tests/api_app/analyzers_manager/unit_tests/observable_analyzers/test_urldna.py

@@ -0,0 +1,40 @@
+from unittest.mock import patch
+
+from api_app.analyzers_manager.observable_analyzers.urldna import UrlDNA
+from tests.api_app.analyzers_manager.unit_tests.observable_analyzers.base_test_class import (
+    BaseAnalyzerTest,
+)
+from tests.mock_utils import MockUpResponse
+
+
+class UrlDNATestCase(BaseAnalyzerTest):
+    analyzer_class = UrlDNA
+
+    @staticmethod
+    def get_mocked_response():
+        post_scan_response = {"id": "scan123"}
+        get_result_response = {"scan": {"status": "DONE"}, "result": {"score": "clean"}}
+        search_response = {"results": [{"domain": "example.com"}]}
+
+        return [
+            patch(
+                "requests.Session.post",
+                side_effect=[
+                    MockUpResponse(search_response, 200),  # For SEARCH
+                    MockUpResponse(
+                        post_scan_response, 200
+                    ),  # For NEW_SCAN - scan request
+                ],
+            ),
+            patch(
+                "requests.Session.get",
+                return_value=MockUpResponse(get_result_response, 200),
+            ),
+        ]
+
+    @classmethod
+    def get_extra_config(cls):
+        return {
+            "urldna_analysis": "SEARCH",  # change to "NEW_SCAN" if testing that mode
+            "_api_key_name": "Bearer dummykey",
+        }

tests/api_app/analyzers_manager/unit_tests/observable_analyzers/test_urlscan.py

@@ -0,0 +1,35 @@
+from unittest.mock import patch
+
+from api_app.analyzers_manager.observable_analyzers.urlscan import UrlScan
+from tests.api_app.analyzers_manager.unit_tests.observable_analyzers.base_test_class import (
+    BaseAnalyzerTest,
+)
+from tests.mock_utils import MockUpResponse
+
+
+class UrlScanTestCase(BaseAnalyzerTest):
+    analyzer_class = UrlScan
+
+    @staticmethod
+    def get_mocked_response():
+        return [
+            patch(
+                "requests.Session.post",
+                return_value=MockUpResponse(
+                    {"api": "https://urlscan.io/result/abc"}, 200
+                ),
+            ),
+            patch(
+                "requests.Session.get",
+                return_value=MockUpResponse({"results": [{"task": "completed"}]}, 200),
+            ),
+        ]
+
+    @classmethod
+    def get_extra_config(cls):
+        return {
+            "urlscan_analysis": "search",  # or "submit_result"
+            "visibility": "public",
+            "search_size": 10,
+            "_api_key_name": "dummy_api_key",
+        }

tests/api_app/analyzers_manager/unit_tests/observable_analyzers/test_validin.py

@@ -0,0 +1,32 @@
+from unittest.mock import patch
+
+from api_app.analyzers_manager.observable_analyzers.validin import Validin
+from tests.api_app.analyzers_manager.unit_tests.observable_analyzers.base_test_class import (
+    BaseAnalyzerTest,
+)
+from tests.mock_utils import MockUpResponse
+
+
+class ValidinTestCase(BaseAnalyzerTest):
+    analyzer_class = Validin
+
+    @staticmethod
+    def get_mocked_response():
+        response = {
+            "key": "191.121.10.0",
+            "effective_opts": {"type": "ip4", "limit": 100, "wildcard": False},
+            "status": "finished",
+            "query_key": "191.121.10.0",
+            "records": {},
+            "records_returned": 0,
+            "limited": False,
+            "error": None,
+        }
+        return patch("requests.get", return_value=MockUpResponse(response, 200))
+
+    @classmethod
+    def get_extra_config(cls):
+        return {
+            "scan_choice": "default",  # or a specific key like "a_records" if needed
+            "_api_key_name": "dummy_validin_token",
+        }

tests/api_app/analyzers_manager/unit_tests/observable_analyzers/test_vireshee.py

@@ -0,0 +1,21 @@
+from unittest.mock import patch
+
+from api_app.analyzers_manager.observable_analyzers.virushee import VirusheeCheckHash
+from tests.api_app.analyzers_manager.unit_tests.observable_analyzers.base_test_class import (
+    BaseAnalyzerTest,
+)
+from tests.mock_utils import MockUpResponse
+
+
+class VirusheeCheckHashTestCase(BaseAnalyzerTest):
+    analyzer_class = VirusheeCheckHash
+
+    @staticmethod
+    def get_mocked_response():
+        return patch(
+            "requests.Session.get", return_value=MockUpResponse({"success": True}, 200)
+        )
+
+    @classmethod
+    def get_extra_config(cls):
+        return {"_api_key_name": "dummy_api_key"}