Upload 11 files

tests/conftest.py

@@ -1,60 +1,60 @@
-"""Test bootstrap for the ResearchBee Release A regression suite."""
-from __future__ import annotations
-
-import sys
-import types
-from pathlib import Path
-
-ROOT = Path(__file__).resolve().parents[1]
-if str(ROOT) not in sys.path:
-    sys.path.insert(0, str(ROOT))
-
-# The production image installs tenacity from requirements.txt. The execution
-# environment used for static/offline review may not have it, so provide a tiny
-# decorator-compatible fallback only when the real package is unavailable.
-try:  # pragma: no cover - exercised only in minimal review environments
-    import tenacity  # noqa: F401
-except ModuleNotFoundError:  # pragma: no cover
-    module = types.ModuleType("tenacity")
-
-    def retry(*_args, **_kwargs):
-        def decorator(func):
-            return func
-        return decorator
-
-    def option(*_args, **_kwargs):
-        return object()
-
-    module.retry = retry
-    module.stop_after_attempt = option
-    module.wait_exponential = option
-    module.retry_if_exception_type = option
-    module.before_sleep_log = option
-    sys.modules["tenacity"] = module
-
-# Same offline-review fallback for the OpenAI SDK. Production installs the real
-# package from requirements.txt; these tests never make a model call.
-try:  # pragma: no cover
-    from openai import AsyncOpenAI  # noqa: F401
-except ModuleNotFoundError:  # pragma: no cover
-    module = types.ModuleType("openai")
-
-    class OpenAIError(Exception):
-        pass
-
-    class RateLimitError(OpenAIError):
-        pass
-
-    class APIStatusError(OpenAIError):
-        def __init__(self, *args, status_code=500, **kwargs):
-            super().__init__(*args)
-            self.status_code = status_code
-
-    class AsyncOpenAI:
-        def __init__(self, *args, **kwargs):
-            self.chat = types.SimpleNamespace()
-
-    module.AsyncOpenAI = AsyncOpenAI
-    module.RateLimitError = RateLimitError
-    module.APIStatusError = APIStatusError
-    sys.modules["openai"] = module
+"""Test bootstrap for the ResearchBee Release A regression suite."""
+from __future__ import annotations
+
+import sys
+import types
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+
+# The production image installs tenacity from requirements.txt. The execution
+# environment used for static/offline review may not have it, so provide a tiny
+# decorator-compatible fallback only when the real package is unavailable.
+try:  # pragma: no cover - exercised only in minimal review environments
+    import tenacity  # noqa: F401
+except ModuleNotFoundError:  # pragma: no cover
+    module = types.ModuleType("tenacity")
+
+    def retry(*_args, **_kwargs):
+        def decorator(func):
+            return func
+        return decorator
+
+    def option(*_args, **_kwargs):
+        return object()
+
+    module.retry = retry
+    module.stop_after_attempt = option
+    module.wait_exponential = option
+    module.retry_if_exception_type = option
+    module.before_sleep_log = option
+    sys.modules["tenacity"] = module
+
+# Same offline-review fallback for the OpenAI SDK. Production installs the real
+# package from requirements.txt; these tests never make a model call.
+try:  # pragma: no cover
+    from openai import AsyncOpenAI  # noqa: F401
+except ModuleNotFoundError:  # pragma: no cover
+    module = types.ModuleType("openai")
+
+    class OpenAIError(Exception):
+        pass
+
+    class RateLimitError(OpenAIError):
+        pass
+
+    class APIStatusError(OpenAIError):
+        def __init__(self, *args, status_code=500, **kwargs):
+            super().__init__(*args)
+            self.status_code = status_code
+
+    class AsyncOpenAI:
+        def __init__(self, *args, **kwargs):
+            self.chat = types.SimpleNamespace()
+
+    module.AsyncOpenAI = AsyncOpenAI
+    module.RateLimitError = RateLimitError
+    module.APIStatusError = APIStatusError
+    sys.modules["openai"] = module

tests/test_frontend_links.cjs

