flaskpaste/tests/test_url_shortener.py

"""Tests for URL shortener endpoints."""

import json
import time


def _create_short_url(client, url="https://example.com", headers=None):
    """Helper to create a short URL and return response data."""
    payload = json.dumps({"url": url})
    return client.post(
        "/s",
        data=payload,
        content_type="application/json",
        headers=headers or {},
    )


class TestShortURLCreate:
    """Tests for POST /s endpoint."""

    def test_create_short_url_json(self, client):
        """Create short URL with JSON body."""
        response = _create_short_url(client)
        assert response.status_code == 201
        data = json.loads(response.data)
        assert "id" in data
        assert len(data["id"]) == 8
        assert data["target_url"] == "https://example.com"
        assert data["url"].endswith(f"/s/{data['id']}")
        assert "created_at" in data

    def test_create_short_url_raw(self, client):
        """Create short URL with raw text body."""
        response = client.post(
            "/s",
            data="https://example.com",
            content_type="text/plain",
        )
        assert response.status_code == 201
        data = json.loads(response.data)
        assert data["target_url"] == "https://example.com"

    def test_create_short_url_no_url(self, client):
        """Reject empty body."""
        response = client.post("/s", data="", content_type="text/plain")
        assert response.status_code == 400

    def test_create_short_url_invalid_scheme(self, client):
        """Reject non-http(s) schemes."""
        response = _create_short_url(client, url="ftp://example.com/file")
        assert response.status_code == 400
        data = json.loads(response.data)
        assert "scheme" in data["error"].lower()

    def test_create_short_url_no_host(self, client):
        """Reject URL without host."""
        response = _create_short_url(client, url="https://")
        assert response.status_code == 400

    def test_create_short_url_with_expiry(self, client):
        """Create short URL with custom expiry."""
        response = client.post(
            "/s",
            data=json.dumps({"url": "https://example.com"}),
            content_type="application/json",
            headers={"X-Expiry": "3600"},
        )
        assert response.status_code == 201
        data = json.loads(response.data)
        assert "expires_at" in data
        assert data["expires_at"] > int(time.time())

    def test_create_short_url_with_auth(self, client, auth_header):
        """Authenticated creation tracks owner."""
        response = _create_short_url(client, headers=auth_header)
        assert response.status_code == 201
        data = json.loads(response.data)
        assert data["owner"] == "a" * 40

    def test_create_short_url_anonymous(self, client):
        """Anonymous creation has no owner."""
        response = _create_short_url(client)
        assert response.status_code == 201
        data = json.loads(response.data)
        assert "owner" not in data

    def test_create_short_url_http(self, client):
        """HTTP URLs are accepted."""
        response = _create_short_url(client, url="http://example.com")
        assert response.status_code == 201

    def test_create_short_url_https_with_path(self, client):
        """URLs with paths are accepted."""
        response = _create_short_url(client, url="https://example.com/some/path?q=1")
        assert response.status_code == 201
        data = json.loads(response.data)
        assert data["target_url"] == "https://example.com/some/path?q=1"


class TestShortURLRedirect:
    """Tests for GET /s/<id> endpoint."""

    def test_redirect_302(self, client):
        """GET /s/<id> returns 302 with Location header."""
        create = _create_short_url(client, url="https://example.com")
        short_id = json.loads(create.data)["id"]

        response = client.get(f"/s/{short_id}")
        assert response.status_code == 302
        assert response.headers["Location"] == "https://example.com"

    def test_redirect_increments_counter(self, client):
        """Redirect increments access_count."""
        create = _create_short_url(client, url="https://example.com")
        short_id = json.loads(create.data)["id"]

        # Access twice
        client.get(f"/s/{short_id}")
        client.get(f"/s/{short_id}")

        # Check info
        info = client.get(f"/s/{short_id}/info")
        data = json.loads(info.data)
        assert data["access_count"] == 2

    def test_redirect_not_found(self, client):
        """Redirect returns 404 for unknown ID."""
        response = client.get("/s/AbCdEfGh")
        assert response.status_code == 404

    def test_redirect_invalid_id(self, client):
        """Redirect returns 400 for invalid ID format."""
        response = client.get("/s/bad!")
        assert response.status_code == 400

    def test_redirect_expired(self, client, app):
        """Redirect returns 404 for expired short URL."""
        # Create with very short expiry
        response = client.post(
            "/s",
            data=json.dumps({"url": "https://example.com"}),
            content_type="application/json",
            headers={"X-Expiry": "1"},
        )
        assert response.status_code == 201
        short_id = json.loads(response.data)["id"]

        # Manually expire it by setting expires_at in the past
        with app.app_context():
            from app.database import get_db

            db = get_db()
            db.execute(
                "UPDATE short_urls SET expires_at = ? WHERE id = ?",
                (int(time.time()) - 10, short_id),
            )
            db.commit()

        result = client.get(f"/s/{short_id}")
        assert result.status_code == 404

    def test_head_redirect(self, client):
        """HEAD /s/<id> returns 302."""
        create = _create_short_url(client, url="https://example.com")
        short_id = json.loads(create.data)["id"]

        response = client.head(f"/s/{short_id}")
        assert response.status_code == 302
        assert response.headers["Location"] == "https://example.com"


