Add integration tests

docs/dev.md

@@ -26,10 +26,22 @@ uv run lefthook install
 
 This project uses [pytest](https://docs.pytest.org/en/stable/) as a testing framework.
 
+### Unit Test
+
+Unit tests use a mock HTTP server ([csernazs/pytest-httpserver](https://github.com/csernazs/pytest-httpserver)) and are located under `<root>/tests/unit/`
+
 ```bash
 uv run pytest
 ```
 
+### Integration Test
+
+Integration tests require the environment variable `URLSCAN_API_KEY` and located under `<root>/tests/integration/`
+
+```bash
+uv run pytest --run-optional-tests=integration
+```
+
 ## Docs
 
 This project uses [MkDocs](https://www.mkdocs.org/) as a documentation tool and uses [Mike](https://github.com/jimporter/mike) for versioning.

pyproject.toml

@@ -23,10 +23,12 @@ dev = [
   "packaging>=26.0",
   "pytest-freezer>=0.4.9",
   "pytest-httpserver>=1.1.3",
+  "pytest-optional-tests==0.1.1",
   "pytest-pretty>=1.3.0",
   "pytest-randomly>=4.0.1",
   "pytest-timeout>=2.4.0",
   "pytest>=9.0.2",
+  "python-dotenv>=1.2.1",
   "ruff>=0.14.14",
   "uv-sort>=0.7.0",
 ]

pytest.ini

@@ -0,0 +1,3 @@
+[pytest]
+optional_tests=
+  integration: tests requiring real API access (require URLSCAN_API_KEY env variable)

src/urlscan/pro/saved_search.py

@@ -61,7 +61,7 @@ def create(
             https://docs.urlscan.io/apis/urlscan-openapi/saved-searches/savedsearches-post
 
         """
-        data: dict[str, Any] = _compact(
+        search: dict[str, Any] = _compact(
             {
                 "datasource": datasource,
                 "query": query,
@@ -73,6 +73,7 @@ def create(
                 "permissions": permissions,
             }
         )
+        data: dict[str, Any] = {"search": search}
 
         res = self._post("/api/v1/user/searches/", json=data)
         return self._response_to_json(res)
@@ -119,7 +120,7 @@ def update(
             https://docs.urlscan.io/apis/urlscan-openapi/saved-searches/savedsearches-put
 
         """
-        data: dict[str, Any] = _compact(
+        search: dict[str, Any] = _compact(
             {
                 "datasource": datasource,
                 "query": query,
@@ -131,6 +132,7 @@ def update(
                 "permissions": permissions,
             }
         )
+        data: dict[str, Any] = {"search": search}
 
         res = self._put(f"/api/v1/user/searches/{search_id}/", json=data)
         return self._response_to_json(res)

tests/integration/conftest.py

@@ -0,0 +1,31 @@
+import os
+
+import pytest
+from dotenv import load_dotenv
+
+from urlscan import Client, Pro
+
+
+@pytest.fixture
+def api_key() -> str:
+    load_dotenv()
+    key = os.getenv("URLSCAN_API_KEY")
+    assert key
+    return key
+
+
+@pytest.fixture
+def client(api_key: str):
+    with Client(api_key) as client:
+        yield client
+
+
+@pytest.fixture
+def pro(api_key: str):
+    with Pro(api_key) as client:
+        yield client
+
+
+@pytest.fixture
+def url() -> str:
+    return "https://httpbin.org/html"

tests/integration/pro/test_brand.py

@@ -0,0 +1,19 @@
+import pytest
+
+from urlscan import Pro
+
+
+@pytest.mark.integration
+def test_get_available_brands(pro: Pro):
+    brands = pro.brand.get_available_brands()
+    assert isinstance(brands, dict)
+    assert "kits" in brands
+    assert len(brands["kits"]) > 0
+
+
+@pytest.mark.integration
+def test_get_brands(pro: Pro):
+    brands = pro.brand.get_brands()
+    assert isinstance(brands, dict)
+    assert "responses" in brands
+    assert len(brands["responses"]) > 0

tests/integration/pro/test_channel.py

@@ -0,0 +1,56 @@
+import os
+
+import pytest
+
+from urlscan import Pro
+
+
+@pytest.fixture
+def channel_id() -> str:
+    """Get channel ID from environment variable.
+
+    Set CHANNEL_ID environment variable to test channel operations.
+    """
+    channel_id = os.getenv("CHANNEL_ID")
+    if not channel_id:
+        pytest.skip("CHANNEL_ID environment variable not set")
+
+    return channel_id
+
+
+@pytest.mark.integration
+def test_get_channels(pro: Pro):
+    channels = pro.channel.get_channels()
+    assert isinstance(channels, dict)
+    assert "channels" in channels
+    assert len(channels["channels"]) > 0
+
+
+@pytest.mark.integration
+def test_get_channel(pro: Pro, channel_id: str):
+    channel = pro.channel.get(channel_id)
+    assert channel is not None
+    assert "channel" in channel
+
+
+@pytest.mark.integration
+def test_update_channel(pro: Pro, channel_id: str):
+    # Get the current channel details
+    channel = pro.channel.get(channel_id)
+    original_name = channel["channel"]["name"]
+    channel_type = channel["channel"]["type"]
+
+    try:
+        # Update the channel name
+        updated = pro.channel.update(
+            channel_id, channel_type=channel_type, name="integration-test"
+        )
+        assert updated is not None
+
+        # Verify the update
+        retrieved = pro.channel.get(channel_id)
+        assert retrieved["channel"]["name"] == "integration-test"
+
+    finally:
+        # Revert the change
+        pro.channel.update(channel_id, channel_type=channel_type, name=original_name)

tests/integration/pro/test_datadump.py

@@ -0,0 +1,30 @@
+import datetime
+
+import pytest
+
+from urlscan import Pro
+
+
+@pytest.fixture
+def today() -> str:
+    return datetime.datetime.now(tz=datetime.timezone.utc).strftime("%Y%m%d")
+
+
+@pytest.mark.integration
+def test_datadump_list(pro: Pro, today: str):
+    result = pro.datadump.get_list(f"hours/api/{today}")
+    assert isinstance(result, dict)
+    assert "files" in result
+    assert len(result["files"]) > 0
+
+
+@pytest.mark.integration
+def test_datadump_download(pro: Pro, today: str, tmp_path):
+    # Download a file from the hours/search path
+    output_path = tmp_path / f"{today}.gz"
+    with open(output_path, "wb") as f:
+        pro.datadump.download_file(
+            f"hours/search/{today}/{today}-00.gz",
+            file=f,
+        )
+    assert output_path.exists()

tests/integration/pro/test_hostname.py

@@ -0,0 +1,17 @@
+import pytest
+
+from urlscan import Pro
+
+
+@pytest.mark.integration
+def test_hostname_with_limit(pro: Pro):
+    it = pro.hostname("example.com", limit=1)
+    results = list(it)
+    assert len(results) == 1
+
+
+@pytest.mark.integration
+def test_hostname_with_limit_and_size(pro: Pro):
+    it = pro.hostname("example.com", limit=10, size=10)
+    results = list(it)
+    assert len(results) == 10

tests/integration/pro/test_livescan.py

@@ -0,0 +1,39 @@
+import pytest
+
+from urlscan import Pro
+
+
+@pytest.mark.integration
+def test_get_scanners(pro: Pro):
+    scanners = pro.livescan.get_scanners()
+    assert isinstance(scanners, dict)
+    assert "scanners" in scanners
+    assert len(scanners["scanners"]) > 0
+
+
+@pytest.fixture
+def scanner_id() -> str:
+    return "us01"
+
+
+@pytest.mark.integration
+def test_scan_get_result_dom_and_purge(pro: Pro, url: str, scanner_id: str):
+    # Scan a URL
+    result = pro.livescan.scan(url, scanner_id=scanner_id)
+    uuid = result["uuid"]
+
+    # Get the result
+    scan_result = pro.livescan.get_resource(
+        scanner_id=scanner_id, resource_type="result", resource_id=uuid
+    )
+    assert scan_result is not None
+
+    # Get the DOM
+    dom = pro.livescan.get_resource(
+        scanner_id=scanner_id, resource_type="dom", resource_id=uuid
+    )
+    assert dom is not None
+
+    # Purge the scan
+    purge_result = pro.livescan.purge(scanner_id=scanner_id, scan_id=uuid)
+    assert purge_result is not None

tests/integration/pro/test_saved_search.py

@@ -0,0 +1,40 @@
+import pytest
+
+from urlscan import Pro
+
+
+@pytest.mark.integration
+def test_saved_search_crud(pro: Pro):
+    # Create a saved search
+    created = pro.saved_search.create(
+        datasource="scans",
+        query="page.domain:example.com",
+        name="integration-test",
+    )
+    search_id = created["search"]["_id"]
+
+    try:
+        # Get the saved search results
+        results = pro.saved_search.get_results(search_id)
+        assert results is not None
+
+        # Update the saved search
+        updated = pro.saved_search.update(
+            search_id,
+            datasource="scans",
+            query="page.domain:example.net",
+            name="integration-test-updated",
+        )
+        assert updated is not None
+
+    finally:
+        # Delete the saved search
+        pro.saved_search.remove(search_id)
+
+
+@pytest.mark.integration
+def test_saved_search_list(pro: Pro):
+    result = pro.saved_search.get_list()
+    assert isinstance(result, dict)
+    assert "searches" in result
+    assert len(result["searches"]) > 0

tests/integration/pro/test_structure_search.py

@@ -0,0 +1,21 @@
+import pytest
+
+from urlscan import Pro
+
+# Reference UUID from urlscan.io documentation
+# https://docs.urlscan.io/apis/urlscan-openapi/search/similarsearch
+REFERENCE_UUID = "68e26c59-2eae-437b-aeb1-cf750fafe7d7"
+
+
+@pytest.mark.integration
+def test_structure_search(pro: Pro):
+    it = pro.structure_search(REFERENCE_UUID, size=10, limit=10)
+    results = list(it)
+    assert len(results) >= 1
+
+
+@pytest.mark.integration
+def test_structure_search_with_query(pro: Pro):
+    it = pro.structure_search(REFERENCE_UUID, q="page.domain:*", size=10, limit=10)
+    results = list(it)
+    assert len(results) >= 1

tests/integration/pro/test_subscription.py

@@ -0,0 +1,55 @@
+import pytest
+
+from urlscan import Pro
+
+
+@pytest.mark.integration
+def test_subscription_crud(pro: Pro):
+    # First, create a saved search to use for the subscription
+    search_result = pro.saved_search.create(
+        datasource="scans",
+        query="page.domain:example.com",
+        name="integration-test-subscription",
+    )
+    search_id = search_result["search"]["_id"]
+
+    try:
+        # Create a subscription
+        created = pro.subscription.create(
+            search_ids=[search_id],
+            frequency="daily",
+            name="integration-test",
+            email_addresses=["test@example.com"],
+            is_active=True,
+            ignore_time=False,
+        )
+        subscription_id = created["subscription"]["_id"]
+
+        try:
+            # Update the subscription
+            updated = pro.subscription.update(
+                subscription_id=subscription_id,
+                search_ids=[search_id],
+                frequency="hourly",
+                name="integration-test-updated",
+                email_addresses=["test@example.com"],
+                is_active=True,
+                ignore_time=False,
+            )
+            assert updated is not None
+
+        finally:
+            # Delete the subscription
+            pro.subscription.delete_subscription(subscription_id=subscription_id)
+
+    finally:
+        # Clean up the saved search
+        pro.saved_search.remove(search_id)
+
+
+@pytest.mark.integration
+def test_subscription_list(pro: Pro):
+    result = pro.subscription.get_subscriptions()
+    assert isinstance(result, dict)
+    assert "subscriptions" in result
+    assert len(result["subscriptions"]) > 0