@@ -0,0 +1,44 @@
+/* Node-only behavioral tests against the actual embedded functions.
+   Does not replace a real-browser visual test. */
+const fs=require('node:fs'),vm=require('node:vm'),assert=require('node:assert/strict'),path=require('node:path');
+const html=fs.readFileSync(path.join(__dirname,'../ResearchBee_LibGuides.html'),'utf8');
+function segment(start,end){return html.slice(html.indexOf(start),html.indexOf(end,html.indexOf(start)));}
+const context={RB_API_BASE:"https://example.org",console,URL,AbortController,setTimeout,clearTimeout,Map,Set,document:{createElement:()=>({textContent:''})},RB_DISCOVERY_STATE:{generation:0,all:[]},RB_ROOT:{addEventListener:()=>{},querySelectorAll:()=>[]},rbText:x=>x,rbV18JournalKey:j=>j.id};
+vm.createContext(context);
+vm.runInContext('const RB_REQUESTS=new Map();const RB_ACTIVE_REQUESTS=new Set();let RB_SESSION_GENERATION=0;'+segment('function rbCancelAllRequests()','function rbRenderChatText(')+segment('function safeHttpsUrl(','function makeExternalLink(')+segment('function esc(str)','function confidenceBadge('),context);
+let count=0;
+function check(fn){fn();count++;}
+(async()=>{
+check(()=>assert.equal(vm.runInContext('safeHttpsUrl("javascript:alert(1)")',context),''));
+check(()=>assert.equal(vm.runInContext('safeHttpsUrl("https://127.0.0.1")',context),''));
+check(()=>assert.equal(vm.runInContext('rbJournalTitle({name:"Nature",journal_links:{homepage:{url:"https://nature.com/"}}}).includes("href=\\\"https://nature.com/\\\"")',context),true));
+check(()=>assert.ok(vm.runInContext('rbJournalLinks({journal_links:{homepage:{status:"metadata-only",url:"https://nature.com"}}})',context).includes('Metadata only')));
+vm.runInContext(segment('function renderFieldQuartiles(', '/* A journal too new'),context);
+check(()=>{let out=vm.runInContext('renderFieldQuartiles([{field:"A",quartile:"Q2",profile_url:"https://www.scimagojr.com/journalsearch.php?q=123&tip=sid",category_url:"https://www.scimagojr.com/journalrank.php?category=2201",source:"SCImago SJR",year:"2025"},{field:"B",quartile:"Q4",profile_url:"https://www.scimagojr.com/journalsearch.php?q=123&tip=sid",source:"SCImago SJR",year:"2025"}])',context);assert.ok(out.includes('Q2'));assert.ok(out.includes('Q4'));assert.ok(out.includes('Not available'));assert.ok(!out.includes('Q1'));});
+context.fetch=()=>new Promise(resolve=>context.resolveFetch=resolve);
+const late=vm.runInContext('rbFetchJson("https://example.org")',context);
+vm.runInContext('rbCancelAllRequests()',context);
+context.resolveFetch({ok:true,json:async()=>({value:'stale'})});
+await assert.rejects(late,e=>e.stale===true);count++;
+// Exercise actual click handler: reset generation changes while fetch is pending.
+let replaced=0;
+const host={isConnected:true,dataset:{linkKey:'21206'},append:()=>{}};
+const button={disabled:false,dataset:{canonicalId:'21206'},closest:()=>host};
+context.event={target:{closest:()=>button}};
+context.RB_ROOT.querySelectorAll=()=>[{dataset:{linkKey:'21206'},set outerHTML(v){replaced++;}}];
+context.RB_DISCOVERY_STATE.all=[{journal_links:{canonical_id:'21206'}}];
+let lookup=vm.runInContext('rbFindGuidelines(event)',context);
+context.RB_DISCOVERY_STATE.generation++;
+context.resolveFetch({ok:true,json:async()=>({canonical_id:'21206',guidelines:{url:'https://nature.com/authors'}})});
+await lookup;check(()=>assert.equal(replaced,0));check(()=>assert.equal(context.RB_DISCOVERY_STATE.all[0].journal_links.guidelines,undefined));
+// Successful click updates state and visible copies.
+button.disabled=false;
+context.fetch=async()=>({ok:true,json:async()=>({canonical_id:'21206',guidelines:{url:'https://nature.com/authors',status:'checked'}})});
+await vm.runInContext('rbFindGuidelines(event)',context);
+check(()=>assert.equal(replaced,1));check(()=>assert.equal(context.RB_DISCOVERY_STATE.all[0].journal_links.guidelines.status,'checked'));
+// Failure preserves existing markup/homepage, reenables the action.
+button.disabled=false; context.fetch=async()=>({ok:false,status:503,json:async()=>({detail:'Unavailable'})});
+await vm.runInContext('rbFindGuidelines(event)',context);
+check(()=>assert.equal(button.disabled,false));check(()=>assert.equal(replaced,1));
+console.log(`${count} frontend behavioral checks passed (Node VM; no browser).`);
+})().catch(e=>{console.error(e);process.exit(1)});

tests/test_frontend_release_a.py

