Spaces:
Running
Running
| """Offline regression tests; runnable via pytest or stdlib unittest.""" | |
| import asyncio | |
| import sys | |
| import types | |
| import unittest | |
| from pathlib import Path | |
| from unittest.mock import patch, AsyncMock | |
| sys.path.insert(0,str(Path(__file__).resolve().parents[1])) | |
| import trusted_links as tl | |
| from journal_lookup import get_index | |
| from repositories import hydrate_selection, hydrate_deposit_strategy | |
| class LinkTests(unittest.IsolatedAsyncioTestCase): | |
| def setUp(self): | |
| tl._cache.clear(); tl._metadata_cache.clear(); tl._locks.clear(); tl._gate=asyncio.Semaphore(4) | |
| self.rec=get_index().lookup_exact(title='Nature') | |
| self.j={'name':self.rec.title,'scopus_id':self.rec.scopus_id,'issn':self.rec.issn_e,'ranking':{}} | |
| self.oa={'issn_l':self.rec.issn_e,'issns':[self.rec.issn_print], 'homepage_url':'https://www.nature.com/nature/', 'openalex_id':'https://openalex.org/S137773608','source_updated_at':'2025-12-01','retrieved_at':'2026-09-20'} | |
| self.mock_oa=types.SimpleNamespace(enrich_journal=AsyncMock(return_value=self.oa)) | |
| self.modules=patch.dict(sys.modules,{'openalex':self.mock_oa});self.modules.start() | |
| def tearDown(self): self.modules.stop() | |
| def test_unsafe_urls(self): | |
| for url in ['javascript:alert(1)','http://nature.com','https://localhost/','https://127.0.0.1/','https://169.254.169.254/','https://[::1]/','https://user:pass@nature.com/','https://nature.com:444/','https://nature.com\\@evil.com/','https://nature.com/\n']: | |
| self.assertIsNone(tl.safe_url(url),url) | |
| self.assertFalse(tl.allowed('https://nature.com.evil.com',['nature.com'])) | |
| self.assertTrue(tl.allowed('https://www.nature.com/nature/',['nature.com'])) | |
| def test_identity_rejects_conflicting_id_issn_or_title(self): | |
| self.assertIsNotNone(tl.canonical(self.j)) | |
| self.assertIsNone(tl.canonical({**self.j,'issn':'1234-5678'})) | |
| self.assertIsNone(tl.canonical({**self.j,'name':'Not Nature'})) | |
| self.assertIsNone(tl.canonical({'name':'Nature made up journal'})) | |
| self.assertIsNone(tl.canonical({'scopus_id':'invalid','issn':self.rec.issn_e})) | |
| async def test_homepage_metadata_dates_and_no_automatic_discovery(self): | |
| with patch.object(tl,'fetch_page',side_effect=AssertionError('must not fetch')): | |
| links=await tl.journal_links(self.j) | |
| self.assertEqual(links['homepage']['status'],'metadata-only') | |
| self.assertIsNone(links['homepage']['reviewed_at']) | |
| self.assertEqual(links['homepage']['source_updated_at'],'2025-12-01') | |
| self.assertIsNone(links['guidelines']['url']) | |
| async def test_wrong_openalex_identity_rejected(self): | |
| self.mock_oa.enrich_journal.return_value={**self.oa,'issn_l':'12345678','issns':[]} | |
| self.assertIsNone((await tl.journal_links(self.j))['homepage']['url']) | |
| async def test_category_quartiles_and_identifiers(self): | |
| data=tl.metric_data() | |
| record=next(r for r in data.values() if len({f['quartile'] for f in r['fields']})>1) | |
| self.assertGreater(len({f['quartile'] for f in record['fields']}),1) | |
| self.assertEqual(record['year'],'2025') | |
| await tl.journal_links(self.j) | |
| self.assertIn('tip=sid',self.j['verify_links']['scimago']) | |
| self.assertEqual(self.j['verify_links']['scopus'],'https://www.scopus.com/sourceid/'+self.rec.scopus_id) | |
| for f in self.j['quartiles_by_field']: | |
| self.assertEqual(f['profile_url'],self.j['verify_links']['scimago']) | |
| if f['category_url']: self.assertIn('category='+f['code'],f['category_url']) | |
| async def test_missing_canonical_data(self): | |
| j={'name':'Imaginary Journal','ranking':{}} | |
| links=await tl.journal_links(j) | |
| self.assertFalse(links['can_discover']); self.assertIsNone(links['homepage']['url']) | |
| self.mock_oa.enrich_journal.assert_not_called() | |
| async def test_success_cache_and_actual_relative_link(self): | |
| home='<h1>Nature</h1><a href="/nature/for-authors">Author guidelines</a>' | |
| guide='<title>Nature — author instructions</title>' | |
| with patch.object(tl,'fetch_page',side_effect=[(self.oa['homepage_url'],home),('https://www.nature.com/nature/for-authors',guide)]) as fetch: | |
| first=await tl.discover(self.j); second=await tl.discover(self.j) | |
| self.assertEqual(fetch.call_count,2) | |
| self.assertEqual(first['guidelines'],second['guidelines']) | |
| self.assertEqual(first['guidelines']['scope'],'journal-specific') | |
| self.assertEqual(first['guidelines']['status'],'checked') | |
| self.assertTrue(first['guidelines']['reviewed_at']) | |
| async def test_general_guidance_label(self): | |
| with patch.object(tl,'fetch_page',side_effect=[(self.oa['homepage_url'],'Nature <a href="/authors">Guide for authors</a>'),('https://www.nature.com/authors','<title>Publisher author services</title>')]): | |
| links=await tl.discover(self.j) | |
| self.assertEqual(links['guidelines']['scope'],'general-publisher') | |
| async def test_failed_request_negative_cache_preserves_homepage(self): | |
| with patch.object(tl,'fetch_page',side_effect=TimeoutError) as fetch: | |
| first=await tl.discover(self.j); second=await tl.discover(self.j) | |
| self.assertEqual(fetch.call_count,1) | |
| self.assertEqual(first,second) | |
| self.assertEqual(first['homepage']['url'],self.oa['homepage_url']) | |
| self.assertIsNone(first['guidelines']['reviewed_at']) | |
| self.assertEqual(first['guidelines']['status'],'unavailable') | |
| async def test_page_identity_required(self): | |
| with patch.object(tl,'fetch_page',return_value=(self.oa['homepage_url'],'<h1>Another Journal</h1><a href="/authors">Guide for authors</a>')) as fetch: | |
| links=await tl.discover(self.j) | |
| self.assertIsNone(links['guidelines']['url']); self.assertEqual(fetch.call_count,1) | |
| async def test_unsafe_discovered_link_rejected(self): | |
| with patch.object(tl,'fetch_page',return_value=(self.oa['homepage_url'],'Nature <a href="https://evil.com">Guide for authors</a>')) as fetch: | |
| links=await tl.discover(self.j) | |
| self.assertIsNone(links['guidelines']['url']); self.assertEqual(fetch.call_count,1) | |
| def test_private_dns_and_redirects(self): | |
| with patch.object(tl.socket,'getaddrinfo',return_value=[(2,1,6,'',('127.0.0.1',443))]), patch.object(tl,'PinnedHTTPS') as connect: | |
| with self.assertRaises(ValueError): tl.fetch_page('https://nature.com',['nature.com']) | |
| connect.assert_not_called() | |
| fake=types.SimpleNamespace(status=302,getheader=lambda k,*args:'https://evil.com/' if k=='Location' else 'text/html') | |
| conn=types.SimpleNamespace(request=lambda *a,**k:None,getresponse=lambda:fake,close=lambda:None) | |
| with patch.object(tl.socket,'getaddrinfo',return_value=[(2,1,6,'',('8.8.8.8',443))]),patch.object(tl,'PinnedHTTPS',return_value=conn) as connect: | |
| with self.assertRaises(ValueError):tl.fetch_page('https://nature.com',['nature.com']) | |
| self.assertEqual(connect.call_count,1) | |
| def test_curated_records_preserve_review_dates(self): | |
| entry={'issns':[self.rec.issn_e],'url':'https://nature.com/authors','source':'https://nature.com/nature','reviewed_at':'2025-10-01','scope':'general-publisher'} | |
| with patch.object(tl,'curated_records',return_value={self.rec.scopus_id:entry}): | |
| self.assertEqual(tl.curated_guidance(self.rec)['reviewed_at'],'2025-10-01') | |
| self.assertEqual(tl.curated_guidance(self.rec)['status'],'curated') | |
| with patch.object(tl,'curated_records',return_value={self.rec.scopus_id:{**entry,'issns':['12345678']}}): | |
| self.assertIsNone(tl.curated_guidance(self.rec)) | |
| async def test_missing_metadata_is_cached(self): | |
| self.mock_oa.enrich_journal.return_value=None | |
| await tl.discover(self.j); await tl.discover(self.j) | |
| self.assertEqual(self.mock_oa.enrich_journal.await_count,1) | |
| async def test_concurrent_discovery_singleflight(self): | |
| import time | |
| def failed(*args): | |
| time.sleep(0.02) | |
| raise TimeoutError() | |
| with patch.object(tl,'fetch_page',side_effect=failed) as fetch: | |
| results=await asyncio.gather(tl.discover(dict(self.j)),tl.discover(dict(self.j))) | |
| self.assertEqual(fetch.call_count,1) | |
| self.assertEqual(results[0]['guidelines'],results[1]['guidelines']) | |
| def test_expired_cache_and_size_limit(self): | |
| tl.cache_put('a',tl.unavailable()) | |
| with patch.object(tl.time,'monotonic',return_value=10**15): self.assertIsNone(tl.cached('a')) | |
| for i in range(600): tl.cache_put(str(i),tl.unavailable()) | |
| self.assertEqual(len(tl._cache),512) | |
| def test_repository_choices_are_curated(self): | |
| repos=hydrate_selection([{'id':'zenodo','url':'https://evil.com'}]) | |
| strategy=hydrate_deposit_strategy({'primary_repository_id':'zenodo','primary_url':'https://evil.com'},repos) | |
| self.assertEqual(strategy['primary_url'],'https://zenodo.org') | |
| self.assertIsNone(strategy['secondary_url']) | |
| if __name__=='__main__': unittest.main() | |