test_credentials.py

"""The credential store shared with DeckHTML.

The behaviour worth guarding is not "can we read JSON" but the three ways this
integration can silently break: a browser login being invisible because only
``apiKey`` is read, DeckHTML's own keys being dropped when we write, and a
caller that withheld credentials on purpose getting them back from the file.
"""

from __future__ import annotations

import contextlib
import io
import json
import os
import socket
import tempfile
import threading
import time
import unittest
import urllib.request
from pathlib import Path
from unittest import mock

import _helpers  # noqa: F401  (sys.path side effect)

from deckflow_extract import registry
from deckflow_extract.cloud import credentials


class _IsolatedConfig(unittest.TestCase):
    """Each test gets its own config dir; nothing touches the real one."""

    def setUp(self) -> None:
        self.root = Path(tempfile.mkdtemp(prefix="deckflow-credentials-"))
        patcher = mock.patch.dict(
            os.environ, {"DECKFLOW_CONFIG_DIR": str(self.root)}, clear=False
        )
        patcher.start()
        self.addCleanup(patcher.stop)
        for name in (
            "DECKFLOW_API_KEY", "DECKOPS_API_KEY", "DECKFLOW_TOKEN", "DECKOPS_TOKEN",
            "DECKFLOW_SPACE_ID", "DECKOPS_SPACE_ID", "DECKFLOW_API_BASE",
            "DECKOPS_API_BASE", "DECKFLOW_NO_STORED_CREDENTIALS", "DECKHTML_CONFIG_DIR",
        ):
            os.environ.pop(name, None)

    def write_config(self, **values: object) -> Path:
        path = self.root / "credentials"
        path.write_text(json.dumps(values, indent=2), encoding="utf-8")
        return path


class DeckhtmlCompatibilityTest(_IsolatedConfig):
    def test_reads_the_file_deckhtml_writes(self):
        self.write_config(apiKey="worker-secret", spaceId="space-1")
        resolved = credentials.resolve()
        self.assertEqual(resolved.api_key, "worker-secret")
        self.assertEqual(resolved.space_id, "space-1")
        self.assertEqual(resolved.api_key_source, credentials.SOURCE_FILE)
        self.assertTrue(resolved.configured)

    def test_a_browser_login_token_alone_counts_as_configured(self):
        """`deckhtml auth login` stores a token and no api key at all.

        Treating that as unconfigured is the bug this whole module exists to
        fix: the user had working credentials and was told to authenticate.
        """
        self.write_config(token="session-token", spaceId="space-1")
        resolved = credentials.resolve()
        self.assertIsNone(resolved.api_key)
        self.assertEqual(resolved.token, "session-token")
        self.assertTrue(resolved.configured)
        self.assertEqual(resolved.source, credentials.SOURCE_FILE)

    def test_default_path_is_the_one_deckhtml_uses(self):
        os.environ.pop("DECKFLOW_CONFIG_DIR")
        self.assertEqual(credentials.config_path(), Path.home() / ".deckflow" / "credentials")

    def test_deckhtml_config_dir_override_is_honoured(self):
        os.environ.pop("DECKFLOW_CONFIG_DIR")
        os.environ["DECKHTML_CONFIG_DIR"] = str(self.root)
        self.write_config(apiKey="k")
        self.assertEqual(credentials.resolve().api_key, "k")

    def test_write_preserves_keys_this_tool_does_not_use(self):
        """DeckHTML owns `webhook` and `retentionHours`.

        A whole-file rewrite that dropped them would silently reconfigure the
        other tool every time someone ran `config set` here.
        """
        self.write_config(apiKey="k", webhook="https://hook.test", retentionHours=9)
        credentials.save_file({"spaceId": "space-2"})
        stored = json.loads((self.root / "credentials").read_text(encoding="utf-8"))
        self.assertEqual(stored["webhook"], "https://hook.test")
        self.assertEqual(stored["retentionHours"], 9)
        self.assertEqual(stored["apiKey"], "k")
        self.assertEqual(stored["spaceId"], "space-2")

    def test_refuses_keys_deckhtml_would_delete(self):
        # sanitizeConfig keeps a fixed key list and rewrites the file when it
        # sees anything else, so an invented key is data loss on a delay.
        with self.assertRaises(ValueError):
            credentials.save_file({"extractOnlyKey": "x"})

    def test_written_file_is_not_world_readable(self):
        credentials.save_file({"apiKey": "secret"})
        mode = (self.root / "credentials").stat().st_mode & 0o777
        self.assertEqual(mode, 0o600)

    def test_write_reloads_before_merging(self):
        """The other tool may have logged in since this process started."""
        credentials.load_file()
        self.write_config(token="written-by-deckhtml")
        credentials.save_file({"spaceId": "space-3"})
        stored = json.loads((self.root / "credentials").read_text(encoding="utf-8"))
        self.assertEqual(stored["token"], "written-by-deckhtml")