@@ -1,51 +1,51 @@
-from __future__ import annotations
-
-from pathlib import Path
-
-FRONTEND = Path(__file__).resolve().parents[1] / "ResearchBee_LibGuides.html"
-TEXT = FRONTEND.read_text(encoding="utf-8")
-
-
-def between(start: str, end: str) -> str:
-    return TEXT.split(start, 1)[1].split(end, 1)[0]
-
-
-def test_citation_loader_preserves_accordion_container_and_tabs():
-    loader = between(
-        "window.ResearchBee._loadRelatedWorks = async function(uid)",
-        "// ── Deeper Analysis Strip",
-    )
-    assert "rwpanel-cited-${uid}" in loader
-    assert "document.getElementById(`rw-${uid}`)" not in loader
-    assert "panel.replaceChildren" not in loader or "renderWorkList(panel" in loader
-    assert "rwpanel-chain-${uid}" in TEXT
-
-
-def test_cover_letter_rendering_has_no_inline_event_handlers():
-    renderer = between("function renderCoverLetterBtn", "// ── Altmetric badge")
-    generator = between("// ── Cover letter generator", "// ── Subject browse cover letter")
-    subject = between("function renderSubjectQuartileCell", "// ── Export citations")
-    assert "onclick=" not in renderer
-    assert "onclick=" not in generator
-    assert "onclick=" not in subject
-    assert "addEventListener" in renderer
-    assert "textContent" in generator
-    assert "data-cover-id" in renderer
-
-
-def test_related_work_rendering_has_no_inline_handlers_or_raw_html_injection():
-    related = between("// ── Related works accordion", "// ── Deeper Analysis Strip")
-    citation = between("// ── Citation chaining", "// ── KU Journal Finder Guide card")
-    assert "onclick=" not in related
-    assert "onclick=" not in citation
-    assert "api.openalex.org/works?filter=cites" not in TEXT
-    assert "/api/openalex-cited-by" in citation
-    assert "title.textContent" in related
-    assert "authorLine.textContent" in related
-
-
-def test_subject_table_renders_all_field_quartiles():
-    assert "function renderSubjectQuartileCell" in TEXT
-    assert "journal.quartiles_by_field" in TEXT
-    assert "Quartile by field" in TEXT
-    assert "subject-q-item" in TEXT
+from __future__ import annotations
+
+from pathlib import Path
+
+FRONTEND = Path(__file__).resolve().parents[1] / "ResearchBee_LibGuides.html"
+TEXT = FRONTEND.read_text(encoding="utf-8")
+
+
+def between(start: str, end: str) -> str:
+    return TEXT.split(start, 1)[1].split(end, 1)[0]
+
+
+def test_citation_loader_preserves_accordion_container_and_tabs():
+    loader = between(
+        "window.ResearchBee._loadRelatedWorks = async function(uid)",
+        "// ── Deeper Analysis Strip",
+    )
+    assert "rwpanel-cited-${uid}" in loader
+    assert "document.getElementById(`rw-${uid}`)" not in loader
+    assert "panel.replaceChildren" not in loader or "renderWorkList(panel" in loader
+    assert "rwpanel-chain-${uid}" in TEXT
+
+
+def test_cover_letter_rendering_has_no_inline_event_handlers():
+    renderer = between("function renderCoverLetterBtn", "// ── Altmetric badge")
+    generator = between("// ── Cover letter generator", "// ── Subject browse cover letter")
+    subject = between("function renderSubjectQuartileCell", "// ── Export citations")
+    assert "onclick=" not in renderer
+    assert "onclick=" not in generator
+    assert "onclick=" not in subject
+    assert "addEventListener" in renderer
+    assert "textContent" in generator
+    assert "data-cover-id" in renderer
+
+
+def test_related_work_rendering_has_no_inline_handlers_or_raw_html_injection():
+    related = between("// ── Related works accordion", "// ── Deeper Analysis Strip")
+    citation = between("// ── Citation chaining", "// ── KU Journal Finder Guide card")
+    assert "onclick=" not in related
+    assert "onclick=" not in citation
+    assert "api.openalex.org/works?filter=cites" not in TEXT
+    assert "/api/openalex-cited-by" in citation
+    assert "title.textContent" in related
+    assert "authorLine.textContent" in related
+
+
+def test_subject_table_renders_all_field_quartiles():
+    assert "function renderSubjectQuartileCell" in TEXT
+    assert "journal.quartiles_by_field" in TEXT
+    assert "Quartile by field" in TEXT
+    assert "subject-q-item" in TEXT

tests/test_journal_policy_guard.py

