url_shortening_service/site/links/tests.py

"""Tests required by Implementation.md §12."""

import json
from datetime import timedelta

from django.contrib.auth import get_user_model
from django.test import TestCase, override_settings
from django.utils import timezone

from django.conf import settings

from links.models import Click, ShortLink

User = get_user_model()

AUTH = "Bearer monica:dev-only-token"
TOKENS = [("monica", "dev-only-token")]

SETTINGS = dict(
    SHORTENER_API_TOKENS=TOKENS,
    SHORT_DOMAIN="piha.lc",
    PUBLIC_SHORT_URL="https://piha.lc",
    SHORT_PUBLIC_HOSTS=["piha.lc"],
    SHORT_API_HOSTS=["testserver", "localhost", "127.0.0.1", "piha.lc"],
    SHORT_ADMIN_HOSTS=["localhost", "127.0.0.1"],
    SHORT_ALLOWED_HOSTS=["mkdrealtor.com"],
    CLICK_IP_PEPPER="test-pepper-not-the-secret-key",
    ALLOWED_HOSTS=[
        "testserver",
        "localhost",
        "127.0.0.1",
        "piha.lc",
        "shortener.example.com",
    ],
    SHORT_CODE_LENGTH=6,
)


def _json(response):
    return json.loads(response.content.decode())


@override_settings(**SETTINGS)
class AuthTests(TestCase):
    def test_missing_bearer_401(self):
        response = self.client.post(
            "/api/links/",
            data=json.dumps({"target_url": "https://mkdrealtor.com/x"}),
            content_type="application/json",
        )
        self.assertEqual(response.status_code, 401)
        self.assertEqual(response["WWW-Authenticate"], "Bearer")

    def test_wrong_token_401(self):
        response = self.client.post(
            "/api/links/",
            data=json.dumps({"target_url": "https://mkdrealtor.com/x"}),
            content_type="application/json",
            HTTP_AUTHORIZATION="Bearer monica:wrong-secret",
        )
        self.assertEqual(response.status_code, 401)
        self.assertEqual(response["WWW-Authenticate"], "Bearer")
        self.assertEqual(_json(response)["detail"], "Unauthorized")

    def test_matching_named_token_201(self):
        response = self.client.post(
            "/api/links/",
            data=json.dumps({"target_url": "https://mkdrealtor.com/x"}),
            content_type="application/json",
            HTTP_AUTHORIZATION=AUTH,
        )
        self.assertEqual(response.status_code, 201)
        body = _json(response)
        self.assertEqual(body["target_url"], "https://mkdrealtor.com/x")
        self.assertTrue(body["is_active"])
        self.assertEqual(body["click_count"], 0)

    @override_settings(SHORTENER_API_TOKENS=[])
    def test_empty_tokens_503(self):
        response = self.client.post(
            "/api/links/",
            data=json.dumps({"target_url": "https://mkdrealtor.com/x"}),
            content_type="application/json",
            HTTP_AUTHORIZATION=AUTH,
        )
        self.assertEqual(response.status_code, 503)


