test_app.py

"""Tests for WPIRIresolve app."""

import pytest
import responses

from app import app, _cache, SPARQL_ENDPOINT, linkify_iris, safe_int_env


SAMPLE_TURTLE = """\
@prefix wp: <http://vocabularies.wikipathways.org/wp#> .
<http://rdf.wikipathways.org/Pathway/WP4846_r140186>
    a wp:Pathway ;
    wp:organism <http://purl.obolibrary.org/obo/NCBITaxon_9606> .
"""

EMPTY_TURTLE = "# Empty TURTLE\n"


@pytest.fixture(autouse=True)
def clear_cache():
    """Clear the in-memory cache before each test."""
    _cache.clear()
    yield
    _cache.clear()


@pytest.fixture
def client():
    app.config["TESTING"] = True
    with app.test_client() as c:
        yield c


# ---------------------------------------------------------------------------
# Root redirect
# ---------------------------------------------------------------------------
def test_root_redirects(client):
    resp = client.get("/")
    assert resp.status_code == 302
    assert resp.headers["Location"] == "https://www.wikipathways.org/rdf.html"


# Landing / about page
# ---------------------------------------------------------------------------
def test_about_page(client):
    resp = client.get("/about")
    assert resp.status_code == 200
    assert b"WikiPathways IRI Resolver" in resp.data
    assert b"text/html" in resp.content_type.encode()


# ---------------------------------------------------------------------------
# Health check
# ---------------------------------------------------------------------------
def test_health(client):
    resp = client.get("/_health")
    assert resp.status_code == 200
    data = resp.get_json()
    assert data["status"] == "ok"


@responses.activate
def test_health_with_sparql_check_ok(client):
    responses.get(SPARQL_ENDPOINT, json={"boolean": True}, status=200)
    resp = client.get("/_health?check_sparql=true")
    assert resp.status_code == 200
    data = resp.get_json()
    assert data["sparql"] == "ok"


@responses.activate
def test_health_with_sparql_check_fail(client):
    responses.get(SPARQL_ENDPOINT, body="error", status=500)
    resp = client.get("/_health?check_sparql=true")
    assert resp.status_code == 503
    data = resp.get_json()
    assert "error" in data["sparql"]


# ---------------------------------------------------------------------------
# Content negotiation — HTML (default)
# ---------------------------------------------------------------------------
@responses.activate
def test_resolve_html_default(client):
    responses.get(SPARQL_ENDPOINT, body=SAMPLE_TURTLE, status=200)
    resp = client.get("/Pathway/WP4846_r140186")
    assert resp.status_code == 200
    assert "text/html" in resp.content_type
    # Pygments wraps tokens in spans, so check for text that survives highlighting
    assert b"Pathway" in resp.data
    assert b"@prefix" in resp.data


@responses.activate
def test_html_contains_clickable_links(client):
    responses.get(SPARQL_ENDPOINT, body=SAMPLE_TURTLE, status=200)
    resp = client.get("/Pathway/WP4846_r140186")
    html = resp.get_data(as_text=True)
    # Internal link should be relative
    assert 'href="/Pathway/WP4846_r140186"' in html
    # External link should be absolute
    assert 'href="http://purl.obolibrary.org/' in html


# ---------------------------------------------------------------------------
# Content negotiation — Turtle via Accept header
# ---------------------------------------------------------------------------
@responses.activate
def test_resolve_turtle_accept(client):
    responses.get(SPARQL_ENDPOINT, body=SAMPLE_TURTLE, status=200)
    resp = client.get("/Pathway/WP4846_r140186", headers={"Accept": "text/turtle"})
    assert resp.status_code == 200
    assert "text/turtle" in resp.content_type
    assert b"wp:Pathway" in resp.data


# ---------------------------------------------------------------------------
# Content negotiation — file extensions
# ---------------------------------------------------------------------------
@responses.activate
def test_resolve_ttl_extension(client):
    responses.get(SPARQL_ENDPOINT, body=SAMPLE_TURTLE, status=200)
    resp = client.get("/Pathway/WP4846_r140186.ttl")
    assert resp.status_code == 200
    assert "text/turtle" in resp.content_type


