From 5edeecd51633aa3626a35f9263d497b82113a89b Mon Sep 17 00:00:00 2001 From: Rob D'Aveta Date: Wed, 3 Jun 2026 10:19:29 -0400 Subject: [PATCH] add search_fraud_checks endpoint and unit tests for sync and async FlashpointConnector Co-Authored-By: Claude Sonnet 4.6 --- dev_env/api/test_flashpoint.py | 30 +++-------- dev_env/api/test_ipqs.py | 2 +- dev_env/api/test_urlscan.py | 2 +- src/pyapiary/api_connectors/flashpoint.py | 20 ++++++++ .../test_unit_async_flashpoint.py | 51 +++++++++++++++++++ .../test_flashpoint/test_unit_flashpoint.py | 49 ++++++++++++++++++ 6 files changed, 129 insertions(+), 25 deletions(-) diff --git a/dev_env/api/test_flashpoint.py b/dev_env/api/test_flashpoint.py index 451f776..6afa8ad 100644 --- a/dev_env/api/test_flashpoint.py +++ b/dev_env/api/test_flashpoint.py @@ -1,32 +1,16 @@ -from pyapiary.api_connectors.flashpoint import FlashpointConnector +import asyncio import httpx +from pyapiary.api_connectors.flashpoint import FlashpointConnector, AsyncFlashpointConnector -fp = FlashpointConnector( +fp = AsyncFlashpointConnector( load_env_vars=True, trust_env=True, verify=False, follow_redirects=True ) -# res = fp.search_communities('telegram') -# print(res) +async def main(): + # Insert test here + pass - -try: - resp = fp.get_media_image( - "gs://kraken-datalake-media/artifacts/67/6739eed871575b5ad1864ea66d42ebf2430eda5f34d40715c9e40efe90232aa6", - headers={}, - request_kwargs={"follow_redirects": True}, - ) - print("Final URL:", resp.url) - print("Final status:", resp.status_code) -except httpx.HTTPStatusError as e: - resp = e.response - print("Final error:", resp.status_code, resp.url) - - # Show the chain of redirects - for hop in resp.history: - print("Redirect:", hop.status_code, hop.url, "->", hop.headers.get("location")) - - # Show body of final error if any - print("Response body:", resp.text[:500]) \ No newline at end of file +asyncio.run(main()) \ No newline at end of file diff --git a/dev_env/api/test_ipqs.py b/dev_env/api/test_ipqs.py index 60e65ff..ef630fd 100644 --- a/dev_env/api/test_ipqs.py +++ b/dev_env/api/test_ipqs.py @@ -17,7 +17,7 @@ async def main(): ) res = await ipqs.malicious_url( - query="cracked.to" + query="bad.url" ) print(res) diff --git a/dev_env/api/test_urlscan.py b/dev_env/api/test_urlscan.py index cbd9bcc..30008a9 100644 --- a/dev_env/api/test_urlscan.py +++ b/dev_env/api/test_urlscan.py @@ -18,7 +18,7 @@ # ---------- sync ---------- -res = urlscan.search("bclubs.cc") +res = urlscan.search("google.com") print(res.status_code) res = urlscan.scan("google.com") diff --git a/src/pyapiary/api_connectors/flashpoint.py b/src/pyapiary/api_connectors/flashpoint.py index 311d523..155c47a 100644 --- a/src/pyapiary/api_connectors/flashpoint.py +++ b/src/pyapiary/api_connectors/flashpoint.py @@ -44,6 +44,16 @@ def search_fraud(self, query: str, **kwargs) -> httpx.Response: """ return self.post("/sources/v2/fraud", json={"query": query, **kwargs}) + @log_method_call + def search_fraud_checks(self, query: str, **kwargs) -> httpx.Response: + """Checks search provides the ability to search and read data from our Checks dataset. + + Args: + query (str): The search string used in the API query. + **kwargs: Additional query logic per the Flashpoint API documentation. + """ + return self.post("/sources/v2/fraud/checks", json={"query": query, **kwargs}) + @log_method_call def search_marketplaces(self, query: str, **kwargs) -> httpx.Response: """ @@ -140,6 +150,16 @@ async def search_fraud(self, query: str, **kwargs) -> httpx.Response: """ return await self.post("/sources/v2/fraud", json={"query": query, **kwargs}) + @log_method_call + async def search_fraud_checks(self, query: str, **kwargs) -> httpx.Response: + """Checks search provides the ability to search and read data from our Checks dataset. + + Args: + query (str): The search string used in the API query. + **kwargs: Additional query logic per the Flashpoint API documentation. + """ + return await self.post("/sources/v2/fraud/checks", json={"query": query, **kwargs}) + @log_method_call async def search_marketplaces(self, query: str, **kwargs) -> httpx.Response: """ diff --git a/src/pyapiary/tests/test_flashpoint/test_unit_async_flashpoint.py b/src/pyapiary/tests/test_flashpoint/test_unit_async_flashpoint.py index 0176f08..1d95225 100644 --- a/src/pyapiary/tests/test_flashpoint/test_unit_async_flashpoint.py +++ b/src/pyapiary/tests/test_flashpoint/test_unit_async_flashpoint.py @@ -26,6 +26,57 @@ async def test_async_init_missing_key(): AsyncFlashpointConnector() +@patch("pyapiary.api_connectors.flashpoint.AsyncFlashpointConnector.post", new_callable=AsyncMock) +@pytest.mark.asyncio +async def test_async_search_fraud_checks(mock_post): + import json + + request = httpx.Request("POST", "https://api.flashpoint.io/mock") + payload = {"success": True, "data": []} + mock_response = httpx.Response( + 200, + request=request, + content=json.dumps(payload).encode("utf-8"), + headers={"Content-Type": "application/json"}, + ) + mock_post.return_value = mock_response + + connector = AsyncFlashpointConnector(api_key="mock_token") + result = await connector.search_fraud_checks("stolen checks") + + assert isinstance(result, httpx.Response) + assert result.json() == payload + mock_post.assert_awaited_once_with( + "/sources/v2/fraud/checks", json={"query": "stolen checks"} + ) + + +@patch("pyapiary.api_connectors.flashpoint.AsyncFlashpointConnector.post", new_callable=AsyncMock) +@pytest.mark.asyncio +async def test_async_search_fraud_checks_with_kwargs(mock_post): + import json + + request = httpx.Request("POST", "https://api.flashpoint.io/mock") + payload = {"success": True, "data": [{"id": "abc123"}]} + mock_response = httpx.Response( + 200, + request=request, + content=json.dumps(payload).encode("utf-8"), + headers={"Content-Type": "application/json"}, + ) + mock_post.return_value = mock_response + + connector = AsyncFlashpointConnector(api_key="mock_token") + result = await connector.search_fraud_checks("routing number", size=10, from_=0) + + assert isinstance(result, httpx.Response) + assert result.json() == payload + mock_post.assert_awaited_once_with( + "/sources/v2/fraud/checks", + json={"query": "routing number", "size": 10, "from_": 0}, + ) + + @patch("pyapiary.api_connectors.flashpoint.AsyncFlashpointConnector.post", new_callable=AsyncMock) @pytest.mark.asyncio async def test_async_search_fraud(mock_post): diff --git a/src/pyapiary/tests/test_flashpoint/test_unit_flashpoint.py b/src/pyapiary/tests/test_flashpoint/test_unit_flashpoint.py index 220d39d..342f5f9 100644 --- a/src/pyapiary/tests/test_flashpoint/test_unit_flashpoint.py +++ b/src/pyapiary/tests/test_flashpoint/test_unit_flashpoint.py @@ -22,6 +22,55 @@ def test_init_missing_key(): FlashpointConnector() +@patch("pyapiary.api_connectors.flashpoint.FlashpointConnector.post") +def test_search_fraud_checks(mock_post): + import json + + request = httpx.Request("POST", "https://api.flashpoint.io/mock") + payload = {"success": True, "data": []} + mock_response = httpx.Response( + 200, + request=request, + content=json.dumps(payload).encode("utf-8"), + headers={"Content-Type": "application/json"}, + ) + mock_post.return_value = mock_response + + connector = FlashpointConnector(api_key="mock_token") + result = connector.search_fraud_checks("stolen checks") + + assert isinstance(result, httpx.Response) + assert result.json() == payload + mock_post.assert_called_once_with( + "/sources/v2/fraud/checks", json={"query": "stolen checks"} + ) + + +@patch("pyapiary.api_connectors.flashpoint.FlashpointConnector.post") +def test_search_fraud_checks_with_kwargs(mock_post): + import json + + request = httpx.Request("POST", "https://api.flashpoint.io/mock") + payload = {"success": True, "data": [{"id": "abc123"}]} + mock_response = httpx.Response( + 200, + request=request, + content=json.dumps(payload).encode("utf-8"), + headers={"Content-Type": "application/json"}, + ) + mock_post.return_value = mock_response + + connector = FlashpointConnector(api_key="mock_token") + result = connector.search_fraud_checks("routing number", size=10, from_=0) + + assert isinstance(result, httpx.Response) + assert result.json() == payload + mock_post.assert_called_once_with( + "/sources/v2/fraud/checks", + json={"query": "routing number", "size": 10, "from_": 0}, + ) + + @patch("pyapiary.api_connectors.flashpoint.FlashpointConnector.post") def test_search_fraud(mock_post): import json