@@ -1,44 +1,44 @@
-from __future__ import annotations
-
-import app
-
-
-def test_journal_finder_discards_model_generated_policy_facts():
-    model_output = {
-        "journals": [
-            {
-                "name": "Example Journal",
-                "issn": "1234-5678",
-                "oa_compliance_note": "Deposit the AAM immediately under CC BY.",
-                "green_oa": {
-                    "policy_status": "Confirmed",
-                    "postprint": {
-                        "allowed": "Yes",
-                        "embargo": "0 months",
-                        "licence": "CC BY",
-                    },
-                },
-            }
-        ],
-        "repository_recommendation": {
-            "immediate_open_deposit_possible": "Yes",
-        },
-        "next_actions": [
-            "Deposit the accepted manuscript immediately.",
-            "Read the journal author guidelines.",
-        ],
-        "global_notes": "The journal permits Green OA.",
-    }
-
-    guarded = app.enforce_journal_policy_guard(model_output)
-    journal = guarded["journals"][0]
-    green = journal["green_oa"]
-    assert green["policy_status"] == "Not confirmed"
-    assert green["permission_status"] == "unknown"
-    assert green["embargo_status"] == "unknown"
-    assert green["postprint"]["allowed"] == "Unclear"
-    assert "DOI-based" in journal["oa_compliance_note"]
-    assert guarded["repository_recommendation"]["immediate_open_deposit_possible"] == "Unknown"
-    assert all("Deposit the accepted" not in action for action in guarded["next_actions"])
-    assert any("author guidelines" in action for action in guarded["next_actions"])
-    assert guarded["policy_source_mode"] == "not_checked_in_journal_finder"
+from __future__ import annotations
+
+import app
+
+
+def test_journal_finder_discards_model_generated_policy_facts():
+    model_output = {
+        "journals": [
+            {
+                "name": "Example Journal",
+                "issn": "1234-5678",
+                "oa_compliance_note": "Deposit the AAM immediately under CC BY.",
+                "green_oa": {
+                    "policy_status": "Confirmed",
+                    "postprint": {
+                        "allowed": "Yes",
+                        "embargo": "0 months",
+                        "licence": "CC BY",
+                    },
+                },
+            }
+        ],
+        "repository_recommendation": {
+            "immediate_open_deposit_possible": "Yes",
+        },
+        "next_actions": [
+            "Deposit the accepted manuscript immediately.",
+            "Read the journal author guidelines.",
+        ],
+        "global_notes": "The journal permits Green OA.",
+    }
+
+    guarded = app.enforce_journal_policy_guard(model_output)
+    journal = guarded["journals"][0]
+    green = journal["green_oa"]
+    assert green["policy_status"] == "Not confirmed"
+    assert green["permission_status"] == "unknown"
+    assert green["embargo_status"] == "unknown"
+    assert green["postprint"]["allowed"] == "Unclear"
+    assert "DOI-based" in journal["oa_compliance_note"]
+    assert guarded["repository_recommendation"]["immediate_open_deposit_possible"] == "Unknown"
+    assert all("Deposit the accepted" not in action for action in guarded["next_actions"])
+    assert any("author guidelines" in action for action in guarded["next_actions"])
+    assert guarded["policy_source_mode"] == "not_checked_in_journal_finder"

tests/test_khazna_reload_auth.py

