last30days-skill/tests/test_codex_auth.py

"""Tests for Codex auth integration (env.py + openai_reddit.py)."""

import base64
import json
import os
import sys
import time
import unittest
from pathlib import Path
from unittest.mock import patch

# Add scripts directory to path
sys.path.insert(0, str(Path(__file__).parent.parent / "scripts"))

from lib import env, openai_reddit


def _make_jwt(payload: dict) -> str:
    """Build a fake JWT with the given payload (no signature verification)."""
    header = base64.urlsafe_b64encode(json.dumps({"alg": "none"}).encode()).rstrip(b"=")
    body = base64.urlsafe_b64encode(json.dumps(payload).encode()).rstrip(b"=")
    return f"{header.decode()}.{body.decode()}.fakesig"


class TestDecodeJwtPayload(unittest.TestCase):

    def test_valid_jwt(self):
        token = _make_jwt({"sub": "user123", "exp": 9999999999})
        result = env._decode_jwt_payload(token)
        self.assertEqual(result["sub"], "user123")

    def test_invalid_jwt(self):
        self.assertIsNone(env._decode_jwt_payload("not-a-jwt"))

    def test_empty_string(self):
        self.assertIsNone(env._decode_jwt_payload(""))


class TestTokenExpired(unittest.TestCase):

    def test_not_expired(self):
        token = _make_jwt({"exp": int(time.time()) + 3600})
        self.assertFalse(env._token_expired(token))

    def test_expired(self):
        token = _make_jwt({"exp": int(time.time()) - 100})
        self.assertTrue(env._token_expired(token))

    def test_no_exp_claim(self):
        token = _make_jwt({"sub": "user"})
        self.assertFalse(env._token_expired(token))


class TestExtractChatgptAccountId(unittest.TestCase):

    def test_extracts_account_id(self):
        token = _make_jwt({
            "https://api.openai.com/auth": {
                "chatgpt_account_id": "acct_abc123"
            }
        })
        self.assertEqual(env.extract_chatgpt_account_id(token), "acct_abc123")

    def test_missing_auth_claim(self):
        token = _make_jwt({"sub": "user"})
        self.assertIsNone(env.extract_chatgpt_account_id(token))

    def test_missing_account_id_in_claim(self):
        token = _make_jwt({
            "https://api.openai.com/auth": {"other_field": "value"}
        })
        self.assertIsNone(env.extract_chatgpt_account_id(token))


class TestGetOpenaiAuth(unittest.TestCase):

    @patch.dict(os.environ, {}, clear=True)
    def test_api_key_takes_priority(self):
        """OPENAI_API_KEY in env file is used when env var is not set."""
        file_env = {"OPENAI_API_KEY": "sk-test123"}
        auth = env.get_openai_auth(file_env)
        self.assertEqual(auth.source, "api_key")
        self.assertEqual(auth.status, "ok")
        self.assertEqual(auth.token, "sk-test123")
        self.assertIsNone(auth.account_id)

    @patch.dict(os.environ, {"OPENAI_API_KEY": "sk-from-env"}, clear=False)
    def test_env_var_takes_priority(self):
        """OPENAI_API_KEY env var should be preferred over file."""
        file_env = {}
        auth = env.get_openai_auth(file_env)
        self.assertEqual(auth.source, "api_key")
        self.assertEqual(auth.token, "sk-from-env")

    def test_no_keys_returns_none_source(self):
        """No API key and no Codex auth → source=none."""
        fake_path = Path("/tmp/nonexistent_codex_auth_test.json")
        with patch.object(env, 'CODEX_AUTH_FILE', fake_path):
            # Also patch get_codex_access_token to avoid reading real auth file
            with patch.object(env, 'get_codex_access_token', return_value=(None, "missing")):
                environ_copy = {k: v for k, v in os.environ.items() if k != "OPENAI_API_KEY"}
                with patch.dict(os.environ, environ_copy, clear=True):
                    auth = env.get_openai_auth({})
                    self.assertEqual(auth.source, "none")
                    self.assertIsNone(auth.token)


class TestLoadCodexAuth(unittest.TestCase):

    def test_nonexistent_file(self):
        result = env.load_codex_auth(Path("/tmp/nonexistent_codex_auth.json"))
        self.assertEqual(result, {})

    def test_valid_json(self):
        import tempfile
        data = {"tokens": {"access_token": "tok123"}}
        with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f:
            json.dump(data, f)
            f.flush()
            result = env.load_codex_auth(Path(f.name))
        os.unlink(f.name)
        self.assertEqual(result["tokens"]["access_token"], "tok123")


class TestGetAvailableSourcesWithAuth(unittest.TestCase):

    def test_codex_auth_ok_counts_as_openai(self):
        config = {
            "OPENAI_API_KEY": "codex-token",
            "OPENAI_AUTH_STATUS": "ok",
            "XAI_API_KEY": None,
        }
        result = env.get_available_sources(config)
        self.assertIn("reddit", result)

    def test_codex_auth_expired_not_counted(self):
        config = {
            "OPENAI_API_KEY": None,
            "OPENAI_AUTH_STATUS": "expired",
            "XAI_API_KEY": None,
        }
        result = env.get_available_sources(config)
        # Reddit is available via public JSON fallback even without OpenAI auth
        self.assertEqual(result, "reddit")


class TestParseCodexStream(unittest.TestCase):

    def test_response_completed_event(self):
        """Should extract response from response.completed SSE event."""
        sse = (
            'data: {"type":"response.created","response":{"id":"r1"}}\n\n'
            'data: {"type":"response.completed","response":{"id":"r1","output":[{"type":"message","content":[{"type":"output_text","text":"hello"}]}]}}\n\n'
        )
        result = openai_reddit._parse_codex_stream(sse)
        self.assertIn("output", result)

    def test_delta_fallback(self):
        """Should reconstruct text from delta events."""
        sse = (
            'data: {"delta":"hel"}\n\n'
            'data: {"delta":"lo"}\n\n'
        )
        result = openai_reddit._parse_codex_stream(sse)
        self.assertIn("output", result)
        text = result["output"][0]["content"][0]["text"]
        self.assertEqual(text, "hello")

    def test_empty_stream(self):
        result = openai_reddit._parse_codex_stream("")
        self.assertEqual(result, {})


class TestBuildPayload(unittest.TestCase):

    def test_api_key_payload(self):
        payload = openai_reddit._build_payload(
            "gpt-4o", "instructions", "input text", "api_key"
        )
        self.assertEqual(payload["model"], "gpt-4o")
        self.assertEqual(payload["input"], "input text")
        self.assertNotIn("stream", payload)

    def test_codex_payload_has_stream(self):
        payload = openai_reddit._build_payload(
            "gpt-4o", "instructions", "input text", env.AUTH_SOURCE_CODEX
        )
        self.assertTrue(payload["stream"])
        # Input should be structured message format for Codex
        self.assertIsInstance(payload["input"], list)
        self.assertEqual(payload["input"][0]["role"], "user")

    def test_codex_payload_has_store_false(self):
        payload = openai_reddit._build_payload(
            "gpt-4o", "inst", "text", env.AUTH_SOURCE_CODEX
        )
        self.assertFalse(payload["store"])


if __name__ == "__main__":
    unittest.main()