@override_settings(**SETTINGS)
class HostSplitTests(TestCase):
    def test_short_host_with_bearer_201(self):
        response = self.client.post(
            "/api/links/",
            data=json.dumps({"target_url": "https://mkdrealtor.com/x"}),
            content_type="application/json",
            HTTP_AUTHORIZATION=AUTH,
            HTTP_HOST="piha.lc",
        )
        self.assertEqual(response.status_code, 201)

    def test_public_host_admin_404(self):
        response = self.client.get("/admin/", HTTP_HOST="piha.lc")
        self.assertEqual(response.status_code, 404)

    def test_short_host_without_bearer_401(self):
        response = self.client.post(
            "/api/links/",
            data=json.dumps({"target_url": "https://mkdrealtor.com/x"}),
            content_type="application/json",
            HTTP_HOST="piha.lc",
        )
        self.assertEqual(response.status_code, 401)
        self.assertEqual(response["WWW-Authenticate"], "Bearer")

    def test_short_host_wrong_token_401(self):
        response = self.client.post(
            "/api/links/",
            data=json.dumps({"target_url": "https://mkdrealtor.com/x"}),
            content_type="application/json",
            HTTP_AUTHORIZATION="Bearer monica:wrong-secret",
            HTTP_HOST="piha.lc",
        )
        self.assertEqual(response.status_code, 401)

    @override_settings(
        SHORT_API_HOSTS=["testserver", "localhost", "127.0.0.1"],
    )
    def test_short_host_serves_api_even_if_not_in_api_hosts(self):
        response = self.client.post(
            "/api/links/",
            data=json.dumps({"target_url": "https://mkdrealtor.com/x"}),
            content_type="application/json",
            HTTP_AUTHORIZATION=AUTH,
            HTTP_HOST="piha.lc",
        )
        self.assertEqual(response.status_code, 201)

    @override_settings(
        ALLOWED_HOSTS=[*SETTINGS["ALLOWED_HOSTS"], "other.example.com"]
    )
    def test_allowed_host_not_public_or_api_404(self):
        response = self.client.post(
            "/api/links/",
            data=json.dumps({"target_url": "https://mkdrealtor.com/x"}),
            content_type="application/json",
            HTTP_AUTHORIZATION=AUTH,
            HTTP_HOST="other.example.com",
        )
        self.assertEqual(response.status_code, 404)

    def test_healthz_on_public_and_api(self):
        for host in ("piha.lc", "testserver", "shortener.example.com"):
            response = self.client.get("/healthz/", HTTP_HOST=host)
            self.assertEqual(response.status_code, 200)
            self.assertEqual(_json(response), {"status": "ok"})


@override_settings(**SETTINGS)
class AllowlistTests(TestCase):
    def _post(self, target_url):
        return self.client.post(
            "/api/links/",
            data=json.dumps({"target_url": target_url}),
            content_type="application/json",
            HTTP_AUTHORIZATION=AUTH,
        )

    def test_https_allowlisted_ok(self):
        self.assertEqual(self._post("https://mkdrealtor.com/x").status_code, 201)
        self.assertEqual(self._post("https://www.mkdrealtor.com/x").status_code, 201)

    @override_settings(SHORT_ALLOWED_HOSTS=["mkdrealtor.com", "*.aimloperations.com"])
    def test_glob_suffix_and_apex(self):
        self.assertEqual(
            self._post("https://aimloperations.com/web_design").status_code, 201
        )
        self.assertEqual(
            self._post("https://www.aimloperations.com/web_design").status_code, 201
        )

    def test_http_rejected(self):
        self.assertEqual(self._post("http://mkdrealtor.com/x").status_code, 400)

    def test_evil_host_rejected(self):
        self.assertEqual(self._post("https://evil.com").status_code, 400)

    def test_javascript_rejected(self):
        self.assertEqual(self._post("javascript:alert(1)").status_code, 400)

    def test_protocol_relative_rejected(self):
        self.assertEqual(self._post("//evil.com").status_code, 400)

    def test_credentials_rejected(self):
        self.assertEqual(
            self._post("https://user:pass@mkdrealtor.com/x").status_code, 400
        )

    def test_suffix_does_not_match_cousin_domain(self):
        self.assertEqual(self._post("https://notmkdrealtor.com/x").status_code, 400)