@@ -1,62 +1,62 @@
-from __future__ import annotations
-
-import inspect
-
-import pytest
-from fastapi import HTTPException
-from starlette.requests import Request
-
-import app
-
-
-def request_with(headers=None):
-    raw_headers = [
-        (str(key).lower().encode("latin-1"), str(value).encode("latin-1"))
-        for key, value in (headers or {}).items()
-    ]
-    return Request({
-        "type": "http",
-        "method": "POST",
-        "path": "/api/admin/reload-khazna",
-        "headers": raw_headers,
-        "client": ("127.0.0.1", 12345),
-        "server": ("testserver", 80),
-        "scheme": "http",
-        "query_string": b"",
-    })
-
-
-def test_reload_secret_must_be_configured(monkeypatch):
-    monkeypatch.delenv("KHAZNA_RELOAD_TOKEN", raising=False)
-    with pytest.raises(HTTPException) as exc:
-        app.require_khazna_reload_auth(request_with())
-    assert exc.value.status_code == 503
-
-
-def test_reload_rejects_missing_and_invalid_tokens(monkeypatch):
-    monkeypatch.setenv("KHAZNA_RELOAD_TOKEN", "correct-secret")
-    with pytest.raises(HTTPException) as missing:
-        app.require_khazna_reload_auth(request_with())
-    assert missing.value.status_code == 401
-
-    with pytest.raises(HTTPException) as invalid:
-        app.require_khazna_reload_auth(request_with({
-            app.KHAZNA_RELOAD_HEADER: "wrong-secret",
-        }))
-    assert invalid.value.status_code == 403
-
-
-def test_reload_accepts_header_or_bearer_token(monkeypatch):
-    monkeypatch.setenv("KHAZNA_RELOAD_TOKEN", "correct-secret")
-    app.require_khazna_reload_auth(request_with({
-        app.KHAZNA_RELOAD_HEADER: "correct-secret",
-    }))
-    app.require_khazna_reload_auth(request_with({
-        "Authorization": "Bearer correct-secret",
-    }))
-
-
-def test_reload_endpoint_calls_auth_guard_before_reload():
-    source = inspect.getsource(app.reload_khazna)
-    assert "require_khazna_reload_auth(request)" in source
-    assert source.index("require_khazna_reload_auth(request)") < source.index("khazna_index.reload")
+from __future__ import annotations
+
+import inspect
+
+import pytest
+from fastapi import HTTPException
+from starlette.requests import Request
+
+import app
+
+
+def request_with(headers=None):
+    raw_headers = [
+        (str(key).lower().encode("latin-1"), str(value).encode("latin-1"))
+        for key, value in (headers or {}).items()
+    ]
+    return Request({
+        "type": "http",
+        "method": "POST",
+        "path": "/api/admin/reload-khazna",
+        "headers": raw_headers,
+        "client": ("127.0.0.1", 12345),
+        "server": ("testserver", 80),
+        "scheme": "http",
+        "query_string": b"",
+    })
+
+
+def test_reload_secret_must_be_configured(monkeypatch):
+    monkeypatch.delenv("KHAZNA_RELOAD_TOKEN", raising=False)
+    with pytest.raises(HTTPException) as exc:
+        app.require_khazna_reload_auth(request_with())
+    assert exc.value.status_code == 503
+
+
+def test_reload_rejects_missing_and_invalid_tokens(monkeypatch):
+    monkeypatch.setenv("KHAZNA_RELOAD_TOKEN", "correct-secret")
+    with pytest.raises(HTTPException) as missing:
+        app.require_khazna_reload_auth(request_with())
+    assert missing.value.status_code == 401
+
+    with pytest.raises(HTTPException) as invalid:
+        app.require_khazna_reload_auth(request_with({
+            app.KHAZNA_RELOAD_HEADER: "wrong-secret",
+        }))
+    assert invalid.value.status_code == 403
+
+
+def test_reload_accepts_header_or_bearer_token(monkeypatch):
+    monkeypatch.setenv("KHAZNA_RELOAD_TOKEN", "correct-secret")
+    app.require_khazna_reload_auth(request_with({
+        app.KHAZNA_RELOAD_HEADER: "correct-secret",
+    }))
+    app.require_khazna_reload_auth(request_with({
+        "Authorization": "Bearer correct-secret",
+    }))
+
+
+def test_reload_endpoint_calls_auth_guard_before_reload():
+    source = inspect.getsource(app.reload_khazna)
+    assert "require_khazna_reload_auth(request)" in source
+    assert source.index("require_khazna_reload_auth(request)") < source.index("khazna_index.reload")

tests/test_oaworks_tri_state.py

