test: Add unit, integration, and e2e tests for scanner components and configure CI workflow.

Author: knowlet

.github/workflows/test.yml

@@ -0,0 +1,29 @@
+name: Test
+
+on:
+    push:
+        branches: [main]
+    pull_request:
+        branches: [main]
+
+jobs:
+    test:
+        runs-on: ubuntu-latest
+        steps:
+            - uses: actions/checkout@v4
+
+            - name: Install uv
+              uses: astral-sh/setup-uv@v5
+              with:
+                  version: "0.5.11"
+
+            - name: Set up Python
+              uses: actions/setup-python@v5
+              with:
+                  python-version-file: "pyproject.toml"
+
+            - name: Install dependencies
+              run: uv sync --all-extras --dev
+
+            - name: Run tests
+              run: uv run pytest

pyproject.toml

@@ -18,6 +18,10 @@ dependencies = [
 dev = [
     "ruff>=0.9.0",
     "pre-commit>=4.0.1",
+    "pytest>=8.0.0",
+    "pytest-asyncio>=0.23.0",
+    "pytest-httpx>=0.30.0",
+    "pytest-cov>=7.0.0",
 ]
 
 [build-system]

tests/test_analyzer.py

@@ -0,0 +1,83 @@
+import json
+import tempfile
+
+import pytest
+
+from scanner.analyzer import detect_api_prefix
+
+
+@pytest.fixture
+def har_data():
+    return {
+        "log": {
+            "entries": [
+                {
+                    "request": {"url": "https://api.example.com/v1/users", "method": "GET"},
+                    "response": {"status": 200, "content": {"mimeType": "application/json"}},
+                },
+                {
+                    "request": {"url": "https://api.example.com/v1/posts", "method": "GET"},
+                    "response": {"status": 200, "content": {"mimeType": "application/json"}},
+                },
+                {
+                    "request": {"url": "https://google.com/analytics", "method": "POST"},
+                    "response": {"status": 200, "content": {"mimeType": "application/json"}},
+                },
+                {
+                    "request": {"url": "https://legacy-app.com/login.php", "method": "POST"},
+                    "response": {"status": 200, "content": {"mimeType": "text/html"}},
+                },
+            ]
+        }
+    }
+
+
+def create_har_file(data):
+    with tempfile.NamedTemporaryFile(mode="w", suffix=".har", delete=False) as f:
+        json.dump(data, f)
+        return f.name
+
+
+def test_detect_api_prefix_json(har_data):
+    har_file = create_har_file(har_data)
+    # targeting the main API
+    prefix = detect_api_prefix(har_file, target_url="https://api.example.com")
+    assert prefix == "https://api.example.com/v1"
+
+
+def test_detect_api_prefix_filter_third_party(har_data):
+    har_file = create_har_file(har_data)
+    # Should ignore google.com even though it has JSON
+    prefix = detect_api_prefix(har_file, target_url="https://api.example.com")
+    assert "google.com" not in prefix
+
+
+def test_detect_api_prefix_html_legacy():
+    # specifically test the fix for text/html content types
+    data = {
+        "log": {
+            "entries": [
+                {
+                    "request": {"url": "https://legacy-app.com/auth/login", "method": "POST"},
+                    "response": {"status": 200, "content": {"mimeType": "text/html"}},
+                },
+                {
+                    "request": {"url": "https://legacy-app.com/auth/register", "method": "POST"},
+                    "response": {"status": 200, "content": {"mimeType": "application/x-www-form-urlencoded"}},
+                },
+            ]
+        }
+    }
+    har_file = create_har_file(data)
+    prefix = detect_api_prefix(har_file, target_url="https://legacy-app.com")
+    assert prefix == "https://legacy-app.com/auth"
+
+
+def test_fallback_behavior(har_data):
+    # If no target URL matches, it might return None or fallback.
+    # Current implementation falls back to all traffic if target filter yields nothing.
+    har_file = create_har_file(har_data)
+    # Searching for a domain not in HAR
+    prefix = detect_api_prefix(har_file, target_url="https://missing.com")
+    # It should fall back to finding the most common prefix in *all* traffic (api.example.com)
+    assert prefix == "https://api.example.com/v1"

tests/test_crawler.py

@@ -0,0 +1,177 @@
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from scanner.crawler import AsyncCrawler
+
+
+@pytest.fixture
+def mock_page():
+    page = AsyncMock()
+    # Mock locator for forms
+    page.locator.return_value.all.return_value = []
+    # Mock eval_on_selector_all for links
+    page.eval_on_selector_all.return_value = []
+    page.url = "http://test-site.com"
+    return page
+
+
+@pytest.fixture
+def mock_context(mock_page):
+    context = AsyncMock()
+    context.new_page.return_value = mock_page
+    return context
+
+
+@pytest.fixture
+def mock_browser(mock_context):
+    browser = AsyncMock()
+    browser.new_context.return_value = mock_context
+    return browser
+
+
+@pytest.fixture
+def mock_playwright(mock_browser):
+    pw = AsyncMock()
+    pw.chromium.launch.return_value = mock_browser
+    return pw
+
+
+@pytest.mark.asyncio
+async def test_crawler_initialization():
+    crawler = AsyncCrawler("http://test.com", max_depth=3)
+    assert crawler.start_url == "http://test.com"
+    assert crawler.max_depth == 3
+    assert crawler.domain == "test.com"
+    assert "http://test.com" in [q[0] for q in crawler.queue]
+
+
+@pytest.mark.asyncio
+async def test_is_valid_url():
+    crawler = AsyncCrawler("http://test.com")
+
+    assert crawler.is_valid_url("http://test.com/path")
+    assert not crawler.is_valid_url("http://google.com")  # External
+    assert not crawler.is_valid_url("http://test.com/image.png")  # Static
+
+    crawler.visited.add("http://test.com/visited")
+    assert not crawler.is_valid_url("http://test.com/visited")  # Visited
+
+
+@patch("scanner.crawler.async_playwright")
+@pytest.mark.asyncio
+async def test_crawl_flow(mock_pw_func, mock_playwright, mock_page):
+    # Setup mock
+    mock_pw_context = AsyncMock()
+    mock_pw_context.__aenter__.return_value = mock_playwright
+    mock_pw_context.__aexit__.return_value = None
+    mock_pw_func.return_value = mock_pw_context
+
+    # Specifics for this test
+    # 1. First visit http://test.com -> found link http://test.com/page2
+    # 2. Visit http://test.com/page2 -> found no links
+
+    async def side_effect_eval(selector, fn):
+        if selector == "a":
+            if mock_page.url == "http://test.com":
+                return ["http://test.com/page2", "http://google.com", "http://test.com/image.png"]
+            elif mock_page.url == "http://test.com/page2":
+                return []
+        return []
+
+    mock_page.eval_on_selector_all.side_effect = side_effect_eval
+
+    # We also need to update mock_page.url when goto is called to simulate navigation
+    async def side_effect_goto(url, **kwargs):
+        mock_page.url = url
+
+    mock_page.goto.side_effect = side_effect_goto
+
+    crawler = AsyncCrawler("http://test.com", max_depth=2, output_har="test.har")
+    await crawler.crawl()
+
+    # Verify visited
+    assert "http://test.com" in crawler.visited
+    assert "http://test.com/page2" in crawler.visited
+
+    # Verify ignored
+    assert "http://google.com" not in crawler.visited
+
+    # Verify calls
+    assert mock_page.goto.call_count == 2
+
+
+@patch("scanner.crawler.async_playwright")
+@pytest.mark.asyncio
+async def test_crawl_form_detection(mock_pw_func, mock_playwright, mock_page):
+    # Setup mock
+    mock_pw_context = AsyncMock()
+    mock_pw_context.__aenter__.return_value = mock_playwright
+    mock_pw_context.__aexit__.return_value = None
+    mock_pw_func.return_value = mock_pw_context
+
+    mock_page.url = "http://test.com/login"
+
+    # Define mock form and input
+    mock_form = AsyncMock()
+    mock_input = AsyncMock()
+    mock_input.is_visible.return_value = True
+
+    # Side effects for attributes
+    async def attr_side_effect(attr):
+        if attr == "type":
+            return "text"
+        if attr == "name":
+            return "username"
+        return ""
+
+    mock_input.get_attribute.side_effect = attr_side_effect
+
+    # Mock submit button
+    mock_submit = AsyncMock()
+    mock_submit.count.return_value = 1
+    mock_submit.is_visible.return_value = True
+
+    # Let's construct a Mock Locator for input
+    # Locator.all() IS async
+    mock_input_locator = AsyncMock()
+    mock_input_locator.all.return_value = [mock_input]
+
+    # Mock Locator for submit
+    mock_submit_locator = AsyncMock()
+    mock_submit_locator.first = mock_submit
+    mock_submit_locator.count.return_value = 1
+
+    # form.locator("...") is SYNCHRONOUS, returns a locator
+    # We use MagicMock for synchronous callable
+    # form.locator("...") should return the appropriate locator
+    mock_form.locator = MagicMock()
+
+    def form_locator_side_effect(selector):
+        # The submit selector contains 'submit'
+        if "submit" in selector or "Login" in selector:
+            return mock_submit_locator
+        # The input selector is mostly implicit or generic 'input'
+        elif "input" in selector or "textarea" in selector:
+            return mock_input_locator
+        else:
+            # Fallback
+            return MagicMock()
+
+    mock_form.locator.side_effect = form_locator_side_effect
+
+    # page.locator("form") is SYNCHRONOUS
+    mock_form_locator = AsyncMock()
+    mock_form_locator.all.return_value = [mock_form]
+
+    mock_page.locator = MagicMock()
+    mock_page.locator.return_value = mock_form_locator
+
+    crawler = AsyncCrawler("http://test.com/login", max_depth=1)
+    await crawler.process_page(mock_page, "http://test.com/login", 0)
+
+    # Verify input was filled
+    mock_input.fill.assert_called_with("testuser")
+
+    # Verify submit clicked
+    mock_submit.click.assert_called()

tests/test_e2e_mock.py

@@ -0,0 +1,71 @@
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from scanner.main import async_main
+
+
+@patch("scanner.main.run_crawler")
+@patch("scanner.main.run_prober")
+@patch("scanner.main.detect_api_prefix")
+@patch("scanner.main.process_to_spec")
+@patch("scanner.main.subprocess.Popen")
+@patch("scanner.main.argparse.ArgumentParser.parse_args")
+@pytest.mark.asyncio
+async def test_main_e2e_flow(
+    mock_parse_args, mock_popen, mock_process_to_spec, mock_detect_prefix, mock_run_prober, mock_run_crawler
+):
+    # Setup args
+    mock_args = MagicMock()
+    mock_args.url = "http://target.com"
+    mock_args.depth = 2
+    mock_args.header = ["Authorization: Bearer test"]
+    mock_args.cookie = ["session=123"]
+    mock_args.proxy_port = 8080
+    mock_args.har_file = "traffic.har"
+    mock_args.initial_spec = "init.yaml"
+    mock_args.fuzzing_dump = "fuzz.mitm"
+    mock_args.final_spec = "final.yaml"
+    mock_args.resume = False
+    mock_args.state_file = "state.json"
+
+    mock_parse_args.return_value = mock_args
+
+    # Setup detect_api_prefix return
+    mock_detect_prefix.return_value = "http://target.com/api"
+
+    # Setup Popen for mitmdump
+    mock_process = MagicMock()
+    mock_process.poll.return_value = None  # Process running
+    mock_popen.return_value = mock_process
+
+    # Run main
+    await async_main()
+
+    # Verify Step 1: Crawler
+    mock_run_crawler.assert_called_once()
+    args, kwargs = mock_run_crawler.call_args
+    assert args[0] == "http://target.com"
+    assert kwargs["headers"] == {"Authorization": "Bearer test"}
+    assert kwargs["cookies"][0]["name"] == "session"
+
+    # Verify Step 2: Proxy Start
+    mock_popen.assert_called_once()
+    assert "mitmdump" in mock_popen.call_args[0][0]
+
+    # Verify Step 3: Probing
+    mock_run_prober.assert_called_once()
+    # Check that parsed prefix was passed
+    # async_main calls: run_prober(args.initial_spec, target_prefix, proxy_url, ...)
+    # where target_prefix comes from detect_api_prefix or fallback
+    call_args = mock_run_prober.call_args
+    # call_args could be positional or keyword
+    # Signature: run_prober(spec_file, api_prefix, proxy, ...)
+    # args: ('init.yaml', 'http://target.com/api', 'http://127.0.0.1:8080')
+    assert call_args[0][1] == "http://target.com/api"
+
+    # Verify Step 4: Spec Gen
+    mock_process_to_spec.assert_called_once()
+    path_gen_kwargs = mock_process_to_spec.call_args[1]
+    assert path_gen_kwargs["api_prefix"] == "http://target.com/api"
+    assert path_gen_kwargs["output_file"] == "final.yaml"