@responses.activate
def test_resolve_jsonld_extension(client):
    jsonld_body = '{"@context": {}}'
    responses.get(SPARQL_ENDPOINT, body=jsonld_body, status=200)
    resp = client.get("/Pathway/WP4846_r140186.jsonld")
    assert resp.status_code == 200
    assert "application/ld+json" in resp.content_type


@responses.activate
def test_resolve_rdf_extension(client):
    rdfxml_body = (
        '<rdf:RDF xmlns:rdf="http://www.w3.org/1999/02/22-rdf-syntax-ns#"></rdf:RDF>'
    )
    responses.get(SPARQL_ENDPOINT, body=rdfxml_body, status=200)
    resp = client.get("/Pathway/WP4846_r140186.rdf")
    assert resp.status_code == 200
    assert "application/rdf+xml" in resp.content_type


@responses.activate
def test_resolve_nt_extension(client):
    nt_body = "<http://example.org/s> <http://example.org/p> <http://example.org/o> .\n"
    responses.get(SPARQL_ENDPOINT, body=nt_body, status=200)
    resp = client.get("/Pathway/WP4846_r140186.nt")
    assert resp.status_code == 200
    assert "application/n-triples" in resp.content_type


# ---------------------------------------------------------------------------
# Empty results → 404
# ---------------------------------------------------------------------------
@responses.activate
def test_empty_result_returns_404(client):
    responses.get(SPARQL_ENDPOINT, body=EMPTY_TURTLE, status=200)
    resp = client.get("/Nonexistent/Resource")
    assert resp.status_code == 404


@responses.activate
def test_empty_result_404_html(client):
    responses.get(SPARQL_ENDPOINT, body=EMPTY_TURTLE, status=200)
    resp = client.get("/Nonexistent/Resource")
    assert resp.status_code == 404
    assert b"No triples found" in resp.data


@responses.activate
def test_empty_result_404_turtle(client):
    responses.get(SPARQL_ENDPOINT, body=EMPTY_TURTLE, status=200)
    resp = client.get("/Nonexistent/Resource", headers={"Accept": "text/turtle"})
    assert resp.status_code == 404
    assert b"# Error 404" in resp.data


# ---------------------------------------------------------------------------
# SPARQL errors
# ---------------------------------------------------------------------------
@responses.activate
def test_sparql_connection_error_502(client):
    from requests.exceptions import ConnectionError as RequestsConnectionError

    responses.get(SPARQL_ENDPOINT, body=RequestsConnectionError("refused"))
    resp = client.get("/Pathway/WP4846_r140186")
    assert resp.status_code == 502


@responses.activate
def test_sparql_timeout_504(client):
    from requests.exceptions import ReadTimeout

    responses.get(SPARQL_ENDPOINT, body=ReadTimeout("timed out"))
    resp = client.get("/Pathway/WP4846_r140186")
    assert resp.status_code == 504


@responses.activate
def test_sparql_http_error_502(client):
    responses.get(SPARQL_ENDPOINT, body="Internal Server Error", status=500)
    resp = client.get("/Pathway/WP4846_r140186")
    assert resp.status_code == 502


# ---------------------------------------------------------------------------
# CORS headers
# ---------------------------------------------------------------------------
@responses.activate
def test_cors_headers(client):
    responses.get(SPARQL_ENDPOINT, body=SAMPLE_TURTLE, status=200)
    resp = client.get("/Pathway/WP4846_r140186")
    assert resp.headers["Access-Control-Allow-Origin"] == "*"
    assert "Accept" in resp.headers["Access-Control-Allow-Headers"]


# ---------------------------------------------------------------------------
# HTTP caching headers
# ---------------------------------------------------------------------------
@responses.activate
def test_vary_header(client):
    responses.get(SPARQL_ENDPOINT, body=SAMPLE_TURTLE, status=200)
    resp = client.get("/Pathway/WP4846_r140186")
    assert "Accept" in resp.headers.get("Vary", "")


@responses.activate
def test_etag_and_cache_control(client):
    responses.get(SPARQL_ENDPOINT, body=SAMPLE_TURTLE, status=200)
    resp = client.get("/Pathway/WP4846_r140186")
    assert "ETag" in resp.headers
    assert "max-age=" in resp.headers.get("Cache-Control", "")