@@ -1,65 +1,65 @@
-from __future__ import annotations
-
-import oaworks
-
-
-def permission_payload(*, can_archive=True, embargo="absent"):
-    permission = {
-        "can_archive": can_archive,
-        "version": "acceptedVersion",
-        "versions": ["acceptedVersion"],
-        "locations": ["Institutional Repository"],
-        "issuer": {"id": ["1234-5678"]},
-        "score": 100,
-    }
-    if embargo != "absent":
-        permission["embargo_months"] = embargo
-    return {"best_permission": permission, "all_permissions": [permission]}
-
-
-def test_allowed_permission_with_absent_embargo_is_unknown_not_immediate():
-    green = oaworks.parse_permissions(permission_payload(embargo="absent"))
-    assert green["permission_status"] == "allowed"
-    assert green["embargo_status"] == "unknown"
-    recommendation = oaworks.recommendation_from(green)
-    assert recommendation["immediate_open_deposit_possible"] == "Unknown"
-    assert recommendation["embargoed_deposit_needed"] == "Unknown"
-    assert recommendation["metadata_only_first"] == "Unknown"
-
-
-def test_explicit_zero_embargo_allows_immediate_open_deposit():
-    green = oaworks.parse_permissions(permission_payload(embargo=0))
-    assert green["embargo_status"] == "zero"
-    assert green["_embargo_months"] == 0
-    recommendation = oaworks.recommendation_from(green)
-    assert recommendation["immediate_open_deposit_possible"] == "Yes"
-    assert recommendation["embargoed_deposit_needed"] == "No"
-
-
-def test_positive_embargo_requires_delayed_opening():
-    green = oaworks.parse_permissions(permission_payload(embargo=12))
-    assert green["embargo_status"] == "stated"
-    recommendation = oaworks.recommendation_from(green)
-    assert recommendation["immediate_open_deposit_possible"] == "No"
-    assert recommendation["embargoed_deposit_needed"] == "Yes"
-    assert recommendation["best_timing"] == "After a 12-month embargo"
-
-
-def test_unverified_permission_is_unknown_not_denied():
-    green = oaworks.not_confirmed("Example Journal", "1234-5678")
-    assert green["_can_archive"] is None
-    assert green["permission_status"] == "unknown"
-    recommendation = oaworks.recommendation_from(green)
-    assert recommendation["permission_status"] == "unknown"
-    assert recommendation["immediate_open_deposit_possible"] == "Unknown"
-    actions = oaworks.next_actions_from(green, "")
-    assert "could not determine" in actions[0]
-    assert "does not appear to be permitted" not in actions[0]
-
-
-def test_confirmed_denial_remains_denied():
-    green = oaworks.parse_permissions(permission_payload(can_archive=False))
-    assert green["permission_status"] == "denied"
-    recommendation = oaworks.recommendation_from(green)
-    assert recommendation["immediate_open_deposit_possible"] == "No"
-    assert recommendation["permission_status"] == "denied"
+from __future__ import annotations
+
+import oaworks
+
+
+def permission_payload(*, can_archive=True, embargo="absent"):
+    permission = {
+        "can_archive": can_archive,
+        "version": "acceptedVersion",
+        "versions": ["acceptedVersion"],
+        "locations": ["Institutional Repository"],
+        "issuer": {"id": ["1234-5678"]},
+        "score": 100,
+    }
+    if embargo != "absent":
+        permission["embargo_months"] = embargo
+    return {"best_permission": permission, "all_permissions": [permission]}
+
+
+def test_allowed_permission_with_absent_embargo_is_unknown_not_immediate():
+    green = oaworks.parse_permissions(permission_payload(embargo="absent"))
+    assert green["permission_status"] == "allowed"
+    assert green["embargo_status"] == "unknown"
+    recommendation = oaworks.recommendation_from(green)
+    assert recommendation["immediate_open_deposit_possible"] == "Unknown"
+    assert recommendation["embargoed_deposit_needed"] == "Unknown"
+    assert recommendation["metadata_only_first"] == "Unknown"
+
+
+def test_explicit_zero_embargo_allows_immediate_open_deposit():
+    green = oaworks.parse_permissions(permission_payload(embargo=0))
+    assert green["embargo_status"] == "zero"
+    assert green["_embargo_months"] == 0
+    recommendation = oaworks.recommendation_from(green)
+    assert recommendation["immediate_open_deposit_possible"] == "Yes"
+    assert recommendation["embargoed_deposit_needed"] == "No"
+
+
+def test_positive_embargo_requires_delayed_opening():
+    green = oaworks.parse_permissions(permission_payload(embargo=12))
+    assert green["embargo_status"] == "stated"
+    recommendation = oaworks.recommendation_from(green)
+    assert recommendation["immediate_open_deposit_possible"] == "No"
+    assert recommendation["embargoed_deposit_needed"] == "Yes"
+    assert recommendation["best_timing"] == "After a 12-month embargo"
+
+
+def test_unverified_permission_is_unknown_not_denied():
+    green = oaworks.not_confirmed("Example Journal", "1234-5678")
+    assert green["_can_archive"] is None
+    assert green["permission_status"] == "unknown"
+    recommendation = oaworks.recommendation_from(green)
+    assert recommendation["permission_status"] == "unknown"
+    assert recommendation["immediate_open_deposit_possible"] == "Unknown"
+    actions = oaworks.next_actions_from(green, "")
+    assert "could not determine" in actions[0]
+    assert "does not appear to be permitted" not in actions[0]
+
+
+def test_confirmed_denial_remains_denied():
+    green = oaworks.parse_permissions(permission_payload(can_archive=False))
+    assert green["permission_status"] == "denied"
+    recommendation = oaworks.recommendation_from(green)
+    assert recommendation["immediate_open_deposit_possible"] == "No"
+    assert recommendation["permission_status"] == "denied"

tests/test_repository_schema.py

