58 lines
3.8 KiB
Python
58 lines
3.8 KiB
Python
import unittest
|
|
from datetime import datetime, timedelta, timezone
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock, patch
|
|
from backend.app.services import public_urls, newsletter_store, recap_store, newsletters, newsletter_catalog
|
|
from backend.tests.test_newsletters import NewsletterFixture
|
|
|
|
|
|
class PublicUrlTests(unittest.TestCase):
|
|
def resolve(self, application=None, proxy=None, enabled=False, legacy='https://legacy.test'):
|
|
with patch.object(public_urls,'get_runtime_settings',return_value=SimpleNamespace(
|
|
magent_application_url=application,magent_proxy_base_url=proxy,magent_proxy_enabled=enabled)):
|
|
return public_urls.magent_public_url(legacy)
|
|
|
|
def test_hosting_is_authoritative_with_proxy_and_path_support(self):
|
|
self.assertEqual(self.resolve('https://prod.test/'),'https://prod.test')
|
|
self.assertEqual(self.resolve('http://internal:3000','https://public.test/magent/',True),'https://public.test/magent')
|
|
self.assertEqual(self.resolve('https://prod.test','https://old-proxy.test',False),'https://prod.test')
|
|
self.assertEqual(self.resolve(),'https://legacy.test')
|
|
|
|
def test_invalid_configured_address_does_not_use_stale_legacy(self):
|
|
for value in ['javascript:alert(1)','https://user:password@host.test','https://host.test?key=secret','https://host.test/#fragment','https://host.test:99999','https://host.test/ bad']:
|
|
self.assertEqual(self.resolve(value),'')
|
|
|
|
|
|
class NewsletterHostingTests(NewsletterFixture, unittest.IsolatedAsyncioTestCase):
|
|
async def test_existing_draft_previews_using_hosting_without_duplicate_url(self):
|
|
draft=self.draft()
|
|
with newsletter_store.transaction() as c:
|
|
c.execute("UPDATE newsletter_settings SET public_url=''")
|
|
self.runtime.magent_application_url='https://prod.example.test'
|
|
with patch.object(public_urls,'get_runtime_settings',return_value=self.runtime),patch.object(newsletter_catalog,'posters',new=AsyncMock(return_value={})):
|
|
rendered=await newsletters.preview(draft['id'],draft['revision'])
|
|
self.assertIn('https://prod.example.test/profile#newsletters',rendered['body_html'])
|
|
self.assertIn('https://watch.example.test',rendered['body_html'])
|
|
self.assertNotIn('https://beta.example.test',rendered['body_html'])
|
|
self.assertEqual(recap_store.settings()['public_url'],'https://prod.example.test')
|
|
|
|
def test_scheduled_delivery_uses_current_hosting_address(self):
|
|
self.subscribe(when=100)
|
|
draft=self.draft()
|
|
newsletter_store.publish(draft['id'],draft['revision'],200,150)
|
|
self.runtime.magent_application_url='https://prod.example.test'
|
|
with patch.object(public_urls,'get_runtime_settings',return_value=self.runtime):
|
|
newsletter_store.enqueue_due(201)
|
|
delivery=newsletter_store.read_one('SELECT * FROM newsletter_deliveries WHERE edition_id=?',(draft['id'],))
|
|
self.assertEqual(delivery['public_url'],'https://prod.example.test')
|
|
self.runtime.magent_application_url='https://new.example.test'
|
|
self.assertEqual(newsletter_store.settings()['public_url'],'https://new.example.test')
|
|
|
|
def test_saving_schedule_uses_hosting_instead_of_client_address(self):
|
|
self.runtime.magent_application_url='https://prod.example.test'
|
|
with patch.object(public_urls,'get_runtime_settings',return_value=self.runtime):
|
|
result=newsletter_store.save_settings({**self.config,'public_url':'https://stale.example.test'},datetime.now(timezone.utc))
|
|
self.assertEqual(result['public_url'],'https://prod.example.test')
|
|
result=recap_store.save_settings({'enabled':False,'day':2,'hour':9,'public_url':''},datetime.now(timezone.utc))
|
|
self.assertEqual(result['public_url'],'https://prod.example.test')
|