@responses.activate
def test_if_none_match_304(client):
    responses.get(SPARQL_ENDPOINT, body=SAMPLE_TURTLE, status=200)
    # First request to get the ETag
    resp1 = client.get("/Pathway/WP4846_r140186")
    etag = resp1.headers["ETag"]
    # Second request with If-None-Match
    resp2 = client.get("/Pathway/WP4846_r140186", headers={"If-None-Match": etag})
    assert resp2.status_code == 304


# ---------------------------------------------------------------------------
# Caching
# ---------------------------------------------------------------------------
@responses.activate
def test_sparql_response_cached(client):
    responses.get(SPARQL_ENDPOINT, body=SAMPLE_TURTLE, status=200)
    # First request hits SPARQL
    resp1 = client.get("/Pathway/WP4846_r140186")
    assert resp1.status_code == 200
    # Second request should use cache (only one SPARQL call was made)
    resp2 = client.get("/Pathway/WP4846_r140186")
    assert resp2.status_code == 200
    assert len(responses.calls) == 1


# ---------------------------------------------------------------------------
# Metrics endpoint
# ---------------------------------------------------------------------------
def test_metrics_endpoint(client):
    resp = client.get("/metrics")
    assert resp.status_code == 200
    assert b"wpiri_" in resp.data


# ---------------------------------------------------------------------------
# Rate limiting (429)
# ---------------------------------------------------------------------------
def test_rate_limit_handler(client):
    """The 429 error handler returns a proper response."""
    with app.test_request_context():
        from app import handle_429

        resp = handle_429(None)
        assert resp.status_code == 429
        assert b"Rate limit" in resp.data


# ---------------------------------------------------------------------------
# TEST-01: linkify_iris edge cases
# ---------------------------------------------------------------------------
def test_linkify_single_external_iri():
    result = linkify_iris("See http://example.org/thing for details")
    assert 'href="http://example.org/thing"' in result
    assert 'rel="noopener"' in result


def test_linkify_multiple_iris_same_line():
    result = linkify_iris("a http://example.org/a b http://example.org/b")
    assert 'href="http://example.org/a"' in result
    assert 'href="http://example.org/b"' in result


def test_linkify_internal_iri_becomes_relative():
    from app import BASE_IRI

    result = linkify_iris(f"{BASE_IRI}/Pathway/WP1234")
    assert 'href="/Pathway/WP1234"' in result
    assert 'rel="noopener"' not in result


def test_linkify_external_iri_has_noopener():
    result = linkify_iris("http://external.org/thing")
    assert 'rel="noopener"' in result


def test_linkify_semicolon_stops_iri():
    # Regex [^\s<>"&;]+ stops at semicolon — documented boundary behavior
    result = linkify_iris("http://example.org/thing; more text")
    assert 'href="http://example.org/thing"' in result
    assert "; more text" in result


def test_linkify_period_included_in_href():
    # Current behavior: trailing period is NOT stripped from the matched IRI
    result = linkify_iris("See http://example.org/thing.")
    assert 'href="http://example.org/thing."' in result


def test_linkify_ampersand_in_html_entity_stops_iri():
    # &gt; contains & which stops the regex — angle bracket entities work correctly
    result = linkify_iris("&lt;http://example.org/thing&gt;")
    assert 'href="http://example.org/thing"' in result


# ---------------------------------------------------------------------------
# TEST-03: safe_int_env environment variable parsing
# ---------------------------------------------------------------------------
def test_safe_int_env_valid_positive(monkeypatch):
    monkeypatch.setenv("SPARQL_TIMEOUT", "45")
    assert safe_int_env("SPARQL_TIMEOUT", 30) == 45


def test_safe_int_env_non_integer_returns_default(monkeypatch):
    monkeypatch.setenv("SPARQL_TIMEOUT", "not_a_number")
    assert safe_int_env("SPARQL_TIMEOUT", 30) == 30


def test_safe_int_env_negative_returns_default(monkeypatch):
    monkeypatch.setenv("SPARQL_TIMEOUT", "-5")
    assert safe_int_env("SPARQL_TIMEOUT", 30) == 30