@@ -1,62 +1,62 @@
-from __future__ import annotations
-
-from prompts import REPO_SYSTEM_PROMPT
-from repositories import hydrate_deposit_strategy, hydrate_selection
-
-
-def test_repository_prompt_requires_exact_id_minimal_schema():
-    assert '"id": "exact registry id"' in REPO_SYSTEM_PROMPT
-    assert "Every repositories item MUST contain an exact registry id" in REPO_SYSTEM_PROMPT
-    assert "id, fit_reason, fit_confidence" in REPO_SYSTEM_PROMPT
-
-
-def test_hydration_requires_id_and_discards_model_facts():
-    selected = [
-        {
-            "id": "zenodo",
-            "fit_reason": "Good multidisciplinary fallback for these files.",
-            "fit_confidence": "High",
-            # All of these must be ignored in favour of the registry.
-            "name": "Fake Zenodo",
-            "url": "javascript:alert(1)",
-            "cost": "Expensive",
-            "access_model": "Invented",
-            "licences_supported": "Invented",
-        },
-        {"name": "Figshare", "fit_reason": "Missing id must not hydrate."},
-        {"id": "not-in-registry", "fit_reason": "Unknown id must not hydrate."},
-        {"id": "zenodo", "fit_reason": "Duplicate must not hydrate twice."},
-    ]
-
-    hydrated = hydrate_selection(selected)
-    assert len(hydrated) == 1
-    repository = hydrated[0]
-    assert repository["id"] == "zenodo"
-    assert repository["name"] == "Zenodo"
-    assert repository["url"] == "https://zenodo.org"
-    assert repository["cost"] == "Free"
-    assert repository["fit_reason"].startswith("Good multidisciplinary")
-    assert repository["fit_confidence"] == "High"
-    assert "access_model" not in repository
-    assert "licences_supported" not in repository
-
-
-def test_deposit_strategy_resolves_only_hydrated_ids():
-    repositories = hydrate_selection([
-        {"id": "zenodo", "fit_reason": "Primary", "fit_confidence": "High"},
-        {"id": "figshare", "fit_reason": "Secondary", "fit_confidence": "Medium"},
-    ])
-    strategy = hydrate_deposit_strategy(
-        {
-            "primary_repository_id": "zenodo",
-            "secondary_repository_id": "invented",
-            "metadata_standard": "Check repository guidance",
-            "deposit_workflow": ["Review terms", "Deposit files"],
-        },
-        repositories,
-    )
-    assert strategy["primary_repository_id"] == "zenodo"
-    assert strategy["primary_choice"] == "Zenodo"
-    # Invalid model id falls back to the next actually hydrated result.
-    assert strategy["secondary_repository_id"] == "figshare"
-    assert strategy["secondary_choice"] == "Figshare"
+from __future__ import annotations
+
+from prompts import REPO_SYSTEM_PROMPT
+from repositories import hydrate_deposit_strategy, hydrate_selection
+
+
+def test_repository_prompt_requires_exact_id_minimal_schema():
+    assert '"id": "exact registry id"' in REPO_SYSTEM_PROMPT
+    assert "Every repositories item MUST contain an exact registry id" in REPO_SYSTEM_PROMPT
+    assert "id, fit_reason, fit_confidence" in REPO_SYSTEM_PROMPT
+
+
+def test_hydration_requires_id_and_discards_model_facts():
+    selected = [
+        {
+            "id": "zenodo",
+            "fit_reason": "Good multidisciplinary fallback for these files.",
+            "fit_confidence": "High",
+            # All of these must be ignored in favour of the registry.
+            "name": "Fake Zenodo",
+            "url": "javascript:alert(1)",
+            "cost": "Expensive",
+            "access_model": "Invented",
+            "licences_supported": "Invented",
+        },
+        {"name": "Figshare", "fit_reason": "Missing id must not hydrate."},
+        {"id": "not-in-registry", "fit_reason": "Unknown id must not hydrate."},
+        {"id": "zenodo", "fit_reason": "Duplicate must not hydrate twice."},
+    ]
+
+    hydrated = hydrate_selection(selected)
+    assert len(hydrated) == 1
+    repository = hydrated[0]
+    assert repository["id"] == "zenodo"
+    assert repository["name"] == "Zenodo"
+    assert repository["url"] == "https://zenodo.org"
+    assert repository["cost"] == "Free"
+    assert repository["fit_reason"].startswith("Good multidisciplinary")
+    assert repository["fit_confidence"] == "High"
+    assert "access_model" not in repository
+    assert "licences_supported" not in repository
+
+
+def test_deposit_strategy_resolves_only_hydrated_ids():
+    repositories = hydrate_selection([
+        {"id": "zenodo", "fit_reason": "Primary", "fit_confidence": "High"},
+        {"id": "figshare", "fit_reason": "Secondary", "fit_confidence": "Medium"},
+    ])
+    strategy = hydrate_deposit_strategy(
+        {
+            "primary_repository_id": "zenodo",
+            "secondary_repository_id": "invented",
+            "metadata_standard": "Check repository guidance",
+            "deposit_workflow": ["Review terms", "Deposit files"],
+        },
+        repositories,
+    )
+    assert strategy["primary_repository_id"] == "zenodo"
+    assert strategy["primary_choice"] == "Zenodo"
+    # Invalid model id falls back to the next actually hydrated result.
+    assert strategy["secondary_repository_id"] == "figshare"
+    assert strategy["secondary_choice"] == "Figshare"

