" not in result
- assert "Hello" in result
- assert "world" in result
-
- def test_removes_navigation_artifacts(self):
- """Should remove navigation/menu text artifacts."""
- content = "Get Started with Kubeflow. Home Menu Navigation overview."
- result = clean_content(content)
- assert "Get Started" not in result
- assert "Menu" not in result
- assert "Navigation" not in result
- assert "Kubeflow" in result
-
- def test_removes_urls(self):
- """Should remove HTTP/HTTPS URLs."""
- content = "Visit https://kubeflow.org/docs for more info."
- result = clean_content(content)
- assert "https://kubeflow.org" not in result
- assert "Visit" in result
- assert "for more info." in result
-
- def test_converts_markdown_links_to_text(self):
- """Should convert [text](url) markdown links to just text."""
- content = "See [the documentation](https://kubeflow.org/docs) for details."
- result = clean_content(content)
- assert "the documentation" in result
- assert "(https://kubeflow.org/docs)" not in result
-
- def test_normalizes_whitespace(self):
- """Should collapse multiple spaces into single space."""
- content = "Hello world test"
- result = clean_content(content)
- assert " " not in result
- assert "Hello world test" in result
-
- def test_strips_leading_trailing_whitespace(self):
- """Should strip leading and trailing whitespace."""
- content = " Hello world "
- result = clean_content(content)
- assert result == "Hello world"
-
- def test_returns_empty_for_frontmatter_only(self):
- """Content that is only frontmatter should return empty string."""
- content = "---\ntitle: Nothing\ndate: 2026-01-01\n---"
- result = clean_content(content)
- assert result == ""
-
- def test_preserves_meaningful_content(self):
- """Should not remove meaningful documentation content."""
- content = (
- "Kubeflow Pipelines is a platform for building and deploying "
- "portable, scalable ML workflows based on Docker containers."
- )
- result = clean_content(content)
- assert "Kubeflow Pipelines" in result
- assert "ML workflows" in result
- assert "Docker containers" in result
-
- def test_handles_multiline_html_comment(self):
- """Should remove HTML comments that span multiple lines."""
- content = "Before\n\nAfter"
- result = clean_content(content)
- assert "multi-line comment" not in result
- assert "Before" in result
- assert "After" in result
-
- def test_handles_multiline_hugo_template(self):
- """Should remove Hugo templates that span multiple lines."""
- content = "Before {{ if .IsHome }}\nstuff\n{{ end }} After"
- result = clean_content(content)
- assert "{{ if" not in result
- assert "{{ end }}" not in result
-
- def test_real_world_kubeflow_doc_snippet(self):
- """Test with a realistic Kubeflow documentation snippet."""
- content = """---
-title: KServe Overview
-description: Overview of KServe
-weight: 1
----
-
-
-
-
-
-## What is KServe?
-
-[KServe](https://kserve.github.io/website/) enables serverless inference
-on Kubernetes. Visit https://kubeflow.org/docs/external-add-ons/kserve/
-for more details.
-
-{{ partial "section-index.html" . }}
-
-
-"""
- result = clean_content(content)
- assert "title:" not in result
- assert "auto-generated" not in result
- assert "section-index" not in result
- assert "KServe" in result
- assert "serverless inference" in result
- assert "Kubernetes" in result
-
-
-class TestCleanContentEdgeCases:
- """Edge cases for content cleaning."""
-
- def test_empty_string(self):
- """Should handle empty string input."""
- assert clean_content("") == ""
-
- def test_whitespace_only(self):
- """Should handle whitespace-only input."""
- assert clean_content(" \n\n \t ") == ""
-
- def test_no_cleaning_needed(self):
- """Plain text should pass through with minimal changes."""
- content = "Simple plain text with no special formatting."
- result = clean_content(content)
- assert result == content
-
- def test_nested_html_tags(self):
- """Should handle deeply nested HTML tags."""
- content = ""
- result = clean_content(content)
- assert "Deep content" in result
- assert "<" not in result
-
-
-class TestTruncateForTei:
- def test_truncates_long_text(self):
- long = "word " * 500
- assert len(truncate_for_tei(long)) == 1000
-
-
-class TestEmbedTexts:
- def test_batches_requests(self, monkeypatch):
- calls = []
-
- class FakeResponse:
- def raise_for_status(self):
- return None
-
- def json(self):
- return [[0.1, 0.2], [0.3, 0.4]]
-
- def fake_post(url, json, headers, timeout):
- calls.append((url, len(json["inputs"]), [len(t) for t in json["inputs"]]))
- return FakeResponse()
-
- monkeypatch.setattr("utils.requests.post", fake_post)
-
- vectors = embed_texts(["a", "b"], "http://embeddings/embed", batch_size=2)
- assert len(vectors) == 2
- assert calls[0][1] == 2
-
- def test_truncates_inputs_for_tei(self, monkeypatch):
- sent = []
-
- class FakeResponse:
- def raise_for_status(self):
- return None
-
- def json(self):
- return [[0.1]]
-
- def fake_post(url, json, headers, timeout):
- sent.extend(json["inputs"])
- return FakeResponse()
-
- monkeypatch.setattr("utils.requests.post", fake_post)
- long = "x" * 5000
- embed_texts([long], "http://embeddings/embed", batch_size=1)
- assert len(sent[0]) == 1000
+"""Tests for pipeline utility functions and component logic."""
+
+import sys
+from pathlib import Path
+
+# Add pipelines directory to path
+PIPELINES_DIR = Path(__file__).parent.parent / "kagent-feast-mcp" / "pipelines"
+sys.path.insert(0, str(PIPELINES_DIR))
+
+from utils import clean_content, embed_texts, resolve_github_token, truncate_for_tei
+
+
+class TestResolveGithubToken:
+ """Tests for resolve_github_token."""
+
+ def test_prefers_explicit_parameter(self, monkeypatch):
+ monkeypatch.setenv("Github_Pat", "env-pat")
+ monkeypatch.setenv("GITHUB_TOKEN", "fallback")
+ assert resolve_github_token("param-pat") == "param-pat"
+
+ def test_falls_back_to_github_pat_env(self, monkeypatch):
+ monkeypatch.setenv("Github_Pat", "env-pat")
+ monkeypatch.setenv("GITHUB_TOKEN", "fallback")
+ assert resolve_github_token("") == "env-pat"
+
+ def test_falls_back_to_github_token_env(self, monkeypatch):
+ monkeypatch.delenv("Github_Pat", raising=False)
+ monkeypatch.setenv("GITHUB_TOKEN", "fallback")
+ assert resolve_github_token("") == "fallback"
+
+ def test_returns_empty_when_unset(self, monkeypatch):
+ monkeypatch.delenv("Github_Pat", raising=False)
+ monkeypatch.delenv("GITHUB_TOKEN", raising=False)
+ assert resolve_github_token("") == ""
+ assert resolve_github_token(" ") == ""
+
+
+class TestCleanContent:
+ """Tests for the clean_content utility function."""
+
+ def test_removes_yaml_frontmatter(self):
+ """Should remove --- delimited YAML frontmatter."""
+ content = "---\ntitle: Test\ndate: 2026-01-01\n---\n\nActual content here."
+ result = clean_content(content)
+ assert "title:" not in result
+ assert "Actual content here." in result
+
+ def test_removes_toml_frontmatter(self):
+ """Should remove +++ delimited TOML frontmatter."""
+ content = "+++\ntitle = 'Test'\n+++\n\nActual content here."
+ result = clean_content(content)
+ assert "title =" not in result
+ assert "Actual content here." in result
+
+ def test_removes_hugo_template_syntax(self):
+ """Should remove {{ ... }} Hugo template expressions."""
+ content = 'Install with {{ .Get "name" }} command. Then run it.'
+ result = clean_content(content)
+ assert "{{" not in result
+ assert "Install with" in result
+ assert "command" in result
+
+ def test_removes_html_comments(self):
+ """Should remove HTML comments."""
+ content = "Before After"
+ result = clean_content(content)
+ assert "this is a comment" not in result
+ assert "Before" in result
+ assert "After" in result
+
+ def test_removes_html_tags(self):
+ """Should remove HTML tags but keep their text content."""
+ content = ""
+ result = clean_content(content)
+ assert "" not in result
+ assert "Hello" in result
+ assert "world" in result
+
+ def test_removes_navigation_artifacts(self):
+ """Should remove navigation/menu text artifacts."""
+ content = "Get Started with Kubeflow. Home Menu Navigation overview."
+ result = clean_content(content)
+ assert "Get Started" not in result
+ assert "Menu" not in result
+ assert "Navigation" not in result
+ assert "Kubeflow" in result
+
+ def test_removes_urls(self):
+ """Should remove HTTP/HTTPS URLs."""
+ content = "Visit https://kubeflow.org/docs for more info."
+ result = clean_content(content)
+ assert "https://kubeflow.org" not in result
+ assert "Visit" in result
+ assert "for more info." in result
+
+ def test_converts_markdown_links_to_text(self):
+ """Should convert [text](url) markdown links to just text."""
+ content = "See [the documentation](https://kubeflow.org/docs) for details."
+ result = clean_content(content)
+ assert "the documentation" in result
+ assert "(https://kubeflow.org/docs)" not in result
+
+ def test_normalizes_whitespace(self):
+ """Should collapse multiple spaces into single space."""
+ content = "Hello world test"
+ result = clean_content(content)
+ assert " " not in result
+ assert "Hello world test" in result
+
+ def test_strips_leading_trailing_whitespace(self):
+ """Should strip leading and trailing whitespace."""
+ content = " Hello world "
+ result = clean_content(content)
+ assert result == "Hello world"
+
+ def test_returns_empty_for_frontmatter_only(self):
+ """Content that is only frontmatter should return empty string."""
+ content = "---\ntitle: Nothing\ndate: 2026-01-01\n---"
+ result = clean_content(content)
+ assert result == ""
+
+ def test_preserves_meaningful_content(self):
+ """Should not remove meaningful documentation content."""
+ content = (
+ "Kubeflow Pipelines is a platform for building and deploying "
+ "portable, scalable ML workflows based on Docker containers."
+ )
+ result = clean_content(content)
+ assert "Kubeflow Pipelines" in result
+ assert "ML workflows" in result
+ assert "Docker containers" in result
+
+ def test_handles_multiline_html_comment(self):
+ """Should remove HTML comments that span multiple lines."""
+ content = "Before\n\nAfter"
+ result = clean_content(content)
+ assert "multi-line comment" not in result
+ assert "Before" in result
+ assert "After" in result
+
+ def test_handles_multiline_hugo_template(self):
+ """Should remove Hugo templates that span multiple lines."""
+ content = "Before {{ if .IsHome }}\nstuff\n{{ end }} After"
+ result = clean_content(content)
+ assert "{{ if" not in result
+ assert "{{ end }}" not in result
+
+ def test_real_world_kubeflow_doc_snippet(self):
+ """Test with a realistic Kubeflow documentation snippet."""
+ content = """---
+title: KServe Overview
+description: Overview of KServe
+weight: 1
+---
+
+
+
+
+
+## What is KServe?
+
+[KServe](https://kserve.github.io/website/) enables serverless inference
+on Kubernetes. Visit https://kubeflow.org/docs/external-add-ons/kserve/
+for more details.
+
+{{ partial "section-index.html" . }}
+
+
+"""
+ result = clean_content(content)
+ assert "title:" not in result
+ assert "auto-generated" not in result
+ assert "section-index" not in result
+ assert "KServe" in result
+ assert "serverless inference" in result
+ assert "Kubernetes" in result
+
+
+class TestCleanContentEdgeCases:
+ """Edge cases for content cleaning."""
+
+ def test_empty_string(self):
+ """Should handle empty string input."""
+ assert clean_content("") == ""
+
+ def test_whitespace_only(self):
+ """Should handle whitespace-only input."""
+ assert clean_content(" \n\n \t ") == ""
+
+ def test_no_cleaning_needed(self):
+ """Plain text should pass through with minimal changes."""
+ content = "Simple plain text with no special formatting."
+ result = clean_content(content)
+ assert result == content
+
+ def test_nested_html_tags(self):
+ """Should handle deeply nested HTML tags."""
+ content = ""
+ result = clean_content(content)
+ assert "Deep content" in result
+ assert "<" not in result
+
+
+class TestTruncateForTei:
+ def test_truncates_long_text(self):
+ long = "word " * 500
+ assert len(truncate_for_tei(long)) == 1000
+
+
+class TestEmbedTexts:
+ def test_batches_requests(self, monkeypatch):
+ calls = []
+
+ class FakeResponse:
+ def raise_for_status(self):
+ return None
+
+ def json(self):
+ return [[0.1, 0.2], [0.3, 0.4]]
+
+ def fake_post(url, json, headers, timeout):
+ calls.append((url, len(json["inputs"]), [len(t) for t in json["inputs"]]))
+ return FakeResponse()
+
+ monkeypatch.setattr("utils.requests.post", fake_post)
+
+ vectors = embed_texts(["a", "b"], "http://embeddings/embed", batch_size=2)
+ assert len(vectors) == 2
+ assert calls[0][1] == 2
+
+ def test_truncates_inputs_for_tei(self, monkeypatch):
+ sent = []
+
+ class FakeResponse:
+ def raise_for_status(self):
+ return None
+
+ def json(self):
+ return [[0.1]]
+
+ def fake_post(url, json, headers, timeout):
+ sent.extend(json["inputs"])
+ return FakeResponse()
+
+ monkeypatch.setattr("utils.requests.post", fake_post)
+ long = "x" * 5000
+ embed_texts([long], "http://embeddings/embed", batch_size=1)
+ assert len(sent[0]) == 1000
diff --git a/tests/test_session_issuer.py b/tests/test_session_issuer.py
index 22746ab..7f15000 100644
--- a/tests/test_session_issuer.py
+++ b/tests/test_session_issuer.py
@@ -1,140 +1,140 @@
-"""Tests for the session issuer core (docs-agent-mcp/session-issuer/issuer_core.py).
-
-Pure-unit: no FastAPI/uvicorn imports, no network. Verifies the JWT
-mint -> JWKS -> verify roundtrip that Istio RequestAuthentication performs
-at the ingress gateway.
-"""
-
-import base64
-import importlib.util
-import time
-from pathlib import Path
-
-import jwt
-import pytest
-
-ISSUER_DIR = Path(__file__).parent.parent / "docs-agent-mcp" / "session-issuer"
-ISSUER_CORE_PATH = ISSUER_DIR / "issuer_core.py"
-
-spec = importlib.util.spec_from_file_location("issuer_core", ISSUER_CORE_PATH)
-issuer_core = importlib.util.module_from_spec(spec)
-spec.loader.exec_module(issuer_core)
-
-
-@pytest.fixture(scope="module")
-def private_key():
- return issuer_core.load_or_generate_private_key(path=None)
-
-
-@pytest.fixture(scope="module")
-def kid(private_key):
- return issuer_core.compute_kid(private_key)
-
-
-@pytest.mark.unit
-class TestMintAndVerify:
- def test_token_verifies_against_public_key(self, private_key, kid):
- """The roundtrip Istio performs: verify RS256 signature + expiry."""
- resp = issuer_core.mint_token(private_key, kid)
- claims = jwt.decode(
- resp["access_token"],
- private_key.public_key(),
- algorithms=["RS256"],
- options={"require": ["iss", "sub", "iat", "exp"]},
- )
- assert claims["iss"] == issuer_core.DEFAULT_ISSUER
- assert claims["sub"].startswith("session-")
-
- def test_token_response_shape(self, private_key, kid):
- resp = issuer_core.mint_token(private_key, kid, ttl_seconds=60)
- assert resp["token_type"] == "Bearer"
- assert resp["expires_in"] == 60
-
- def test_expiry_honors_ttl(self, private_key, kid):
- resp = issuer_core.mint_token(private_key, kid, ttl_seconds=120)
- claims = jwt.decode(resp["access_token"], private_key.public_key(), algorithms=["RS256"])
- assert claims["exp"] - claims["iat"] == 120
- assert abs(claims["iat"] - time.time()) < 10
-
- def test_expired_token_rejected(self, private_key, kid):
- resp = issuer_core.mint_token(private_key, kid, ttl_seconds=-10)
- with pytest.raises(jwt.ExpiredSignatureError):
- jwt.decode(resp["access_token"], private_key.public_key(), algorithms=["RS256"])
-
- def test_sessions_are_unique(self, private_key, kid):
- subs = set()
- for _ in range(5):
- resp = issuer_core.mint_token(private_key, kid)
- claims = jwt.decode(resp["access_token"], private_key.public_key(), algorithms=["RS256"])
- subs.add(claims["sub"])
- assert len(subs) == 5
-
- def test_wrong_key_rejected(self, private_key, kid):
- """A token signed by another key must not verify (forged tokens)."""
- other_key = issuer_core.load_or_generate_private_key(path=None)
- resp = issuer_core.mint_token(other_key, kid)
- with pytest.raises(jwt.InvalidSignatureError):
- jwt.decode(resp["access_token"], private_key.public_key(), algorithms=["RS256"])
-
- def test_kid_in_header(self, private_key, kid):
- resp = issuer_core.mint_token(private_key, kid)
- header = jwt.get_unverified_header(resp["access_token"])
- assert header["kid"] == kid
- assert header["alg"] == "RS256"
-
-
-@pytest.mark.unit
-class TestJwks:
- def test_jwks_shape(self, private_key, kid):
- jwks = issuer_core.build_jwks(private_key, kid)
- assert len(jwks["keys"]) == 1
- key = jwks["keys"][0]
- assert key["kty"] == "RSA"
- assert key["alg"] == "RS256"
- assert key["use"] == "sig"
- assert key["kid"] == kid
-
- def test_jwks_b64url_no_padding(self, private_key, kid):
- """RFC 7518: base64url without padding; Envoy's JWKS parser is strict."""
- key = issuer_core.build_jwks(private_key, kid)["keys"][0]
- for field in ("n", "e"):
- assert "=" not in key[field]
- assert "+" not in key[field]
- assert "/" not in key[field]
-
- def test_jwks_verifies_token(self, private_key, kid):
- """Full path: reconstruct the public key from JWKS and verify a token."""
- jwks = issuer_core.build_jwks(private_key, kid)
- public_key = jwt.algorithms.RSAAlgorithm.from_jwk(jwks["keys"][0])
- resp = issuer_core.mint_token(private_key, kid)
- claims = jwt.decode(resp["access_token"], public_key, algorithms=["RS256"])
- assert claims["iss"] == issuer_core.DEFAULT_ISSUER
-
- def test_e_is_65537(self, private_key, kid):
- key = issuer_core.build_jwks(private_key, kid)["keys"][0]
- e = int.from_bytes(base64.urlsafe_b64decode(key["e"] + "=="), "big")
- assert e == 65537
-
-
-@pytest.mark.unit
-class TestKeyLoading:
- def test_stable_kid(self, private_key):
- assert issuer_core.compute_kid(private_key) == issuer_core.compute_kid(private_key)
-
- def test_load_from_pem(self, private_key, tmp_path):
- """Key round-trips through PEM (the Secret mount path in-cluster)."""
- from cryptography.hazmat.primitives import serialization
-
- pem = private_key.private_bytes(
- encoding=serialization.Encoding.PEM,
- format=serialization.PrivateFormat.PKCS8,
- encryption_algorithm=serialization.NoEncryption(),
- )
- key_file = tmp_path / "private.pem"
- key_file.write_bytes(pem)
- loaded = issuer_core.load_or_generate_private_key(str(key_file))
- assert issuer_core.compute_kid(loaded) == issuer_core.compute_kid(private_key)
-
- def test_missing_path_generates_ephemeral(self, tmp_path):
- key = issuer_core.load_or_generate_private_key(str(tmp_path / "nope.pem"))
- assert key.key_size == 2048
+"""Tests for the session issuer core (kagent-feast-mcp/session-issuer/issuer_core.py).
+
+Pure-unit: no FastAPI/uvicorn imports, no network. Verifies the JWT
+mint -> JWKS -> verify roundtrip that Istio RequestAuthentication performs
+at the ingress gateway.
+"""
+
+import base64
+import importlib.util
+import time
+from pathlib import Path
+
+import jwt
+import pytest
+
+ISSUER_DIR = Path(__file__).parent.parent / "kagent-feast-mcp" / "session-issuer"
+ISSUER_CORE_PATH = ISSUER_DIR / "issuer_core.py"
+
+spec = importlib.util.spec_from_file_location("issuer_core", ISSUER_CORE_PATH)
+issuer_core = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(issuer_core)
+
+
+@pytest.fixture(scope="module")
+def private_key():
+ return issuer_core.load_or_generate_private_key(path=None)
+
+
+@pytest.fixture(scope="module")
+def kid(private_key):
+ return issuer_core.compute_kid(private_key)
+
+
+@pytest.mark.unit
+class TestMintAndVerify:
+ def test_token_verifies_against_public_key(self, private_key, kid):
+ """The roundtrip Istio performs: verify RS256 signature + expiry."""
+ resp = issuer_core.mint_token(private_key, kid)
+ claims = jwt.decode(
+ resp["access_token"],
+ private_key.public_key(),
+ algorithms=["RS256"],
+ options={"require": ["iss", "sub", "iat", "exp"]},
+ )
+ assert claims["iss"] == issuer_core.DEFAULT_ISSUER
+ assert claims["sub"].startswith("session-")
+
+ def test_token_response_shape(self, private_key, kid):
+ resp = issuer_core.mint_token(private_key, kid, ttl_seconds=60)
+ assert resp["token_type"] == "Bearer"
+ assert resp["expires_in"] == 60
+
+ def test_expiry_honors_ttl(self, private_key, kid):
+ resp = issuer_core.mint_token(private_key, kid, ttl_seconds=120)
+ claims = jwt.decode(resp["access_token"], private_key.public_key(), algorithms=["RS256"])
+ assert claims["exp"] - claims["iat"] == 120
+ assert abs(claims["iat"] - time.time()) < 10
+
+ def test_expired_token_rejected(self, private_key, kid):
+ resp = issuer_core.mint_token(private_key, kid, ttl_seconds=-10)
+ with pytest.raises(jwt.ExpiredSignatureError):
+ jwt.decode(resp["access_token"], private_key.public_key(), algorithms=["RS256"])
+
+ def test_sessions_are_unique(self, private_key, kid):
+ subs = set()
+ for _ in range(5):
+ resp = issuer_core.mint_token(private_key, kid)
+ claims = jwt.decode(resp["access_token"], private_key.public_key(), algorithms=["RS256"])
+ subs.add(claims["sub"])
+ assert len(subs) == 5
+
+ def test_wrong_key_rejected(self, private_key, kid):
+ """A token signed by another key must not verify (forged tokens)."""
+ other_key = issuer_core.load_or_generate_private_key(path=None)
+ resp = issuer_core.mint_token(other_key, kid)
+ with pytest.raises(jwt.InvalidSignatureError):
+ jwt.decode(resp["access_token"], private_key.public_key(), algorithms=["RS256"])
+
+ def test_kid_in_header(self, private_key, kid):
+ resp = issuer_core.mint_token(private_key, kid)
+ header = jwt.get_unverified_header(resp["access_token"])
+ assert header["kid"] == kid
+ assert header["alg"] == "RS256"
+
+
+@pytest.mark.unit
+class TestJwks:
+ def test_jwks_shape(self, private_key, kid):
+ jwks = issuer_core.build_jwks(private_key, kid)
+ assert len(jwks["keys"]) == 1
+ key = jwks["keys"][0]
+ assert key["kty"] == "RSA"
+ assert key["alg"] == "RS256"
+ assert key["use"] == "sig"
+ assert key["kid"] == kid
+
+ def test_jwks_b64url_no_padding(self, private_key, kid):
+ """RFC 7518: base64url without padding; Envoy's JWKS parser is strict."""
+ key = issuer_core.build_jwks(private_key, kid)["keys"][0]
+ for field in ("n", "e"):
+ assert "=" not in key[field]
+ assert "+" not in key[field]
+ assert "/" not in key[field]
+
+ def test_jwks_verifies_token(self, private_key, kid):
+ """Full path: reconstruct the public key from JWKS and verify a token."""
+ jwks = issuer_core.build_jwks(private_key, kid)
+ public_key = jwt.algorithms.RSAAlgorithm.from_jwk(jwks["keys"][0])
+ resp = issuer_core.mint_token(private_key, kid)
+ claims = jwt.decode(resp["access_token"], public_key, algorithms=["RS256"])
+ assert claims["iss"] == issuer_core.DEFAULT_ISSUER
+
+ def test_e_is_65537(self, private_key, kid):
+ key = issuer_core.build_jwks(private_key, kid)["keys"][0]
+ e = int.from_bytes(base64.urlsafe_b64decode(key["e"] + "=="), "big")
+ assert e == 65537
+
+
+@pytest.mark.unit
+class TestKeyLoading:
+ def test_stable_kid(self, private_key):
+ assert issuer_core.compute_kid(private_key) == issuer_core.compute_kid(private_key)
+
+ def test_load_from_pem(self, private_key, tmp_path):
+ """Key round-trips through PEM (the Secret mount path in-cluster)."""
+ from cryptography.hazmat.primitives import serialization
+
+ pem = private_key.private_bytes(
+ encoding=serialization.Encoding.PEM,
+ format=serialization.PrivateFormat.PKCS8,
+ encryption_algorithm=serialization.NoEncryption(),
+ )
+ key_file = tmp_path / "private.pem"
+ key_file.write_bytes(pem)
+ loaded = issuer_core.load_or_generate_private_key(str(key_file))
+ assert issuer_core.compute_kid(loaded) == issuer_core.compute_kid(private_key)
+
+ def test_missing_path_generates_ephemeral(self, tmp_path):
+ key = issuer_core.load_or_generate_private_key(str(tmp_path / "nope.pem"))
+ assert key.key_size == 2048