def test_safe_int_env_missing_returns_default(monkeypatch):
    monkeypatch.delenv("SPARQL_TIMEOUT", raising=False)
    assert safe_int_env("SPARQL_TIMEOUT", 30) == 30


def test_safe_int_env_zero_returns_zero(monkeypatch):
    # Zero is a valid non-negative integer (timeout of 0 is allowed by requests library)
    monkeypatch.setenv("SPARQL_TIMEOUT", "0")
    assert safe_int_env("SPARQL_TIMEOUT", 30) == 0


def test_safe_int_env_float_string_returns_default(monkeypatch):
    # "3.5" cannot be parsed by int() — must return default
    monkeypatch.setenv("CACHE_TTL", "3.5")
    assert safe_int_env("CACHE_TTL", 300) == 300


# ---------------------------------------------------------------------------
# TEST-02: Non-Turtle format 404 responses
# ---------------------------------------------------------------------------
@responses.activate
def test_empty_result_jsonld_accept_returns_404(client):
    responses.get(SPARQL_ENDPOINT, body=EMPTY_TURTLE, status=200)
    resp = client.get(
        "/Nonexistent/Resource", headers={"Accept": "application/ld+json"}
    )
    assert resp.status_code == 404
    # Via Accept header: error_response returns text/turtle comment body
    assert b"# Error 404" in resp.data


@responses.activate
def test_empty_result_rdfxml_accept_returns_404(client):
    responses.get(SPARQL_ENDPOINT, body=EMPTY_TURTLE, status=200)
    resp = client.get(
        "/Nonexistent/Resource", headers={"Accept": "application/rdf+xml"}
    )
    assert resp.status_code == 404
    assert b"# Error 404" in resp.data


@responses.activate
def test_empty_result_nt_accept_returns_404(client):
    responses.get(SPARQL_ENDPOINT, body=EMPTY_TURTLE, status=200)
    resp = client.get(
        "/Nonexistent/Resource", headers={"Accept": "application/n-triples"}
    )
    assert resp.status_code == 404
    assert b"# Error 404" in resp.data


@responses.activate
def test_empty_result_jsonld_extension_returns_404_html(client):
    # Extension is stripped before error_response(); no Accept header set by test client.
    # negotiate_media_type() falls back to HTML — confirmed behavior per RESEARCH.md Pitfall 2.
    responses.get(SPARQL_ENDPOINT, body=EMPTY_TURTLE, status=200)
    resp = client.get("/Nonexistent/Resource.jsonld")
    assert resp.status_code == 404
    assert "text/html" in resp.content_type


# ---------------------------------------------------------------------------
# TEST-05: HEAD and OPTIONS HTTP compliance
# ---------------------------------------------------------------------------
@responses.activate
def test_head_request_returns_200_no_body(client):
    # Flask runs full view for HEAD but strips body — SPARQL mock required (see RESEARCH.md Pitfall 3)
    responses.get(SPARQL_ENDPOINT, body=SAMPLE_TURTLE, status=200)
    resp = client.head("/Pathway/WP4846_r140186")
    assert resp.status_code == 200
    assert len(resp.data) == 0


@responses.activate
def test_head_request_returns_etag_and_content_type(client):
    responses.get(SPARQL_ENDPOINT, body=SAMPLE_TURTLE, status=200)
    resp = client.head("/Pathway/WP4846_r140186")
    assert resp.status_code == 200
    assert "ETag" in resp.headers
    assert "text/html" in resp.content_type


def test_options_request_returns_200(client):
    resp = client.options("/Pathway/WP4846_r140186")
    assert resp.status_code == 200


def test_options_request_cors_headers(client):
    resp = client.options("/Pathway/WP4846_r140186")
    assert resp.headers.get("Access-Control-Allow-Origin") == "*"
    assert "GET" in resp.headers.get("Access-Control-Allow-Methods", "")
    assert "HEAD" in resp.headers.get("Access-Control-Allow-Methods", "")
    assert "OPTIONS" in resp.headers.get("Access-Control-Allow-Methods", "")
    assert "Accept" in resp.headers.get("Access-Control-Allow-Headers", "")