tests/test_subject_quartiles.py

@@ -1,38 +1,38 @@
-from __future__ import annotations
-
-import app
-
-
-def test_subject_result_includes_all_field_quartiles(monkeypatch):
-    monkeypatch.setattr(app, "lookup_any", lambda _issns: {
-        "matched_issn": "1234-5678",
-        "quartile": "Q1",
-        "h_index": "88",
-        "publisher": "Example Publisher",
-        "open_access": "Yes",
-        "country": "UAE",
-        "categories": "Engineering (Q1); Energy (Q2)",
-        "sourceid": "999",
-        "quartiles_by_field": [
-            {"code": "2201", "field": "Engineering", "quartile": "Q1"},
-            {"code": "2102", "field": "Energy", "quartile": "Q2"},
-        ],
-        "state": "indexed_ranked",
-        "currently_indexed": True,
-        "warn": False,
-    })
-    monkeypatch.setattr(app, "build_verify_links", lambda *_args: {"scimago": "https://example.test"})
-    monkeypatch.setattr(app, "is_ku_apc_covered", lambda *_args: False)
-
-    result = app.build_subject_journal_result({
-        "title": "Example Journal",
-        "issn": "1234-5678",
-        "quartile": "Q1",
-        "h_index": 80,
-    })
-    assert result["quartile"] == "Q1"
-    assert result["quartiles_by_field"] == [
-        {"code": "2201", "field": "Engineering", "quartile": "Q1"},
-        {"code": "2102", "field": "Energy", "quartile": "Q2"},
-    ]
-    assert result["h_index"] == 88
+from __future__ import annotations
+
+import app
+
+
+def test_subject_result_includes_all_field_quartiles(monkeypatch):
+    monkeypatch.setattr(app, "lookup_any", lambda _issns: {
+        "matched_issn": "1234-5678",
+        "quartile": "Q1",
+        "h_index": "88",
+        "publisher": "Example Publisher",
+        "open_access": "Yes",
+        "country": "UAE",
+        "categories": "Engineering (Q1); Energy (Q2)",
+        "sourceid": "999",
+        "quartiles_by_field": [
+            {"code": "2201", "field": "Engineering", "quartile": "Q1"},
+            {"code": "2102", "field": "Energy", "quartile": "Q2"},
+        ],
+        "state": "indexed_ranked",
+        "currently_indexed": True,
+        "warn": False,
+    })
+    monkeypatch.setattr(app, "build_verify_links", lambda *_args: {"scimago": "https://example.test"})
+    monkeypatch.setattr(app, "is_ku_apc_covered", lambda *_args: False)
+
+    result = app.build_subject_journal_result({
+        "title": "Example Journal",
+        "issn": "1234-5678",
+        "quartile": "Q1",
+        "h_index": 80,
+    })
+    assert result["quartile"] == "Q1"
+    assert result["quartiles_by_field"] == [
+        {"code": "2201", "field": "Engineering", "quartile": "Q1"},
+        {"code": "2102", "field": "Energy", "quartile": "Q2"},
+    ]
+    assert result["h_index"] == 88

tests/test_trusted_links.py

@@ -0,0 +1,146 @@
+"""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()

tests/test_trusted_links_api.py

@@ -0,0 +1,23 @@
+"""Framework integration tests, executed by the Docker pytest build gate."""
+from fastapi.testclient import TestClient
+import app
+
+
+def test_discovery_endpoint_accepts_identity_only(monkeypatch):
+    captured=[]
+    async def fake_discover(value):
+        captured.append(value)
+        return {'homepage':{'url':None},'guidelines':{'url':None,'status':'unavailable'}}
+    monkeypatch.setattr(app,'discover',fake_discover)
+    monkeypatch.setattr(app,'check_rate_limit',lambda *args:True)
+    response=TestClient(app.app).post('/api/journal-links',json={
+        'scopus_id':'21206','url':'https://127.0.0.1/private'})
+    assert response.status_code==200
+    assert captured==[{'scopus_id':'21206','issn':''}]
+    assert response.json()['guidelines']['status']=='unavailable'
+
+
+def test_discovery_endpoint_rate_limited(monkeypatch):
+    monkeypatch.setattr(app,'check_rate_limit',lambda *args:False)
+    response=TestClient(app.app).post('/api/journal-links',json={'scopus_id':'21206'})
+    assert response.status_code==429