class TestShortURLInfo:
    """Tests for GET /s/<id>/info endpoint."""

    def test_info_returns_metadata(self, client, auth_header):
        """Info endpoint returns full metadata."""
        create = _create_short_url(client, url="https://example.com", headers=auth_header)
        short_id = json.loads(create.data)["id"]

        response = client.get(f"/s/{short_id}/info")
        assert response.status_code == 200
        data = json.loads(response.data)
        assert data["id"] == short_id
        assert data["target_url"] == "https://example.com"
        assert data["access_count"] == 0
        assert "created_at" in data
        assert data["owner"] == "a" * 40

    def test_info_does_not_increment_counter(self, client):
        """Info endpoint does not increment access_count."""
        create = _create_short_url(client, url="https://example.com")
        short_id = json.loads(create.data)["id"]

        # Access info multiple times
        client.get(f"/s/{short_id}/info")
        client.get(f"/s/{short_id}/info")
        client.get(f"/s/{short_id}/info")

        response = client.get(f"/s/{short_id}/info")
        data = json.loads(response.data)
        assert data["access_count"] == 0

    def test_info_not_found(self, client):
        """Info returns 404 for unknown ID."""
        response = client.get("/s/AbCdEfGh/info")
        assert response.status_code == 404


class TestShortURLDelete:
    """Tests for DELETE /s/<id> endpoint."""

    def test_delete_requires_auth(self, client):
        """Delete requires authentication."""
        create = _create_short_url(client)
        short_id = json.loads(create.data)["id"]

        response = client.delete(f"/s/{short_id}")
        assert response.status_code == 401

    def test_delete_owner_can_delete(self, client, auth_header):
        """Owner can delete their short URL."""
        create = _create_short_url(client, headers=auth_header)
        short_id = json.loads(create.data)["id"]

        response = client.delete(f"/s/{short_id}", headers=auth_header)
        assert response.status_code == 200

        # Verify deletion
        redirect = client.get(f"/s/{short_id}")
        assert redirect.status_code == 404

    def test_delete_non_owner_rejected(self, client, auth_header, other_auth_header):
        """Non-owner cannot delete."""
        create = _create_short_url(client, headers=auth_header)
        short_id = json.loads(create.data)["id"]

        response = client.delete(f"/s/{short_id}", headers=other_auth_header)
        assert response.status_code == 403

    def test_delete_not_found(self, client, auth_header):
        """Delete returns 404 for unknown ID."""
        response = client.delete("/s/AbCdEfGh", headers=auth_header)
        assert response.status_code == 404


class TestShortURLsList:
    """Tests for GET /s endpoint."""

    def test_list_requires_auth(self, client):
        """List requires authentication."""
        response = client.get("/s")
        assert response.status_code == 401

    def test_list_own_urls(self, client, auth_header):
        """Lists only URLs owned by authenticated user."""
        # Create URLs with different owners
        _create_short_url(client, url="https://mine.com", headers=auth_header)
        _create_short_url(client, url="https://anon.com")  # anonymous

        response = client.get("/s", headers=auth_header)
        assert response.status_code == 200
        data = json.loads(response.data)
        assert data["count"] == 1
        assert data["urls"][0]["target_url"] == "https://mine.com"

    def test_list_pagination(self, client, auth_header):
        """Pagination works correctly."""
        for i in range(3):
            _create_short_url(client, url=f"https://example.com/{i}", headers=auth_header)

        response = client.get("/s?limit=2&offset=0", headers=auth_header)
        data = json.loads(response.data)
        assert data["count"] == 2
        assert data["total"] == 3

    def test_list_empty(self, client, auth_header):
        """Empty list for user with no URLs."""
        response = client.get("/s", headers=auth_header)
        data = json.loads(response.data)
        assert data["count"] == 0
        assert data["urls"] == []


class TestShortURLSecurity:
    """Security-focused tests for URL shortener."""

    def test_reject_javascript_scheme(self, client):
        """Reject javascript: URLs."""
        response = _create_short_url(client, url="javascript:alert(1)")
        assert response.status_code == 400

    def test_reject_data_scheme(self, client):
        """Reject data: URLs."""
        response = _create_short_url(client, url="data:text/html,<script>alert(1)</script>")
        assert response.status_code == 400

    def test_reject_ftp_scheme(self, client):
        """Reject ftp: URLs."""
        response = _create_short_url(client, url="ftp://files.example.com/secret")
        assert response.status_code == 400

    def test_reject_file_scheme(self, client):
        """Reject file: URLs."""
        response = _create_short_url(client, url="file:///etc/passwd")
        assert response.status_code == 400

    def test_reject_no_netloc(self, client):
        """Reject URLs without network location."""
        response = _create_short_url(client, url="https:///path/only")
        assert response.status_code == 400

    def test_reject_url_too_long(self, client):
        """Reject URLs exceeding max length."""
        long_url = "https://example.com/" + "a" * 2050
        response = _create_short_url(client, url=long_url)
        assert response.status_code == 400

    def test_cache_control_on_redirect(self, client):
        """Redirect includes Cache-Control: no-cache."""
        create = _create_short_url(client, url="https://example.com")
        short_id = json.loads(create.data)["id"]

        response = client.get(f"/s/{short_id}")
        assert "no-cache" in response.headers.get("Cache-Control", "")

    def test_short_ids_are_base62(self, client):
        """Generated IDs contain only base62 characters."""
        for _ in range(5):
            create = _create_short_url(client, url="https://example.com")
            short_id = json.loads(create.data)["id"]
            assert len(short_id) == 8
            assert short_id.isalnum()