"""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", "shortener.example.com"], 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_public_host_api_404_even_with_bearer(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, 404) @override_settings( SHORT_API_HOSTS=[ "testserver", "localhost", "127.0.0.1", "shortener.example.com", "piha.lc", ] ) def test_short_host_never_serves_api_even_if_also_listed_as_api(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, 404) def test_public_host_admin_404(self): response = self.client.get("/admin/", HTTP_HOST="piha.lc") self.assertEqual(response.status_code, 404) def test_public_api_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="shortener.example.com", ) self.assertEqual(response.status_code, 401) self.assertEqual(response["WWW-Authenticate"], "Bearer") def test_public_api_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="shortener.example.com", ) self.assertEqual(response.status_code, 401) def test_public_api_host_valid_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="shortener.example.com", ) self.assertEqual(response.status_code, 201) def test_public_api_host_admin_404(self): response = self.client.get("/admin/", HTTP_HOST="shortener.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, "aimloperations.com") self.assertContains(response, "Contact") 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"]} @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)