@override_settings(**SETTINGS)
class RedirectTests(TestCase):
    def setUp(self):
        self.link = ShortLink.objects.create(
            code="a3k9xm",
            target_url="https://mkdrealtor.com/listings/oak-st",
            created_by_token="monica",
        )

    def test_active_code_302_to_target(self):
        response = self.client.get(
            "/a3k9xm", HTTP_HOST="piha.lc", follow=False
        )
        self.assertEqual(response.status_code, 302)
        self.assertNotEqual(response.status_code, 301)
        self.assertEqual(
            response["Location"], "https://mkdrealtor.com/listings/oak-st"
        )

    def test_head_also_302_without_click(self):
        response = self.client.head(
            "/a3k9xm", HTTP_HOST="piha.lc", follow=False
        )
        self.assertEqual(response.status_code, 302)
        self.assertEqual(
            response["Location"], "https://mkdrealtor.com/listings/oak-st"
        )
        self.link.refresh_from_db()
        self.assertEqual(self.link.click_count, 0)
        self.assertEqual(Click.objects.filter(link=self.link).count(), 0)

    def test_inactive_404(self):
        self.link.is_active = False
        self.link.save()
        response = self.client.get("/a3k9xm", HTTP_HOST="piha.lc")
        self.assertEqual(response.status_code, 404)

    def test_expired_404(self):
        self.link.expires_at = timezone.now() - timedelta(minutes=1)
        self.link.save()
        response = self.client.get("/a3k9xm", HTTP_HOST="piha.lc")
        self.assertEqual(response.status_code, 404)

    def test_unknown_404(self):
        response = self.client.get("/zzzzzz", HTTP_HOST="piha.lc")
        self.assertEqual(response.status_code, 404)

    def test_invalid_code_shape_404(self):
        response = self.client.get("/AB", HTTP_HOST="piha.lc")
        self.assertEqual(response.status_code, 404)

    def test_public_root_is_landing(self):
        response = self.client.get("/", HTTP_HOST="piha.lc")
        self.assertEqual(response.status_code, 200)
        self.assertContains(response, "piha.lc")
        self.assertContains(response, "Request access")
        self.assertContains(response, "Why Piha?")
        self.assertContains(response, "Quenya")
        self.assertContains(response, "aimloperations.com")
        self.assertContains(response, "Contact")
        self.assertNotContains(response, "<nav")
        self.assertNotContains(response, "Create a short link")

    def test_click_row_and_count(self):
        self.client.get(
            "/a3k9xm",
            HTTP_HOST="piha.lc",
            HTTP_USER_AGENT="sms-client",
        )
        self.link.refresh_from_db()
        self.assertEqual(self.link.click_count, 1)
        self.assertEqual(Click.objects.filter(link=self.link).count(), 1)
        click = Click.objects.get(link=self.link)
        self.assertEqual(click.user_agent, "sms-client")
        self.assertTrue(click.ip_hash)
        self.assertNotIn("127.0.0.1", click.ip_hash)


@override_settings(**SETTINGS)
class CreateTests(TestCase):
    def _post(self, payload, **headers):
        return self.client.post(
            "/api/links/",
            data=json.dumps(payload),
            content_type="application/json",
            HTTP_AUTHORIZATION=AUTH,
            **headers,
        )

    def test_short_url_uses_public_origin_not_api_host(self):
        response = self._post(
            {"target_url": "https://mkdrealtor.com/x"},
            HTTP_HOST="testserver",
        )
        self.assertEqual(response.status_code, 201)
        body = _json(response)
        self.assertTrue(body["short_url"].startswith("https://piha.lc/"))
        self.assertNotIn("testserver", body["short_url"])

    def test_idempotent_same_target_and_external_ref(self):
        payload = {
            "target_url": "https://mkdrealtor.com/listings/oak-st?utm_source=monica",
            "title": "Oak St",
            "external_ref": "campaign-1",
        }
        first = self._post(payload)
        second = self._post(payload)
        self.assertEqual(first.status_code, 201)
        self.assertEqual(second.status_code, 200)
        self.assertEqual(_json(first)["code"], _json(second)["code"])
        self.assertEqual(ShortLink.objects.count(), 1)

    def test_empty_external_ref_always_mints(self):
        payload = {"target_url": "https://mkdrealtor.com/x"}
        first = self._post(payload)
        second = self._post(payload)
        self.assertEqual(first.status_code, 201)
        self.assertEqual(second.status_code, 201)
        self.assertNotEqual(_json(first)["code"], _json(second)["code"])

    def test_code_uses_unambiguous_alphabet(self):
        alphabet = set(settings.CODE_ALPHABET)
        forbidden = set("01iloILO")
        for _ in range(20):
            response = self._post({"target_url": "https://mkdrealtor.com/x"})
            code = _json(response)["code"]
            self.assertTrue(set(code) <= alphabet)
            self.assertFalse(set(code) & forbidden)
            self.assertEqual(len(code), 6)

    def test_stores_token_name_not_secret(self):
        self._post({"target_url": "https://mkdrealtor.com/x"})
        link = ShortLink.objects.get()
        self.assertEqual(link.created_by_token, "monica")
        self.assertNotIn("dev-only-token", link.created_by_token)

    def test_disable_is_idempotent(self):
        code = _json(self._post({"target_url": "https://mkdrealtor.com/x"}))["code"]
        first = self.client.post(
            f"/api/links/{code}/disable/",
            HTTP_AUTHORIZATION=AUTH,
        )
        second = self.client.post(
            f"/api/links/{code}/disable/",
            HTTP_AUTHORIZATION=AUTH,
        )
        self.assertEqual(first.status_code, 200)
        self.assertEqual(second.status_code, 200)
        self.assertFalse(_json(second)["is_active"])
        self.client.get(f"/{code}", HTTP_HOST="piha.lc")
        # disabled → no redirect
        response = self.client.get(f"/{code}", HTTP_HOST="piha.lc")
        self.assertEqual(response.status_code, 404)