class PrecedenceTest(_IsolatedConfig):
    def test_environment_beats_the_stored_file(self):
        self.write_config(apiKey="from-file")
        os.environ["DECKFLOW_API_KEY"] = "from-env"
        resolved = credentials.resolve()
        self.assertEqual(resolved.api_key, "from-env")
        self.assertEqual(resolved.api_key_source, credentials.SOURCE_ENV)

    def test_deckops_variable_names_still_work(self):
        os.environ["DECKOPS_API_KEY"] = "legacy"
        self.assertEqual(credentials.resolve().api_key, "legacy")

    def test_api_base_falls_back_to_the_shared_default(self):
        resolved = credentials.resolve()
        self.assertEqual(resolved.api_base, credentials.DEFAULT_API_BASE)
        self.assertEqual(resolved.api_base_source, credentials.SOURCE_DEFAULT)

    def test_stored_credentials_can_be_refused_entirely(self):
        """`deckflow-core` strips the variables to keep a run local.

        Without this switch a stored credential would quietly restore what the
        caller just removed, and the caller's guarantee would be false.
        """
        self.write_config(apiKey="from-file", token="also-from-file")
        os.environ["DECKFLOW_NO_STORED_CREDENTIALS"] = "1"
        resolved = credentials.resolve()
        self.assertIsNone(resolved.api_key)
        self.assertIsNone(resolved.token)
        self.assertFalse(resolved.configured)

    def test_refusing_stored_credentials_does_not_erase_them(self):
        # "Do not trust the file" must not become "overwrite the file".
        self.write_config(apiKey="from-file", webhook="https://hook.test")
        os.environ["DECKFLOW_NO_STORED_CREDENTIALS"] = "1"
        credentials.save_file({"spaceId": "space-9"})
        stored = json.loads((self.root / "credentials").read_text(encoding="utf-8"))
        self.assertEqual(stored["apiKey"], "from-file")
        self.assertEqual(stored["webhook"], "https://hook.test")


class DegradationTest(_IsolatedConfig):
    def test_missing_file_is_not_an_error(self):
        self.assertEqual(credentials.load_file(), {})
        self.assertFalse(credentials.resolve().configured)

    def test_corrupt_file_degrades_to_unconfigured(self):
        # A local parse needs no credentials at all; a broken file must not
        # take one down.
        (self.root / "credentials").write_text("{not json", encoding="utf-8")
        self.assertEqual(credentials.load_file(), {})
        self.assertFalse(credentials.resolve().configured)

    def test_secrets_are_masked_in_reports(self):
        self.write_config(apiKey="0123456789abcdef", token="fedcba9876543210")
        report = credentials.resolve().redacted()
        self.assertNotIn("0123456789abcdef", json.dumps(report))
        self.assertNotIn("fedcba9876543210", json.dumps(report))
        self.assertTrue(report["configured"])