@override_settings(**SETTINGS)
class UseCaseTests(TestCase):
    """monica_site mints via Bearer; a phone hits the public short URL and 302s."""

    def test_caller_creates_then_public_redirects(self):
        create = self.client.post(
            "/api/links/",
            data=json.dumps(
                {
                    "target_url": (
                        "https://mkdrealtor.com/listings/oak-st"
                        "?utm_source=monica&utm_medium=sms"
                    ),
                    "title": "Oak St listing",
                    "external_ref": "campaign-uuid-optional",
                }
            ),
            content_type="application/json",
            HTTP_AUTHORIZATION="Bearer monica:dev-only-token",
            HTTP_HOST="testserver",
        )
        self.assertEqual(create.status_code, 201)
        body = _json(create)
        self.assertEqual(
            body["short_url"], f"https://piha.lc/{body['code']}"
        )

        follow = self.client.get(
            f"/{body['code']}",
            HTTP_HOST="piha.lc",
            follow=False,
        )
        self.assertEqual(follow.status_code, 302)
        self.assertEqual(
            follow["Location"],
            "https://mkdrealtor.com/listings/oak-st?utm_source=monica&utm_medium=sms",
        )


@override_settings(**SETTINGS)
class DebugCreateTests(TestCase):
    def test_hidden_when_not_debug(self):
        response = self.client.get("/debug/")
        self.assertEqual(response.status_code, 404)

    @override_settings(DEBUG=True)
    def test_form_when_debug(self):
        response = self.client.get("/debug/")
        self.assertEqual(response.status_code, 200)
        self.assertContains(response, "Create a short link")

    @override_settings(DEBUG=True)
    def test_hidden_on_public_short_host_even_in_debug(self):
        response = self.client.get("/debug/", HTTP_HOST="piha.lc")
        self.assertEqual(response.status_code, 404)

    @override_settings(DEBUG=True)
    def test_post_mints_link(self):
        response = self.client.post(
            "/debug/",
            {
                "target_url": "https://mkdrealtor.com/listings/oak-st",
                "title": "Oak St",
            },
        )
        self.assertEqual(response.status_code, 200)
        self.assertEqual(ShortLink.objects.count(), 1)
        link = ShortLink.objects.get()
        self.assertEqual(link.created_by_token, "debug")
        self.assertContains(response, link.public_short_url)


ADMIN_SETTINGS = {
    **SETTINGS,
    "SHORT_ADMIN_HOSTS": ["testserver", "localhost", "10.0.0.128"],
    "ALLOWED_HOSTS": [*SETTINGS["ALLOWED_HOSTS"], "10.0.0.128"],
}


@override_settings(**ADMIN_SETTINGS)
class AdminTests(TestCase):
    def setUp(self):
        self.user = User.objects.create_superuser("admin", "admin@example.com", "pass")
        self.client.force_login(self.user)
        self.link = ShortLink.objects.create(
            code="a3k9xm",
            target_url="https://mkdrealtor.com/listings/oak-st",
            title="Oak St",
            created_by_token="monica",
        )

    def test_changelist(self):
        response = self.client.get("/admin/links/shortlink/")
        self.assertEqual(response.status_code, 200)
        self.assertContains(response, "a3k9xm")
        self.assertContains(response, "Oak St")

    def test_add_page(self):
        response = self.client.get("/admin/links/shortlink/add/")
        self.assertEqual(response.status_code, 200)
        self.assertContains(response, "Target url")

    def test_add_mints_code(self):
        response = self.client.post(
            "/admin/links/shortlink/add/",
            {
                "target_url": "https://mkdrealtor.com/new",
                "title": "From admin",
                "is_active": "on",
                "external_ref": "",
                "clicks-TOTAL_FORMS": "0",
                "clicks-INITIAL_FORMS": "0",
                "clicks-MIN_NUM_FORMS": "0",
                "clicks-MAX_NUM_FORMS": "0",
                "_save": "Save",
            },
        )
        self.assertEqual(response.status_code, 302)
        created = ShortLink.objects.exclude(code="a3k9xm").get()
        self.assertEqual(created.target_url, "https://mkdrealtor.com/new")
        self.assertEqual(created.created_by_token, "admin")
        self.assertEqual(len(created.code), 6)

    def test_disable_action(self):
        response = self.client.post(
            "/admin/links/shortlink/",
            {
                "action": "disable_links",
                "_selected_action": [str(self.link.pk)],
            },
        )
        self.assertEqual(response.status_code, 302)
        self.link.refresh_from_db()
        self.assertFalse(self.link.is_active)

    def test_click_changelist(self):
        Click.objects.create(link=self.link, user_agent="sms-client")
        response = self.client.get("/admin/links/click/")
        self.assertEqual(response.status_code, 200)
        self.assertContains(response, "a3k9xm")

    def test_admin_404_on_public_hosts(self):
        self.client.logout()
        response = self.client.get("/admin/", HTTP_HOST="piha.lc")
        self.assertEqual(response.status_code, 404)
        response = self.client.get("/admin/", HTTP_HOST="shortener.example.com")
        self.assertEqual(response.status_code, 404)

    def test_admin_200_on_lan_ip(self):
        response = self.client.get("/admin/", HTTP_HOST="10.0.0.128")
        self.assertEqual(response.status_code, 200)
        self.assertContains(response, "Create a short link")
        self.assertContains(response, "Domain")
        self.assertContains(response, "Campaign")
        self.assertContains(response, "Source")
        self.assertContains(response, "Metric")

    def test_quick_mint_creates_tracked_url(self):
        response = self.client.post(
            "/admin/",
            {
                "domain": "mkdrealtor.com",
                "campaign": "open-house",
                "source": "sms",
                "metric": "listing-click",
            },
            HTTP_HOST="10.0.0.128",
        )
        self.assertEqual(response.status_code, 200)
        created = ShortLink.objects.exclude(code="a3k9xm").get()
        self.assertEqual(
            created.target_url,
            "https://mkdrealtor.com/?utm_campaign=open-house&utm_source=sms&utm_medium=listing-click",
        )
        self.assertEqual(created.title, "open-house")
        self.assertEqual(created.created_by_token, "admin")
        self.assertContains(response, created.public_short_url)
        self.assertContains(response, "Copy")

    def test_quick_mint_rejects_unknown_host(self):
        response = self.client.post(
            "/admin/",
            {
                "domain": "evil.example",
                "campaign": "spam",
                "source": "sms",
                "metric": "click",
            },
        )
        self.assertEqual(response.status_code, 200)
        self.assertEqual(ShortLink.objects.exclude(code="a3k9xm").count(), 0)
        self.assertContains(response, "host not allowlisted")


class TrackedUrlTests(TestCase):
    @override_settings(**SETTINGS)
    def test_build_tracked_url(self):
        from links.services import build_tracked_url

        url = build_tracked_url(
            domain="https://mkdrealtor.com/ignored",
            campaign="open house",
            source="sms",
            metric="listing-click",
        )
        self.assertEqual(
            url,
            "https://mkdrealtor.com/?utm_campaign=open+house&utm_source=sms&utm_medium=listing-click",
        )

    @override_settings(**SETTINGS)
    def test_build_tracked_url_rejects_unknown_host(self):
        from links.services import ValidationError, build_tracked_url

        with self.assertRaises(ValidationError):
            build_tracked_url(
                domain="evil.example",
                campaign="c",
                source="s",
                metric="m",
            )