class CliTest(_IsolatedConfig):
    """`auth` / `config` mirror DeckHTML's command and key names."""

    def _run(self, *argv: str) -> tuple[int, dict]:
        from deckflow_extract.cli import main

        stdout = io.StringIO()
        with contextlib.redirect_stdout(stdout):
            code = main(list(argv))
        return code, json.loads(stdout.getvalue())

    def test_status_reports_the_shared_file_and_the_source(self):
        self.write_config(token="session-token")
        code, payload = self._run("auth", "status")
        self.assertEqual(code, 0)
        self.assertTrue(payload["cloud"]["configured"])
        self.assertEqual(payload["cloud"]["source"], "file")
        self.assertEqual(payload["cloud"]["config_file"], str(self.root / "credentials"))
        self.assertEqual(payload["shared_with"], "deckhtml")

    def test_status_succeeds_under_strict(self):
        # The strict exit map falls back to `blocked`; a successful status
        # command must not exit 4 because "ok" was missing from the table.
        code, _ = self._run("auth", "status", "--strict")
        self.assertEqual(code, 0)

    def test_config_set_uses_deckhtmls_key_names(self):
        code, payload = self._run("config", "set", "api-key", "worker-secret")
        self.assertEqual(code, 0)
        stored = json.loads((self.root / "credentials").read_text(encoding="utf-8"))
        self.assertEqual(stored["apiKey"], "worker-secret")
        self.assertEqual(payload["key"], "api-key")

    def test_config_set_reads_a_secret_from_stdin(self):
        with mock.patch("sys.stdin", io.StringIO("worker-secret\n")):
            code, payload = self._run("config", "set", "api-key", "--stdin")
        self.assertEqual(code, 0)
        stored = json.loads((self.root / "credentials").read_text(encoding="utf-8"))
        self.assertEqual(stored["apiKey"], "worker-secret")
        self.assertEqual(payload["key"], "api-key")

    def test_config_set_rejects_an_empty_value(self):
        code, payload = self._run("config", "set", "api-key", "", "--strict")
        self.assertEqual(code, 4)
        self.assertEqual(payload["diagnostics"][0]["code"], "missing-config-value")
        self.assertFalse((self.root / "credentials").exists())

    def test_config_set_rejects_two_value_sources(self):
        code, payload = self._run(
            "config", "set", "api-key", "worker-secret", "--stdin", "--strict"
        )
        self.assertEqual(code, 4)
        self.assertEqual(payload["diagnostics"][0]["code"], "multiple-config-values")
        self.assertFalse((self.root / "credentials").exists())

    def test_config_set_warns_when_the_environment_shadows_the_write(self):
        os.environ["DECKFLOW_API_KEY"] = "from-env"
        _, payload = self._run("config", "set", "api-key", "from-file")
        codes = [d["code"] for d in payload.get("diagnostics", [])]
        self.assertIn("shadowed-by-environment", codes)

    def test_unknown_config_key_is_reported_not_written(self):
        code, payload = self._run("config", "set", "retention-hours", "9", "--strict")
        self.assertEqual(code, 4)
        self.assertEqual(payload["diagnostics"][0]["code"], "unknown-config-key")
        self.assertFalse((self.root / "credentials").exists())

    def test_logout_clears_both_credential_shapes(self):
        self.write_config(apiKey="k", token="t", spaceId="space-1")
        self._run("auth", "logout")
        stored = json.loads((self.root / "credentials").read_text(encoding="utf-8"))
        self.assertNotIn("apiKey", stored)
        self.assertNotIn("token", stored)
        self.assertEqual(stored["spaceId"], "space-1")

    def test_logout_admits_an_environment_credential_it_cannot_remove(self):
        self.write_config(token="t")
        os.environ["DECKFLOW_API_KEY"] = "from-env"
        _, payload = self._run("auth", "logout")
        codes = [d["code"] for d in payload.get("diagnostics", [])]
        self.assertIn("credential-still-in-environment", codes)

    def test_login_url_matches_the_one_deckhtml_opens(self):
        from deckflow_extract.cloud import auth

        self.assertEqual(
            auth.login_url("https://app.deckflow.com/v1", "http://localhost:3737"),
            "https://app.deckflow.com/cli/auth?redirect_url=http%3A%2F%2Flocalhost%3A3737",
        )

    def test_login_stores_the_callback_token_in_the_shared_file(self):
        from deckflow_extract.cloud import auth

        port = _free_port()
        received: dict = {}

        def drive() -> None:
            deadline = time.time() + 10
            while time.time() < deadline:
                try:
                    with urllib.request.urlopen(
                        f"http://localhost:{port}/?token=browser-token&spaceId=space-9", timeout=2
                    ) as response:
                        received["status"] = response.status
                        return
                except Exception:  # noqa: BLE001 - the server may not be up yet
                    time.sleep(0.05)

        driver = threading.Thread(target=drive, daemon=True)
        driver.start()
        result = auth.login(port=port, timeout=15, open_browser=False)
        driver.join(5)

        self.assertEqual(received.get("status"), 200)
        self.assertEqual(result["space_id"], "space-9")
        stored = json.loads((self.root / "credentials").read_text(encoding="utf-8"))
        self.assertEqual(stored["token"], "browser-token")
        self.assertEqual(stored["spaceId"], "space-9")

    def test_login_that_never_receives_a_callback_fails_loudly(self):
        from deckflow_extract.cloud import auth

        with self.assertRaises(auth.LoginError):
            auth.login(port=_free_port(), timeout=0.3, open_browser=False)


def _free_port() -> int:
    with socket.socket() as probe:
        probe.bind(("127.0.0.1", 0))
        return probe.getsockname()[1]


class RoutingAgreementTest(_IsolatedConfig):
    """Routing and execution must not disagree about what "configured" means."""

    def _cloud_engine(self):
        return next(
            engine for engine in registry.engines_for("pdf")
            if engine.tier == registry.TIER_CLOUD
        )

    def test_a_stored_token_makes_the_cloud_engine_available(self):
        self.write_config(token="session-token")
        with mock.patch.object(registry, "has_module", lambda name: name == "deckops"):
            self.assertEqual(registry.unmet(self._cloud_engine()), [])

    def test_no_credential_blocks_the_cloud_engine(self):
        with mock.patch.object(registry, "has_module", lambda name: name == "deckops"):
            self.assertEqual(registry.unmet(self._cloud_engine()), ["auth:cloud"])


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